[CloudRun] Implement Cloudrun to Cloudrun test

.kokoro/psm_interop_kokoro_lib.sh

@@ -179,6 +179,7 @@ psm::cloud_run::setup() {
 psm::cloud_run::get_tests() {
   TESTS=(
     "cloud_run_csm_inbound_test"
+    "cloud_run_csm_outbound_test"
   )
 }
 

bin/lib/common.py

@@ -27,6 +27,7 @@
 from framework.infrastructure.gcp import cloud_run
 from framework.test_app import client_app
 from framework.test_app import server_app
+from framework.test_app.runners.cloud_run import cloud_run_xds_client_runner
 from framework.test_app.runners.cloud_run import cloud_run_xds_server_runner
 from framework.test_app.runners.k8s import gamma_server_runner
 from framework.test_app.runners.k8s import k8s_xds_client_runner
@@ -48,6 +49,7 @@
 KubernetesServerRunner = k8s_xds_server_runner.KubernetesServerRunner
 GammaServerRunner = gamma_server_runner.GammaServerRunner
 CloudRunServerRunner = cloud_run_xds_server_runner.CloudRunServerRunner
+CloudRunClientRunner = cloud_run_xds_client_runner.CloudRunClientRunner
 _XdsTestServer = server_app.XdsTestServer
 _XdsTestClient = client_app.XdsTestClient
 
@@ -177,6 +179,21 @@ def make_cloud_run_server_runner() -> CloudRunServerRunner:
     server_runner = CloudRunServerRunner
     return server_runner(**runner_kwargs)
 
+def make_cloud_run_client_runner(mesh_name: str, server_target: str) -> CloudRunClientRunner:
+    # CloudRunClientRunner arguments.
+    runner_kwargs = dict(
+        project=xds_flags.PROJECT.value,
+        service_name=xds_flags.SERVER_NAME.value,
+        image_name=xds_k8s_flags.SERVER_IMAGE.value,
+        network=xds_flags.NETWORK.value,
+        region=xds_flags.CLOUD_RUN_REGION.value,
+        gcp_api_manager=gcp.api.GcpApiManager(),
+        mesh_name=mesh_name,
+        server_target=server_target,
+    )
+    client_runner = CloudRunClientRunner
+    return client_runner(**runner_kwargs)
+
 def _ensure_atexit(signum, frame):
     """Needed to handle signals or atexit handler won't be called."""
     del frame

bin/run_test_client_cloud_run.py

@@ -0,0 +1,72 @@
+# Copyright 2025 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Run test xds client.
+
+Typical usage examples:
+
+    # Help.
+    ./run.sh ./bin/run_test_client_cloud_run.py --help
+
+"""
+import logging
+
+from absl import app
+from absl import flags
+
+from bin.lib import common
+from framework import xds_flags
+from framework import xds_k8s_flags
+from framework.infrastructure.mesh_resource_manager import (
+    cloud_run_mesh_manager,
+)
+
+logger = logging.getLogger(__name__)
+
+flags.adopt_module_key_flags(xds_flags)
+flags.adopt_module_key_flags(xds_k8s_flags)
+flags.adopt_module_key_flags(common)
+
+
+def main(argv):
+    if len(argv) > 1:
+        raise app.UsageError("Too many command-line arguments.")
+
+    xds_flags.set_socket_default_timeout_from_flag()
+
+    run_kwargs = dict()
+    td = cloud_run_mesh_manager.CloudRunMeshManager(
+        **common.td_attrs(), region=xds_flags.CLOUD_RUN_REGION.value
+    )
+
+    # Default server target pattern.
+    server_target = f"xds:///{xds_flags.SERVER_XDS_HOST.value}"
+    if xds_flags.SERVER_XDS_PORT.value != 80:
+        server_target = f"{server_target}:{xds_flags.SERVER_XDS_PORT.value}"
+
+    run_kwargs["server_target"] = server_target
+
+    mesh = td.create_mesh()
+    run_kwargs[
+        "config_mesh"
+    ] = f"projects/{td.project}/locations/global/meshes/{mesh.name}"
+    logger.info("Config Mesh: %s", mesh.name)
+    client_runner = common.make_cloud_run_client_runner(
+        run_kwargs["config_mesh"], server_target
+    )
+    client_runner.run(**run_kwargs)
+
+
+if __name__ == "__main__":
+    app.run(main)

config/common-cloudrun.cfg

@@ -1,2 +1,2 @@
 # Common config file for Cloudrun PSM tests.
---cloud_run_region=us-central1
+--cloud_run_region=us-east7

framework/infrastructure/gcp/api.py

@@ -581,6 +581,29 @@ def _delete_resource(
                 logger.warning("Failed to delete %s, %r", full_name, error)
         return False
 
+    def _set_iam_policy(
+        self,
+        collection: discovery.Resource,
+        body: dict,
+        full_name: str,
+        **kwargs,
+    ):
+        logger.info(
+            "Setting IAM policy for %s resource:\n%s",
+            self.api_name,
+            self.resource_pretty_format(body),
+        )
+        try:
+            collection.setIamPolicy(
+                resource=full_name, body=body, **kwargs
+            ).execute()
+            return True
+        except _HttpError as error:
+            logger.warning(
+                "Failed to set IAM policy for %s, %r", full_name, error
+            )
+            raise
+
     # TODO(sergiitk): Use ResponseError and TransportError
     def _execute(  # pylint: disable=arguments-differ
         self,

framework/infrastructure/gcp/cloud_run.py

@@ -88,3 +88,12 @@ def delete_service(self, service_name: str):
                 service_name, self.SERVICES, self.region
             ),
         )
+
+    def setIamPolicy(self, service_name: str, policy_body: dict):
+        self._set_iam_policy(
+            collection=self._services_collection,
+            full_name=self.resource_full_name(
+                service_name, self.SERVICES, self.region
+            ),
+            body=policy_body,
+        )

framework/rpc/grpc.py

@@ -95,10 +95,15 @@ def __init__(self, rpc_host):
         # Cache gRPC channels per port
         self.channels = dict()
 
-    def _make_channel(self, port) -> grpc.Channel:
+    def _make_channel(self, port, secure_mode=False) -> grpc.Channel:
         if port not in self.channels:
             target = f"{self.rpc_host}:{port}"
-            self.channels[port] = grpc.insecure_channel(target)
+            if secure_mode:
+                self.channels[port] = grpc.secure_channel(
+                    target, grpc.ssl_channel_credentials()
+                )
+            else:
+                self.channels[port] = grpc.insecure_channel(target)
         return self.channels[port]
 
     def close(self):

framework/test_app/client_app.py

@@ -49,6 +49,7 @@
 DEFAULT_TD_XDS_URI = "trafficdirector.googleapis.com:443"
 
 
+# pylint: disable=too-many-public-methods
 class XdsTestClient(framework.rpc.grpc.GrpcApp):
     """
     Represents RPC services implemented in Client component of the xds test app.
@@ -85,6 +86,18 @@ def load_balancer_stats(self) -> _LoadBalancerStatsServiceClient:
             log_target=f"{self.hostname}:{self.rpc_port}",
         )
 
