cleanup fixes

framework/infrastructure/k8s.py

@@ -327,6 +327,13 @@ def api_session_affinity_policy(self) -> dynamic_res.Resource:
             "GCPSessionAffinityPolicy",
         )
 
+    @functools.cached_property  # pylint: disable=no-member
+    def pod_monitoring(self) -> dynamic_res.Resource:
+        return self._get_dynamic_api(
+            "monitoring.googleapis.com/v1",
+            "PodMonitoring",
+        )
+
     @functools.cached_property  # pylint: disable=no-member
     def api_backend_policy(self) -> dynamic_res.Resource:
         return self._get_dynamic_api(
@@ -649,6 +656,19 @@ def delete_session_affinity_filter(
             grace_period_seconds=grace_period_seconds,
         )
 
+    def delete_pod_monitoring(
+        self,
+        name: str,
+        grace_period_seconds=DELETE_GRACE_PERIOD_SEC,
+    ) -> None:
+        self._execute(
+            self.pod_monitoring.delete,  # pylint: disable=no-member
+            name=name,
+            namespace=self.name,
+            propagation_policy="Foreground",
+            grace_period_seconds=grace_period_seconds,
+        )
+
     def delete_backend_policy(
         self,
         name: str,

framework/test_app/runners/k8s/k8s_base_runner.py

@@ -307,7 +307,7 @@ def _create_pod_monitoring(
         deployment_id: str,
         pod_monitoring_name: str,
         **kwargs,
-    ) -> None:
+    ) -> k8s.PodMonitoring:
         pod_monitoring = self._create_from_template(
             template,
             custom_object=True,
@@ -678,6 +678,20 @@ def _delete_session_affinity_filter(self, name, wait_for_deletion=True):
             )
         logger.info("GCPSessionAffinityFilter %s deleted", name)
 
+    def _delete_pod_monitoring(self, name):
+        logger.info("Deleting PodMonitoring %s", name)
+        try:
+            self.k8s_namespace.delete_pod_monitoring(name)
+        except k8s.NotFound:
+            logger.debug(
+                "PodMonitoring %s not deleted since it doesn't exist", name
+            )
+            return
+        except retryers.RetryError as e:
+            logger.warning("PodMonitoring %s deletion failed: %s", name, e)
+            return
+        logger.info("PodMonitoring %s deleted", name)
+
     def _delete_backend_policy(self, name, wait_for_deletion=True):
         logger.info("Deleting GCPBackendPolicy %s", name)
         try:

framework/test_app/runners/k8s/k8s_xds_client_runner.py

@@ -39,6 +39,7 @@ class KubernetesClientRunner(k8s_base_runner.KubernetesBaseRunner):
     service_account_name: Optional[str] = None
     service_account_template: Optional[str] = None
     gcp_iam: Optional[gcp.iam.IamV1] = None
+    pod_monitoring: Optional[k8s.PodMonitoring] = None
 
     def __init__(  # pylint: disable=too-many-locals
         self,
@@ -165,7 +166,7 @@ def run(  # pylint: disable=arguments-differ
         # Create a PodMonitoring resource if CSM Observability is enabled
         # This is GMP (Google Managed Prometheus)
         if enable_csm_observability:
-            self._create_pod_monitoring(
+            self.pod_monitoring = self._create_pod_monitoring(
                 "csm/pod-monitoring.yaml",
                 namespace_name=self.k8s_namespace.name,
                 deployment_id=self.deployment_id,
@@ -218,6 +219,11 @@ def cleanup(self, *, force=False, force_namespace=False):
                 )
                 self._delete_service_account(self.service_account_name)
                 self.service_account = None
+            if self.pod_monitoring:
+                self._delete_pod_monitoring(
+                    self.pod_monitoring.metadata.name
+                )
+                self.pod_monitoring = None
             self._cleanup_namespace(force=force_namespace and force)
         finally:
             self._stop()