xds/cleanup: list leaked gcloud resources using API instead of gcloud command (#28720)

tools/run_tests/xds_k8s_test_driver/bin/cleanup/cleanup.py

@@ -334,8 +334,13 @@ def main(argv):
     # target proxy cannot deleted unless the forwarding rule is deleted). The
     # leaked target-proxy is guaranteed to be a super set of leaked
     # forwarding-rule.
-    leakedHealthChecks = exec_gcloud(project, 'compute', 'health-checks',
-                                     'list')
+    compute = gcp.compute.ComputeV1(gcp.api.GcpApiManager(), project)
+    leakedHealthChecks = []
+    for item in compute.list_health_check()['items']:
+        if datetime.datetime.fromisoformat(
+                item['creationTimestamp']) <= get_expire_timestamp():
+            leakedHealthChecks.append(item)
+
     delete_leaked_td_resources(dry_run, td_resource_rules, project, network,
                                leakedHealthChecks)
 

tools/run_tests/xds_k8s_test_driver/framework/infrastructure/gcp/compute.py

@@ -80,6 +80,9 @@ class ComputeV1(gcp.api.GcpProjectApiResource):
                 health_check_field: health_check_settings,
             })
 
+    def list_health_check(self):
+        return self._list_resource(self.api.healthChecks())
+
     def delete_health_check(self, name):
         self._delete_resource(self.api.healthChecks(), 'healthCheck', name)
 
@@ -391,6 +394,10 @@ class ComputeV1(gcp.api.GcpProjectApiResource):
         self._execute(
             collection.patch(project=self.project, body=body, **kwargs))
 
+    def _list_resource(self, collection: discovery.Resource):
+        return collection.list(project=self.project).execute(
+            num_retries=self._GCP_API_RETRIES)
+
     def _delete_resource(self, collection: discovery.Resource,
                          resource_type: str, resource_name: str) -> bool:
         try: