| # Copyright 2021 The 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. |
| """A test framework built for urlMap related xDS test cases.""" |
| |
| import functools |
| import inspect |
| from typing import Any, Iterable, Mapping, Tuple |
| |
| from absl import flags |
| from absl import logging |
| |
| from framework import xds_flags |
| from framework import xds_k8s_flags |
| import framework.helpers.rand |
| from framework.infrastructure import gcp |
| from framework.infrastructure import k8s |
| from framework.infrastructure import traffic_director |
| from framework.test_app import client_app |
| from framework.test_app import server_app |
| |
| flags.adopt_module_key_flags(xds_flags) |
| flags.adopt_module_key_flags(xds_k8s_flags) |
| |
| STRATEGY = flags.DEFINE_enum('strategy', |
| default='reuse', |
| enum_values=['create', 'keep', 'reuse'], |
| help='Strategy of GCP resources management') |
| |
| # Type alias |
| UrlMapType = Any |
| HostRule = Any |
| PathMatcher = Any |
| |
| |
| class _UrlMapChangeAggregator: |
| """Where all the urlMap change happens.""" |
| |
| def __init__(self, url_map_name: str): |
| self._map = { |
| "name": url_map_name, |
| "defaultService": GcpResourceManager().default_backend_service(), |
| "hostRules": [], |
| "pathMatchers": [], |
| } |
| |
| def get_map(self) -> UrlMapType: |
| return self._map |
| |
| def apply_change(self, test_case: 'XdsUrlMapTestCase') -> None: |
| logging.info('Apply urlMap change for test case: %s.%s', |
| test_case.short_module_name, test_case.__name__) |
| url_map_parts = test_case.url_map_change( |
| *self._get_test_case_url_map(test_case)) |
| self._set_test_case_url_map(*url_map_parts) |
| |
| @staticmethod |
| def _get_test_case_url_map( |
| test_case: 'XdsUrlMapTestCase') -> Tuple[HostRule, PathMatcher]: |
| host_rule = { |
| "hosts": [test_case.hostname()], |
| "pathMatcher": test_case.path_matcher_name(), |
| } |
| path_matcher = { |
| "name": test_case.path_matcher_name(), |
| "defaultService": GcpResourceManager().default_backend_service(), |
| } |
| return host_rule, path_matcher |
| |
| def _set_test_case_url_map(self, host_rule: HostRule, |
| path_matcher: PathMatcher) -> None: |
| self._map["hostRules"].append(host_rule) |
| self._map["pathMatchers"].append(path_matcher) |
| |
| |
| def _package_flags() -> Mapping[str, Any]: |
| """Automatically parse Abseil flags into a dictionary. |
| |
| Abseil flag is only available after the Abseil app initialization. If we use |
| __new__ in our metaclass, the flag value parse will happen during the |
| initialization of modules, hence will fail. That's why we are using __call__ |
| to inject metaclass magics, and the flag parsing will be delayed until the |
| class is about to be instantiated. |
| """ |
| res = {} |
| for flag_module in [xds_flags, xds_k8s_flags]: |
| for key, value in inspect.getmembers(flag_module): |
| if isinstance(value, flags.FlagHolder): |
| res[key.lower()] = value.value |
| res['strategy'] = STRATEGY.value |
| return res |
| |
| |
| class _MetaSingletonAndAbslFlags(type): |
| """Ensures singleton and injects flag values.""" |
| |
| # Allow different subclasses to create different singletons. |
| _instances = {} |
| # But we only parse Abseil flags once. |
| _flags = None |
| |
| def __call__(cls, *args, **kwargs): |
| if cls not in cls._instances: |
| if cls._flags is None: |
| cls._flags = _package_flags() |
| obj = super().__call__(cls._flags, *args, **kwargs) |
| cls._instances[cls] = obj |
| return obj |
| return cls._instances[cls] |
| |
| |
| class GcpResourceManager(metaclass=_MetaSingletonAndAbslFlags): |
| """Manages the lifecycle of GCP resources. |
| |
| The GCP resources including: |
| - 3 K8s deployment (client, default backends, alternative backends) |
| - Full set of the Traffic Director stuff |
| - Merged gigantic urlMap from all imported test cases |
| |
| All resources are intended to be used across test cases and multiple runs |
| (except the client K8s deployment). |
| """ |
| |
| # This class dynamically set, so disable "no-member" check. |
| # pylint: disable=no-member |
| |
| def __init__(self, absl_flags: Mapping[str, Any] = None): |
| if absl_flags is not None: |
| for key in absl_flags: |
| setattr(self, key, absl_flags[key]) |
| # Pick a client_namespace_suffix if not set |
| if getattr(self, 'resource_suffix', None) is None: |
| self.resource_suffix = "" |
| else: |
| raise NotImplementedError( |
| 'Predefined resource_suffix is not supported for UrlMap tests') |
| logging.info('GcpResourceManager: resource prefix=%s, suffix=%s', |
| self.resource_prefix, self.resource_suffix) |
| # API managers |
| self.k8s_api_manager = k8s.KubernetesApiManager(self.kube_context) |
| self.gcp_api_manager = gcp.api.GcpApiManager() |
| self.td = traffic_director.TrafficDirectorManager( |
| self.gcp_api_manager, |
| self.project, |
| resource_prefix=self.resource_prefix, |
| resource_suffix=(self.resource_suffix or ""), |
| network=self.network, |
| compute_api_version=self.compute_api_version, |
| ) |
| # Kubernetes namespace |
| self.k8s_namespace = k8s.KubernetesNamespace(self.k8s_api_manager, |
| self.resource_prefix) |
| # Kubernetes Test Servers |
| self.test_server_runner = server_app.KubernetesServerRunner( |
| self.k8s_namespace, |
| deployment_name=self.server_name, |
| image_name=self.server_image, |
| gcp_project=self.project, |
| gcp_api_manager=self.gcp_api_manager, |
| gcp_service_account=self.gcp_service_account, |
| td_bootstrap_image=self.td_bootstrap_image, |
| xds_server_uri=self.xds_server_uri, |
| network=self.network, |
| enable_workload_identity=self.enable_workload_identity) |
| self.test_server_alternative_runner = server_app.KubernetesServerRunner( |
| self.k8s_namespace, |
| deployment_name=self.server_name + '-alternative', |
| image_name=self.server_image, |
| gcp_project=self.project, |
| gcp_api_manager=self.gcp_api_manager, |
| gcp_service_account=self.gcp_service_account, |
| td_bootstrap_image=self.td_bootstrap_image, |
| xds_server_uri=self.xds_server_uri, |
| network=self.network, |
| enable_workload_identity=self.enable_workload_identity, |
| reuse_namespace=True) |
| self.test_server_affinity_runner = server_app.KubernetesServerRunner( |
| self.k8s_namespace, |
| deployment_name=self.server_name + '-affinity', |
| image_name=self.server_image, |
| gcp_project=self.project, |
| gcp_api_manager=self.gcp_api_manager, |
| gcp_service_account=self.gcp_service_account, |
| td_bootstrap_image=self.td_bootstrap_image, |
| xds_server_uri=self.xds_server_uri, |
| network=self.network, |
| enable_workload_identity=self.enable_workload_identity, |
| reuse_namespace=True) |
| logging.info('Strategy of GCP resources management: %s', self.strategy) |
| |
| def create_test_client_runner(self): |
| if self.resource_suffix: |
| client_namespace_suffix = self.resource_suffix |
| else: |
| client_namespace_suffix = framework.helpers.rand.random_resource_suffix( |
| ) |
| logging.info('GcpResourceManager: client_namespace_suffix=%s', |
| client_namespace_suffix) |
| # Kubernetes Test Client |
| return client_app.KubernetesClientRunner( |
| k8s.KubernetesNamespace( |
| self.k8s_api_manager, |
| client_app.KubernetesClientRunner.make_namespace_name( |
| self.resource_prefix, client_namespace_suffix)), |
| deployment_name=self.client_name, |
| image_name=self.client_image, |
| gcp_project=self.project, |
| gcp_api_manager=self.gcp_api_manager, |
| gcp_service_account=self.gcp_service_account, |
| td_bootstrap_image=self.td_bootstrap_image, |
| xds_server_uri=self.xds_server_uri, |
| network=self.network, |
| debug_use_port_forwarding=self.debug_use_port_forwarding, |
| enable_workload_identity=self.enable_workload_identity, |
| stats_port=self.client_port) |
| |
| def _pre_cleanup(self): |
| # Cleanup existing debris |
| logging.info('GcpResourceManager: pre clean-up') |
| self.td.cleanup(force=True) |
| self.test_server_runner.delete_namespace() |
| |
| def setup(self, test_case_classes: Iterable['XdsUrlMapTestCase']) -> None: |
| if self.strategy not in ['create', 'keep']: |
| logging.info('GcpResourceManager: skipping setup for strategy [%s]', |
| self.strategy) |
| return |
| # Clean up debris from previous runs |
| self._pre_cleanup() |
| # Start creating GCP resources |
| logging.info('GcpResourceManager: start setup') |
| # Firewall |
| if self.ensure_firewall: |
| self.td.create_firewall_rule( |
| allowed_ports=self.firewall_allowed_ports) |
| # Health Checks |
| self.td.create_health_check() |
| # Backend Services |
| self.td.create_backend_service() |
| self.td.create_alternative_backend_service() |
| self.td.create_affinity_backend_service() |
| # Construct UrlMap from test classes |
| aggregator = _UrlMapChangeAggregator( |
| url_map_name=self.td.make_resource_name(self.td.URL_MAP_NAME)) |
| for test_case_class in test_case_classes: |
| aggregator.apply_change(test_case_class) |
| final_url_map = aggregator.get_map() |
| # UrlMap |
| self.td.create_url_map_with_content(final_url_map) |
| # Target Proxy |
| self.td.create_target_proxy() |
| # Forwarding Rule |
| self.td.create_forwarding_rule(self.server_xds_port) |
| # Kubernetes Test Server |
| self.test_server_runner.run( |
| test_port=self.server_port, |
| maintenance_port=self.server_maintenance_port) |
| # Kubernetes Test Server Alternative |
| self.test_server_alternative_runner.run( |
| test_port=self.server_port, |
| maintenance_port=self.server_maintenance_port) |
| # Kubernetes Test Server Affinity. 3 endpoints to test that only the |
| # picked sub-channel is connected. |
| self.test_server_affinity_runner.run( |
| test_port=self.server_port, |
| maintenance_port=self.server_maintenance_port, |
| replica_count=3) |
| # Add backend to default backend service |
| neg_name, neg_zones = self.k8s_namespace.get_service_neg( |
| self.test_server_runner.service_name, self.server_port) |
| self.td.backend_service_add_neg_backends(neg_name, neg_zones) |
| # Add backend to alternative backend service |
| neg_name_alt, neg_zones_alt = self.k8s_namespace.get_service_neg( |
| self.test_server_alternative_runner.service_name, self.server_port) |
| self.td.alternative_backend_service_add_neg_backends( |
| neg_name_alt, neg_zones_alt) |
| # Add backend to affinity backend service |
| neg_name_affinity, neg_zones_affinity = self.k8s_namespace.get_service_neg( |
| self.test_server_affinity_runner.service_name, self.server_port) |
| self.td.affinity_backend_service_add_neg_backends( |
| neg_name_affinity, neg_zones_affinity) |
| # Wait for healthy backends |
| self.td.wait_for_backends_healthy_status() |
| self.td.wait_for_alternative_backends_healthy_status() |
| self.td.wait_for_affinity_backends_healthy_status() |
| |
| def cleanup(self) -> None: |
| if self.strategy not in ['create']: |
| logging.info( |
| 'GcpResourceManager: skipping tear down for strategy [%s]', |
| self.strategy) |
| return |
| logging.info('GcpResourceManager: start tear down') |
| if hasattr(self, 'td'): |
| self.td.cleanup(force=True) |
| if hasattr(self, 'test_server_runner'): |
| self.test_server_runner.cleanup(force=True) |
| if hasattr(self, 'test_server_alternative_runner'): |
| self.test_server_alternative_runner.cleanup(force=True, |
| force_namespace=True) |
| if hasattr(self, 'test_server_affinity_runner'): |
| self.test_server_affinity_runner.cleanup(force=True, |
| force_namespace=True) |
| |
| @functools.lru_cache(None) |
| def default_backend_service(self) -> str: |
| """Returns default backend service URL.""" |
| self.td.load_backend_service() |
| return self.td.backend_service.url |
| |
| @functools.lru_cache(None) |
| def alternative_backend_service(self) -> str: |
| """Returns alternative backend service URL.""" |
| self.td.load_alternative_backend_service() |
| return self.td.alternative_backend_service.url |
| |
| @functools.lru_cache(None) |
| def affinity_backend_service(self) -> str: |
| """Returns affinity backend service URL.""" |
| self.td.load_affinity_backend_service() |
| return self.td.affinity_backend_service.url |
