Add Envoy service mesh example

docs/examples.md

@@ -17,3 +17,4 @@ Examples of different use cases provided by Testcontainers can be found below:
 - [Zookeeper](https://github.com/testcontainers/testcontainers-java/tree/main/examples/zookeeper)
 - [NATS](https://github.com/testcontainers/testcontainers-java/tree/main/examples/nats)
 - [SFTP](https://github.com/testcontainers/testcontainers-java/tree/main/examples/sftp)
+- [Envoy](https://github.com/testcontainers/testcontainers-java/tree/main/examples/envoy)

examples/envoy/build.gradle

@@ -0,0 +1,24 @@
+plugins {
+    id 'java'
+}
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    testImplementation 'org.assertj:assertj-core:3.27.4'
+    testImplementation 'org.testcontainers:testcontainers'
+    testImplementation 'org.testcontainers:testcontainers-junit-jupiter'
+    testImplementation 'io.envoyproxy.controlplane:server:1.0.55'
+    // keep in sync with the gRPC version used by java-control-plane
+    testImplementation 'io.grpc:grpc-netty-shaded:1.62.2'
+    testImplementation 'org.awaitility:awaitility:4.3.0'
+    testImplementation 'ch.qos.logback:logback-classic:1.3.15'
+    testImplementation 'org.junit.jupiter:junit-jupiter:5.13.4'
+    testRuntimeOnly 'org.junit.platform:junit-platform-launcher:1.11.0'
+}
+
+test {
+    useJUnitPlatform()
+}

examples/envoy/src/test/java/com/example/EnvoyContainerTest.java

@@ -0,0 +1,196 @@
+package com.example;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.Testcontainers;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.junit.jupiter.Container;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Shows how to set up a service mesh style environment with Testcontainers: every service gets an Envoy sidecar,
+ * service-to-service traffic goes through the sidecars, and their configuration is delivered over xDS by a control
+ * plane running in the test JVM (see {@link XdsControlPlane}).
+ * <p>
+ * The assertions only demonstrate that the example is wired correctly. They are not a test of Envoy itself.
+ *
+ * <pre>
+ * test (client app) -> client-sidecar :15001 -> server-sidecar :15006 -> server-app :8080
+ *                             ^                        ^
+ *                             +------ xDS (ADS) -------+
+ *                              control plane in test JVM
+ * </pre>
+ */
+@org.testcontainers.junit.jupiter.Testcontainers
+class EnvoyContainerTest {
+
+    private static final String ENVOY_IMAGE = "envoyproxy/envoy:v1.39.1";
+
+    private static final int OUTBOUND_PORT = 15001;
+
+    private static final int INBOUND_PORT = 15006;
+
+    private static final int ADMIN_PORT = 9901;
+
+    // client-sidecar adds this value as the x-service-caller header to every outbound request
+    private static final String CLIENT_CALLER = "client";
+
+    private static final XdsControlPlane controlPlane = startControlPlane();
+
+    private static final Network network = Network.newNetwork();
+
+    @Container
+    private static final GenericContainer<?> serverApp = new GenericContainer<>("mendhak/http-https-echo:41")
+        .withNetwork(network)
+        .withNetworkAliases("server-app")
+        .withEnv("HTTP_PORT", "8080")
+        .withExposedPorts(8080)
+        .waitingFor(Wait.forHttp("/").forPort(8080));
+
+    @Container
+    private static final GenericContainer<?> serverSidecar = sidecar("server-sidecar", INBOUND_PORT)
+        .dependsOn(serverApp);
+
+    @Container
+    private static final GenericContainer<?> clientSidecar = sidecar("client-sidecar", OUTBOUND_PORT)
+        .dependsOn(serverSidecar);
+
+    private final HttpClient httpClient = HttpClient.newHttpClient();
+
+    private static XdsControlPlane startControlPlane() {
+        XdsControlPlane controlPlane = new XdsControlPlane();
+        // lets the sidecar containers reach the control plane through host.testcontainers.internal
+        Testcontainers.exposeHostPorts(controlPlane.getPort());
+        controlPlane.setSnapshot(
+            "client-sidecar",
+            XdsControlPlane.listener("xds/client-sidecar-listener.json", Collections.emptyMap()),
+            XdsControlPlane.cluster("xds/client-sidecar-cluster.json")
+        );
+        allowCaller(controlPlane, CLIENT_CALLER);
+        return controlPlane;
+    }
+
+    private static void allowCaller(XdsControlPlane controlPlane, String caller) {
+        controlPlane.setSnapshot(
+            "server-sidecar",
+            XdsControlPlane.listener(
+                "xds/server-sidecar-listener.json",
+                Collections.singletonMap("ALLOWED_CALLER", caller)
+            ),
+            XdsControlPlane.cluster("xds/server-sidecar-cluster.json")
+        );
+    }
+
+    private static GenericContainer<?> sidecar(String nodeId, int trafficPort) {
+        String bootstrap = XdsControlPlane.render(
+            "bootstrap.yaml",
+            Collections.singletonMap("XDS_PORT", String.valueOf(controlPlane.getPort()))
+        );
+        return new GenericContainer<>(ENVOY_IMAGE)
+            .withNetwork(network)
+            .withNetworkAliases(nodeId)
+            .withCopyToContainer(Transferable.of(bootstrap), "/etc/envoy/bootstrap.yaml")
+            .withCommand("-c", "/etc/envoy/bootstrap.yaml", "--service-node", nodeId, "--service-cluster", nodeId)
+            .withExposedPorts(trafficPort, ADMIN_PORT)
+            // /ready returns 200 only after the listeners and clusters were received over xDS
+            .waitingFor(Wait.forHttp("/ready").forPort(ADMIN_PORT));
+    }
+
+    @AfterAll
+    static void stopControlPlane() {
+        controlPlane.close();
+    }
+
+    @BeforeEach
+    void resetCounters() throws Exception {
+        send(
+            HttpRequest.newBuilder(adminUri(clientSidecar, "/reset_counters")).POST(HttpRequest.BodyPublishers.noBody())
+        );
+    }
+
+    @Test
+    void sendsRequestThroughTheSidecars() throws Exception {
+        HttpResponse<String> response = callServer("/hello");
+
+        // the request reached server-app through client-sidecar and server-sidecar
+        assertThat(response.statusCode()).isEqualTo(200);
+        // client-sidecar marks outbound requests with the caller name, which server-sidecar authorizes
+        // the echo application returns the received request as pretty-printed JSON
+        String body = response.body().replaceAll("\\s", "");
+        assertThat(body).contains("\"path\":\"/hello\"").contains("\"x-service-caller\":\"client\"");
+    }
+
+    /**
+     * Configuration is not baked into the containers: the control plane can change it while the sidecars run,
+     * which is how a mesh rolls out a new policy.
+     */
+    @Test
+    void pushesConfigurationUpdateToARunningSidecar() throws Exception {
+        allowCaller(controlPlane, "another-client");
+        try {
+            await()
+                .atMost(Duration.ofSeconds(30))
+                .untilAsserted(() -> assertThat(callServer("/hello").statusCode()).isEqualTo(403));
+        } finally {
+            allowCaller(controlPlane, CLIENT_CALLER);
+            await()
+                .atMost(Duration.ofSeconds(30))
+                .untilAsserted(() -> assertThat(callServer("/hello").statusCode()).isEqualTo(200));
+        }
+    }
+
+    @Test
+    void configuresRetriesOnTheClientSidecar() throws Exception {
+        // the echo application responds with the status code requested in this header, so the upstream service
+        // can be made to fail on demand
+        HttpResponse<String> response = callServer("/hello", "x-set-response-status-code", "503");
+
+        assertThat(response.statusCode()).isEqualTo(503);
+        // 1 original request + 2 retries configured in the client-sidecar route
+        HttpResponse<String> stats = send(
+            HttpRequest.newBuilder(adminUri(clientSidecar, "/stats?filter=cluster.server.upstream_rq_retry$"))
+        );
+        assertThat(stats.body()).contains("cluster.server.upstream_rq_retry: 2");
+    }
+
+    @Test
+    void configuresTimeoutOnTheClientSidecar() throws Exception {
+        // the echo application delays its response by this header value, longer than the 1s route timeout
+        // configured for the client-sidecar
+        HttpResponse<String> response = callServer("/hello", "x-set-response-delay-ms", "3000");
+
+        assertThat(response.statusCode()).isEqualTo(504);
+    }
+
+    private HttpResponse<String> callServer(String path, String... headers) throws Exception {
+        URI uri = URI.create(
+            String.format("http://%s:%d%s", clientSidecar.getHost(), clientSidecar.getMappedPort(OUTBOUND_PORT), path)
+        );
+        HttpRequest.Builder request = HttpRequest.newBuilder(uri);
+        if (headers.length > 0) {
+            request.headers(headers);
+        }
+        return send(request);
+    }
+
+    private static URI adminUri(GenericContainer<?> sidecar, String path) {
+        return URI.create(String.format("http://%s:%d%s", sidecar.getHost(), sidecar.getMappedPort(ADMIN_PORT), path));
+    }
+
+    private HttpResponse<String> send(HttpRequest.Builder request) throws Exception {
+        return httpClient.send(request.build(), HttpResponse.BodyHandlers.ofString());
+    }
+}

examples/envoy/src/test/java/com/example/XdsControlPlane.java

@@ -0,0 +1,125 @@
+package com.example;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.google.protobuf.util.JsonFormat;
+import io.envoyproxy.controlplane.cache.v3.SimpleCache;
+import io.envoyproxy.controlplane.cache.v3.Snapshot;
+import io.envoyproxy.controlplane.server.V3DiscoveryServer;
+import io.envoyproxy.envoy.config.cluster.v3.Cluster;
+import io.envoyproxy.envoy.config.core.v3.Node;
+import io.envoyproxy.envoy.config.listener.v3.Listener;
+import io.envoyproxy.envoy.extensions.filters.http.rbac.v3.RBAC;
+import io.envoyproxy.envoy.extensions.filters.http.router.v3.Router;
+import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
+import io.grpc.Server;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A minimal xDS control plane based on envoyproxy/java-control-plane.
+ * <p>
+ * Each sidecar is identified by its Envoy node id and receives its own snapshot of listeners and clusters.
+ * Calling {@link #setSnapshot(String, Listener, Cluster)} again pushes the new configuration to the running sidecar.
+ */
+class XdsControlPlane implements AutoCloseable {
+
+    private static final JsonFormat.Parser JSON_PARSER = JsonFormat
+        .parser()
+        .usingTypeRegistry(
+            JsonFormat.TypeRegistry
+                .newBuilder()
+                .add(HttpConnectionManager.getDescriptor())
+                .add(Router.getDescriptor())
+                .add(RBAC.getDescriptor())
+                .build()
+        );
+
+    private final SimpleCache<String> cache = new SimpleCache<>(Node::getId);
+
+    private final AtomicLong version = new AtomicLong();
+
+    private final Server server;
+
+    XdsControlPlane() {
+        V3DiscoveryServer discoveryServer = new V3DiscoveryServer(cache);
+        server = NettyServerBuilder.forPort(0).addService(discoveryServer.getAggregatedDiscoveryServiceImpl()).build();
+        try {
+            server.start();
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
+    int getPort() {
+        return server.getPort();
+    }
+
+    void setSnapshot(String nodeId, Listener listener, Cluster cluster) {
+        cache.setSnapshot(
+            nodeId,
+            Snapshot.create(
+                Collections.singletonList(cluster),
+                Collections.emptyList(),
+                Collections.singletonList(listener),
+                Collections.emptyList(),
+                Collections.emptyList(),
+                String.valueOf(version.incrementAndGet())
+            )
+        );
+    }
+
+    static Listener listener(String resource, Map<String, String> variables) {
+        Listener.Builder builder = Listener.newBuilder();
+        mergeJson(render(resource, variables), builder);
+        return builder.build();
+    }
+
+    static Cluster cluster(String resource) {
+        Cluster.Builder builder = Cluster.newBuilder();
+        mergeJson(render(resource, Collections.emptyMap()), builder);
+        return builder.build();
+    }
+
+    /**
+     * Reads a classpath resource and replaces every {@code ${name}} placeholder with its value.
+     */
+    static String render(String resource, Map<String, String> variables) {
+        String content = readResource(resource);
+        for (Map.Entry<String, String> variable : variables.entrySet()) {
+            content = content.replace("${" + variable.getKey() + "}", variable.getValue());
+        }
+        return content;
+    }
+
+    private static void mergeJson(String json, Message.Builder builder) {
+        try {
+            JSON_PARSER.merge(json, builder);
+        } catch (InvalidProtocolBufferException e) {
+            throw new IllegalArgumentException("Invalid xDS resource: " + json, e);
+        }
+    }
+
+    private static String readResource(String resource) {
+        try (InputStream in = XdsControlPlane.class.getClassLoader().getResourceAsStream(resource)) {
+            if (in == null) {
+                throw new IllegalArgumentException("Resource not found: " + resource);
+            }
+            return new String(in.readAllBytes(), StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
+    @Override
+    public void close() {
+        server.shutdownNow();
+    }
+}

examples/envoy/src/test/resources/bootstrap.yaml

@@ -0,0 +1,41 @@
+# Static bootstrap shared by every sidecar.
+# Listeners and clusters are delivered dynamically by the xDS control plane running in the test JVM.
+admin:
+  address:
+    socket_address:
+      address: 0.0.0.0
+      port_value: 9901
+
+dynamic_resources:
+  ads_config:
+    api_type: GRPC
+    transport_api_version: V3
+    grpc_services:
+      - envoy_grpc:
+          cluster_name: xds_cluster
+  lds_config:
+    resource_api_version: V3
+    ads: {}
+  cds_config:
+    resource_api_version: V3
+    ads: {}
+
+static_resources:
+  clusters:
+    - name: xds_cluster
+      type: STRICT_DNS
+      connect_timeout: 1s
+      typed_extension_protocol_options:
+        envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+          "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+          explicit_http_config:
+            http2_protocol_options: {}
+      load_assignment:
+        cluster_name: xds_cluster
+        endpoints:
+          - lb_endpoints:
+              - endpoint:
+                  address:
+                    socket_address:
+                      address: host.testcontainers.internal
+                      port_value: ${XDS_PORT}