Ring hash fix for affinity test (#211)

Start a temporary client to verify all endpoints are reported by TD.
This is to address flakiness where the main client starts with a partial
list of endpoints. Causing issue with RING_HASH no of sub channel
validations

Bug b/374693206

Author: pawbhard

framework/xds_k8s_testcase.py

@@ -1116,6 +1116,9 @@ def initKubernetesServerRunner(self, **kwargs) -> KubernetesServerRunner:
         )
 
     def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
+        reuse_namespace = kwargs.pop("reuse_namespace", False) or (
+            self.server_namespace == self.client_namespace
+        )
         return KubernetesClientRunner(
             k8s.KubernetesNamespace(
                 self.k8s_api_manager, self.client_namespace
@@ -1131,7 +1134,7 @@ def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
             debug_use_port_forwarding=self.debug_use_port_forwarding,
             enable_workload_identity=self.enable_workload_identity,
             stats_port=self.client_port,
-            reuse_namespace=self.server_namespace == self.client_namespace,
+            reuse_namespace=reuse_namespace,
             **kwargs,
         )
 

tests/affinity_test.py

@@ -17,12 +17,14 @@
 
 from absl import flags
 from absl.testing import absltest
+from typing_extensions import override
 
 from framework import xds_k8s_flags
 from framework import xds_k8s_testcase
 from framework.helpers import skips
 from framework.rpc import grpc_channelz
 from framework.rpc import grpc_testing
+from framework.test_app.runners.k8s import k8s_xds_client_runner
 
 logger = logging.getLogger(__name__)
 flags.adopt_module_key_flags(xds_k8s_testcase)
@@ -31,6 +33,7 @@
 _XdsTestServer = xds_k8s_testcase.XdsTestServer
 _XdsTestClient = xds_k8s_testcase.XdsTestClient
 _ChannelzChannelState = grpc_channelz.ChannelState
+KubernetesClientRunner = k8s_xds_client_runner.KubernetesClientRunner
 _Lang = skips.Lang
 
 # Testing consts
@@ -61,6 +64,10 @@ def is_supported(config: skips.TestConfig) -> bool:
             return config.version_gte("v1.10.x")
         return True
 
+    @override
+    def initKubernetesClientRunner(self, **kwargs) -> KubernetesClientRunner:
+        return super().initKubernetesClientRunner(reuse_namespace=True)
+
     def test_affinity(self) -> None:  # pylint: disable=too-many-statements
         with self.subTest("00_create_health_check"):
             self.td.create_health_check()
@@ -86,6 +93,14 @@ def test_affinity(self) -> None:  # pylint: disable=too-many-statements
         with self.subTest("06_add_server_backends_to_backend_services"):
             self.setupServerBackends()
 
+        with self.subTest("06a_wait_for_endpoints_propagation"):
+            # Start a temporary client to verify all endpoints are reported by TD.
+            # This is to address flakiness where the main client starts with
+            # a partial list of endpoints.
+            temp_client = self.startTestClient(test_servers[0])
+            self.assertHealthyEndpointsCount(temp_client, _REPLICA_COUNT)
+            self.client_runner.cleanup(force_namespace=False)
+
         test_client: _XdsTestClient
         with self.subTest("07_start_test_client"):
             test_client = self.startTestClient(
@@ -174,7 +189,7 @@ def test_affinity(self) -> None:  # pylint: disable=too-many-statements
             finally:
                 logging.info("Client received CSDS response: %s", parsed)
 
-        with self.subTest("12_next_100_affinity_rpcs_pick_different_backend"):
+        with self.subTest("13_next_100_affinity_rpcs_pick_different_backend"):
             rpc_stats = self.getClientRpcStats(test_client, _RPC_COUNT)
             rpc_distribution = grpc_testing.RpcDistributionStats.from_message(
                 rpc_stats