[Fallback] Initial version of the interop test.

.github/workflows/psm-interop.yaml

@@ -53,6 +53,8 @@ jobs:
           protos/grpc/testing/empty.proto
           protos/grpc/testing/messages.proto
           protos/grpc/testing/test.proto
+          protos/grpc/testing/xdsconfig/control.proto
+          protos/grpc/testing/xdsconfig/service.proto
 
       - name: "Run unit tests"
         run: python -m tests.unit

README.md

@@ -210,7 +210,7 @@ pip install -r requirements.lock
 # Generate protos
 python -m grpc_tools.protoc --proto_path=. \
   --python_out=. --grpc_python_out=. --pyi_out=. \
-  protos/grpc/testing/*.proto
+  protos/grpc/testing/*.proto protos/grpc/testing/xdsconfig/*.proto 
 ```
 
 # Basic usage

framework/helpers/docker.py

@@ -0,0 +1,301 @@
+import datetime
+import logging
+import math
+import pathlib
+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 protos.grpc.testing import messages_pb2
+from protos.grpc.testing import test_pb2_grpc
+from protos.grpc.testing.xdsconfig.control_pb2 import StopOnRequestRequest
+from protos.grpc.testing.xdsconfig.control_pb2 import UpsertResourcesRequest
+from protos.grpc.testing.xdsconfig.service_pb2_grpc import (
+    XdsConfigControlServiceStub,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class Bootstrap:
+    def __init__(self, base: pathlib.Path, ports: list[int], host_name: str):
+        self.base = base
+        self.host_name = host_name
+        self.ports = ports
+        self.mount_dir: pathlib.Path = None
+        self.make_working_dir(self.base)
+        # Use Mako
+        template = mako.template.Template(filename="templates/bootstrap.json")
+        file = template.render(
+            servers=[f"{self.host_name}:{port}" for port in self.ports]
+        )
+        destination = self.get_mount_dir() / "bootstrap.json"
+        with open(destination, "w", encoding="utf-8") as f:
+            f.write(file)
+            logger.debug("Generated bootstrap file at %s", destination)
+
+    def make_working_dir(self, base: str):
+        # Date time to string
+        run_id = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+        self.mount_dir = base / f"testrun_{run_id}"
+        if not self.mount_dir.exists():
+            logger.debug("Creating %s", self.mount_dir)
+            self.get_mount_dir().mkdir(parents=True)
+
+    def get_mount_dir(self):
+        if self.mount_dir is None:
+            raise RuntimeError("Working dir was not created yet")
+        return self.mount_dir.absolute()
+
+    def xds_config_server_port(self, n: int):
+        return self.ports[n]
+
+
+class ChildProcessEvent:
+    def __init__(self, source: str, data: str):
+        self.source = source
+        self.data = data
+
+    def __str__(self) -> str:
+        return f"{self.data}, {self.source}"
+
+    def __repr__(self) -> str:
+        return f"data={self.data}, source={self.source}"
+
+
+class ProcessManager:
+    def __init__(
+        self,
+        bootstrap: Bootstrap,
+        node_id: str,
+        verbosity="info",
+    ):
+        self.docker_client = DockerClient.from_env()
+        self.node_id = node_id
+        self.outputs = {}
+        self.queue = queue.Queue()
+        self.bootstrap = bootstrap
+        self.verbosity = verbosity
+
+    def next_event(self, timeout: int) -> ChildProcessEvent:
+        event: ChildProcessEvent = self.queue.get(timeout=timeout)
+        source = event.source
+        message = event.data
+        logger.info("[%s] %s", source, message)
+        if not source in self.outputs:
+            self.outputs[source] = []
+        self.outputs[source].append(message)
+        return event
+
+    def expect_output(self, source: str, message: str, timeout_s=5) -> bool:
+        logger.debug('Waiting for message "%s" on %s', message, source)
+        if source in self.outputs:
+            for m in self.outputs[source]:
+                if m.find(message) >= 0:
+                    return True
+        deadline = datetime.datetime.now() + datetime.timedelta(
+            seconds=timeout_s
+        )
+        while datetime.datetime.now() <= deadline:
+            event = self.next_event(timeout_s)
+            if event.source == source and event.data.find(message) >= 0:
+                return True
+        return False
+
+    def on_message(self, source: str, message: str):
+        self.queue.put(ChildProcessEvent(source, message))
+
+    def get_mount_dir(self):
+        return self.bootstrap.get_mount_dir()
+
+
+def _Sanitize(l: str) -> str:
+    if l.find("\0") < 0:
+        return l
+    return l.replace("\0", "�")
+
+
+def Configure(config, image: str, name: str, verbosity: str):
+    config["detach"] = True
+    config["environment"] = {
+        "GRPC_EXPERIMENTAL_XDS_FALLBACK": "true",
+        "GRPC_TRACE": "xds_client",
+        "GRPC_VERBOSITY": verbosity,
+        "GRPC_XDS_BOOTSTRAP": "/grpc/bootstrap.json",
+    }
+    config["extra_hosts"] = {"host.docker.internal": "host-gateway"}
+    config["image"] = image
+    config["hostname"] = name
+    config["remove"] = True
+    return config
+
+
+class DockerProcess:
+    def __init__(
+        self,
+        image: str,
+        name: str,
+        manager: ProcessManager,
+        **config: ContainerConfig,
+    ):
+        self.name = name
+        self.config = Configure(
+            config, image=image, name=name, verbosity=manager.verbosity
+        )
+        self.container = None
+        self.manager = manager
+        self.thread = None
+
+    def __enter__(self):
+        self.container = self.manager.docker_client.containers.run(
+            **self.config
+        )
+        self.thread = threading.Thread(
+            target=lambda process: process.log_reader_loop(),
+            args=(self,),
+        )
+        self.thread.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        try:
+            self.container.stop()
+            self.container.wait()
+        except NotFound:
+            # Ok, container was auto removed
+            pass
+        finally:
+            self.thread.join()
+
+    def log_reader_loop(self):
+        prefix = ""
+        for log in self.container.logs(stream=True):
+            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():
+                self.manager.on_message(self.name, _Sanitize(l))
+
+
+class GrpcProcess:
+    def __init__(
+        self,
+        manager: ProcessManager,
+        name: str,
+        port: int,
+        ports,
+        image: str,
+        command: list[str],
+        volumes=None,
+    ):
+        self.process = DockerProcess(
+            image,
+            name,
+            manager,
+            command=" ".join(command),
+            hostname=name,
+            ports=ports,
+            volumes=volumes if volumes is not None else {},
+        )
+        self.manager = manager
+        self.port = port
+        self.grpc_channel: grpc.Channel = None
+
+    def __enter__(self):
+        self.process.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.grpc_channel is not None:
+            self.grpc_channel.close()
+        self.process.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)
+
+    def expect_output(self, message: str, timeout_s=5) -> bool:
+        return self.manager.expect_output(self.process.name, message, timeout_s)
+
+    def channel(self) -> grpc.Channel:
+        if self.grpc_channel is None:
+            self.grpc_channel = grpc.insecure_channel(f"localhost:{self.port}")
+        return self.grpc_channel
+
+    def get_port(self):
+        return self.port
+
+
+class ControlPlane(GrpcProcess):
+    def __init__(
+        self,
+        manager: ProcessManager,
+        name: str,
+        port: int,
+        upstream: str,
+        image: str,
+    ):
+        super().__init__(
+            manager=manager,
+            name=name,
+            port=port,
+            image=image,
+            ports={3333: port},
+            command=["--upstream", str(upstream), "--node", manager.node_id],
+        )
+
+    def stop_on_resource_request(self, resource_type: str, resource_name: str):
+        stub = XdsConfigControlServiceStub(self.channel())
+        res = stub.StopOnRequest(
+            StopOnRequestRequest(
+                resource_type=resource_type, resource_name=resource_name
+            )
+        )
+        return res
+
+    def update_resources(
+        self, cluster: str, upstream_port: int, upstream_host="localhost"
+    ):
+        stub = XdsConfigControlServiceStub(self.channel())
+        return stub.UpsertResources(
+            UpsertResourcesRequest(
+                cluster=cluster,
+                upstream_host=upstream_host,
+                upstream_port=upstream_port,
+            )
+        )
+
+
+class Client(GrpcProcess):
+    def __init__(
+        self,
+        manager: ProcessManager,
+        port: int,
+        name: str,
+        url: str,
+        image: str,
+    ):
+        super().__init__(
+            manager=manager,
+            port=port,
+            image=image,
+            name=name,
+            command=[f"--server={url}", "--print_response"],
+            ports={50052: port},
+            volumes={
+                manager.get_mount_dir(): {
+                    "bind": "/grpc",
+                    "mode": "ro",
+                }
+            },
+        )
+
+    def get_stats(self, num_rpcs: int):
+        logger.debug("Sending %d requests", num_rpcs)
+        stub = test_pb2_grpc.LoadBalancerStatsServiceStub(self.channel())
+        res = stub.GetClientStats(
+            messages_pb2.LoadBalancerStatsRequest(
+                num_rpcs=num_rpcs, timeout_sec=math.ceil(num_rpcs * 1.5)
+            )
+        )
+        return res

protos/grpc/testing/xdsconfig/control.proto

@@ -0,0 +1,47 @@
+// Copyright 2024 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.
+
+syntax = "proto3";
+
+package grpc.testing.xdsconfig;
+
+option go_package = "grpc/interop/grpc_testing/xdsconfig";
+
+import "google/protobuf/any.proto";
+
+
+message StopOnRequestRequest {
+  string resource_type = 1;
+  string resource_name = 2;
+};
+
+message StopOnRequestResponse {
+  message ResourceFilter {
+    string resource_type = 1;
+    string resource_name = 2;
+  };
+
+  repeated ResourceFilter filters = 1;
+};
+
+message UpsertResourcesRequest {
+  optional string listener = 1;
+  string cluster = 2;
+  string upstream_host = 3;
+  uint32 upstream_port = 4;
+};
+
+message UpsertResourcesResponse {
+  repeated google.protobuf.Any resource = 1;
+};

protos/grpc/testing/xdsconfig/service.proto

@@ -0,0 +1,26 @@
+// Copyright 2024 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.
+
+syntax = "proto3";
+
+package grpc.testing.xdsconfig;
+
+option go_package = "grpc/interop/grpc_testing/xdsconfig";
+
+import "protos/grpc/testing/xdsconfig/control.proto";
+
+service XdsConfigControlService {
+  rpc StopOnRequest(StopOnRequestRequest) returns (StopOnRequestResponse);
+  rpc UpsertResources(UpsertResourcesRequest) returns (UpsertResourcesResponse);
+};