+    # For fetching stats from gRPC client containers requiring secure
+    # communication, like Cloud run where, a proxy is involved. The proxy uses
+    # HTTP/1.1 for plaintext connections. However, since gRPC requires HTTP/2,
+    # we must use encrypted communication (HTTPS) to ensure compatibility.
+    @property
+    @functools.lru_cache(None)
+    def secure_load_balancer_stats(self) -> _LoadBalancerStatsServiceClient:
+        return _LoadBalancerStatsServiceClient(
+            self._make_channel(self.rpc_port, secure_mode=True),
+            log_target=f"{self.hostname}:{self.rpc_port}",
+        )
+
     @property
     @functools.lru_cache(None)
     def update_config(self):
@@ -109,6 +122,18 @@ def csds(self) -> _CsdsClient:
             log_target=f"{self.hostname}:{self.maintenance_port}",
         )
 
+    # For fetching stats from gRPC client containers requiring secure
+    # communication, like Cloud run where, a proxy is involved. The proxy uses
+    # HTTP/1.1 for plaintext connections. However, since gRPC requires HTTP/2,
+    # we must use encrypted communication (HTTPS) to ensure compatibility.
+    @property
+    @functools.lru_cache(None)
+    def secure_csds(self) -> _CsdsClient:
+        return _CsdsClient(
+            self._make_channel(self.maintenance_port, secure_mode=True),
+            log_target=f"{self.hostname}:{self.maintenance_port}",
+        )
+
     def get_csds_parsed(self, **kwargs) -> Optional[grpc_csds.DumpedXdsConfig]:
         return self.csds.fetch_client_status_parsed(**kwargs)
 
@@ -118,15 +143,23 @@ def get_load_balancer_stats(
         num_rpcs: int,
         metadata_keys: Optional[tuple[str, ...]] = None,
         timeout_sec: Optional[int] = None,
+        secure_mode: bool = False,
     ) -> grpc_testing.LoadBalancerStatsResponse:
         """
         Shortcut to LoadBalancerStatsServiceClient.get_client_stats()
         """
-        return self.load_balancer_stats.get_client_stats(
-            num_rpcs=num_rpcs,
-            timeout_sec=timeout_sec,
-            metadata_keys=metadata_keys,
-        )
+        if secure_mode:
+            return self.secure_load_balancer_stats.get_client_stats(
+                num_rpcs=num_rpcs,
+                timeout_sec=timeout_sec,
+                metadata_keys=metadata_keys,
+            )
+        else:
+            return self.load_balancer_stats.get_client_stats(
+                num_rpcs=num_rpcs,
+                timeout_sec=timeout_sec,
+                metadata_keys=metadata_keys,
+            )
 
     def get_load_balancer_accumulated_stats(
         self,

framework/test_app/runners/cloud_run/cloud_run_base_runner.py

@@ -99,7 +99,7 @@ def _initalize_cloud_run_api_manager(self):
             api_manager=self.gcp_api_manager,
         )
 
-    def run(self, **kwargs):
+    def run(self, **kwargs) -> None:
         if self.time_start_requested and not self.time_stopped:
             if self.time_start_completed:
                 raise RuntimeError(