diff --git a/sdk-tests/pom.xml b/sdk-tests/pom.xml index 75cd6d2c9d..333b744eb1 100644 --- a/sdk-tests/pom.xml +++ b/sdk-tests/pom.xml @@ -270,6 +270,8 @@ + + diff --git a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeIT.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/MethodInvokeIT.java similarity index 72% rename from sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeIT.java rename to sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/MethodInvokeIT.java index f1be9f354c..7a58c35099 100644 --- a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/MethodInvokeIT.java @@ -11,22 +11,29 @@ limitations under the License. */ -package io.dapr.it.methodinvoke.grpc; +package io.dapr.it.testcontainers.methodinvoke.grpc; import io.dapr.client.DaprClient; import io.dapr.client.resiliency.ResiliencyOptions; -import io.dapr.it.AppRun; import io.dapr.it.MethodInvokeServiceGrpc; -import io.dapr.it.containers.BaseContainerIT; -import io.dapr.testcontainers.DaprContainer; +import io.dapr.it.testcontainers.methodinvoke.grpc.app.MethodInvokeService; +import io.dapr.it.testcontainers.support.DaprTestEnvironment; import io.dapr.testcontainers.DaprProtocol; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; import io.grpc.Status; import io.grpc.StatusRuntimeException; -import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; import java.time.Duration; import java.util.Map; +import java.util.concurrent.TimeUnit; import static io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequest; import static io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest; @@ -36,7 +43,10 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -public class MethodInvokeIT extends BaseContainerIT { +@Testcontainers +@Tag("testcontainers") +@Execution(ExecutionMode.SAME_THREAD) +public class MethodInvokeIT { private static final String APP_NAME = "methodinvoke-grpc-it"; private static final int NUM_MESSAGES = 10; @@ -44,29 +54,32 @@ public class MethodInvokeIT extends BaseContainerIT { private static final ResiliencyOptions RESILIENCY_OPTIONS = new ResiliencyOptions() .setTimeout(Duration.ofMillis(TIMEOUT_MS)); - private static DaprContainer dapr; - private static AppRun app; - - @BeforeAll - public static void init() throws Exception { - var pair = startAppAndAttach( - APP_NAME, - MethodInvokeService.class, - AppRun.AppProtocol.GRPC, - appPort -> { - DaprContainer d = daprBuilder(APP_NAME) - .withAppPort(appPort) - .withAppChannelAddress("host.testcontainers.internal") - .withAppProtocol(DaprProtocol.GRPC); - return d; - }); - dapr = pair.dapr(); - app = pair.app(); + @Container + private static final DaprTestEnvironment ENVIRONMENT = + new DaprTestEnvironment(APP_NAME, MethodInvokeService.class, DaprProtocol.GRPC); + + @BeforeEach + public void reset() throws InterruptedException { + ManagedChannel channel = ManagedChannelBuilder.forTarget(ENVIRONMENT.appAddress()).usePlaintext().build(); + + try { + var stub = MethodInvokeServiceGrpc.newBlockingStub(channel).withDeadlineAfter(10, TimeUnit.SECONDS); + var request = GetMessagesRequest.getDefaultInstance(); + + for (int id : stub.getMessages(request).getMessagesMap().keySet()) { + stub.deleteMessage(DeleteMessageRequest.newBuilder().setId(id).build()); + } + + assertTrue(stub.getMessages(request).getMessagesMap().isEmpty()); + } finally { + channel.shutdownNow(); + assertTrue(channel.awaitTermination(5, TimeUnit.SECONDS), "reset channel did not terminate"); + } } @Test public void testInvoke() throws Exception { - try (DaprClient client = newDaprClient(dapr)) { + try (DaprClient client = ENVIRONMENT.clientBuilder().build()) { client.waitForSidecar(10000).block(); MethodInvokeServiceGrpc.MethodInvokeServiceBlockingStub stub = createGrpcStub(client); @@ -97,7 +110,7 @@ public void testInvoke() throws Exception { @Test public void testInvokeTimeout() throws Exception { - try (DaprClient client = newDaprClientBuilder(dapr).withResiliencyOptions(RESILIENCY_OPTIONS).build()) { + try (DaprClient client = ENVIRONMENT.clientBuilder().withResiliencyOptions(RESILIENCY_OPTIONS).build()) { client.waitForSidecar(10000).block(); MethodInvokeServiceGrpc.MethodInvokeServiceBlockingStub stub = createGrpcStub(client); @@ -114,7 +127,7 @@ public void testInvokeTimeout() throws Exception { @Test public void testInvokeException() throws Exception { - try (DaprClient client = newDaprClient(dapr)) { + try (DaprClient client = ENVIRONMENT.clientBuilder().build()) { client.waitForSidecar(10000).block(); MethodInvokeServiceGrpc.MethodInvokeServiceBlockingStub stub = createGrpcStub(client); diff --git a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/app/MethodInvokeService.java similarity index 90% rename from sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java rename to sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/app/MethodInvokeService.java index 90d3739c82..0740f57a10 100644 --- a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/app/MethodInvokeService.java @@ -11,7 +11,7 @@ limitations under the License. */ -package io.dapr.it.methodinvoke.grpc; +package io.dapr.it.testcontainers.methodinvoke.grpc.app; import io.dapr.grpc.GrpcHealthCheckService; import io.dapr.it.MethodInvokeServiceGrpc; @@ -83,7 +83,7 @@ private void awaitTermination() throws InterruptedException { * {@inheritDoc} */ public void postMessage(io.dapr.it.MethodInvokeServiceProtos.PostMessageRequest request, - io.grpc.stub.StreamObserver responseObserver) { + io.grpc.stub.StreamObserver responseObserver) { this.messages.put(request.getId(), request.getMessage()); io.dapr.it.MethodInvokeServiceProtos.PostMessageResponse.Builder responseBuilder = @@ -96,7 +96,7 @@ public void postMessage(io.dapr.it.MethodInvokeServiceProtos.PostMessageRequest * {@inheritDoc} */ public void deleteMessage(io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequest request, - io.grpc.stub.StreamObserver responseObserver) { + io.grpc.stub.StreamObserver responseObserver) { this.messages.remove(request.getId()); io.dapr.it.MethodInvokeServiceProtos.DeleteMessageResponse.Builder responseBuilder = @@ -109,7 +109,7 @@ public void deleteMessage(io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequ * {@inheritDoc} */ public void getMessages(io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest request, - io.grpc.stub.StreamObserver responseObserver) { + io.grpc.stub.StreamObserver responseObserver) { GetMessagesResponse res = GetMessagesResponse.newBuilder().putAllMessages(this.messages).build(); io.dapr.it.MethodInvokeServiceProtos.GetMessagesResponse.Builder responseBuilder @@ -122,7 +122,7 @@ public void getMessages(io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest * {@inheritDoc} */ public void sleep(io.dapr.it.MethodInvokeServiceProtos.SleepRequest request, - io.grpc.stub.StreamObserver responseObserver) { + io.grpc.stub.StreamObserver responseObserver) { SleepResponse res = this.sleep(request); io.dapr.it.MethodInvokeServiceProtos.SleepResponse.Builder responseBuilder = diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/MethodInvokeIT.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/MethodInvokeIT.java index 3920713533..9e03938c0f 100644 --- a/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/MethodInvokeIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/MethodInvokeIT.java @@ -65,6 +65,7 @@ public void reset() throws Exception { public void testInvoke() throws Exception { try (DaprClient client = ENVIRONMENT.clientBuilder().build()) { client.waitForSidecar(10000).block(); + for (int i = 0; i < NUM_MESSAGES; i++) { String message = String.format("This is message #%d", i); client.invokeMethod(APP_NAME, "messages", message.getBytes(), HttpExtension.POST).block(); @@ -90,6 +91,7 @@ public void testInvoke() throws Exception { public void testInvokeWithObjects() throws Exception { try (DaprClient client = ENVIRONMENT.clientBuilder().build()) { client.waitForSidecar(10000).block(); + for (int i = 0; i < NUM_MESSAGES; i++) { Person person = new Person(); person.setName(String.format("Name %d", i)); diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/app/MethodInvokeController.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/app/MethodInvokeController.java index 29b123d417..28d8d872c8 100644 --- a/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/app/MethodInvokeController.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/app/MethodInvokeController.java @@ -80,6 +80,7 @@ public void postPerson(@RequestBody Person person){ @PutMapping(path = "/persons/{personId}") public void putPerson(@PathVariable("personId") Integer personId, @RequestBody Person person){ final Optional auxPerson = persons.stream().filter(person1 -> person1.getId() == personId).findFirst(); + if(auxPerson.isPresent()){ auxPerson.get().setName(person.getName()); auxPerson.get().setLastName(person.getLastName()); @@ -90,6 +91,7 @@ public void putPerson(@PathVariable("personId") Integer personId, @RequestBody P @DeleteMapping(path = "/persons/{personId}") public void deletePerson(@PathVariable("personId") Integer personId){ final Optional auxPerson = persons.stream().filter(person1 -> person1.getId() == personId).findFirst(); + if(auxPerson.isPresent()) { persons.remove(auxPerson.get()); } @@ -105,6 +107,7 @@ public void sleep(@RequestBody int seconds) throws InterruptedException { if (seconds < 0) { throw new IllegalArgumentException("Sleep time cannot be negative."); } + Thread.sleep(seconds * 1000); } diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironment.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironment.java index 6f589d6932..e0b1057307 100644 --- a/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironment.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironment.java @@ -17,11 +17,13 @@ import io.dapr.testcontainers.DaprContainer; import io.dapr.testcontainers.DaprContainerConstants; import io.dapr.testcontainers.DaprPlacementContainer; +import io.dapr.testcontainers.DaprProtocol; import io.dapr.testcontainers.DaprSchedulerContainer; import org.testcontainers.containers.BindMode; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.Network; import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.containers.wait.strategy.WaitStrategy; import org.testcontainers.lifecycle.Startable; import reactor.util.retry.Retry; @@ -43,6 +45,7 @@ public final class DaprTestEnvironment implements Startable { private static final String APP_RUNTIME = "/it-runtime"; private final String identity; private final String mainClass; + private final DaprProtocol protocol; private final Network network; private final OwnedResources resources; private final DaprPlacementContainer placement; @@ -53,12 +56,31 @@ public final class DaprTestEnvironment implements Startable { /** Defines an isolated topology; no container is started during construction. */ public DaprTestEnvironment(String appId, Class mainClass) { - this(appId, mainClass.getName(), Duration.ofMinutes(2)); + this(appId, mainClass, DaprProtocol.HTTP); + } + + /** + * Defines an isolated topology with the specified application callback protocol. + * @param appId Dapr application ID. + * @param mainClass Application entry point. + * @param protocol Application callback protocol. + */ + public DaprTestEnvironment(String appId, Class mainClass, DaprProtocol protocol) { + this(appId, mainClass.getName(), protocol, Duration.ofMinutes(2)); } DaprTestEnvironment(String appId, String mainClass, Duration appReadinessTimeout) { + this(appId, mainClass, DaprProtocol.HTTP, appReadinessTimeout); + } + + DaprTestEnvironment(String appId, String mainClass, DaprProtocol protocol, Duration appReadinessTimeout) { + if (protocol != DaprProtocol.HTTP && protocol != DaprProtocol.GRPC) { + throw new IllegalArgumentException("Unsupported application protocol: " + protocol); + } + identity = appId + "-" + UUID.randomUUID(); this.mainClass = mainClass; + this.protocol = protocol; Map labels = Map.of("io.dapr.it.environment", identity); network = Network.builder().createNetworkCmdModifier(cmd -> cmd.withLabels(labels)).build(); resources = new OwnedResources(network); @@ -68,11 +90,13 @@ public DaprTestEnvironment(String appId, Class mainClass) { scheduler = new DaprSchedulerContainer(DaprContainerConstants.DAPR_SCHEDULER_IMAGE_TAG) .withNetwork(network).withNetworkAliases("scheduler").withLabels(labels) .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2))); + WaitStrategy appWait = protocol == DaprProtocol.GRPC + ? new GrpcCallbackWaitStrategy(APP_PORT).withStartupTimeout(appReadinessTimeout) + : Wait.forHttp("/health").forPort(APP_PORT).forStatusCode(200).withStartupTimeout(appReadinessTimeout); app = new GenericContainer<>(JDK_17_TEMURIN_JAMMY) .withNetwork(network).withNetworkAliases("app").withLabels(labels) .withExposedPorts(APP_PORT) - .waitingFor(Wait.forHttp("/health").forPort(APP_PORT).forStatusCode(200) - .withStartupTimeout(appReadinessTimeout)) + .waitingFor(appWait) .withEnv("DAPR_HTTP_ENDPOINT", Endpoints.http("dapr", 3500).toString()) .withEnv("DAPR_GRPC_ENDPOINT", "dapr:50001") .withEnv("DAPR_SIDECAR_IP", "dapr") @@ -81,7 +105,7 @@ public DaprTestEnvironment(String appId, Class mainClass) { dapr = new DaprContainer(DaprContainerConstants.DAPR_RUNTIME_IMAGE_TAG) .withNetwork(network).withNetworkAliases("dapr").withLabels(labels) .withAccessToHost(false) - .withAppName(appId).withAppPort(APP_PORT).withAppChannelAddress("app") + .withAppName(appId).withAppPort(APP_PORT).withAppChannelAddress("app").withAppProtocol(protocol) .withPlacementContainer(placement).withSchedulerContainer(scheduler) .waitingFor(Wait.forHttp("/v1.0/healthz/outbound").forPort(3500) .forStatusCodeMatching(status -> status >= 200 && status <= 399) @@ -95,9 +119,11 @@ public DaprTestEnvironment(String appId, Class mainClass) { private void attachLogs(GenericContainer container, String alias) { container.withLogConsumer(frame -> { String text = frame.getUtf8String(); + if ("app".equals(alias)) { appOutput.append(text); } + if (!text.isEmpty()) { System.out.print("[" + identity + " " + alias + "] " + text); } @@ -110,6 +136,7 @@ private void configureApplication() { List classpath = new ArrayList<>(); classpath.add(APP_RUNTIME + "/test-classes"); classpath.add(APP_RUNTIME + "/classes"); + try (var files = Files.walk(lib)) { classpath.addAll(files.filter(path -> path.toString().endsWith(".jar")) .sorted().map(path -> APP_RUNTIME + "/lib/" + lib.relativize(path).toString().replace('\\', '/')) @@ -117,6 +144,7 @@ private void configureApplication() { } catch (IOException e) { throw new IllegalStateException("Cannot read staged IT dependencies", e); } + app.withFileSystemBind(runtime.toString(), APP_RUNTIME, BindMode.READ_ONLY); app.withCommand("java", "-cp", String.join(":", classpath), mainClass, Integer.toString(APP_PORT)); } @@ -130,18 +158,21 @@ public void start() { System.out.printf("[%s] scheduler id=%s%n", identity, scheduler.getContainerId()); configureApplication(); resources.start(app); - System.out.printf("[%s] app id=%s endpoint=%s%n", identity, app.getContainerId(), appEndpoint()); + System.out.printf("[%s] app id=%s image=%s protocol=%s address=%s%n", identity, app.getContainerId(), + JDK_17_TEMURIN_JAMMY, protocol, appAddress()); resources.start(dapr); System.out.printf("[%s] dapr id=%s endpoint=%s grpc=%s:%d%n", identity, dapr.getContainerId(), Endpoints.http(dapr.getHost(), dapr.getHttpPort()), dapr.getHost(), dapr.getGrpcPort()); + try (DaprClient client = clientBuilder().build()) { client.waitForSidecar(30_000).block(Duration.ofSeconds(35)); client.getMetadata().retryWhen(Retry.fixedDelay(30, Duration.ofSeconds(1))).block(Duration.ofSeconds(35)); } catch (Exception e) { throw new IllegalStateException("Sidecar client readiness failed", e); } - System.out.printf("[%s] app=%s (%s), dapr=%s (%s), placement=%s, scheduler=%s%n", identity, - appEndpoint(), app.getContainerId(), Endpoints.http(dapr.getHost(), dapr.getHttpPort()), + + System.out.printf("[%s] app=%s (%s, %s), dapr=%s (%s), placement=%s, scheduler=%s%n", identity, + appAddress(), protocol, app.getContainerId(), Endpoints.http(dapr.getHost(), dapr.getHttpPort()), dapr.getContainerId(), placement.getContainerId(), scheduler.getContainerId()); } catch (RuntimeException | Error failure) { try { @@ -149,6 +180,7 @@ public void start() { } catch (RuntimeException | Error cleanup) { failure.addSuppressed(cleanup); } + throw failure; } } @@ -158,11 +190,23 @@ public void stop() { resources.stop(); } - /** Returns the app address reachable by the test JVM. */ + /** Returns the HTTP application endpoint reachable by the test JVM. */ public URI appEndpoint() { + if (protocol != DaprProtocol.HTTP) { + throw new IllegalStateException("Application is not HTTP: " + protocol); + } + return Endpoints.http(app.getHost(), app.getMappedPort(APP_PORT)); } + /** + * Returns the mapped host and port as an IPv6-safe authority. + * @return Application authority reachable by the test JVM. + */ + public String appAddress() { + return Endpoints.http(app.getHost(), app.getMappedPort(APP_PORT)).getRawAuthority(); + } + /** Returns a client builder configured for the mapped sidecar ports. */ public DaprClientBuilder clientBuilder() { return Endpoints.client(dapr.getHost(), dapr.getHttpPort(), dapr.getGrpcPort()); @@ -177,6 +221,10 @@ String appOutput() { return appOutput.toString(); } + DaprProtocol appProtocol() { + return dapr.getAppProtocol(); + } + String appContainerId() { return app.getContainerId(); } diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironmentIT.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironmentIT.java index 9e16ea4fac..b1889808a9 100644 --- a/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironmentIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironmentIT.java @@ -12,9 +12,16 @@ */ package io.dapr.it.testcontainers.support; +import com.google.protobuf.Empty; import io.dapr.client.DaprClient; import io.dapr.client.domain.HttpExtension; +import io.dapr.it.MethodInvokeServiceGrpc; +import io.dapr.it.MethodInvokeServiceProtos; import io.dapr.it.testcontainers.methodinvoke.http.app.MethodInvokeService; +import io.dapr.testcontainers.DaprProtocol; +import io.dapr.v1.AppCallbackHealthCheckGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.testcontainers.DockerClientFactory; @@ -22,6 +29,7 @@ import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -34,6 +42,7 @@ class DaprTestEnvironmentIT { @Test void startsExactlyFourChildrenAndStopsOnlyItsOwnResources() throws Exception { DaprTestEnvironment environment = new DaprTestEnvironment("foundation-it", MethodInvokeService.class); + try { environment.start(); assertEquals(4, containers(environment.identity())); @@ -49,6 +58,7 @@ void startsExactlyFourChildrenAndStopsOnlyItsOwnResources() throws Exception { .findFirst().orElseThrow(() -> new AssertionError("application runtime is not mounted")); assertEquals(Boolean.FALSE, runtime.getRW(), "application runtime must be read-only"); assertTrue(runtime.getSource().replace('\\', '/').endsWith("/sdk-tests/target/it-runtime")); + try (DaprClient client = environment.clientBuilder().build()) { client.invokeMethod("foundation-it", "health", null, HttpExtension.GET).block(); } @@ -56,6 +66,7 @@ void startsExactlyFourChildrenAndStopsOnlyItsOwnResources() throws Exception { environment.stop(); environment.stop(); } + assertClean(environment.identity()); } @@ -71,6 +82,44 @@ void failedApplicationRollsBackAllOwnedResources() { assertClean(environment.identity()); } + @Test + void grpcCallbackAndRoutedInvocationAreReady() throws Exception { + DaprTestEnvironment environment = new DaprTestEnvironment("foundation-grpc-it", + io.dapr.it.testcontainers.methodinvoke.grpc.app.MethodInvokeService.class, DaprProtocol.GRPC); + + try { + environment.start(); + assertEquals(4, containers(environment.identity())); + assertEquals(1, networks(environment.identity())); + var network = DockerClientFactory.instance().client().listNetworksCmd() + .withFilter("label", List.of(LABEL + "=" + environment.identity())).exec().get(0); + assertEquals(4, DockerClientFactory.instance().client().inspectNetworkCmd() + .withNetworkId(network.getId()).exec().getContainers().size()); + assertTrue(environment.appOutput().contains("Artificial delay of 10 seconds")); + assertThrows(IllegalStateException.class, environment::appEndpoint); + ManagedChannel channel = ManagedChannelBuilder.forTarget(environment.appAddress()).usePlaintext().build(); + + try { + AppCallbackHealthCheckGrpc.newBlockingStub(channel).withDeadlineAfter(5, TimeUnit.SECONDS) + .healthCheck(Empty.getDefaultInstance()); + } finally { + channel.shutdownNow(); + assertTrue(channel.awaitTermination(5, TimeUnit.SECONDS)); + } + + try (DaprClient client = environment.clientBuilder().build()) { + var stub = client.newGrpcStub("foundation-grpc-it", MethodInvokeServiceGrpc::newBlockingStub); + assertTrue(stub.getMessages(MethodInvokeServiceProtos.GetMessagesRequest.getDefaultInstance()) + .getMessagesMap().isEmpty()); + } + } finally { + environment.stop(); + environment.stop(); + } + + assertClean(environment.identity()); + } + private static long containers(String identity) { return DockerClientFactory.instance().client().listContainersCmd().withShowAll(true) .withLabelFilter(Map.of(LABEL, identity)).exec().size(); diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/GrpcCallbackWaitStrategy.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/GrpcCallbackWaitStrategy.java new file mode 100644 index 0000000000..8369c7d760 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/GrpcCallbackWaitStrategy.java @@ -0,0 +1,112 @@ +/* + * Copyright 2026 The Dapr 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. + */ +package io.dapr.it.testcontainers.support; + +import com.google.protobuf.Empty; +import io.dapr.v1.AppCallbackHealthCheckGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import org.testcontainers.containers.wait.strategy.AbstractWaitStrategy; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** Waits for the Dapr callback health RPC, not merely a listening socket. */ +final class GrpcCallbackWaitStrategy extends AbstractWaitStrategy { + private final int port; + + GrpcCallbackWaitStrategy(int port) { + this.port = port; + } + + @Override + protected void waitUntilReady() { + String host = waitStrategyTarget.getHost(); + int mappedPort = waitStrategyTarget.getMappedPort(port); + + try { + awaitCallback(ManagedChannelBuilder.forAddress(host, mappedPort).usePlaintext().build(), startupTimeout); + } catch (RuntimeException e) { + throw new IllegalStateException("gRPC callback readiness failed for " + + Endpoints.http(host, mappedPort).getRawAuthority() + " container=" + waitStrategyTarget.getContainerId() + + " alias=app image=" + waitStrategyTarget.getContainerInfo().getConfig().getImage() + + " logs:\n" + waitStrategyTarget.getLogs(), e); + } + } + + /** Owns and closes the supplied probe channel. */ + static void awaitCallback(ManagedChannel channel, Duration timeout) { + long end = System.nanoTime() + timeout.toNanos(); + StatusRuntimeException last = null; + + try { + while (System.nanoTime() < end) { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException("gRPC callback readiness interrupted"); + } + + long remaining = end - System.nanoTime(); + + if (remaining <= 0) { + break; + } + + try { + AppCallbackHealthCheckGrpc.newBlockingStub(channel) + .withDeadlineAfter(Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(500)), TimeUnit.NANOSECONDS) + .healthCheck(Empty.getDefaultInstance()); + return; + } catch (StatusRuntimeException e) { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException("gRPC callback readiness interrupted"); + } + + if (e.getStatus().getCode() != Status.Code.UNAVAILABLE + && e.getStatus().getCode() != Status.Code.DEADLINE_EXCEEDED) { + throw e; + } + + last = e; + } + + remaining = end - System.nanoTime(); + + if (remaining > 0) { + TimeUnit.NANOSECONDS.sleep(Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(100))); + } + } + + throw new IllegalStateException("Timed out waiting for gRPC callback health after " + timeout, last); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted waiting for gRPC callback health", e); + } finally { + channel.shutdownNow(); + boolean interrupted = Thread.interrupted(); + + try { + if (!channel.awaitTermination(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("gRPC callback probe channel did not terminate"); + } + } catch (InterruptedException e) { + interrupted = true; + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/GrpcCallbackWaitStrategyTest.java b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/GrpcCallbackWaitStrategyTest.java new file mode 100644 index 0000000000..329c021fd7 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/testcontainers/support/GrpcCallbackWaitStrategyTest.java @@ -0,0 +1,137 @@ +/* + * Copyright 2026 The Dapr 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. + */ +package io.dapr.it.testcontainers.support; + +import com.google.protobuf.Empty; +import io.dapr.grpc.GrpcHealthCheckService; +import io.dapr.testcontainers.DaprProtocol; +import io.dapr.v1.AppCallbackHealthCheckGrpc; +import io.dapr.v1.DaprAppCallbackProtos.HealthCheckResponse; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.Server; +import io.grpc.ServerBuilder; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class GrpcCallbackWaitStrategyTest { + @Test + void protocolsAndIpv6Address() { + assertEquals(DaprProtocol.HTTP, new DaprTestEnvironment("http", getClass()).appProtocol()); + assertEquals(DaprProtocol.GRPC, new DaprTestEnvironment("grpc", getClass(), DaprProtocol.GRPC).appProtocol()); + assertThrows(IllegalArgumentException.class, () -> new DaprTestEnvironment("bad", getClass(), null)); + assertEquals("[::1]:8080", Endpoints.http("::1", 8080).getRawAuthority()); + } + + @Test + void successfulHealthClosesProbeChannel() throws Exception { + Server server = ServerBuilder.forPort(0).addService(new GrpcHealthCheckService()).build().start(); + ManagedChannel channel = channel(server); + + try { + GrpcCallbackWaitStrategy.awaitCallback(channel, Duration.ofSeconds(5)); + assertTrue(channel.isTerminated()); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void retriesUnavailableCallbackUntilItSucceeds() throws Exception { + AtomicInteger attempts = new AtomicInteger(); + Server server = ServerBuilder.forPort(0).addService( + new AppCallbackHealthCheckGrpc.AppCallbackHealthCheckImplBase() { + @Override + public void healthCheck(Empty request, StreamObserver response) { + if (attempts.incrementAndGet() == 1) { + response.onError(Status.UNAVAILABLE.asRuntimeException()); + } else { + response.onNext(HealthCheckResponse.getDefaultInstance()); + response.onCompleted(); + } + } + }).build().start(); + ManagedChannel channel = channel(server); + + try { + GrpcCallbackWaitStrategy.awaitCallback(channel, Duration.ofSeconds(5)); + assertTrue(attempts.get() >= 2); + assertTrue(channel.isTerminated()); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void missingCallbackFailsAndClosesProbeChannel() throws Exception { + Server server = ServerBuilder.forPort(0).build().start(); + ManagedChannel channel = channel(server); + + try { + assertThrows(StatusRuntimeException.class, + () -> GrpcCallbackWaitStrategy.awaitCallback(channel, Duration.ofSeconds(5))); + assertTrue(channel.isTerminated()); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void stalledCallbackTimesOutAndClosesProbeChannel() throws Exception { + Server server = ServerBuilder.forPort(0).addService( + new AppCallbackHealthCheckGrpc.AppCallbackHealthCheckImplBase() { + @Override + public void healthCheck(Empty request, StreamObserver response) { } + }).build().start(); + ManagedChannel channel = channel(server); + + try { + long start = System.nanoTime(); + assertThrows(IllegalStateException.class, + () -> GrpcCallbackWaitStrategy.awaitCallback(channel, Duration.ofMillis(300))); + assertTrue(Duration.ofNanos(System.nanoTime() - start).compareTo(Duration.ofSeconds(5)) < 0); + assertTrue(channel.isTerminated()); + } finally { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void interruptionIsPreservedAndChannelIsClosed() { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 1).usePlaintext().build(); + + try { + Thread.currentThread().interrupt(); + assertThrows(IllegalStateException.class, + () -> GrpcCallbackWaitStrategy.awaitCallback(channel, Duration.ofSeconds(5))); + assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(channel.isTerminated()); + } finally { + Thread.interrupted(); + } + } + + private static ManagedChannel channel(Server server) { + return ManagedChannelBuilder.forAddress("localhost", server.getPort()).usePlaintext().build(); + } +}