Merge branch 'main' into cursor-timestamp

framework/test_app/runners/k8s/gamma_server_runner.py

@@ -39,9 +39,13 @@ class GammaServerRunner(KubernetesServerRunner):
     be_policy: Optional[k8s.GcpBackendPolicy] = None
     termination_grace_period_seconds: Optional[int] = None
     pre_stop_hook: bool = False
+    pod_monitoring: Optional[k8s.PodMonitoring] = None
+    pod_monitoring_name: Optional[str] = None
 
     route_name: str
     frontend_service_name: str
+    csm_workload_name: str
+    csm_canonical_service_name: str
 
     def __init__(
         self,
@@ -73,6 +77,8 @@ def __init__(
         bepolicy_name: str = "backend-policy",
         termination_grace_period_seconds: int = 0,
         pre_stop_hook: bool = False,
+        csm_workload_name: str = "",
+        csm_canonical_service_name: str = "",
     ):
         # pylint: disable=too-many-locals
         super().__init__(
@@ -105,6 +111,8 @@ def __init__(
         self.bepolicy_name = bepolicy_name
         self.termination_grace_period_seconds = termination_grace_period_seconds
         self.pre_stop_hook = pre_stop_hook
+        self.csm_workload_name = csm_workload_name
+        self.csm_canonical_service_name = csm_canonical_service_name
 
     def run(  # pylint: disable=arguments-differ
         self,
@@ -204,8 +212,21 @@ def run(  # pylint: disable=arguments-differ
             pre_stop_hook=self.pre_stop_hook,
             enable_csm_observability=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:
+            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,
+            )
+
         servers = self._make_servers_for_deployment(
             replica_count,
             test_port=test_port,
@@ -292,6 +313,12 @@ def cleanup(self, *, force=False, force_namespace=False):
                 self._delete_service_account(self.service_account_name)
                 self.service_account = None
 
+            # Pod monitoring name is only set when CSM observability is enabled.
+            if self.pod_monitoring_name and (self.pod_monitoring or force):
+                self._delete_pod_monitoring(self.pod_monitoring_name)
+                self.pod_monitoring = None
+                self.pod_monitoring_name = None
+
             self._cleanup_namespace(force=(force_namespace and force))
         finally:
             self._stop()

framework/test_app/runners/k8s/k8s_xds_client_runner.py

@@ -34,6 +34,8 @@ class KubernetesClientRunner(k8s_base_runner.KubernetesBaseRunner):
     debug_use_port_forwarding: bool
     td_bootstrap_image: str
     network: str
+    csm_workload_name: str
+    csm_canonical_service_name: str
 
     # Optional fields.
     service_account_name: Optional[str] = None
@@ -62,6 +64,8 @@ def __init__(  # pylint: disable=too-many-locals
         namespace_template: Optional[str] = None,
         debug_use_port_forwarding: bool = False,
         enable_workload_identity: bool = True,
+        csm_workload_name: str = "",
+        csm_canonical_service_name: str = "",
     ):
         super().__init__(
             k8s_namespace,
@@ -79,6 +83,8 @@ 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.csm_workload_name = csm_workload_name
+        self.csm_canonical_service_name = csm_canonical_service_name
 
         # Used by the TD bootstrap generator.
         self.td_bootstrap_image = td_bootstrap_image
@@ -166,6 +172,8 @@ def run(  # pylint: disable=arguments-differ
             generate_mesh_id=generate_mesh_id,
             print_response=print_response,
             enable_csm_observability=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

framework/xds_gamma_testcase.py

@@ -94,7 +94,7 @@ def initTrafficDirectorManager(
             compute_api_version=self.compute_api_version,
         )
 
-    def initKubernetesServerRunner(self) -> GammaServerRunner:
+    def initKubernetesServerRunner(self, **kwargs) -> GammaServerRunner:
         return GammaServerRunner(
             k8s.KubernetesNamespace(
                 self.k8s_api_manager, self.server_namespace
@@ -112,6 +112,7 @@ def initKubernetesServerRunner(self) -> GammaServerRunner:
             enable_workload_identity=self.enable_workload_identity,
             termination_grace_period_seconds=self.termination_grace_period_seconds,
             pre_stop_hook=self.pre_stop_hook,
+            **kwargs,
         )
 
     def startTestClient(

framework/xds_k8s_testcase.py

@@ -683,11 +683,11 @@ def initTrafficDirectorManager(self) -> TrafficDirectorManager:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def initKubernetesServerRunner(self) -> KubernetesServerRunner:
+    def initKubernetesServerRunner(self, **kwargs) -> KubernetesServerRunner:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def initKubernetesClientRunner(self) -> KubernetesClientRunner:
+    def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
         raise NotImplementedError
 
     def tearDown(self):
@@ -795,7 +795,7 @@ def initTrafficDirectorManager(self) -> TrafficDirectorManager:
             compute_api_version=self.compute_api_version,
         )
 
-    def initKubernetesServerRunner(self) -> KubernetesServerRunner:
+    def initKubernetesServerRunner(self, **kwargs) -> KubernetesServerRunner:
         return KubernetesServerRunner(
             k8s.KubernetesNamespace(
                 self.k8s_api_manager, self.server_namespace
@@ -810,9 +810,10 @@ def initKubernetesServerRunner(self) -> KubernetesServerRunner:
             network=self.network,
             debug_use_port_forwarding=self.debug_use_port_forwarding,
             enable_workload_identity=self.enable_workload_identity,
+            **kwargs,
         )
 
-    def initKubernetesClientRunner(self) -> KubernetesClientRunner:
+    def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
         return KubernetesClientRunner(
             k8s.KubernetesNamespace(
                 self.k8s_api_manager, self.client_namespace
@@ -829,6 +830,7 @@ def initKubernetesClientRunner(self) -> KubernetesClientRunner:
             enable_workload_identity=self.enable_workload_identity,
             stats_port=self.client_port,
             reuse_namespace=self.server_namespace == self.client_namespace,
+            **kwargs,
         )
 
     def startTestServers(
@@ -904,7 +906,7 @@ def initTrafficDirectorManager(self) -> TrafficDirectorSecureManager:
             compute_api_version=self.compute_api_version,
         )
 
-    def initKubernetesServerRunner(self) -> KubernetesServerRunner:
+    def initKubernetesServerRunner(self, **kwargs) -> KubernetesServerRunner:
         return KubernetesServerRunner(
             k8s.KubernetesNamespace(
                 self.k8s_api_manager, self.server_namespace
@@ -919,9 +921,10 @@ def initKubernetesServerRunner(self) -> KubernetesServerRunner:
             xds_server_uri=self.xds_server_uri,
             deployment_template="server-secure.deployment.yaml",
             debug_use_port_forwarding=self.debug_use_port_forwarding,
+            **kwargs,
         )
 
-    def initKubernetesClientRunner(self) -> KubernetesClientRunner:
+    def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
         return KubernetesClientRunner(
             k8s.KubernetesNamespace(
                 self.k8s_api_manager, self.client_namespace
@@ -938,6 +941,7 @@ def initKubernetesClientRunner(self) -> KubernetesClientRunner:
             stats_port=self.client_port,
             reuse_namespace=self.server_namespace == self.client_namespace,
             debug_use_port_forwarding=self.debug_use_port_forwarding,
+            **kwargs,
         )
 
     def startSecureTestServer(self, replica_count=1, **kwargs) -> XdsTestServer:

kubernetes-manifests/client.deployment.yaml

@@ -66,6 +66,26 @@ spec:
             value: "true"
           - name: GRPC_EXPERIMENTAL_XDS_ENABLE_OVERRIDE_HOST
             value: "true"
+          % if csm_workload_name:
+          - name: CSM_WORKLOAD_NAME
+            value: ${csm_workload_name}
+          % endif
+          % if csm_canonical_service_name:
+          - name: CSM_CANONICAL_SERVICE_NAME
+            value: ${csm_canonical_service_name}
+          % endif
+          % if enable_csm_observability:
+          - name: POD_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.name
+          - name: NAMESPACE_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
+          - name: OTEL_RESOURCE_ATTRIBUTES
+            value: k8s.pod.name=$(POD_NAME),k8s.namespace.name=$(NAMESPACE_NAME)
+          % endif
         volumeMounts:
           - mountPath: /tmp/grpc-xds/
             name: grpc-td-conf

kubernetes-manifests/server.deployment.yaml

@@ -52,6 +52,26 @@ spec:
             value: "true"
           - name: GRPC_EXPERIMENTAL_XDS_ENABLE_OVERRIDE_HOST
             value: "true"
+          % if csm_workload_name:
+          - name: CSM_WORKLOAD_NAME
+            value: ${csm_workload_name}
+          % endif
+          % if csm_canonical_service_name:
+          - name: CSM_CANONICAL_SERVICE_NAME
+            value: ${csm_canonical_service_name}
+          % endif
+          % if enable_csm_observability:
+          - name: POD_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.name
+          - name: NAMESPACE_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
+          - name: OTEL_RESOURCE_ATTRIBUTES
+            value: k8s.pod.name=$(POD_NAME),k8s.namespace.name=$(NAMESPACE_NAME)
+          % endif
         volumeMounts:
           - mountPath: /tmp/grpc-xds/
             name: grpc-td-conf