[CSM O11y test] Ping GMP endpoint during test for debugging purpose

framework/test_app/runners/k8s/gamma_server_runner.py

@@ -41,9 +41,13 @@ class GammaServerRunner(KubernetesServerRunner):
     pre_stop_hook: bool = False
     pod_monitoring: Optional[k8s.PodMonitoring] = None
     pod_monitoring_name: Optional[str] = None
+    pod_monitoring_port: int = 9464
+    monitoring_port: Optional[int] = None
+    monitoring_host: Optional[str] = None
 
     route_name: str
     frontend_service_name: str
+    enable_csm_observability: bool
     csm_workload_name: str
     csm_canonical_service_name: str
 
@@ -77,6 +81,7 @@ def __init__(
         bepolicy_name: str = "backend-policy",
         termination_grace_period_seconds: int = 0,
         pre_stop_hook: bool = False,
+        enable_csm_observability: bool = False,
         csm_workload_name: str = "",
         csm_canonical_service_name: str = "",
     ):
@@ -111,6 +116,7 @@ def __init__(
         self.bepolicy_name = bepolicy_name
         self.termination_grace_period_seconds = termination_grace_period_seconds
         self.pre_stop_hook = pre_stop_hook
+        self.enable_csm_observability = enable_csm_observability
         self.csm_workload_name = csm_workload_name
         self.csm_canonical_service_name = csm_canonical_service_name
 
@@ -124,7 +130,6 @@ def run(  # pylint: disable=arguments-differ
         log_to_stdout: bool = False,
         bootstrap_version: Optional[str] = None,
         route_template: str = "gamma/route_http.yaml",
-        enable_csm_observability: bool = False,
         generate_mesh_id: bool = False,
     ) -> List[XdsTestServer]:
         if not maintenance_port:
@@ -210,21 +215,22 @@ def run(  # pylint: disable=arguments-differ
             bootstrap_version=bootstrap_version,
             termination_grace_period_seconds=self.termination_grace_period_seconds,
             pre_stop_hook=self.pre_stop_hook,
-            enable_csm_observability=enable_csm_observability,
+            enable_csm_observability=self.enable_csm_observability,
             generate_mesh_id=generate_mesh_id,
             csm_workload_name=self.csm_workload_name,
             csm_canonical_service_name=self.csm_canonical_service_name,
         )
 
         # Create a PodMonitoring resource if CSM Observability is enabled
         # This is GMP (Google Managed Prometheus)
-        if enable_csm_observability:
+        if self.enable_csm_observability:
             self.pod_monitoring_name = f"{self.deployment_id}-gmp"
             self.pod_monitoring = self._create_pod_monitoring(
                 "csm/pod-monitoring.yaml",
                 namespace_name=self.k8s_namespace.name,
                 deployment_id=self.deployment_id,
                 pod_monitoring_name=self.pod_monitoring_name,
+                pod_monitoring_port=self.pod_monitoring_port,
             )
 
         servers = self._make_servers_for_deployment(
@@ -271,6 +277,32 @@ def createBackendPolicy(self):
             service_name=self.service_name,
         )
 
+    def _xds_test_server_for_pod(
+        self,
+        pod: k8s.V1Pod,
+        *,
+        test_port: int = KubernetesServerRunner.DEFAULT_TEST_PORT,
+        maintenance_port: Optional[int] = None,
+        secure_mode: bool = False,
+    ) -> XdsTestServer:
+        if self.enable_csm_observability:
+            if self.debug_use_port_forwarding:
+                pf = self._start_port_forwarding_pod(
+                    pod, self.pod_monitoring_port
+                )
+                self.monitoring_port = pf.local_port
+                self.monitoring_host = pf.local_address
+            else:
+                self.monitoring_port = self.pod_monitoring_port
+                self.monitoring_host = pod.status.pod_ip
+
+        return super()._xds_test_server_for_pod(
+            pod=pod,
+            test_port=test_port,
+            maintenance_port=maintenance_port,
+            secure_mode=secure_mode,
+        )
+
     # pylint: disable=arguments-differ
     def cleanup(self, *, force=False, force_namespace=False):
         try:

framework/test_app/runners/k8s/k8s_base_runner.py

@@ -306,6 +306,7 @@ def _create_pod_monitoring(
         namespace_name: str,
         deployment_id: str,
         pod_monitoring_name: str,
+        pod_monitoring_port: int,
         **kwargs,
     ) -> k8s.PodMonitoring:
         pod_monitoring = self._create_from_template(
@@ -314,6 +315,7 @@ def _create_pod_monitoring(
             namespace_name=namespace_name,
             deployment_id=deployment_id,
             pod_monitoring_name=pod_monitoring_name,
+            pod_monitoring_port=pod_monitoring_port,
             **kwargs,
         )
         if not (

framework/test_app/runners/k8s/k8s_xds_client_runner.py

@@ -34,6 +34,7 @@ class KubernetesClientRunner(k8s_base_runner.KubernetesBaseRunner):
     debug_use_port_forwarding: bool
     td_bootstrap_image: str
     network: str
+    enable_csm_observability: bool
     csm_workload_name: str
     csm_canonical_service_name: str
 
@@ -43,6 +44,9 @@ class KubernetesClientRunner(k8s_base_runner.KubernetesBaseRunner):
     gcp_iam: Optional[gcp.iam.IamV1] = None
     pod_monitoring: Optional[k8s.PodMonitoring] = None
     pod_monitoring_name: Optional[str] = None
+    pod_monitoring_port: int = 9464
+    monitoring_port: Optional[int] = None
+    monitoring_host: Optional[str] = None
 
     def __init__(  # pylint: disable=too-many-locals
         self,
@@ -64,6 +68,7 @@ def __init__(  # pylint: disable=too-many-locals
         namespace_template: Optional[str] = None,
         debug_use_port_forwarding: bool = False,
         enable_workload_identity: bool = True,
+        enable_csm_observability: bool = False,
         csm_workload_name: str = "",
         csm_canonical_service_name: str = "",
     ):
@@ -83,6 +88,7 @@ def __init__(  # pylint: disable=too-many-locals
         self.deployment_template = deployment_template
         self.enable_workload_identity = enable_workload_identity
         self.debug_use_port_forwarding = debug_use_port_forwarding
+        self.enable_csm_observability = enable_csm_observability
         self.csm_workload_name = csm_workload_name
         self.csm_canonical_service_name = csm_canonical_service_name
 
@@ -112,7 +118,6 @@ def run(  # pylint: disable=arguments-differ
         generate_mesh_id=False,
         print_response=False,
         log_to_stdout: bool = False,
-        enable_csm_observability: bool = False,
         request_payload_size: int = 0,
         response_payload_size: int = 0,
     ) -> XdsTestClient:
@@ -171,20 +176,21 @@ def run(  # pylint: disable=arguments-differ
             config_mesh=config_mesh,
             generate_mesh_id=generate_mesh_id,
             print_response=print_response,
-            enable_csm_observability=enable_csm_observability,
+            enable_csm_observability=self.enable_csm_observability,
             csm_workload_name=self.csm_workload_name,
             csm_canonical_service_name=self.csm_canonical_service_name,
         )
 
         # Create a PodMonitoring resource if CSM Observability is enabled
         # This is GMP (Google Managed Prometheus)
-        if enable_csm_observability:
+        if self.enable_csm_observability:
             self.pod_monitoring_name = f"{self.deployment_id}-gmp"
             self.pod_monitoring = self._create_pod_monitoring(
                 "csm/pod-monitoring.yaml",
                 namespace_name=self.k8s_namespace.name,
                 deployment_id=self.deployment_id,
                 pod_monitoring_name=self.pod_monitoring_name,
+                pod_monitoring_port=self.pod_monitoring_port,
             )
 
         # Load test client pod. We need only one client at the moment
@@ -205,8 +211,17 @@ def _xds_test_client_for_pod(
         if self.debug_use_port_forwarding:
             pf = self._start_port_forwarding_pod(pod, self.stats_port)
             rpc_port, rpc_host = pf.local_port, pf.local_address
+            if self.enable_csm_observability:
+                pf = self._start_port_forwarding_pod(
+                    pod, self.pod_monitoring_port
+                )
+                self.monitoring_port = pf.local_port
+                self.monitoring_host = pf.local_address
         else:
             rpc_port, rpc_host = self.stats_port, None
+            if self.enable_csm_observability:
+                self.monitoring_port = self.pod_monitoring_port
+                self.monitoring_host = pod.status.pod_ip
 
         return XdsTestClient(
             ip=pod.status.pod_ip,

kubernetes-manifests/csm/pod-monitoring.yaml

@@ -11,5 +11,5 @@ spec:
     matchLabels:
       deployment_id: ${deployment_id}
   endpoints:
-  - port: 9464
+  - port: ${pod_monitoring_port}
     interval: 10s

requirements.txt

@@ -15,4 +15,5 @@ packaging~=23.1
 Pygments~=2.9
 python-dateutil~=2.8
 protobuf~=4.24
+requests==2.31.0
 xds-protos==1.58.0rc1

tests/gamma/csm_observability_test.py

@@ -22,6 +22,8 @@
 from google.api_core import exceptions as gapi_errors
 from google.api_core import retry as gapi_retries
 from google.cloud import monitoring_v3
+import requests
+from requests.exceptions import RequestException
 import yaml
 
 from framework import xds_gamma_testcase
@@ -158,6 +160,7 @@ def setUpClass(cls):
     # each run().
     def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
         return super().initKubernetesClientRunner(
+            enable_csm_observability=True,
             csm_workload_name=CSM_WORKLOAD_NAME_CLIENT,
             csm_canonical_service_name=CSM_CANONICAL_SERVICE_NAME_CLIENT,
         )
@@ -166,6 +169,7 @@ def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
     # each run().
     def initKubernetesServerRunner(self, **kwargs) -> GammaServerRunner:
         return super().initKubernetesServerRunner(
+            enable_csm_observability=True,
             csm_workload_name=CSM_WORKLOAD_NAME_SERVER,
             csm_canonical_service_name=CSM_CANONICAL_SERVICE_NAME_SERVER,
         )
@@ -175,14 +179,11 @@ def test_csm_observability(self):
         #   resource creation out of self.startTestServers()
         with self.subTest("1_run_test_server"):
             start_secs = int(time.time())
-            test_server: _XdsTestServer = self.startTestServers(
-                enable_csm_observability=True,
-            )[0]
+            test_server: _XdsTestServer = self.startTestServers()[0]
 
         with self.subTest("2_start_test_client"):
             test_client: _XdsTestClient = self.startTestClient(
                 test_server,
-                enable_csm_observability=True,
                 request_payload_size=REQUEST_PAYLOAD_SIZE,
                 response_payload_size=RESPONSE_PAYLOAD_SIZE,
             )
@@ -195,7 +196,20 @@ def test_csm_observability(self):
                 "Letting test client run for %d seconds to produce metric data",
                 TEST_RUN_SECS,
             )
-            time.sleep(TEST_RUN_SECS)
+            for i in range(0, TEST_RUN_SECS // 10):
+                time.sleep(10)
+                logger.info(
+                    self.ping_gmp_endpoint(
+                        self.server_runner.monitoring_host,
+                        self.server_runner.monitoring_port,
+                    )
+                )
+                logger.info(
+                    self.ping_gmp_endpoint(
+                        self.client_runner.monitoring_host,
+                        self.client_runner.monitoring_port,
+                    )
+                )
             end_secs = int(time.time())
             interval = monitoring_v3.TimeInterval(
                 start_time={"seconds": start_secs},
@@ -483,6 +497,24 @@ def assertAtLeastOnePointWithinRange(
             f"No data point with {ref_bytes}±{tolerance*100}% bytes found"
         )
 
+    def ping_gmp_endpoint(
+        self, monitoring_host: str, monitoring_port: int
+    ) -> str:
+        """
+        A helper function to ping the GMP endpoint to get what GMP sees
+        from the OTel exporter before passing metrics to Cloud Monitoring.
+        """
+        try:
+            gmp_log = requests.get(
+                f"http://{monitoring_host}:{monitoring_port}/metrics"
+            )
+            return "\n".join(gmp_log.text.splitlines())
+        except RequestException as e:
+            logger.error("Http request to GMP endpoint failed: %r", e)
+            # It's OK the caller will receive nothing in case of an exception.
+            # Caller can continue.
+            return ""
+
 
 if __name__ == "__main__":
     absltest.main()