Merge remote-tracking branch 'grpc/main' into next-version

.kokoro/psm_interop_kokoro_lib.sh

@@ -392,7 +392,7 @@ psm::run::test() {
     PSM_TEST_FLAGS+=("--server_image=${SERVER_IMAGE_NAME}:${GIT_COMMIT}")
   elif [[ "${GRPC_LANGUAGE}" == "node"  ]]; then
     # TODO(b/261911148): To be replaced with --server_image_use_canonical when implemented.
-    PSM_TEST_FLAGS+=("--server_image=us-docker.pkg.dev/grpc-testing/psm-interop/java-server:canonical-v1.65")
+    PSM_TEST_FLAGS+=("--server_image=us-docker.pkg.dev/grpc-testing/psm-interop/java-server:canonical-v1.66")
   fi
 
   # So far, only LB test uses secondary GKE cluster.

config/common.cfg

@@ -5,7 +5,7 @@
 # Can be used in tests where language-specific xDS test server does not exist,
 # or missing a feature required for the test.
 # TODO(sergiitk): Update every ~ 6 months; next 2025-01.
---server_image_canonical=us-docker.pkg.dev/grpc-testing/psm-interop/java-server:canonical-v1.65
+--server_image_canonical=us-docker.pkg.dev/grpc-testing/psm-interop/java-server:canonical-v1.66
 
 --logger_levels=__main__:DEBUG,framework:INFO
 --verbosity=0

config/url-map.cfg

@@ -7,7 +7,7 @@
 # 2. All UrlMap tests today are testing client-side logic.
 #
 # TODO(sergiitk): Use --server_image_canonical instead.
---server_image=us-docker.pkg.dev/grpc-testing/psm-interop/java-server:canonical-v1.65
+--server_image=us-docker.pkg.dev/grpc-testing/psm-interop/java-server:canonical-v1.66
 
 # Disables the GCP Workload Identity feature to simplify permission control
 --gcp_service_account=None

framework/helpers/docker.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.
+
 from collections import defaultdict
 import datetime
 import logging
@@ -19,12 +20,12 @@
 import queue
 import threading
 
-from docker.client import DockerClient
-from docker.errors import NotFound
-from docker.types import ContainerConfig
 import grpc
 import mako.template
 
+from docker import client
+from docker import errors
+from docker import types
 import framework.helpers.xds_resources
 from protos.grpc.testing import messages_pb2
 from protos.grpc.testing import test_pb2_grpc
@@ -86,7 +87,7 @@ def __init__(
         node_id: str,
         verbosity="info",
     ):
-        self.docker_client = DockerClient.from_env()
+        self.docker_client = client.DockerClient.from_env()
         self.node_id = node_id
         self.outputs = defaultdict(list)
         self.queue = queue.Queue()
@@ -95,7 +96,9 @@ def __init__(
 
     def next_event(self, timeout: int) -> ChildProcessEvent:
         event: ChildProcessEvent = self.queue.get(timeout=timeout)
-        self.outputs[event.source].append(event.data)
+        source = event.source
+        message = event.data
+        self.outputs[source].append(message)
         return event
 
     def expect_output(
@@ -135,8 +138,8 @@ def expect_output(
                 return True
         return False
 
-    def on_message(self, source: str, msg: str):
-        self.queue.put(ChildProcessEvent(source, msg))
+    def on_message(self, source: str, message: str):
+        self.queue.put(ChildProcessEvent(source, message))
 
 
 def _Sanitize(l: str) -> str:
@@ -166,7 +169,7 @@ def __init__(
         image: str,
         name: str,
         manager: ProcessManager,
-        **config: ContainerConfig,
+        **config: types.ContainerConfig,
     ):
         self.name = name
         self.config = Configure(
@@ -192,7 +195,7 @@ def exit(self):
         try:
             self.container.stop()
             self.container.wait()
-        except NotFound:
+        except errors.NotFound:
             # It is ok, container was auto removed
             logger.debug(
                 "Container %s was autoremoved, most likely because the app crashed",
@@ -212,9 +215,9 @@ def log_reader_loop(self):
             s = str(prefix + log.decode("utf-8"))
             prefix = "" if s[-1] == "\n" else s[s.rfind("\n") :]
             for l in s[: s.rfind("\n")].splitlines():
-                sanitized = _Sanitize(l)
-                logger.info("[%s] %s", self.name, sanitized)
-                self.manager.on_message(self.name, sanitized)
+                message = _Sanitize(l)
+                logger.info("[%s] %s", self.name, message)
+                self.manager.on_message(self.name, message)
 
 
 class GrpcProcess:

tests/fallback_test.py

@@ -118,7 +118,9 @@ def test_fallback_on_startup(self):
             self.assertEqual(client.get_stats(5).num_failures, 5)
             # Fallback control plane start, send traffic to server2
             with self.start_control_plane(
-                "fallback_xds_config", 1, server2.port
+                name="fallback_xds_config",
+                index=1,
+                upstream_port=server2.port,
             ):
                 stats = client.get_stats(5)
                 self.assertGreater(stats.rpcs_by_peer["server2"], 0)
@@ -138,11 +140,11 @@ def test_fallback_on_startup(self):
                     self.assertEqual(stats.num_failures, 0)
                     self.assertIn("server1", stats.rpcs_by_peer)
                     self.assertGreater(stats.rpcs_by_peer["server1"], 0)
-                # Primary control plane down
+                # Primary control plane down, cached value is used
                 stats = client.get_stats(5)
                 self.assertEqual(stats.num_failures, 0)
                 self.assertEqual(stats.rpcs_by_peer["server1"], 5)
-            # Fallback control plane down
+            # Fallback control plane down, cached value is used
             stats = client.get_stats(5)
             self.assertEqual(stats.num_failures, 0)
             self.assertEqual(stats.rpcs_by_peer["server1"], 5)