Add App Net based CSM Observability Test

framework/test_app/runners/k8s/gamma_server_runner.py

@@ -41,15 +41,10 @@ class GammaServerRunner(KubernetesServerRunner):
     session_affinity_filter: Optional[k8s.GcpSessionAffinityFilter] = None
     session_affinity_policy: Optional[k8s.GcpSessionAffinityPolicy] = None
     backend_policy: Optional[k8s.GcpBackendPolicy] = None
-    pod_monitoring: Optional[k8s.PodMonitoring] = None
-    pod_monitoring_name: Optional[str] = None
 
     route_kind: Final[RouteKind]
     route_name: Final[str]
     frontend_service_name: str
-    enable_csm_observability: bool
-    csm_workload_name: str
-    csm_canonical_service_name: str
 
     SESSION_AFFINITY_FILTER_NAME: Final[str] = "ssa-filter"
     SESSION_AFFINITY_POLICY_NAME: Final[str] = "ssa-policy"
@@ -80,9 +75,6 @@ def __init__(
         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 = "",
         deployment_args: Optional[ServerDeploymentArgs] = None,
     ):
         # pylint: disable=too-many-locals
@@ -111,9 +103,6 @@ def __init__(
         )
 
         self.frontend_service_name = frontend_service_name
-        self.enable_csm_observability = enable_csm_observability
-        self.csm_workload_name = csm_workload_name
-        self.csm_canonical_service_name = csm_canonical_service_name
         self.route_kind = route_kind
         self.route_name = f"route-{route_kind.value.lower()}-{deployment_name}"
 
@@ -214,16 +203,13 @@ def run(  # pylint: disable=arguments-differ
             maintenance_port=maintenance_port,
             secure_mode=secure_mode,
             bootstrap_version=bootstrap_version,
-            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,
             **self.deployment_args.as_dict(),
         )
 
         # Create a PodMonitoring resource if CSM Observability is enabled
         # This is GMP (Google Managed Prometheus)
-        if self.enable_csm_observability:
+        if self.deployment_args.enable_csm_observability:
             self.pod_monitoring_name = f"{self.deployment_id}-gmp"
             self.pod_monitoring = self._create_pod_monitoring(
                 "csm/pod-monitoring.yaml",
@@ -305,7 +291,7 @@ def _xds_test_server_for_pod(
         secure_mode: bool = False,
         monitoring_port: Optional[int] = None,
     ) -> XdsTestServer:
-        if self.enable_csm_observability:
+        if self.deployment_args.enable_csm_observability:
             if self.debug_use_port_forwarding:
                 pf = self._start_port_forwarding_pod(
                     pod, self.DEFAULT_MONITORING_PORT

framework/test_app/runners/k8s/k8s_xds_server_runner.py

@@ -34,13 +34,19 @@
 class ServerDeploymentArgs:
     pre_stop_hook: bool = False
     termination_grace_period: dt.timedelta = dt.timedelta()
+    enable_csm_observability: bool = False
+    csm_workload_name: str = ""
+    csm_canonical_service_name: str = ""
 
     def as_dict(self):
         return {
             "pre_stop_hook": self.pre_stop_hook,
             "termination_grace_period_seconds": int(
                 self.termination_grace_period.total_seconds()
             ),
+            "enable_csm_observability": self.enable_csm_observability,
+            "csm_workload_name": self.csm_workload_name,
+            "csm_canonical_service_name": self.csm_canonical_service_name,
         }
 
 
@@ -60,6 +66,8 @@ class KubernetesServerRunner(k8s_base_runner.KubernetesBaseRunner):
     td_bootstrap_image: str
     xds_server_uri: str
     network: str
+    pod_monitoring: Optional[k8s.PodMonitoring] = None
+    pod_monitoring_name: Optional[str] = None
 
     # Server Deployment args
     deployment_args: ServerDeploymentArgs
@@ -253,6 +261,18 @@ def run(  # pylint: disable=arguments-differ,too-many-branches
             **self.deployment_args.as_dict(),
         )
 
+        # Create a PodMonitoring resource if CSM Observability is enabled
+        # This is GMP (Google Managed Prometheus)
+        if self.deployment_args.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.DEFAULT_MONITORING_PORT,
+            )
+
         return self._make_servers_for_deployment(
             replica_count,
             test_port=test_port,
@@ -317,8 +337,15 @@ def _xds_test_server_for_pod(
         if self.debug_use_port_forwarding:
             pf = self._start_port_forwarding_pod(pod, maintenance_port)
             rpc_port, rpc_host = pf.local_port, pf.local_address
+            if self.deployment_args.enable_csm_observability:
+                pf = self._start_port_forwarding_pod(
+                    pod, self.DEFAULT_MONITORING_PORT
+                )
+                monitoring_port = pf.local_port
         else:
             rpc_port, rpc_host = maintenance_port, None
+            if self.deployment_args.enable_csm_observability:
+                monitoring_port = self.DEFAULT_MONITORING_PORT
 
         server = XdsTestServer(
             ip=pod.status.pod_ip,