xds-k8s: Allow multiple instance of the driver to run concurrently (#26542)

tools/run_tests/xds_k8s_test_driver/README.md

@@ -8,9 +8,9 @@ changes to this codebase at the moment.
 
 ### Stabilization roadmap 
 - [ ] Replace retrying with tenacity
-- [ ] Generate namespace for each test to prevent resource name conflicts and
+- [x] Generate namespace for each test to prevent resource name conflicts and
       allow running tests in parallel
-- [ ] Security: run server and client in separate namespaces
+- [x] Security: run server and client in separate namespaces
 - [ ] Make framework.infrastructure.gcp resources [first-class
       citizen](https://en.wikipedia.org/wiki/First-class_citizen), support
       simpler CRUD
@@ -198,23 +198,6 @@ python -m tests.security_test \
   --client_image="gcr.io/grpc-testing/xds-interop/java-client:d22f93e1ade22a1e026b57210f6fc21f7a3ca0cf"
 ```
 
-### Test namespace
-It's possible to run multiple xDS interop test workloads in the same project.
-But we need to ensure the name of the global resources won't conflict. This can
-be solved by supplying `--namespace` and `--server_xds_port`. The xDS port needs
-to be unique across the entire project (default port range is [8080, 8280],
-avoid if possible). Here is an example:
-
-```shell
-python3 -m tests.baseline_test \
-  --flagfile="config/grpc-testing.cfg" \
-  --kube_context="${KUBE_CONTEXT}" \
-  --server_image="gcr.io/grpc-testing/xds-interop/java-server:d22f93e1ade22a1e026b57210f6fc21f7a3ca0cf" \
-  --client_image="gcr.io/grpc-testing/xds-interop/java-client:d22f93e1ade22a1e026b57210f6fc21f7a3ca0cf" \
-  --namespace="box-$(date +"%F-%R")" \
-  --server_xds_port="$(($RANDOM%1000 + 34567))"
-```
-
 ## Local development
 This test driver allows running tests locally against remote GKE clusters, right
 from your dev environment. You need:
@@ -290,7 +273,7 @@ This tool performs the following:
 EXAMPLES:
 ./run.sh bin/run_td_setup.py --help
 ./run.sh bin/run_td_setup.py --helpfull
-XDS_K8S_CONFIG=./path-to-flagfile.cfg ./run.sh bin/run_td_setup.py --namespace=override-namespace
+XDS_K8S_CONFIG=./path-to-flagfile.cfg ./run.sh bin/run_td_setup.py --resource_suffix=override-suffix
 ./run.sh tests/baseline_test.py
 ./run.sh tests/security_test.py --verbosity=1 --logger_levels=__main__:DEBUG,framework:DEBUG
 ./run.sh tests/security_test.py SecurityTest.test_mtls --nocheck_local_certs

tools/run_tests/xds_k8s_test_driver/bin/cleanup.sh

@@ -32,7 +32,7 @@ ENVIRONMENT:
 EXAMPLES:
 $0
 $0 --secure
-XDS_K8S_CONFIG=./path-to-flagfile.cfg $0 --namespace=override-namespace
+XDS_K8S_CONFIG=./path-to-flagfile.cfg $0 --resource_suffix=override-suffix
 EOF
   exit 1
 }

tools/run_tests/xds_k8s_test_driver/bin/run_channelz.py

@@ -58,6 +58,8 @@ _SECURITY = flags.DEFINE_enum('security',
                               help='Show info for a security setup')
 flags.adopt_module_key_flags(xds_flags)
 flags.adopt_module_key_flags(xds_k8s_flags)
+# Running outside of a test suite, so require explicit resource_suffix.
+flags.mark_flag_as_required("resource_suffix")
 
 # Type aliases
 _Channel = grpc_channelz.Channel
@@ -174,9 +176,13 @@ def main(argv):
 
     k8s_api_manager = k8s.KubernetesApiManager(xds_k8s_flags.KUBE_CONTEXT.value)
 
+    # Resource names.
+    resource_prefix: str = xds_flags.RESOURCE_PREFIX.value
+    resource_suffix: str = xds_flags.RESOURCE_SUFFIX.value
+
     # Server
     server_name = xds_flags.SERVER_NAME.value
-    server_namespace = xds_flags.NAMESPACE.value
+    server_namespace = resource_prefix
     server_k8s_ns = k8s.KubernetesNamespace(k8s_api_manager, server_namespace)
     server_pod_ip = get_deployment_pod_ips(server_k8s_ns, server_name)[0]
     test_server: _XdsTestServer = _XdsTestServer(
@@ -188,7 +194,7 @@ def main(argv):
 
     # Client
     client_name = xds_flags.CLIENT_NAME.value
-    client_namespace = xds_flags.NAMESPACE.value
+    client_namespace = resource_prefix
     client_k8s_ns = k8s.KubernetesNamespace(k8s_api_manager, client_namespace)
     client_pod_ip = get_deployment_pod_ips(client_k8s_ns, client_name)[0]
     test_client: _XdsTestClient = _XdsTestClient(

tools/run_tests/xds_k8s_test_driver/bin/run_td_setup.py

@@ -31,13 +31,13 @@ Typical usage examples:
     python -m bin.run_td_setup --helpfull
 """
 import logging
-import uuid
 
 from absl import app
 from absl import flags
 
 from framework import xds_flags
 from framework import xds_k8s_flags
+from framework.helpers import rand
 from framework.infrastructure import gcp
 from framework.infrastructure import k8s
 from framework.infrastructure import traffic_director
@@ -49,7 +49,7 @@ _CMD = flags.DEFINE_enum('cmd',
                          default='create',
                          enum_values=[
                              'cycle', 'create', 'cleanup', 'backends-add',
-                             'backends-cleanup'
+                             'backends-cleanup', 'unused-xds-port'
                          ],
                          help='Command')
 _SECURITY = flags.DEFINE_enum('security',
@@ -61,9 +61,10 @@ _SECURITY = flags.DEFINE_enum('security',
                               help='Configure TD with security')
 flags.adopt_module_key_flags(xds_flags)
 flags.adopt_module_key_flags(xds_k8s_flags)
+# Running outside of a test suite, so require explicit resource_suffix.
+flags.mark_flag_as_required("resource_suffix")
 
-_DEFAULT_SECURE_MODE_MAINTENANCE_PORT = \
-    server_app.KubernetesServerRunner.DEFAULT_SECURE_MODE_MAINTENANCE_PORT
+KubernetesServerRunner = server_app.KubernetesServerRunner
 
 
 def main(argv):
@@ -75,7 +76,10 @@ def main(argv):
 
     project: str = xds_flags.PROJECT.value
     network: str = xds_flags.NETWORK.value
-    namespace = xds_flags.NAMESPACE.value
+
+    # Resource names.
+    resource_prefix: str = xds_flags.RESOURCE_PREFIX.value
+    resource_suffix: str = xds_flags.RESOURCE_SUFFIX.value
 
     # Test server
     server_name = xds_flags.SERVER_NAME.value
@@ -83,22 +87,27 @@ def main(argv):
     server_maintenance_port = xds_flags.SERVER_MAINTENANCE_PORT.value
     server_xds_host = xds_flags.SERVER_XDS_HOST.value
     server_xds_port = xds_flags.SERVER_XDS_PORT.value
+    server_namespace = KubernetesServerRunner.make_namespace_name(
+        resource_prefix, resource_suffix)
 
     gcp_api_manager = gcp.api.GcpApiManager()
 
     if security_mode is None:
-        td = traffic_director.TrafficDirectorManager(gcp_api_manager,
-                                                     project=project,
-                                                     resource_prefix=namespace,
-                                                     network=network)
+        td = traffic_director.TrafficDirectorManager(
+            gcp_api_manager,
+            project=project,
+            network=network,
+            resource_prefix=resource_prefix,
+            resource_suffix=resource_suffix)
     else:
         td = traffic_director.TrafficDirectorSecureManager(
             gcp_api_manager,
             project=project,
-            resource_prefix=namespace,
-            network=network)
+            network=network,
+            resource_prefix=resource_prefix,
+            resource_suffix=resource_suffix)
         if server_maintenance_port is None:
-            server_maintenance_port = _DEFAULT_SECURE_MODE_MAINTENANCE_PORT
+            server_maintenance_port = KubernetesServerRunner.DEFAULT_SECURE_MODE_MAINTENANCE_PORT
 
     try:
         if command in ('create', 'cycle'):
@@ -114,12 +123,12 @@ def main(argv):
                 td.setup_for_grpc(server_xds_host,
                                   server_xds_port,
                                   health_check_port=server_maintenance_port)
-                td.setup_server_security(server_namespace=namespace,
+                td.setup_server_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          server_port=server_port,
                                          tls=True,
                                          mtls=True)
-                td.setup_client_security(server_namespace=namespace,
+                td.setup_client_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          tls=True,
                                          mtls=True)
@@ -129,12 +138,12 @@ def main(argv):
                 td.setup_for_grpc(server_xds_host,
                                   server_xds_port,
                                   health_check_port=server_maintenance_port)
-                td.setup_server_security(server_namespace=namespace,
+                td.setup_server_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          server_port=server_port,
                                          tls=True,
                                          mtls=False)
-                td.setup_client_security(server_namespace=namespace,
+                td.setup_client_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          tls=True,
                                          mtls=False)
@@ -144,12 +153,12 @@ def main(argv):
                 td.setup_for_grpc(server_xds_host,
                                   server_xds_port,
                                   health_check_port=server_maintenance_port)
-                td.setup_server_security(server_namespace=namespace,
+                td.setup_server_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          server_port=server_port,
                                          tls=False,
                                          mtls=False)
-                td.setup_client_security(server_namespace=namespace,
+                td.setup_client_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          tls=False,
                                          mtls=False)
@@ -161,12 +170,12 @@ def main(argv):
                 td.setup_for_grpc(server_xds_host,
                                   server_xds_port,
                                   health_check_port=server_maintenance_port)
-                td.setup_server_security(server_namespace=namespace,
+                td.setup_server_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          server_port=server_port,
                                          tls=True,
                                          mtls=True)
-                td.setup_client_security(server_namespace=namespace,
+                td.setup_client_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          tls=True,
                                          mtls=False)
@@ -180,16 +189,16 @@ def main(argv):
                                   health_check_port=server_maintenance_port)
                 # Regular TLS setup, but with client policy configured using
                 # intentionality incorrect server_namespace.
-                td.setup_server_security(server_namespace=namespace,
+                td.setup_server_security(server_namespace=server_namespace,
                                          server_name=server_name,
                                          server_port=server_port,
                                          tls=True,
                                          mtls=False)
-                incorrect_namespace = f'incorrect-namespace-{uuid.uuid4().hex}'
-                td.setup_client_security(server_namespace=incorrect_namespace,
-                                         server_name=server_name,
-                                         tls=True,
-                                         mtls=False)
+                td.setup_client_security(
+                    server_namespace=f'incorrect-namespace-{rand.rand_string()}',
+                    server_name=server_name,
+                    tls=True,
+                    mtls=False)
 
             logger.info('Works!')
     except Exception:  # noqa pylint: disable=broad-except
@@ -203,7 +212,8 @@ def main(argv):
         logger.info('Adding backends')
         k8s_api_manager = k8s.KubernetesApiManager(
             xds_k8s_flags.KUBE_CONTEXT.value)
-        k8s_namespace = k8s.KubernetesNamespace(k8s_api_manager, namespace)
+        k8s_namespace = k8s.KubernetesNamespace(k8s_api_manager,
+                                                server_namespace)
 
         neg_name, neg_zones = k8s_namespace.get_service_neg(
             server_name, server_port)
@@ -211,10 +221,16 @@ def main(argv):
         td.load_backend_service()
         td.backend_service_add_neg_backends(neg_name, neg_zones)
         td.wait_for_backends_healthy_status()
-        # TODO(sergiitk): wait until client reports rpc health
     elif command == 'backends-cleanup':
         td.load_backend_service()
         td.backend_service_remove_all_backends()
+    elif command == 'unused-xds-port':
+        try:
+            unused_xds_port = td.find_unused_forwarding_rule_port()
+            logger.info('Found unused forwarding rule port: %s',
+                        unused_xds_port)
+        except Exception:  # noqa pylint: disable=broad-except
+            logger.exception("Couldn't find unused forwarding rule port")
 
 
 if __name__ == '__main__':

tools/run_tests/xds_k8s_test_driver/bin/run_test_client.py

@@ -44,20 +44,22 @@ _CLEANUP_NAMESPACE = flags.DEFINE_bool(
     help="Delete namespace during resource cleanup")
 flags.adopt_module_key_flags(xds_flags)
 flags.adopt_module_key_flags(xds_k8s_flags)
+# Running outside of a test suite, so require explicit resource_suffix.
+flags.mark_flag_as_required("resource_suffix")
+
+# Type aliases
+KubernetesClientRunner = client_app.KubernetesClientRunner
 
 
 def main(argv):
     if len(argv) > 1:
         raise app.UsageError('Too many command-line arguments.')
 
-    # Flag shortcuts.
     project: str = xds_flags.PROJECT.value
     # GCP Service Account email
     gcp_service_account: str = xds_k8s_flags.GCP_SERVICE_ACCOUNT.value
-    # Base namespace
-    namespace = xds_flags.NAMESPACE.value
-    client_namespace = namespace
 
+    # KubernetesClientRunner arguments.
     runner_kwargs = dict(
         deployment_name=xds_flags.CLIENT_NAME.value,
         image_name=xds_k8s_flags.CLIENT_IMAGE.value,
@@ -75,7 +77,9 @@ def main(argv):
             deployment_template='client-secure.deployment.yaml')
 
     k8s_api_manager = k8s.KubernetesApiManager(xds_k8s_flags.KUBE_CONTEXT.value)
-    client_runner = client_app.KubernetesClientRunner(
+    client_namespace = KubernetesClientRunner.make_namespace_name(
+        xds_flags.RESOURCE_PREFIX.value, xds_flags.RESOURCE_SUFFIX.value)
+    client_runner = KubernetesClientRunner(
         k8s.KubernetesNamespace(k8s_api_manager, client_namespace),
         **runner_kwargs)
 

tools/run_tests/xds_k8s_test_driver/bin/run_test_server.py

@@ -40,20 +40,25 @@ _CLEANUP_NAMESPACE = flags.DEFINE_bool(
     help="Delete namespace during resource cleanup")
 flags.adopt_module_key_flags(xds_flags)
 flags.adopt_module_key_flags(xds_k8s_flags)
+# Running outside of a test suite, so require explicit resource_suffix.
+flags.mark_flag_as_required("resource_suffix")
+
+KubernetesServerRunner = server_app.KubernetesServerRunner
 
 
 def main(argv):
     if len(argv) > 1:
         raise app.UsageError('Too many command-line arguments.')
 
-    # Flag shortcuts.
     project: str = xds_flags.PROJECT.value
     # GCP Service Account email
     gcp_service_account: str = xds_k8s_flags.GCP_SERVICE_ACCOUNT.value
-    # Base namespace
-    namespace = xds_flags.NAMESPACE.value
-    server_namespace = namespace
 
+    # Resource names.
+    resource_prefix: str = xds_flags.RESOURCE_PREFIX.value
+    resource_suffix: str = xds_flags.RESOURCE_SUFFIX.value
+
+    # KubernetesServerRunner arguments.
     runner_kwargs = dict(
         deployment_name=xds_flags.SERVER_NAME.value,
         image_name=xds_k8s_flags.SERVER_IMAGE.value,
@@ -70,7 +75,9 @@ def main(argv):
             deployment_template='server-secure.deployment.yaml')
 
     k8s_api_manager = k8s.KubernetesApiManager(xds_k8s_flags.KUBE_CONTEXT.value)
-    server_runner = server_app.KubernetesServerRunner(
+    server_namespace = KubernetesServerRunner.make_namespace_name(
+        resource_prefix, resource_suffix)
+    server_runner = KubernetesServerRunner(
         k8s.KubernetesNamespace(k8s_api_manager, server_namespace),
         **runner_kwargs)
 

tools/run_tests/xds_k8s_test_driver/config/common.cfg

@@ -1,4 +1,4 @@
---namespace=interop-psm-security
+--resource_prefix=xds-k8s-security
 --td_bootstrap_image=gcr.io/grpc-testing/td-grpc-bootstrap:2558ec79df06984ed0d37e9e69f34688ffe301bb
 --logger_levels=__main__:DEBUG,framework:INFO
 --verbosity=0

tools/run_tests/xds_k8s_test_driver/config/grpc-testing.cfg

@@ -3,3 +3,5 @@
 --network=default-vpc
 --gcp_service_account=xds-k8s-interop-tests@grpc-testing.iam.gserviceaccount.com
 --private_api_key_secret_name=projects/830293263384/secrets/xds-interop-tests-private-api-access-key
+# Randomize xds port.
+--server_xds_port=0

tools/run_tests/xds_k8s_test_driver/config/local-dev.cfg.example

@@ -11,6 +11,9 @@
 # Uncomment to ensure the allow health check firewall exists before test case runs
 # --ensure_firewall
 
+# Use predictable resource suffix to simplify debugging
+--resource_suffix=dev
+
 # The name of kube context to use. See `gcloud container clusters get-credentials` and `kubectl config`
 --kube_context=context_name
 

tools/run_tests/xds_k8s_test_driver/framework/helpers/datetime.py

@@ -0,0 +1,39 @@
+# Copyright 2021 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.
+"""This contains common helpers for working with dates and time."""
+import datetime
+import re
+from typing import Pattern
+
+RE_ZERO_OFFSET: Pattern[str] = re.compile(r'[+\-]00:?00$')
+
+
+def utc_now() -> datetime.datetime:
+    """Construct a datetime from current time in UTC timezone."""
+    return datetime.datetime.now(datetime.timezone.utc)
+
+
+def datetime_suffix(*, seconds: bool = False) -> str:
+    """Return current UTC date, and time in a format useful for resource naming.
+
+    Examples:
+        - 20210626-1859   (seconds=False)
+        - 20210626-185942 (seconds=True)
+    Use in resources names incompatible with ISO 8601, e.g. some GCP resources
+    that only allow lowercase alphanumeric chars and dashes.
+
+    Hours and minutes are joined together for better readability, so time is
+    visually distinct from dash-separated date.
+    """
+    return utc_now().strftime('%Y%m%d-%H%M' + ('%S' if seconds else ''))

tools/run_tests/xds_k8s_test_driver/framework/helpers/rand.py

@@ -0,0 +1,33 @@
+# Copyright 2021 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.
+"""This contains common helpers for generating randomized data."""
+import random
+import string
+
+# Alphanumeric characters, similar to regex [:alnum:] class, [a-zA-Z0-9]
+ALPHANUM = string.ascii_letters + string.digits
+# Lowercase alphanumeric characters: [a-z0-9]
+# Use ALPHANUM_LOWERCASE alphabet when case-sensitivity is a concern.
+ALPHANUM_LOWERCASE = string.ascii_lowercase + string.digits
+
+
+def rand_string(length: int = 8, *, lowercase: bool = False) -> str:
+    """Return random alphanumeric string of given length.
+
+    Space for default arguments: alphabet^length
+       lowercase and uppercase = (26*2 + 10)^8 = 2.18e14 = 218 trillion.
+       lowercase only = (26 + 10)^8 = 2.8e12 = 2.8 trillion.
+    """
+    alphabet = ALPHANUM_LOWERCASE if lowercase else ALPHANUM
+    return ''.join(random.choices(population=alphabet, k=length))

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

@@ -235,6 +235,17 @@ class ComputeV1(gcp.api.GcpProjectApiResource):
                 'target': target_proxy.url,
             })
 
+    def exists_forwarding_rule(self, src_port) -> bool:
+        # TODO(sergiitk): Better approach for confirming the port is available.
+        #   It's possible a rule allocates actual port range, e.g 8000-9000,
+        #   and this wouldn't catch it. For now, we assume there's no
+        #   port ranges used in the project.
+        filter_str = (f'(portRange eq "{src_port}-{src_port}") '
+                      f'(IPAddress eq "0.0.0.0")'
+                      f'(loadBalancingScheme eq "INTERNAL_SELF_MANAGED")')
+        return self._exists_resource(self.api.globalForwardingRules(),
+                                     filter=filter_str)
+
     def delete_forwarding_rule(self, name):
         self._delete_resource(self.api.globalForwardingRules(),
                               'forwardingRule', name)
@@ -329,6 +340,16 @@ class ComputeV1(gcp.api.GcpProjectApiResource):
                     self.resource_pretty_format(resp))
         return self.GcpResource(resp['name'], resp['selfLink'])
 
+    def _exists_resource(self, collection: discovery.Resource,
+                         filter: str) -> bool:
+        resp = collection.list(
+            project=self.project, filter=filter,
+            maxResults=1).execute(num_retries=self._GCP_API_RETRIES)
+        if 'kind' not in resp:
+            # TODO(sergiitk): better error
+            raise ValueError('List response "kind" is missing')
+        return 'items' in resp and resp['items']
+
     def _insert_resource(self, collection: discovery.Resource,
                          body: Dict[str, Any]) -> GcpResource:
         logger.info('Creating compute resource:\n%s',

tools/run_tests/xds_k8s_test_driver/framework/infrastructure/traffic_director.py

@@ -11,7 +11,9 @@
 # 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.
+import functools
 import logging
+import random
 from typing import Any, List, Optional, Set
 
 from framework import xds_flags
@@ -41,8 +43,11 @@ EndpointConfigSelector = _NetworkServicesV1Alpha1.EndpointConfigSelector
 
 class TrafficDirectorManager:
     compute: _ComputeV1
+    resource_prefix: str
+    resource_suffix: str
+
     BACKEND_SERVICE_NAME = "backend-service"
-    ALTERNATIVE_BACKEND_SERVICE_NAME = "alternative-backend-service"
+    ALTERNATIVE_BACKEND_SERVICE_NAME = "backend-service-alt"
     HEALTH_CHECK_NAME = "health-check"
     URL_MAP_NAME = "url-map"
     URL_MAP_PATH_MATCHER_NAME = "path-matcher"
@@ -56,6 +61,7 @@ class TrafficDirectorManager:
         project: str,
         *,
         resource_prefix: str,
+        resource_suffix: str,
         network: str = 'default',
     ):
         # API
@@ -65,6 +71,7 @@ class TrafficDirectorManager:
         self.project: str = project
         self.network: str = network
         self.resource_prefix: str = resource_prefix
+        self.resource_suffix: str = resource_suffix
 
         # Managed resources
         self.health_check: Optional[GcpResource] = None
@@ -124,8 +131,14 @@ class TrafficDirectorManager:
         self.delete_alternative_backend_service(force=force)
         self.delete_health_check(force=force)
 
-    def _ns_name(self, name):
-        return f'{self.resource_prefix}-{name}'
+    @functools.lru_cache(None)
+    def _make_resource_name(self, name: str) -> str:
+        """Make dash-separated resource name with resource prefix and suffix."""
+        parts = [self.resource_prefix, name]
+        # Avoid trailing dash when the suffix is empty.
+        if self.resource_suffix:
+            parts.append(self.resource_suffix)
+        return '-'.join(parts)
 
     def create_health_check(
             self,
@@ -138,14 +151,14 @@ class TrafficDirectorManager:
         if protocol is None:
             protocol = _HealthCheckGRPC
 
-        name = self._ns_name(self.HEALTH_CHECK_NAME)
+        name = self._make_resource_name(self.HEALTH_CHECK_NAME)
         logger.info('Creating %s Health Check "%s"', protocol.name, name)
         resource = self.compute.create_health_check(name, protocol, port=port)
         self.health_check = resource
 
     def delete_health_check(self, force=False):
         if force:
-            name = self._ns_name(self.HEALTH_CHECK_NAME)
+            name = self._make_resource_name(self.HEALTH_CHECK_NAME)
         elif self.health_check:
             name = self.health_check.name
         else:
@@ -159,7 +172,7 @@ class TrafficDirectorManager:
         if protocol is None:
             protocol = _BackendGRPC
 
-        name = self._ns_name(self.BACKEND_SERVICE_NAME)
+        name = self._make_resource_name(self.BACKEND_SERVICE_NAME)
         logger.info('Creating %s Backend Service "%s"', protocol.name, name)
         resource = self.compute.create_backend_service_traffic_director(
             name, health_check=self.health_check, protocol=protocol)
@@ -167,13 +180,13 @@ class TrafficDirectorManager:
         self.backend_service_protocol = protocol
 
     def load_backend_service(self):
-        name = self._ns_name(self.BACKEND_SERVICE_NAME)
+        name = self._make_resource_name(self.BACKEND_SERVICE_NAME)
         resource = self.compute.get_backend_service_traffic_director(name)
         self.backend_service = resource
 
     def delete_backend_service(self, force=False):
         if force:
-            name = self._ns_name(self.BACKEND_SERVICE_NAME)
+            name = self._make_resource_name(self.BACKEND_SERVICE_NAME)
         elif self.backend_service:
             name = self.backend_service.name
         else:
@@ -213,7 +226,7 @@ class TrafficDirectorManager:
             self, protocol: Optional[BackendServiceProtocol] = _BackendGRPC):
         if protocol is None:
             protocol = _BackendGRPC
-        name = self._ns_name(self.ALTERNATIVE_BACKEND_SERVICE_NAME)
+        name = self._make_resource_name(self.ALTERNATIVE_BACKEND_SERVICE_NAME)
         logger.info('Creating %s Alternative Backend Service "%s"',
                     protocol.name, name)
         resource = self.compute.create_backend_service_traffic_director(
@@ -222,13 +235,14 @@ class TrafficDirectorManager:
         self.alternative_backend_service_protocol = protocol
 
     def load_alternative_backend_service(self):
-        name = self._ns_name(self.ALTERNATIVE_BACKEND_SERVICE_NAME)
+        name = self._make_resource_name(self.ALTERNATIVE_BACKEND_SERVICE_NAME)
         resource = self.compute.get_backend_service_traffic_director(name)
         self.alternative_backend_service = resource
 
     def delete_alternative_backend_service(self, force=False):
         if force:
-            name = self._ns_name(self.ALTERNATIVE_BACKEND_SERVICE_NAME)
+            name = self._make_resource_name(
+                self.ALTERNATIVE_BACKEND_SERVICE_NAME)
         elif self.alternative_backend_service:
             name = self.alternative_backend_service.name
         else:
@@ -272,8 +286,8 @@ class TrafficDirectorManager:
         src_port: int,
     ) -> GcpResource:
         src_address = f'{src_host}:{src_port}'
-        name = self._ns_name(self.URL_MAP_NAME)
-        matcher_name = self._ns_name(self.URL_MAP_PATH_MATCHER_NAME)
+        name = self._make_resource_name(self.URL_MAP_NAME)
+        matcher_name = self._make_resource_name(self.URL_MAP_PATH_MATCHER_NAME)
         logger.info('Creating URL map "%s": %s -> %s', name, src_address,
                     self.backend_service.name)
         resource = self.compute.create_url_map(name, matcher_name,
@@ -290,7 +304,7 @@ class TrafficDirectorManager:
 
     def delete_url_map(self, force=False):
         if force:
-            name = self._ns_name(self.URL_MAP_NAME)
+            name = self._make_resource_name(self.URL_MAP_NAME)
         elif self.url_map:
             name = self.url_map.name
         else:
@@ -300,7 +314,7 @@ class TrafficDirectorManager:
         self.url_map = None
 
     def create_target_proxy(self):
-        name = self._ns_name(self.TARGET_PROXY_NAME)
+        name = self._make_resource_name(self.TARGET_PROXY_NAME)
         if self.backend_service_protocol is BackendServiceProtocol.GRPC:
             target_proxy_type = 'GRPC'
             create_proxy_fn = self.compute.create_target_grpc_proxy
@@ -318,7 +332,7 @@ class TrafficDirectorManager:
 
     def delete_target_grpc_proxy(self, force=False):
         if force:
-            name = self._ns_name(self.TARGET_PROXY_NAME)
+            name = self._make_resource_name(self.TARGET_PROXY_NAME)
         elif self.target_proxy:
             name = self.target_proxy.name
         else:
@@ -330,7 +344,7 @@ class TrafficDirectorManager:
 
     def delete_target_http_proxy(self, force=False):
         if force:
-            name = self._ns_name(self.TARGET_PROXY_NAME)
+            name = self._make_resource_name(self.TARGET_PROXY_NAME)
         elif self.target_proxy:
             name = self.target_proxy.name
         else:
@@ -340,8 +354,21 @@ class TrafficDirectorManager:
         self.target_proxy = None
         self.target_proxy_is_http = False
 
+    def find_unused_forwarding_rule_port(
+            self,
+            *,
+            lo: int = 1024,  # To avoid confusion, skip well-known ports.
+            hi: int = 65535,
+            attempts: int = 25) -> int:
+        for attempts in range(attempts):
+            src_port = random.randint(lo, hi)
+            if not (self.compute.exists_forwarding_rule(src_port)):
+                return src_port
+        # TODO(sergiitk): custom exception
+        raise RuntimeError("Couldn't find unused forwarding rule port")
+
     def create_forwarding_rule(self, src_port: int):
-        name = self._ns_name(self.FORWARDING_RULE_NAME)
+        name = self._make_resource_name(self.FORWARDING_RULE_NAME)
         src_port = int(src_port)
         logging.info(
             'Creating forwarding rule "%s" in network "%s": 0.0.0.0:%s -> %s',
@@ -354,7 +381,7 @@ class TrafficDirectorManager:
 
     def delete_forwarding_rule(self, force=False):
         if force:
-            name = self._ns_name(self.FORWARDING_RULE_NAME)
+            name = self._make_resource_name(self.FORWARDING_RULE_NAME)
         elif self.forwarding_rule:
             name = self.forwarding_rule.name
         else:
@@ -364,7 +391,7 @@ class TrafficDirectorManager:
         self.forwarding_rule = None
 
     def create_firewall_rule(self, allowed_ports: List[str]):
-        name = self._ns_name(self.FIREWALL_RULE_NAME)
+        name = self._make_resource_name(self.FIREWALL_RULE_NAME)
         logging.info(
             'Creating firewall rule "%s" in network "%s" with allowed ports %s',
             name, self.network, allowed_ports)
@@ -376,7 +403,7 @@ class TrafficDirectorManager:
     def delete_firewall_rule(self, force=False):
         """The firewall rule won't be automatically removed."""
         if force:
-            name = self._ns_name(self.FIREWALL_RULE_NAME)
+            name = self._make_resource_name(self.FIREWALL_RULE_NAME)
         elif self.firewall_rule:
             name = self.firewall_rule.name
         else:
@@ -390,7 +417,8 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
     netsec: Optional[_NetworkSecurityV1Alpha1]
     SERVER_TLS_POLICY_NAME = "server-tls-policy"
     CLIENT_TLS_POLICY_NAME = "client-tls-policy"
-    ENDPOINT_CONFIG_SELECTOR_NAME = "endpoint-config-selector"
+    # TODO(sergiitk): Rename to ENDPOINT_POLICY_NAME when upgraded to v1beta
+    ENDPOINT_CONFIG_SELECTOR_NAME = "endpoint-policy"
     CERTIFICATE_PROVIDER_INSTANCE = "google_cloud_private_spiffe"
 
     def __init__(
@@ -399,11 +427,13 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
         project: str,
         *,
         resource_prefix: str,
+        resource_suffix: Optional[str] = None,
         network: str = 'default',
     ):
         super().__init__(gcp_api_manager,
                          project,
                          resource_prefix=resource_prefix,
+                         resource_suffix=resource_suffix,
                          network=network)
 
         # API
@@ -445,7 +475,7 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
         self.delete_client_tls_policy(force=force)
 
     def create_server_tls_policy(self, *, tls, mtls):
-        name = self._ns_name(self.SERVER_TLS_POLICY_NAME)
+        name = self._make_resource_name(self.SERVER_TLS_POLICY_NAME)
         logger.info('Creating Server TLS Policy %s', name)
         if not tls and not mtls:
             logger.warning(
@@ -468,7 +498,7 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
 
     def delete_server_tls_policy(self, force=False):
         if force:
-            name = self._ns_name(self.SERVER_TLS_POLICY_NAME)
+            name = self._make_resource_name(self.SERVER_TLS_POLICY_NAME)
         elif self.server_tls_policy:
             name = self.server_tls_policy.name
         else:
@@ -479,7 +509,7 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
 
     def create_endpoint_config_selector(self, server_namespace, server_name,
                                         server_port):
-        name = self._ns_name(self.ENDPOINT_CONFIG_SELECTOR_NAME)
+        name = self._make_resource_name(self.ENDPOINT_CONFIG_SELECTOR_NAME)
         logger.info('Creating Endpoint Config Selector %s', name)
         endpoint_matcher_labels = [{
             "labelName": "app",
@@ -511,7 +541,7 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
 
     def delete_endpoint_config_selector(self, force=False):
         if force:
-            name = self._ns_name(self.ENDPOINT_CONFIG_SELECTOR_NAME)
+            name = self._make_resource_name(self.ENDPOINT_CONFIG_SELECTOR_NAME)
         elif self.ecs:
             name = self.ecs.name
         else:
@@ -521,7 +551,7 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
         self.ecs = None
 
     def create_client_tls_policy(self, *, tls, mtls):
-        name = self._ns_name(self.CLIENT_TLS_POLICY_NAME)
+        name = self._make_resource_name(self.CLIENT_TLS_POLICY_NAME)
         logger.info('Creating Client TLS Policy %s', name)
         if not tls and not mtls:
             logger.warning(
@@ -542,7 +572,7 @@ class TrafficDirectorSecureManager(TrafficDirectorManager):
 
     def delete_client_tls_policy(self, force=False):
         if force:
-            name = self._ns_name(self.CLIENT_TLS_POLICY_NAME)
+            name = self._make_resource_name(self.CLIENT_TLS_POLICY_NAME)
         elif self.client_tls_policy:
             name = self.client_tls_policy.name
         else:

tools/run_tests/xds_k8s_test_driver/framework/test_app/base_runner.py

@@ -286,3 +286,14 @@ class KubernetesBaseRunner:
             name, service_port)
         logger.info("Service %s: detected NEG=%s in zones=%s", name, neg_name,
                     neg_zones)
+
+    @classmethod
+    def _make_namespace_name(cls, resource_prefix: str, resource_suffix: str,
+                             name: str) -> str:
+        """A helper to make consistent test app kubernetes namespace name
+        for given resource prefix and suffix."""
+        parts = [resource_prefix, name]
+        # Avoid trailing dash when the suffix is empty.
+        if resource_suffix:
+            parts.append(resource_suffix)
+        return '-'.join(parts)

tools/run_tests/xds_k8s_test_driver/framework/test_app/client_app.py

@@ -338,3 +338,17 @@ class KubernetesClientRunner(base_runner.KubernetesBaseRunner):
             self._delete_service_account(self.service_account_name)
             self.service_account = None
         super().cleanup(force=force_namespace and force)
+
+    @classmethod
+    def make_namespace_name(cls,
+                            resource_prefix: str,
+                            resource_suffix: str,
+                            name: str = 'client') -> str:
+        """A helper to make consistent XdsTestClient kubernetes namespace name
+        for given resource prefix and suffix.
+
+        Note: the idea is to intentionally produce different namespace name for
+        the test server, and the test client, as that closely mimics real-world
+        deployments.
+        """
+        return cls._make_namespace_name(resource_prefix, resource_suffix, name)

tools/run_tests/xds_k8s_test_driver/framework/test_app/server_app.py

@@ -307,3 +307,18 @@ class KubernetesServerRunner(base_runner.KubernetesBaseRunner):
             self._delete_service_account(self.service_account_name)
             self.service_account = None
         super().cleanup(force=(force_namespace and force))
+
+    @classmethod
+    def make_namespace_name(cls,
+                            resource_prefix: str,
+                            resource_suffix: str,
+                            name: str = 'server') -> str:
+        """A helper to make consistent XdsTestServer kubernetes namespace name
+        for given resource prefix and suffix.
+
+        Note: the idea is to intentionally produce different namespace name for
+        the test server, and the test client, as that closely mimics real-world
+        deployments.
+        :rtype: object
+        """
+        return cls._make_namespace_name(resource_prefix, resource_suffix, name)

tools/run_tests/xds_k8s_test_driver/framework/xds_flags.py

@@ -17,11 +17,27 @@ import googleapiclient.discovery
 # GCP
 PROJECT = flags.DEFINE_string("project",
                               default=None,
-                              help="GCP Project ID. Required")
+                              help="(required) GCP Project ID.")
+RESOURCE_PREFIX = flags.DEFINE_string(
+    "resource_prefix",
+    default=None,
+    help=("(required) The prefix used to name GCP resources.\n"
+          "Together with `resource_suffix` used to create unique "
+          "resource names."))
+# TODO(sergiitk): remove after all migration to --resource_prefix completed.
+#   Known migration work: url map, staging flagfiles.
 NAMESPACE = flags.DEFINE_string(
     "namespace",
     default=None,
-    help="Isolate GCP resources using given namespace / name prefix. Required")
+    help="Deprecated. Use --resource_prefix instead.")
+RESOURCE_SUFFIX = flags.DEFINE_string(
+    "resource_suffix",
+    default=None,
+    help=("The suffix used to name GCP resources.\n"
+          "Together with `resource_prefix` used to create unique "
+          "resource names.\n"
+          "(default: test suite will generate a random suffix, based on suite "
+          "resource management preferences)"))
 NETWORK = flags.DEFINE_string("network",
                               default="default",
                               help="GCP Network ID")
@@ -29,7 +45,7 @@ NETWORK = flags.DEFINE_string("network",
 XDS_SERVER_URI = flags.DEFINE_string(
     "xds_server_uri",
     default=None,
-    help="Override Traffic Director server uri, for testing")
+    help="Override Traffic Director server URI.")
 ENSURE_FIREWALL = flags.DEFINE_bool(
     "ensure_firewall",
     default=False,
@@ -44,37 +60,62 @@ FIREWALL_ALLOWED_PORTS = flags.DEFINE_list(
     help="Update the allowed ports of the firewall rule.")
 
 # Test server
-SERVER_NAME = flags.DEFINE_string("server_name",
-                                  default="psm-grpc-server",
-                                  help="Server deployment and service name")
-SERVER_PORT = flags.DEFINE_integer("server_port",
-                                   default=8080,
-                                   lower_bound=0,
-                                   upper_bound=65535,
-                                   help="Server test port")
+SERVER_NAME = flags.DEFINE_string(
+    "server_name",
+    default="psm-grpc-server",
+    help="The name to use for test server deployments.")
+SERVER_PORT = flags.DEFINE_integer(
+    "server_port",
+    default=8080,
+    lower_bound=1,
+    upper_bound=65535,
+    help="Server test port.\nMust be within --firewall_allowed_ports.")
 SERVER_MAINTENANCE_PORT = flags.DEFINE_integer(
     "server_maintenance_port",
+    default=None,
+    lower_bound=1,
+    upper_bound=65535,
+    help=("Server port running maintenance services: Channelz, CSDS, Health, "
+          "XdsUpdateHealth, and ProtoReflection (optional).\n"
+          "Must be within --firewall_allowed_ports.\n"
+          "(default: the port is chosen automatically based on "
+          "the security configuration)"))
+SERVER_XDS_HOST = flags.DEFINE_string(
+    "server_xds_host",
+    default="xds-test-server",
+    help=("The xDS hostname of the test server.\n"
+          "Together with `server_xds_port` makes test server target URI, "
+          "xds:///hostname:port"))
+# Note: port 0 known to represent a request for dynamically-allocated port
+# https://en.wikipedia.org/wiki/List_of_TCP_and_UDP_port_numbers#Well-known_ports
+SERVER_XDS_PORT = flags.DEFINE_integer(
+    "server_xds_port",
+    default=8080,
     lower_bound=0,
     upper_bound=65535,
-    default=None,
-    help="Server port running maintenance services: health check, channelz, etc"
-)
-SERVER_XDS_HOST = flags.DEFINE_string("server_xds_host",
-                                      default='xds-test-server',
-                                      help="Test server xDS hostname")
-SERVER_XDS_PORT = flags.DEFINE_integer("server_xds_port",
-                                       default=8000,
-                                       help="Test server xDS port")
+    help=("The xDS port of the test server.\n"
+          "Together with `server_xds_host` makes test server target URI, "
+          "xds:///hostname:port\n"
+          "Must be unique within a GCP project.\n"
+          "Set to 0 to select any unused port."))
 
 # Test client
-CLIENT_NAME = flags.DEFINE_string("client_name",
-                                  default="psm-grpc-client",
-                                  help="Client deployment and service name")
-CLIENT_PORT = flags.DEFINE_integer("client_port",
-                                   default=8079,
-                                   help="Client test port")
+CLIENT_NAME = flags.DEFINE_string(
+    "client_name",
+    default="psm-grpc-client",
+    help="The name to use for test client deployments")
+CLIENT_PORT = flags.DEFINE_integer(
+    "client_port",
+    default=8079,
+    lower_bound=1,
+    upper_bound=65535,
+    help=(
+        "The port test client uses to run gRPC services: Channelz, CSDS, "
+        "XdsStats, XdsUpdateClientConfigure, and ProtoReflection (optional).\n"
+        "Doesn't have to be within --firewall_allowed_ports."))
 
 flags.mark_flags_as_required([
     "project",
-    "namespace",
+    # TODO(sergiitk): Make required when --namespace is removed.
+    # "resource_prefix",
 ])

tools/run_tests/xds_k8s_test_driver/framework/xds_k8s_testcase.py

@@ -11,6 +11,7 @@
 # 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.
+import abc
 import datetime
 import enum
 import hashlib
@@ -25,6 +26,8 @@ from google.protobuf import json_format
 from framework import xds_flags
 from framework import xds_k8s_flags
 from framework.helpers import retryers
+import framework.helpers.datetime
+import framework.helpers.rand
 from framework.infrastructure import gcp
 from framework.infrastructure import k8s
 from framework.infrastructure import traffic_director
@@ -48,21 +51,35 @@ flags.adopt_module_key_flags(xds_flags)
 flags.adopt_module_key_flags(xds_k8s_flags)
 
 # Type aliases
+TrafficDirectorManager = traffic_director.TrafficDirectorManager
+TrafficDirectorSecureManager = traffic_director.TrafficDirectorSecureManager
 XdsTestServer = server_app.XdsTestServer
 XdsTestClient = client_app.XdsTestClient
+KubernetesServerRunner = server_app.KubernetesServerRunner
+KubernetesClientRunner = client_app.KubernetesClientRunner
 LoadBalancerStatsResponse = grpc_testing.LoadBalancerStatsResponse
 _ChannelState = grpc_channelz.ChannelState
 _timedelta = datetime.timedelta
-_DEFAULT_SECURE_MODE_MAINTENANCE_PORT = \
-    server_app.KubernetesServerRunner.DEFAULT_SECURE_MODE_MAINTENANCE_PORT
 
 
-class XdsKubernetesTestCase(absltest.TestCase):
-    k8s_api_manager: k8s.KubernetesApiManager
+class XdsKubernetesTestCase(absltest.TestCase, metaclass=abc.ABCMeta):
+    _resource_suffix_randomize: bool = True
+    client_namespace: str
+    client_runner: KubernetesClientRunner
     gcp_api_manager: gcp.api.GcpApiManager
+    k8s_api_manager: k8s.KubernetesApiManager
+    resource_prefix: str
+    resource_suffix: str = ''
+    server_namespace: str
+    server_runner: KubernetesServerRunner
+    server_xds_port: int
+    td: TrafficDirectorManager
 
     @classmethod
     def setUpClass(cls):
+        """Hook method for setting up class fixture before running tests in
+        the class.
+        """
         # GCP
         cls.project: str = xds_flags.PROJECT.value
         cls.network: str = xds_flags.NETWORK.value
@@ -72,9 +89,17 @@ class XdsKubernetesTestCase(absltest.TestCase):
         cls.ensure_firewall = xds_flags.ENSURE_FIREWALL.value
         cls.firewall_allowed_ports = xds_flags.FIREWALL_ALLOWED_PORTS.value
 
-        # Base namespace
-        # TODO(sergiitk): generate for each test
-        cls.namespace: str = xds_flags.NAMESPACE.value
+        # Resource names.
+        # TODO(sergiitk): Drop namespace parsing when --namespace is removed.
+        cls.resource_prefix = (xds_flags.RESOURCE_PREFIX.value or
+                               xds_flags.NAMESPACE.value)
+        if not cls.resource_prefix:
+            raise flags.IllegalFlagValueError(
+                'Required one of the flags: --resource_prefix or --namespace')
+
+        if xds_flags.RESOURCE_SUFFIX.value is not None:
+            cls._resource_suffix_randomize = False
+            cls.resource_suffix = xds_flags.RESOURCE_SUFFIX.value
 
         # Test server
         cls.server_image = xds_k8s_flags.SERVER_IMAGE.value
@@ -101,15 +126,53 @@ class XdsKubernetesTestCase(absltest.TestCase):
         cls.gcp_api_manager = gcp.api.GcpApiManager()
 
     def setUp(self):
-        # TODO(sergiitk): generate namespace with run id for each test
-        self.server_namespace = self.namespace
-        self.client_namespace = self.namespace
+        """Hook method for setting up the test fixture before exercising it."""
+        super().setUp()
+
+        if self._resource_suffix_randomize:
+            self.resource_suffix = self._random_resource_suffix()
+        logger.info('Test run resource prefix: %s, suffix: %s',
+                    self.resource_prefix, self.resource_suffix)
+
+        # TD Manager
+        self.td = self.initTrafficDirectorManager()
+
+        # Test Server runner
+        self.server_namespace = KubernetesServerRunner.make_namespace_name(
+            self.resource_prefix, self.resource_suffix)
+        self.server_runner = self.initKubernetesServerRunner()
+
+        # Test Client runner
+        self.client_namespace = KubernetesClientRunner.make_namespace_name(
+            self.resource_prefix, self.resource_suffix)
+        self.client_runner = self.initKubernetesClientRunner()
+
+        # Ensures the firewall exist
+        if self.ensure_firewall:
+            self.td.create_firewall_rule(
+                allowed_ports=self.firewall_allowed_ports)
+
+        # Randomize xds port, when it's set to 0
+        if self.server_xds_port == 0:
+            # TODO(sergiitk): this is prone to race conditions:
+            #  The port might not me taken now, but there's not guarantee
+            #  it won't be taken until the tests get to creating
+            #  forwarding rule. This check is better than nothing,
+            #  but we should find a better approach.
+            self.server_xds_port = self.td.find_unused_forwarding_rule_port()
+            logger.info('Found unused xds port: %s', self.server_xds_port)
 
-        # Init this in child class
-        # TODO(sergiitk): consider making a method to be less error-prone
-        self.server_runner = None
-        self.client_runner = None
-        self.td = None
+    @abc.abstractmethod
+    def initTrafficDirectorManager(self) -> TrafficDirectorManager:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def initKubernetesServerRunner(self) -> KubernetesServerRunner:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def initKubernetesClientRunner(self) -> KubernetesClientRunner:
+        raise NotImplementedError
 
     @classmethod
     def tearDownClass(cls):
@@ -132,6 +195,19 @@ class XdsKubernetesTestCase(absltest.TestCase):
         self.server_runner.cleanup(force=self.force_cleanup,
                                    force_namespace=self.force_cleanup)
 
+    @staticmethod
+    def _random_resource_suffix() -> str:
+        # Date and time suffix for debugging. Seconds skipped, not as relevant
+        # Format example: 20210626-1859
+        datetime_suffix: str = framework.helpers.datetime.datetime_suffix()
+        # Use lowercase chars because some resource names won't allow uppercase.
+        # For len 5, total (26 + 10)^5 = 60,466,176 combinations.
+        # Approx. number of test runs needed to start at the same minute to
+        # produce a collision: math.sqrt(math.pi/2 * (26+10)**5) ≈ 9745.
+        # https://en.wikipedia.org/wiki/Birthday_attack#Mathematics
+        unique_hash: str = framework.helpers.rand.rand_string(5, lowercase=True)
+        return f'{datetime_suffix}-{unique_hash}'
+
     def setupTrafficDirectorGrpc(self):
         self.td.setup_for_grpc(self.server_xds_host,
                                self.server_xds_port,
@@ -206,28 +282,22 @@ class RegularXdsKubernetesTestCase(XdsKubernetesTestCase):
 
     @classmethod
     def setUpClass(cls):
+        """Hook method for setting up class fixture before running tests in
+        the class.
+        """
         super().setUpClass()
         if cls.server_maintenance_port is None:
-            cls.server_maintenance_port = \
-                server_app.KubernetesServerRunner.DEFAULT_MAINTENANCE_PORT
+            cls.server_maintenance_port = KubernetesServerRunner.DEFAULT_MAINTENANCE_PORT
 
-    def setUp(self):
-        super().setUp()
-
-        # Traffic Director Configuration
-        self.td = traffic_director.TrafficDirectorManager(
-            self.gcp_api_manager,
-            project=self.project,
-            resource_prefix=self.namespace,
-            network=self.network)
-
-        # Ensures the firewall exist
-        if self.ensure_firewall:
-            self.td.create_firewall_rule(
-                allowed_ports=self.firewall_allowed_ports)
+    def initTrafficDirectorManager(self) -> TrafficDirectorManager:
+        return TrafficDirectorManager(self.gcp_api_manager,
+                                      project=self.project,
+                                      resource_prefix=self.resource_prefix,
+                                      resource_suffix=self.resource_suffix,
+                                      network=self.network)
 
-        # Test Server Runner
-        self.server_runner = server_app.KubernetesServerRunner(
+    def initKubernetesServerRunner(self) -> KubernetesServerRunner:
+        return KubernetesServerRunner(
             k8s.KubernetesNamespace(self.k8s_api_manager,
                                     self.server_namespace),
             deployment_name=self.server_name,
@@ -239,8 +309,8 @@ class RegularXdsKubernetesTestCase(XdsKubernetesTestCase):
             xds_server_uri=self.xds_server_uri,
             network=self.network)
 
-        # Test Client Runner
-        self.client_runner = client_app.KubernetesClientRunner(
+    def initKubernetesClientRunner(self) -> KubernetesClientRunner:
+        return KubernetesClientRunner(
             k8s.KubernetesNamespace(self.k8s_api_manager,
                                     self.client_namespace),
             deployment_name=self.client_name,
@@ -273,6 +343,7 @@ class RegularXdsKubernetesTestCase(XdsKubernetesTestCase):
 
 
 class SecurityXdsKubernetesTestCase(XdsKubernetesTestCase):
+    td: TrafficDirectorSecureManager
 
     class SecurityMode(enum.Enum):
         MTLS = enum.auto()
@@ -281,6 +352,9 @@ class SecurityXdsKubernetesTestCase(XdsKubernetesTestCase):
 
     @classmethod
     def setUpClass(cls):
+        """Hook method for setting up class fixture before running tests in
+        the class.
+        """
         super().setUpClass()
         if cls.server_maintenance_port is None:
             # In secure mode, the maintenance port is different from
@@ -288,25 +362,18 @@ class SecurityXdsKubernetesTestCase(XdsKubernetesTestCase):
             # Health Checks and Channelz tests available.
             # When not provided, use explicit numeric port value, so
             # Backend Health Checks are created on a fixed port.
-            cls.server_maintenance_port = _DEFAULT_SECURE_MODE_MAINTENANCE_PORT
-
-    def setUp(self):
-        super().setUp()
+            cls.server_maintenance_port = KubernetesServerRunner.DEFAULT_SECURE_MODE_MAINTENANCE_PORT
 
-        # Traffic Director Configuration
-        self.td = traffic_director.TrafficDirectorSecureManager(
+    def initTrafficDirectorManager(self) -> TrafficDirectorSecureManager:
+        return TrafficDirectorSecureManager(
             self.gcp_api_manager,
             project=self.project,
-            resource_prefix=self.namespace,
+            resource_prefix=self.resource_prefix,
+            resource_suffix=self.resource_suffix,
             network=self.network)
 
-        # Ensures the firewall exist
-        if self.ensure_firewall:
-            self.td.create_firewall_rule(
-                allowed_ports=self.firewall_allowed_ports)
-
-        # Test Server Runner
-        self.server_runner = server_app.KubernetesServerRunner(
+    def initKubernetesServerRunner(self) -> KubernetesServerRunner:
+        return KubernetesServerRunner(
             k8s.KubernetesNamespace(self.k8s_api_manager,
                                     self.server_namespace),
             deployment_name=self.server_name,
@@ -320,8 +387,8 @@ class SecurityXdsKubernetesTestCase(XdsKubernetesTestCase):
             deployment_template='server-secure.deployment.yaml',
             debug_use_port_forwarding=self.debug_use_port_forwarding)
 
-        # Test Client Runner
-        self.client_runner = client_app.KubernetesClientRunner(
+    def initKubernetesClientRunner(self) -> KubernetesClientRunner:
+        return KubernetesClientRunner(
             k8s.KubernetesNamespace(self.k8s_api_manager,
                                     self.client_namespace),
             deployment_name=self.client_name,

tools/run_tests/xds_k8s_test_driver/run.sh

@@ -42,7 +42,7 @@ This tool performs the following:
 EXAMPLES:
 $0 bin/run_td_setup.py --help      # list script-specific options
 $0 bin/run_td_setup.py --helpfull  # list all available options
-XDS_K8S_CONFIG=./path-to-flagfile.cfg $0 bin/run_td_setup.py --namespace=override-namespace
+XDS_K8S_CONFIG=./path-to-flagfile.cfg ./run.sh bin/run_td_setup.py --resource_suffix=override-suffix
 $0 tests/baseline_test.py
 $0 tests/security_test.py --verbosity=1 --logger_levels=__main__:DEBUG,framework:DEBUG
 $0 tests/security_test.py SecurityTest.test_mtls --nocheck_local_certs

tools/run_tests/xds_k8s_test_driver/tests/security_test.py

@@ -12,12 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-import uuid
 
 from absl import flags
 from absl.testing import absltest
 
 from framework import xds_k8s_testcase
+from framework.helpers import rand
 
 logger = logging.getLogger(__name__)
 flags.adopt_module_key_flags(xds_k8s_testcase)
@@ -161,7 +161,7 @@ class SecurityTest(xds_k8s_testcase.SecurityXdsKubernetesTestCase):
                                       server_port=self.server_port,
                                       tls=True,
                                       mtls=False)
-        incorrect_namespace = f'incorrect-namespace-{uuid.uuid4().hex}'
+        incorrect_namespace = f'incorrect-namespace-{rand.rand_string()}'
         self.td.setup_client_security(server_namespace=incorrect_namespace,
                                       server_name=self.server_name,
                                       tls=True,