diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 8cdbbaf1df..e59e50004a 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -112,17 +112,7 @@ jobs: spring-boot-display-version: 4.0.x experimental: false env: - GOVER: "1.20" - GOOS: linux - GOARCH: amd64 - GOPROXY: https://proxy.golang.org JDK_VER: ${{ matrix.java }} - DAPR_CLI_VER: 1.18.0 - DAPR_RUNTIME_VER: 1.18.0 - DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.18.0/install/install.sh - DAPR_CLI_REF: - DAPR_REF: - TOXIPROXY_URL: https://github.com/Shopify/toxiproxy/releases/download/v2.5.0/toxiproxy-server-linux-amd64 steps: - uses: actions/checkout@v7 - name: Check Docker version @@ -132,69 +122,6 @@ jobs: with: distribution: 'temurin' java-version: ${{ env.JDK_VER }} - - name: Set up Dapr CLI - run: wget -q ${{ env.DAPR_INSTALL_URL }} -O - | /bin/bash -s ${{ env.DAPR_CLI_VER }} - - name: Set up Go ${{ env.GOVER }} - if: env.DAPR_REF != '' || env.DAPR_CLI_REF != '' - uses: actions/setup-go@v6 - with: - go-version: ${{ env.GOVER }} - - name: Checkout Dapr CLI repo to override dapr command. - uses: actions/checkout@v7 - if: env.DAPR_CLI_REF != '' - with: - repository: dapr/cli - ref: ${{ env.DAPR_CLI_REF }} - path: cli - - name: Checkout Dapr repo to override daprd. - uses: actions/checkout@v7 - if: env.DAPR_REF != '' - with: - repository: dapr/dapr - ref: ${{ env.DAPR_REF }} - path: dapr - - name: Build and override dapr cli with referenced commit. - if: env.DAPR_CLI_REF != '' - run: | - cd cli - make - sudo cp dist/linux_amd64/release/dapr /usr/local/bin/dapr - cd .. - - name: Uninstall Dapr runtime ${{ env.DAPR_RUNTIME_VER }} - run: dapr uninstall --all - - name: Ensure Dapr runtime uninstalled - run: | - while [ "$(docker ps -aq --filter 'name=^/dapr')" ]; do - echo "Waiting for Dapr containers to be deleted..." - sleep 5 - done - echo "All dapr containers are deleted." - - name: Initialize Dapr runtime ${{ env.DAPR_RUNTIME_VER }} - run: dapr init --runtime-version ${{ env.DAPR_RUNTIME_VER }} - - name: Build and override daprd with referenced commit. - if: env.DAPR_REF != '' - run: | - cd dapr - make - mkdir -p $HOME/.dapr/bin/ - cp dist/linux_amd64/release/daprd $HOME/.dapr/bin/daprd - cd .. - - name: Override placement service. - if: env.DAPR_REF != '' - run: | - docker stop dapr_placement - cd dapr - ./dist/linux_amd64/release/placement & - - name: Spin local environment - run: | - docker compose -f ./sdk-tests/deploy/local-test.yml up -d kafka - docker ps - - name: Install local ToxiProxy to simulate connectivity issues to Dapr sidecar - run: | - mkdir -p /home/runner/.local/bin - wget -q ${{ env.TOXIPROXY_URL }} -O /home/runner/.local/bin/toxiproxy-server - chmod +x /home/runner/.local/bin/toxiproxy-server - /home/runner/.local/bin/toxiproxy-server --version - name: Clean up and install sdk run: ./mvnw clean install -B -q -DskipTests - name: Integration tests using spring boot 3.x version ${{ matrix.spring-boot-version }} diff --git a/.github/workflows/validate-docs.yml b/.github/workflows/validate-docs.yml index fbc5f78be5..da60f32b0c 100644 --- a/.github/workflows/validate-docs.yml +++ b/.github/workflows/validate-docs.yml @@ -28,6 +28,7 @@ jobs: with: distribution: 'temurin' java-version: ${{ env.JDK_VER }} + cache: 'maven' - name: Install jars run: ./mvnw install -q -B -DskipTests - name: Validate Java docs generation diff --git a/pom.xml b/pom.xml index f1b5f598c0..c9388bd316 100644 --- a/pom.xml +++ b/pom.xml @@ -689,6 +689,10 @@ org.apache.maven.plugins maven-javadoc-plugin ${maven-javadoc-plugin.version} + + false aggregate diff --git a/sdk-tests/components/http_binding.yaml b/sdk-tests/components/http_binding.yaml deleted file mode 100644 index 5a3ca0a1df..0000000000 --- a/sdk-tests/components/http_binding.yaml +++ /dev/null @@ -1,27 +0,0 @@ -apiVersion: dapr.io/v1alpha1 -kind: Component -metadata: - name: github-http-binding-404 -spec: - type: bindings.http - version: v1 - metadata: - - name: url - value: https://api.github.com/unknown_path -scopes: - - bindingit-httpoutputbinding-exception ---- -apiVersion: dapr.io/v1alpha1 -kind: Component -metadata: - name: github-http-binding-404-success -spec: - type: bindings.http - version: v1 - metadata: - - name: url - value: https://api.github.com/unknown_path - - name: errorIfNot2XX - value: "false" -scopes: - - bindingit-httpoutputbinding-ignore-error \ No newline at end of file diff --git a/sdk-tests/components/kafka_bindings.yaml b/sdk-tests/components/kafka_bindings.yaml deleted file mode 100644 index 8735063766..0000000000 --- a/sdk-tests/components/kafka_bindings.yaml +++ /dev/null @@ -1,26 +0,0 @@ -apiVersion: dapr.io/v1alpha1 -kind: Component -metadata: - name: sample123 -spec: - type: bindings.kafka - version: v1 - metadata: - # Kafka broker connection setting - - name: brokers - value: localhost:9092 - # consumer configuration: topic and consumer group - - name: topics - value: "topic-{appID}" - - name: consumerGroup - value: "{appID}" - # publisher configuration: topic - - name: publishTopic - value: "topic-{appID}" - - name: authRequired - value: "false" - - name: initialOffset - value: oldest -scopes: - - bindingit-http-inputbindingservice - - bindingit-grpc-inputbindingservice \ No newline at end of file diff --git a/sdk-tests/deploy/local-test.yml b/sdk-tests/deploy/local-test.yml deleted file mode 100644 index 7160ac25b6..0000000000 --- a/sdk-tests/deploy/local-test.yml +++ /dev/null @@ -1,23 +0,0 @@ -version: '3' -services: - zookeeper: - image: confluentinc/cp-zookeeper:7.4.4 - environment: - ZOOKEEPER_CLIENT_PORT: 2181 - ZOOKEEPER_TICK_TIME: 2000 - ports: - - 2181:2181 - - kafka: - image: confluentinc/cp-kafka:7.4.4 - depends_on: - - zookeeper - ports: - - "9092:9092" - environment: - KAFKA_BROKER_ID: 1 - KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 - KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092 - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT - KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 diff --git a/sdk-tests/pom.xml b/sdk-tests/pom.xml index 9d68559aec..9a0dfcbd65 100644 --- a/sdk-tests/pom.xml +++ b/sdk-tests/pom.xml @@ -217,6 +217,11 @@ testcontainers-toxiproxy test + + org.testcontainers + testcontainers-kafka + test + diff --git a/sdk-tests/src/test/java/io/dapr/it/BaseIT.java b/sdk-tests/src/test/java/io/dapr/it/BaseIT.java deleted file mode 100644 index f50025e217..0000000000 --- a/sdk-tests/src/test/java/io/dapr/it/BaseIT.java +++ /dev/null @@ -1,187 +0,0 @@ -/* - * Copyright 2021 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; - -import io.dapr.actors.client.ActorClient; -import io.dapr.client.resiliency.ResiliencyOptions; -import io.dapr.config.Properties; -import org.apache.commons.lang3.tuple.ImmutablePair; -import org.junit.jupiter.api.AfterAll; - -import java.lang.reflect.Constructor; -import java.lang.reflect.InvocationTargetException; -import java.util.HashMap; -import java.util.LinkedList; -import java.util.Map; -import java.util.Queue; - -import static io.dapr.it.AppRun.AppProtocol.GRPC; -import static io.dapr.it.AppRun.AppProtocol.HTTP; - -public abstract class BaseIT { - - protected static final String STATE_STORE_NAME = "statestore"; - - protected static final String QUERY_STATE_STORE = "mongo-statestore"; - - private static final Map DAPR_RUN_BUILDERS = new HashMap<>(); - - private static final Queue TO_BE_STOPPED = new LinkedList<>(); - - private static final Queue TO_BE_CLOSED = new LinkedList<>(); - - protected static DaprRun startDaprApp( - String testName, - String successMessage, - Class serviceClass, - Boolean useAppPort, - int maxWaitMilliseconds) throws Exception { - return startDaprApp(testName, successMessage, serviceClass, useAppPort, maxWaitMilliseconds, HTTP); - } - - protected static DaprRun startDaprApp( - String testName, - String successMessage, - Class serviceClass, - AppRun.AppProtocol appProtocol, - int maxWaitMilliseconds) throws Exception { - return startDaprApp(testName, successMessage, serviceClass, true, maxWaitMilliseconds, appProtocol); - } - - protected static DaprRun startDaprApp( - String testName, - String successMessage, - Class serviceClass, - Boolean useAppPort, - int maxWaitMilliseconds, - AppRun.AppProtocol appProtocol) throws Exception { - return startDaprApp( - testName, - successMessage, - serviceClass, - useAppPort, - true, - maxWaitMilliseconds, - appProtocol); - } - - protected static DaprRun startDaprApp( - String testName, - int maxWaitMilliseconds) throws Exception { - return startDaprApp( - testName, - "You're up and running!", - null, - false, - true, - maxWaitMilliseconds, - HTTP); - } - - protected static DaprRun startDaprApp( - String testName, - String successMessage, - Class serviceClass, - Boolean useAppPort, - Boolean useDaprPorts, - int maxWaitMilliseconds, - AppRun.AppProtocol appProtocol) throws Exception { - DaprRun.Builder builder = new DaprRun.Builder( - testName, - () -> DaprPorts.build(useAppPort, useDaprPorts, useDaprPorts), - successMessage, - maxWaitMilliseconds, - appProtocol).withServiceClass(serviceClass); - DaprRun run = builder.build(); - TO_BE_STOPPED.add(run); - DAPR_RUN_BUILDERS.put(run.getAppName(), builder); - run.start(); - return run; - } - - protected static ImmutablePair startSplitDaprAndApp( - String testName, - String successMessage, - Class serviceClass, - Boolean useAppPort, - int maxWaitMilliseconds) throws Exception { - return startSplitDaprAndApp( - testName, - successMessage, - serviceClass, - useAppPort, - maxWaitMilliseconds, - HTTP); - } - - protected static ImmutablePair startSplitDaprAndApp( - String testName, - String successMessage, - Class serviceClass, - Boolean useAppPort, - int maxWaitMilliseconds, - AppRun.AppProtocol appProtocol) throws Exception { - DaprRun.Builder builder = new DaprRun.Builder( - testName, - () -> DaprPorts.build(useAppPort, true, true), - successMessage, - maxWaitMilliseconds, - appProtocol).withServiceClass(serviceClass); - ImmutablePair runs = builder.splitBuild(); - TO_BE_STOPPED.add(runs.left); - TO_BE_STOPPED.add(runs.right); - DAPR_RUN_BUILDERS.put(runs.right.getAppName(), builder); - runs.left.start(); - runs.right.start(); - return runs; - } - - protected static T deferClose(T object) { - TO_BE_CLOSED.add(object); - return object; - } - - @AfterAll - public static void cleanUp() throws Exception { - while (!TO_BE_CLOSED.isEmpty()) { - TO_BE_CLOSED.remove().close(); - } - - while (!TO_BE_STOPPED.isEmpty()) { - TO_BE_STOPPED.remove().stop(); - } - } - - protected static ActorClient newActorClient(Properties properties) { - return new ActorClient(properties, null); - } - - protected static ActorClient newActorClient(ResiliencyOptions resiliencyOptions) throws RuntimeException { - try { - Constructor constructor = ActorClient.class.getDeclaredConstructor(ResiliencyOptions.class); - constructor.setAccessible(true); - ActorClient client = constructor.newInstance(resiliencyOptions); - TO_BE_CLOSED.add(client); - return client; - } catch (NoSuchMethodException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - throw new RuntimeException(e); - } catch (InstantiationException e) { - throw new RuntimeException(e); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } - } -} diff --git a/sdk-tests/src/test/java/io/dapr/it/DaprRun.java b/sdk-tests/src/test/java/io/dapr/it/DaprRun.java deleted file mode 100644 index e29c5f1347..0000000000 --- a/sdk-tests/src/test/java/io/dapr/it/DaprRun.java +++ /dev/null @@ -1,436 +0,0 @@ -/* - * Copyright 2021 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; - -import com.google.protobuf.Empty; -import io.dapr.actors.client.ActorClient; -import io.dapr.client.DaprClient; -import io.dapr.client.DaprClientBuilder; -import io.dapr.client.DaprPreviewClient; -import io.dapr.client.resiliency.ResiliencyOptions; -import io.dapr.config.Properties; -import io.dapr.config.Property; -import io.dapr.v1.AppCallbackHealthCheckGrpc; -import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; -import org.apache.commons.lang3.tuple.ImmutablePair; - -import java.io.IOException; -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 java.util.HashMap; -import java.util.Map; -import java.util.UUID; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Supplier; - -import static io.dapr.it.Retry.callWithRetry; - - -public class DaprRun implements Stoppable { - - private static final String DEFAULT_DAPR_API_TOKEN = UUID.randomUUID().toString(); - private static final String DAPR_SUCCESS_MESSAGE = "You're up and running!"; - - private static final String DAPR_RUN = "dapr run --app-id %s --app-protocol %s " + - "--config ./configurations/configuration.yaml " + - "--resources-path ./components"; - - // the arg in -Dexec.args is the app's port - private static final String DAPR_COMMAND = - " -- mvn exec:java -D exec.mainClass=%s -D exec.classpathScope=test -D exec.args=\"%s\""; - - private final DaprPorts ports; - - private final String appName; - - private final AppRun.AppProtocol appProtocol; - - private final int maxWaitMilliseconds; - - private final AtomicBoolean started; - - private final Command startCommand; - - private final Command listCommand; - - private final Command stopCommand; - - private final boolean hasAppHealthCheck; - - private final Map, String> propertyOverrides; - - private DaprRun(String testName, - DaprPorts ports, - String successMessage, - Class serviceClass, - int maxWaitMilliseconds, - AppRun.AppProtocol appProtocol) { - this( - testName, - ports, - successMessage, - serviceClass, - maxWaitMilliseconds, - appProtocol, - resolveDaprApiToken(serviceClass)); - } - - private DaprRun(String testName, - DaprPorts ports, - String successMessage, - Class serviceClass, - int maxWaitMilliseconds, - AppRun.AppProtocol appProtocol, - String daprApiToken) { - // The app name needs to be deterministic since we depend on it to kill previous runs. - this.appName = serviceClass == null ? - testName.toLowerCase() : - String.format("%s-%s", testName, serviceClass.getSimpleName()).toLowerCase(); - this.appProtocol = appProtocol; - this.startCommand = - new Command( - successMessage, - buildDaprCommand(this.appName, serviceClass, ports, appProtocol), - daprApiToken == null ? null : Map.of("DAPR_API_TOKEN", daprApiToken)); - this.listCommand = new Command( - this.appName, - "dapr list"); - this.stopCommand = new Command( - "app stopped successfully", - "dapr stop --app-id " + this.appName); - this.ports = ports; - this.maxWaitMilliseconds = maxWaitMilliseconds; - this.started = new AtomicBoolean(false); - this.hasAppHealthCheck = isAppHealthCheckEnabled(serviceClass); - this.propertyOverrides = daprApiToken == null ? ports.getPropertyOverrides() : - Collections.unmodifiableMap(new HashMap<>(ports.getPropertyOverrides()) {{ - put(Properties.API_TOKEN, daprApiToken); - }}); - } - - public void start() throws InterruptedException, IOException { - long start = System.currentTimeMillis(); - // First, try to stop previous run (if left running). - this.stop(); - // Wait for the previous run to kill the prior process. - long timeLeft = this.maxWaitMilliseconds - (System.currentTimeMillis() - start); - System.out.println("Checking if previous run for Dapr application has stopped ..."); - checkRunState(timeLeft, false); - - System.out.println("Starting dapr application ..."); - this.startCommand.run(); - this.started.set(true); - - timeLeft = this.maxWaitMilliseconds - (System.currentTimeMillis() - start); - System.out.println("Checking if Dapr application has started ..."); - checkRunState(timeLeft, true); - - if (this.ports.getAppPort() != null) { - timeLeft = this.maxWaitMilliseconds - (System.currentTimeMillis() - start); - callWithRetry(() -> { - System.out.println("Checking if app is listening on port ..."); - assertListeningOnPort(this.ports.getAppPort()); - }, timeLeft); - } - - if (this.ports.getHttpPort() != null) { - timeLeft = this.maxWaitMilliseconds - (System.currentTimeMillis() - start); - callWithRetry(() -> { - System.out.println("Checking if Dapr is listening on HTTP port ..."); - assertListeningOnPort(this.ports.getHttpPort()); - }, timeLeft); - } - - if (this.ports.getGrpcPort() != null) { - timeLeft = this.maxWaitMilliseconds - (System.currentTimeMillis() - start); - callWithRetry(() -> { - System.out.println("Checking if Dapr is listening on GRPC port ..."); - assertListeningOnPort(this.ports.getGrpcPort()); - }, timeLeft); - } - System.out.println("Dapr application started."); - } - - @Override - public void stop() throws InterruptedException, IOException { - System.out.println("Stopping dapr application ..."); - try { - this.stopCommand.run(); - System.out.println("Dapr application stopped."); - } catch (RuntimeException e) { - if (e.getMessage() != null && e.getMessage().contains("Could not find success criteria")) { - System.out.println("App " + this.appName + " already stopped or not found (ignored)."); - } else { - System.out.println("Could not stop app " + this.appName + ": " + e.getMessage()); - } - } - } - - public Map, String> getPropertyOverrides() { - return this.propertyOverrides; - } - - public DaprClientBuilder newDaprClientBuilder() { - return new DaprClientBuilder().withPropertyOverrides(this.getPropertyOverrides()); - } - - public ActorClient newActorClient() { - return this.newActorClient(null, null); - } - - public ActorClient newActorClient(Map metadata) { - return this.newActorClient(metadata, null); - } - - public ActorClient newActorClient(ResiliencyOptions resiliencyOptions) { - return this.newActorClient(null, resiliencyOptions); - } - - public ActorClient newActorClient(Map metadata, ResiliencyOptions resiliencyOptions) { - return new ActorClient(new Properties(this.getPropertyOverrides()), metadata, resiliencyOptions); - } - - public void waitForAppHealth(int maxWaitMilliseconds) throws InterruptedException { - if (!this.hasAppHealthCheck) { - return; - } - - if (AppRun.AppProtocol.GRPC.equals(this.appProtocol)) { - ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", this.getAppPort()) - .usePlaintext() - .build(); - try { - AppCallbackHealthCheckGrpc.AppCallbackHealthCheckBlockingStub stub = - AppCallbackHealthCheckGrpc.newBlockingStub(channel); - long maxWait = System.currentTimeMillis() + maxWaitMilliseconds; - while (System.currentTimeMillis() <= maxWait) { - try { - stub.healthCheck(Empty.getDefaultInstance()); - Thread.sleep(2000); - return; - } catch (Exception e) { - Thread.sleep(1000); - } - } - - throw new RuntimeException("timeout: gRPC service is not healthy."); - } finally { - channel.shutdown(); - } - } else { - long maxWait = System.currentTimeMillis() + maxWaitMilliseconds; - HttpClient client = HttpClient.newBuilder() - .version(HttpClient.Version.HTTP_1_1) - .connectTimeout(Duration.ofSeconds(5)) - .build(); - String url = "http://127.0.0.1:" + this.getAppPort() + "/health"; - HttpRequest request = HttpRequest.newBuilder() - .GET() - .uri(URI.create(url)) - .build(); - - while (System.currentTimeMillis() <= maxWait) { - try { - HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); - if (response.statusCode() == 200) { - Thread.sleep(2000); - return; - } - } catch (IOException e) { - // not ready yet - } - Thread.sleep(1000); - } - - throw new RuntimeException("timeout: HTTP service is not healthy."); - } - } - - public Integer getGrpcPort() { - return ports.getGrpcPort(); - } - - public Integer getHttpPort() { - return ports.getHttpPort(); - } - - public Integer getAppPort() { - return ports.getAppPort(); - } - - public String getAppName() { - return appName; - } - - public DaprClient newDaprClient() { - return new DaprClientBuilder() - .withPropertyOverrides(this.getPropertyOverrides()) - .build(); - } - - public DaprPreviewClient newDaprPreviewClient() { - return new DaprClientBuilder() - .withPropertyOverrides(this.getPropertyOverrides()) - .buildPreviewClient(); - } - - public void checkRunState(long timeout, boolean shouldBeRunning) throws InterruptedException { - callWithRetry(() -> { - try { - this.listCommand.run(); - - if (!shouldBeRunning) { - throw new RuntimeException("Previous run for app has not stopped yet!"); - } - } catch (IllegalStateException e) { - // Bad case if the app is supposed to be running. - if (shouldBeRunning) { - throw e; - } - } catch (Exception e) { - throw new RuntimeException(e); - } - }, timeout); - } - - private static String buildDaprCommand( - String appName, Class serviceClass, DaprPorts ports, AppRun.AppProtocol appProtocol) { - StringBuilder stringBuilder = - new StringBuilder(String.format(DAPR_RUN, appName, appProtocol.toString().toLowerCase())) - .append(ports.getAppPort() != null ? " --app-port " + ports.getAppPort() : "") - .append(ports.getHttpPort() != null ? " --dapr-http-port " + ports.getHttpPort() : "") - .append(ports.getGrpcPort() != null ? " --dapr-grpc-port " + ports.getGrpcPort() : "") - .append(isAppHealthCheckEnabled(serviceClass) ? - " --enable-app-health-check --app-health-probe-interval=1" : "") - .append(serviceClass == null ? "" : - String.format(DAPR_COMMAND, serviceClass.getCanonicalName(), - ports.getAppPort() != null ? ports.getAppPort().toString() : "")); - return stringBuilder.toString(); - } - - private static boolean isAppHealthCheckEnabled(Class serviceClass) { - if (serviceClass != null) { - DaprRunConfig daprRunConfig = (DaprRunConfig) serviceClass.getAnnotation(DaprRunConfig.class); - if (daprRunConfig != null) { - return daprRunConfig.enableAppHealthCheck(); - } - } - - return false; - } - - private static String resolveDaprApiToken(Class serviceClass) { - if (serviceClass != null) { - DaprRunConfig daprRunConfig = (DaprRunConfig) serviceClass.getAnnotation(DaprRunConfig.class); - if (daprRunConfig != null) { - if (!daprRunConfig.enableDaprApiToken()) { - return null; - } - // We use the clas name itself as the token. Just needs to be deterministic. - return serviceClass.getCanonicalName(); - } - } - - // By default, we use a token. - return DEFAULT_DAPR_API_TOKEN; - } - - private static void assertListeningOnPort(int port) { - System.out.printf("Checking port %d ...\n", port); - - java.net.SocketAddress socketAddress = new java.net.InetSocketAddress(Properties.SIDECAR_IP.get(), port); - try (java.net.Socket socket = new java.net.Socket()) { - socket.connect(socketAddress, 1000); - } catch (Exception e) { - throw new RuntimeException(e); - } - - System.out.printf("Confirmed listening on port %d.\n", port); - } - - static class Builder { - - private final String testName; - - private final Supplier portsSupplier; - - private final String successMessage; - - private final int maxWaitMilliseconds; - - private Class serviceClass; - - private AppRun.AppProtocol appProtocol; - - private String daprApiToken; - - Builder( - String testName, - Supplier portsSupplier, - String successMessage, - int maxWaitMilliseconds, - AppRun.AppProtocol appProtocol) { - this.testName = testName; - this.portsSupplier = portsSupplier; - this.successMessage = successMessage; - this.maxWaitMilliseconds = maxWaitMilliseconds; - this.appProtocol = appProtocol; - this.daprApiToken = UUID.randomUUID().toString(); - } - - public Builder withServiceClass(Class serviceClass) { - this.serviceClass = serviceClass; - return this; - } - - DaprRun build() { - return new DaprRun( - this.testName, - this.portsSupplier.get(), - this.successMessage, - this.serviceClass, - this.maxWaitMilliseconds, - this.appProtocol); - } - - /** - * Builds app and dapr run separately. It can be useful to force the restart of one of them. - * @return Pair of AppRun and DaprRun. - */ - ImmutablePair splitBuild() { - DaprPorts ports = this.portsSupplier.get(); - AppRun appRun = new AppRun( - ports, - this.successMessage, - this.serviceClass, - this.maxWaitMilliseconds); - - DaprRun daprRun = new DaprRun( - this.testName, - ports, - DAPR_SUCCESS_MESSAGE, - null, - this.maxWaitMilliseconds, - this.appProtocol, - resolveDaprApiToken(serviceClass)); - - return new ImmutablePair<>(appRun, daprRun); - } - } -} diff --git a/sdk-tests/src/test/java/io/dapr/it/DaprRunConfig.java b/sdk-tests/src/test/java/io/dapr/it/DaprRunConfig.java deleted file mode 100644 index 72e9e37318..0000000000 --- a/sdk-tests/src/test/java/io/dapr/it/DaprRunConfig.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright 2021 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; - -import java.lang.annotation.ElementType; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -/** - * Customizes an app run for Dapr. - */ -@Target(ElementType.TYPE) -@Retention(RetentionPolicy.RUNTIME) -public @interface DaprRunConfig { - - boolean enableAppHealthCheck() default false; - - boolean enableDaprApiToken() default true; -} diff --git a/sdk-tests/src/test/java/io/dapr/it/ToxiProxyRun.java b/sdk-tests/src/test/java/io/dapr/it/ToxiProxyRun.java deleted file mode 100644 index ad248ec4ba..0000000000 --- a/sdk-tests/src/test/java/io/dapr/it/ToxiProxyRun.java +++ /dev/null @@ -1,110 +0,0 @@ -/* - * Copyright 2023 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; - -import eu.rekawek.toxiproxy.Proxy; -import eu.rekawek.toxiproxy.ToxiproxyClient; -import eu.rekawek.toxiproxy.model.ToxicDirection; -import io.dapr.actors.client.ActorClient; -import io.dapr.client.DaprClientBuilder; -import io.dapr.client.resiliency.ResiliencyOptions; -import io.dapr.config.Properties; -import io.dapr.config.Property; -import io.dapr.utils.NetworkUtils; - -import java.io.IOException; -import java.time.Duration; -import java.util.Map; - - -public class ToxiProxyRun implements Stoppable { - - private final DaprRun daprRun; - - private final Duration latency; - - private final Duration jitter; - - private final Command toxiProxyServer; - - private final DaprPorts toxiProxyPorts; - - private ToxiproxyClient toxiproxyClient; - - private Proxy grpcProxy; - - private Proxy httpProxy; - - public ToxiProxyRun(DaprRun run, Duration latency, Duration jitter) { - this.daprRun = run; - this.latency = latency; - this.jitter = jitter; - this.toxiProxyPorts = DaprPorts.build(true, true, true); - // artursouza: we use the "appPort" for the ToxiProxy server. - this.toxiProxyServer = new Command( - "Starting HTTP server on endpoint", - "toxiproxy-server --port " - + this.toxiProxyPorts.getAppPort()); - } - - public void start() throws IOException, InterruptedException { - this.toxiProxyServer.run(); - NetworkUtils.waitForSocket("127.0.0.1", this.toxiProxyPorts.getAppPort(), 10000); - this.toxiproxyClient = new ToxiproxyClient("127.0.0.1", this.toxiProxyPorts.getAppPort()); - - if (this.daprRun.getGrpcPort() != null) { - this.grpcProxy = toxiproxyClient.createProxy( - "daprd_grpc", - "127.0.0.1:" + this.toxiProxyPorts.getGrpcPort(), - "127.0.0.1:" + this.daprRun.getGrpcPort()); - this.grpcProxy.toxics() - .latency("latency", ToxicDirection.DOWNSTREAM, this.latency.toMillis()) - .setJitter(this.jitter.toMillis()); - } - - if (this.daprRun.getHttpPort() != null) { - this.httpProxy = toxiproxyClient.createProxy( - "daprd_http", - "127.0.0.1:" + this.toxiProxyPorts.getHttpPort(), - "127.0.0.1:" + this.daprRun.getHttpPort()); - this.httpProxy.toxics() - .latency("latency", ToxicDirection.DOWNSTREAM, this.latency.toMillis()) - .setJitter(this.jitter.toMillis()); - } - } - - public Map, String> getPropertyOverrides() { - return this.toxiProxyPorts.getPropertyOverrides(); - } - - public DaprClientBuilder newDaprClientBuilder() { - return this.daprRun.newDaprClientBuilder().withPropertyOverrides(this.getPropertyOverrides()); - } - - public ActorClient newActorClient() { - return this.newActorClient(null); - } - - public ActorClient newActorClient(ResiliencyOptions resiliencyOptions) { - return new ActorClient(new Properties(this.getPropertyOverrides()), resiliencyOptions); - } - - @Override - public void stop() throws InterruptedException, IOException { - this.toxiProxyServer.stop(); - this.toxiproxyClient = null; - this.grpcProxy = null; - this.httpProxy = null; - } -} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderFailoverIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderFailoverIT.java index 8e7b64c93e..44984d52c8 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderFailoverIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderFailoverIT.java @@ -8,19 +8,20 @@ * 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. + * limitations under the License. */ package io.dapr.it.actors; import io.dapr.actors.ActorId; -import io.dapr.actors.client.ActorClient; import io.dapr.actors.client.ActorProxy; import io.dapr.actors.client.ActorProxyBuilder; -import io.dapr.config.Properties; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; +import io.dapr.it.AppRun; import io.dapr.it.actors.app.MyActorService; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; +import io.dapr.testcontainers.DaprPlacementContainer; +import io.dapr.testcontainers.DaprSchedulerContainer; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -35,46 +36,82 @@ import static io.dapr.it.actors.MyActorTestUtils.validateMethodCalls; import static org.junit.jupiter.api.Assertions.assertNotEquals; -public class ActorReminderFailoverIT extends BaseIT { +/** + * Verifies that actor reminders fail over to a surviving sidecar when the sidecar + * currently hosting the actor is killed. Actor failover is keyed on actor type + * ("MyActorTest"): placement distributes actors of a type across every sidecar that + * registered that type, and rebalances when one host's app dies. This test starts two + * actor-hosting app+sidecar pairs plus a third, client-only sidecar (no app) whose + * ActorClient builds the proxy; all three share one placement, one scheduler, and Redis. + */ +public class ActorReminderFailoverIT extends BaseContainerIT { - private static Logger logger = LoggerFactory.getLogger(ActorReminderFailoverIT.class); + private static final Logger logger = LoggerFactory.getLogger(ActorReminderFailoverIT.class); private static final String METHOD_NAME = "receiveReminder"; - private ActorProxy proxy; + private static final String ACTOR_TYPE = "MyActorTest"; + + private DaprPlacementContainer placement; + + private DaprSchedulerContainer scheduler; - private DaprRun firstAppRun; + private DaprAndApp one; - private DaprRun secondAppRun; + private DaprAndApp two; - private DaprRun clientAppRun; + private DaprContainer client; + + private ActorProxy proxy; @BeforeEach public void init() throws Exception { - firstAppRun = startDaprApp( - ActorReminderFailoverIT.class.getSimpleName() + "One", - "Started MyActorService", + placement = startSharedPlacement(); + scheduler = startSharedScheduler(); + + one = startAppAndAttach( + "actor-reminder-failover-it-one", MyActorService.class, - true, - 60000); - secondAppRun = startDaprApp( - ActorReminderFailoverIT.class.getSimpleName() + "Two", - "Started MyActorService", + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-reminder-failover-it-one") + .withPlacementContainer(placement) + .withSchedulerContainer(scheduler) + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + + two = startAppAndAttach( + "actor-reminder-failover-it-two", MyActorService.class, - true, - 60000); - clientAppRun = startDaprApp( - ActorReminderFailoverIT.class.getSimpleName() + "Client", - 60000); + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-reminder-failover-it-two") + .withPlacementContainer(placement) + .withSchedulerContainer(scheduler) + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + + client = daprBuilder("actor-reminder-failover-it-client") + .withPlacementContainer(placement) + .withSchedulerContainer(scheduler) + .withComponent(redisStateStore(STATE_STORE_NAME)); + client.start(); + deferStop(client); + + // Prove both actor hosts have registered MyActorTest with the shared placement + // service before exercising failover. startAppAndAttach's waitForSidecar only proves + // daprd's gRPC channel is up, not that actor types are registered -- a fixed sleep + // alone is a flakiness risk under CI load. + waitForActorsReady(one.dapr()); + waitForActorsReady(two.dapr()); Thread.sleep(3000); ActorId actorId = new ActorId(UUID.randomUUID().toString()); - String actorType="MyActorTest"; logger.debug("Creating proxy builder"); ActorProxyBuilder proxyBuilder = - new ActorProxyBuilder(actorType, ActorProxy.class, deferClose(clientAppRun.newActorClient())); + new ActorProxyBuilder(ACTOR_TYPE, ActorProxy.class, newActorClient(client)); logger.debug("Creating actorId"); logger.debug("Building proxy"); proxy = proxyBuilder.build(actorId); @@ -104,11 +141,18 @@ public void reminderRecoveryTest() throws Exception { int originalActorHostIdentifier = Integer.parseInt( proxy.invokeMethod("getIdentifier", String.class).block()); - if (originalActorHostIdentifier == firstAppRun.getHttpPort()) { - firstAppRun.stop(); + // Stop BOTH the app and its sidecar for the host currently hosting the actor. Killing + // only the app subprocess leaves daprd connected to placement, so placement never + // reassigns the actor type away from that host (daprd just keeps retrying the dead + // app). The legacy harness's DaprRun.stop() ran `dapr stop --app-id`, which tore down + // the sidecar process along with its supervised app -- this mirrors that. + if (originalActorHostIdentifier == one.dapr().getHttpPort()) { + one.app().stop(); + one.dapr().stop(); } - if (originalActorHostIdentifier == secondAppRun.getHttpPort()) { - secondAppRun.stop(); + if (originalActorHostIdentifier == two.dapr().getHttpPort()) { + two.app().stop(); + two.dapr().stop(); } logger.debug("Pausing 10 seconds to allow failover to take place"); diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderRecoveryIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderRecoveryIT.java index 4ea6a759fd..fbfa8fce4f 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderRecoveryIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReminderRecoveryIT.java @@ -8,7 +8,7 @@ * 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. + * limitations under the License. */ package io.dapr.it.actors; @@ -17,12 +17,12 @@ import io.dapr.actors.client.ActorProxy; import io.dapr.actors.client.ActorProxyBuilder; import io.dapr.it.AppRun; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; import io.dapr.it.actors.app.ActorReminderDataParam; import io.dapr.it.actors.app.MyActorService; -import org.apache.commons.lang3.tuple.ImmutablePair; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -41,12 +41,15 @@ import static io.dapr.it.actors.MyActorTestUtils.validateMessageContent; import static io.dapr.it.actors.MyActorTestUtils.validateMethodCalls; -public class ActorReminderRecoveryIT extends BaseIT { +public class ActorReminderRecoveryIT extends BaseContainerIT { private static final Logger logger = LoggerFactory.getLogger(ActorReminderRecoveryIT.class); private static final String METHOD_NAME = "receiveReminder"; + private static DaprContainer dapr; + private static AppRun app; + /** * Parameters for this test. * Param #1: useGrpc. @@ -81,29 +84,27 @@ public static Stream data() { private ActorProxy proxy; - private ImmutablePair runs; - - private DaprRun clientRun; - - public void setup(String actorType) throws Exception { - runs = startSplitDaprAndApp( - ActorReminderRecoveryIT.class.getSimpleName(), - "Started MyActorService", + @BeforeAll + public static void start() throws Exception { + var pair = startAppAndAttach( + "actor-reminder-recovery-it", MyActorService.class, - true, - 60000); - - // Run that will stay up for integration tests. - // appId must not contain the appId from the other run, otherwise ITs will not run properly. - clientRun = startDaprApp("ActorReminderRecoveryTestClient", 5000); - - Thread.sleep(3000); + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-reminder-recovery-it") + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + dapr = pair.dapr(); + app = pair.app(); + waitForActorsReady(dapr); + } + public void setup(String actorType) { ActorId actorId = new ActorId(UUID.randomUUID().toString()); logger.debug("Creating proxy builder"); ActorProxyBuilder proxyBuilder = - new ActorProxyBuilder(actorType, ActorProxy.class, deferClose(clientRun.newActorClient())); + new ActorProxyBuilder(actorType, ActorProxy.class, newActorClient(dapr)); logger.debug("Creating actorId"); logger.debug("Building proxy"); proxy = proxyBuilder.build(actorId); @@ -144,16 +145,13 @@ public void reminderRecoveryTest( }, 30000); // Restarts runtime only. - logger.info("Stopping Dapr sidecar"); - runs.right.stop(); - // Pause a bit to let placements settle. logger.info("Pausing 10 seconds to let placements settle."); Thread.sleep(Duration.ofSeconds(10).toMillis()); - logger.info("Starting Dapr sidecar"); - runs.right.start(); - logger.info("Dapr sidecar started"); + logger.info("Restarting Dapr sidecar"); + restartSidecar(dapr); + logger.info("Dapr sidecar restarted"); logger.info("Pausing 7 seconds to allow sidecar to be healthy"); Thread.sleep(7000); diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorSdkResiliencyIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorSdkResiliencyIT.java index c9b7f27b76..87b19e7239 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/ActorSdkResiliencyIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorSdkResiliencyIT.java @@ -8,27 +8,34 @@ * 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. + * limitations under the License. */ package io.dapr.it.actors; +import eu.rekawek.toxiproxy.Proxy; +import eu.rekawek.toxiproxy.ToxiproxyClient; +import eu.rekawek.toxiproxy.model.ToxicDirection; import io.dapr.actors.ActorId; import io.dapr.actors.client.ActorClient; import io.dapr.actors.client.ActorProxyBuilder; import io.dapr.client.DaprClient; import io.dapr.client.resiliency.ResiliencyOptions; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; -import io.dapr.it.ToxiProxyRun; +import io.dapr.config.Properties; +import io.dapr.config.Property; +import io.dapr.it.AppRun; import io.dapr.it.actors.services.springboot.DemoActor; import io.dapr.it.actors.services.springboot.DemoActorService; -import org.junit.jupiter.api.AfterAll; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +import org.testcontainers.toxiproxy.ToxiproxyContainer; import java.time.Duration; +import java.util.HashMap; +import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicInteger; @@ -38,7 +45,7 @@ /** * Test SDK resiliency. */ -public class ActorSdkResiliencyIT extends BaseIT { +public class ActorSdkResiliencyIT extends BaseContainerIT { private static final ActorId ACTOR_ID = new ActorId(UUID.randomUUID().toString()); @@ -52,13 +59,15 @@ public class ActorSdkResiliencyIT extends BaseIT { private static final int MAX_RETRIES = -1; // Infinity - private static DaprRun daprRun; + private static DaprContainer dapr; + + private static AppRun app; private static DaprClient daprClient; private static DemoActor demoActor; - private static ToxiProxyRun toxiProxyRun; + private static ToxiproxyContainer toxiproxy; private static DemoActor toxiDemoActor; @@ -68,25 +77,45 @@ public class ActorSdkResiliencyIT extends BaseIT { @BeforeAll public static void init() throws Exception { - daprRun = startDaprApp( - ActorSdkResiliencyIT.class.getSimpleName(), - DemoActorService.SUCCESS_MESSAGE, - DemoActorService.class, - true, - 60000); - - demoActor = buildDemoActorProxy(deferClose(daprRun.newActorClient())); - daprClient = daprRun.newDaprClientBuilder().build(); - - toxiProxyRun = new ToxiProxyRun(daprRun, LATENCY, JITTER); - toxiProxyRun.start(); + var pair = startAppAndAttach( + "actor-sdk-resiliency-it", + DemoActorService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-sdk-resiliency-it") + .withNetworkAliases("dapr") + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + dapr = pair.dapr(); + app = pair.app(); + waitForActorsReady(dapr); + + demoActor = buildDemoActorProxy(newActorClient(dapr)); + daprClient = deferClose(newDaprClient(dapr)); + + toxiproxy = newToxiproxy(); + ToxiproxyClient toxiproxyClient = new ToxiproxyClient(toxiproxy.getHost(), toxiproxy.getControlPort()); + Proxy grpcProxy = toxiproxyClient.createProxy("dapr_grpc", "0.0.0.0:8666", "dapr:50001"); + grpcProxy.toxics() + .latency("latency", ToxicDirection.DOWNSTREAM, LATENCY.toMillis()) + .setJitter(JITTER.toMillis()); toxiDemoActor = buildDemoActorProxy( - toxiProxyRun.newActorClient(new ResiliencyOptions().setTimeout(TIMEOUT))); + newActorClientViaProxy(new ResiliencyOptions().setTimeout(TIMEOUT))); resilientDemoActor = buildDemoActorProxy( - toxiProxyRun.newActorClient(new ResiliencyOptions().setTimeout(TIMEOUT).setMaxRetries(MAX_RETRIES))); + newActorClientViaProxy(new ResiliencyOptions().setTimeout(TIMEOUT).setMaxRetries(MAX_RETRIES))); oneRetryDemoActor = buildDemoActorProxy( - toxiProxyRun.newActorClient(new ResiliencyOptions().setTimeout(TIMEOUT).setMaxRetries(1))); + newActorClientViaProxy(new ResiliencyOptions().setTimeout(TIMEOUT).setMaxRetries(1))); + } + + private static ActorClient newActorClientViaProxy(ResiliencyOptions resiliencyOptions) { + Map, String> overrides = new HashMap<>(); + int mappedPort = toxiproxy.getMappedPort(8666); + overrides.put(Properties.GRPC_ENDPOINT, "127.0.0.1:" + mappedPort); + overrides.put(Properties.GRPC_PORT, String.valueOf(mappedPort)); + ActorClient client = new ActorClient(new Properties(overrides), resiliencyOptions); + deferClose(client); + return client; } private static DemoActor buildDemoActorProxy(ActorClient actorClient) { @@ -94,13 +123,6 @@ private static DemoActor buildDemoActorProxy(ActorClient actorClient) { return builder.build(ACTOR_ID); } - @AfterAll - public static void tearDown() throws Exception { - if (toxiProxyRun != null) { - toxiProxyRun.stop(); - } - } - @Test @Disabled("Flaky when running on GitHub actions") public void retryAndTimeout() { diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorStateIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorStateIT.java index ffd5d3c3dd..991fbca576 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/ActorStateIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorStateIT.java @@ -8,7 +8,7 @@ * 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. + * limitations under the License. */ package io.dapr.it.actors; @@ -16,10 +16,12 @@ import io.dapr.actors.ActorId; import io.dapr.actors.client.ActorProxy; import io.dapr.actors.client.ActorProxyBuilder; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; +import io.dapr.it.AppRun; import io.dapr.it.actors.services.springboot.StatefulActor; import io.dapr.it.actors.services.springboot.StatefulActorService; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -29,21 +31,30 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; -public class ActorStateIT extends BaseIT { +public class ActorStateIT extends BaseContainerIT { private static Logger logger = LoggerFactory.getLogger(ActorStateIT.class); + private static DaprContainer dapr; + private static AppRun app; + + @BeforeAll + public static void start() throws Exception { + var pair = startAppAndAttach( + "actor-state-it", + StatefulActorService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-state-it") + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + dapr = pair.dapr(); + app = pair.app(); + waitForActorsReady(dapr); + } + @Test public void writeReadState() throws Exception { - logger.debug("Starting actor runtime ..."); - // The call below will fail if service cannot start successfully. - DaprRun run = startDaprApp( - this.getClass().getSimpleName(), - StatefulActorService.SUCCESS_MESSAGE, - StatefulActorService.class, - true, - 60000); - String message = "This is a message to be saved and retrieved."; String name = "Jon Doe"; byte[] bytes = new byte[] { 0x1 }; @@ -52,12 +63,12 @@ public void writeReadState() throws Exception { String actorType = "StatefulActorTest"; logger.debug("Building proxy ..."); ActorProxyBuilder proxyBuilder = - new ActorProxyBuilder(actorType, ActorProxy.class, deferClose(run.newActorClient())); + new ActorProxyBuilder(actorType, ActorProxy.class, newActorClient(dapr)); ActorProxy proxy = proxyBuilder.build(actorId); // waiting for actor to be activated Thread.sleep(5000); - + // Validate conditional read works. callWithRetry(() -> { logger.debug("Invoking readMessage where data is not present yet ... "); @@ -126,23 +137,15 @@ public void writeReadState() throws Exception { logger.debug("Waiting, so actor can be deactivated ..."); Thread.sleep(10000); - logger.debug("Stopping service ..."); - run.stop(); - - logger.debug("Starting service ..."); - DaprRun run2 = startDaprApp( - this.getClass().getSimpleName(), - StatefulActorService.SUCCESS_MESSAGE, - StatefulActorService.class, - true, - 60000); + logger.debug("Restarting app ..."); + restartApp(app); // Need new proxy builder because the proxy builder holds the channel. - proxyBuilder = new ActorProxyBuilder(actorType, ActorProxy.class, deferClose(run2.newActorClient())); + proxyBuilder = new ActorProxyBuilder(actorType, ActorProxy.class, newActorClient(dapr)); ActorProxy newProxy = proxyBuilder.build(actorId); // waiting for actor to be activated - Thread.sleep(2000); + Thread.sleep(2000); callWithRetry(() -> { logger.debug("Invoking readMessage where data is not cached ... "); diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorTimerRecoveryIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorTimerRecoveryIT.java index 21910376fa..273a726550 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/ActorTimerRecoveryIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorTimerRecoveryIT.java @@ -8,7 +8,7 @@ * 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. + * limitations under the License. */ package io.dapr.it.actors; @@ -17,10 +17,10 @@ import io.dapr.actors.client.ActorProxy; import io.dapr.actors.client.ActorProxyBuilder; import io.dapr.it.AppRun; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; import io.dapr.it.actors.app.MyActorService; -import org.apache.commons.lang3.tuple.ImmutablePair; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,30 +35,41 @@ import static io.dapr.it.actors.MyActorTestUtils.validateMethodCalls; import static org.junit.jupiter.api.Assertions.assertNotEquals; -public class ActorTimerRecoveryIT extends BaseIT { +public class ActorTimerRecoveryIT extends BaseContainerIT { private static final Logger logger = LoggerFactory.getLogger(ActorTimerRecoveryIT.class); private static final String METHOD_NAME = "clock"; + private static DaprContainer dapr; + private static AppRun app; + + @BeforeAll + public static void start() throws Exception { + var pair = startAppAndAttach( + "actor-timer-recovery-it", + MyActorService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-timer-recovery-it") + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + dapr = pair.dapr(); + app = pair.app(); + waitForActorsReady(dapr); + } + /** * Create an actor, register a timer, validates its content, restarts the Actor and confirms timer continues. * @throws Exception This test is not expected to throw. Thrown exceptions are bugs. */ @Test public void timerRecoveryTest() throws Exception { - ImmutablePair runs = startSplitDaprAndApp( - ActorTimerRecoveryIT.class.getSimpleName(), - "Started MyActorService", - MyActorService.class, - true, - 60000); - - String actorType="MyActorTest"; + String actorType = "MyActorTest"; logger.debug("Creating proxy builder"); ActorProxyBuilder proxyBuilder = - new ActorProxyBuilder(actorType, ActorProxy.class, deferClose(runs.right.newActorClient())); + new ActorProxyBuilder(actorType, ActorProxy.class, newActorClient(dapr)); logger.debug("Creating actorId"); ActorId actorId = new ActorId(UUID.randomUUID().toString()); logger.debug("Building proxy"); @@ -77,11 +88,10 @@ public void timerRecoveryTest() throws Exception { }, 30000); // Restarts app only. - runs.left.stop(); // Cannot sleep between app's stop and start since it can trigger unhealthy actor in runtime and lose timers. // Timers will survive only if the restart is "quick" and survives the runtime's actor health check. // Starting in 1.13, sidecar is more sensitive to an app restart and will not keep actors active for "too long". - runs.left.start(); + restartApp(app); final List newLogs = new ArrayList<>(); callWithRetry(() -> { diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java index 0bb2bac0e2..94f9959822 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java @@ -14,10 +14,7 @@ package io.dapr.it.actors.app; import io.dapr.actors.runtime.ActorRuntime; -import io.dapr.it.DaprRunConfig; -// Enable dapr-api-token once runtime supports it in standalone mode. -@DaprRunConfig(enableDaprApiToken = false) public class MyActorService { public static final String SUCCESS_MESSAGE = "dapr initialized. Status: Running"; diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/StatefulActorService.java b/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/StatefulActorService.java index c96fa05d06..3a15a0c6b8 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/StatefulActorService.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/StatefulActorService.java @@ -14,12 +14,10 @@ package io.dapr.it.actors.services.springboot; import io.dapr.actors.runtime.ActorRuntime; -import io.dapr.it.DaprRunConfig; import io.dapr.serializer.DefaultObjectSerializer; import java.time.Duration; -@DaprRunConfig(enableDaprApiToken = false) public class StatefulActorService { public static final String SUCCESS_MESSAGE = "dapr initialized. Status: Running"; diff --git a/sdk-tests/src/test/java/io/dapr/it/binding/http/BindingIT.java b/sdk-tests/src/test/java/io/dapr/it/binding/http/BindingIT.java index 2bdf7bb3ca..438e8ac374 100644 --- a/sdk-tests/src/test/java/io/dapr/it/binding/http/BindingIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/binding/http/BindingIT.java @@ -15,14 +15,13 @@ import com.fasterxml.jackson.databind.ObjectMapper; import io.dapr.client.DaprClient; -import io.dapr.client.DaprClientBuilder; import io.dapr.client.domain.HttpExtension; import io.dapr.exceptions.DaprException; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; +import io.dapr.it.AppRun; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; import org.junit.jupiter.api.Test; -import java.util.Collections; import java.util.List; import java.util.Map; @@ -34,14 +33,18 @@ /** * Service for input and output binding example. */ -public class BindingIT extends BaseIT { +public class BindingIT extends BaseContainerIT { + + private static final String APP_NAME = "bindingit-grpc"; @Test public void httpOutputBindingError() throws Exception { - var run = startDaprApp( - this.getClass().getSimpleName() + "-httpoutputbinding-exception", - 60000); - try(DaprClient client = run.newDaprClientBuilder().build()) { + DaprContainer dapr = daprBuilder("bindingit-httpoutputbinding-exception") + .withComponent(httpBinding("github-http-binding-404", "https://api.github.com/unknown_path", true)); + dapr.start(); + deferStop(dapr); + + try (DaprClient client = newDaprClient(dapr)) { // Validate error message callWithRetry(() -> { System.out.println("Checking exception handling for output binding ..."); @@ -61,10 +64,13 @@ public void httpOutputBindingError() throws Exception { @Test public void httpOutputBindingErrorIgnoredByComponent() throws Exception { - var run = startDaprApp( - this.getClass().getSimpleName() + "-httpoutputbinding-ignore-error", - 60000); - try(DaprClient client = run.newDaprClientBuilder().build()) { + DaprContainer dapr = daprBuilder("bindingit-httpoutputbinding-ignore-error") + .withComponent( + httpBinding("github-http-binding-404-success", "https://api.github.com/unknown_path", false)); + dapr.start(); + deferStop(dapr); + + try (DaprClient client = newDaprClient(dapr)) { // Validate error message callWithRetry(() -> { System.out.println("Checking exception handling for output binding ..."); @@ -86,20 +92,23 @@ public void httpOutputBindingErrorIgnoredByComponent() throws Exception { @Test public void inputOutputBinding() throws Exception { - DaprRun daprRun = startDaprApp( - this.getClass().getSimpleName() + "-grpc", - InputBindingService.SUCCESS_MESSAGE, - InputBindingService.class, - true, - 60000); - - var bidingName = "sample123"; + var bindingName = "sample123"; - try(DaprClient client = daprRun.newDaprClientBuilder().build()) { + var pair = startAppAndAttach( + APP_NAME, + InputBindingService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder(APP_NAME) + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(kafkaBinding(bindingName))); + DaprContainer dapr = pair.dapr(); + + try (DaprClient client = newDaprClient(dapr)) { callWithRetry(() -> { System.out.println("Checking if input binding is up before publishing events ..."); client.invokeBinding( - bidingName, "create", "ping").block(); + bindingName, "create", "ping").block(); try { Thread.sleep(1000); @@ -108,7 +117,7 @@ public void inputOutputBinding() throws Exception { throw new RuntimeException(e); } - client.invokeMethod(daprRun.getAppName(), "initialized", "", HttpExtension.GET).block(); + client.invokeMethod(APP_NAME, "initialized", "", HttpExtension.GET).block(); }, 120000); // This is an example of sending data in a user-defined object. The input binding will receive: @@ -118,21 +127,21 @@ public void inputOutputBinding() throws Exception { System.out.println("sending first message"); client.invokeBinding( - bidingName, "create", myClass, Map.of("MyMetadata", "MyValue"), Void.class).block(); + bindingName, "create", myClass, Map.of("MyMetadata", "MyValue"), Void.class).block(); // This is an example of sending a plain string. The input binding will receive // cat final String m = "cat"; System.out.println("sending " + m); client.invokeBinding( - bidingName, "create", m, Map.of("MyMetadata", "MyValue"), Void.class).block(); + bindingName, "create", m, Map.of("MyMetadata", "MyValue"), Void.class).block(); // Metadata is not used by Kafka component, so it is not possible to validate. callWithRetry(() -> { System.out.println("Checking results ..."); final List messages = client.invokeMethod( - daprRun.getAppName(), + APP_NAME, "messages", null, HttpExtension.GET, diff --git a/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java b/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java index e6ec233c40..de564c1e3b 100644 --- a/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java @@ -21,12 +21,16 @@ import io.dapr.it.AppRun; import io.dapr.it.DaprPorts; import io.dapr.it.Stoppable; +import io.dapr.it.testcontainers.ContainerConstants; import io.dapr.testcontainers.Component; import io.dapr.testcontainers.DaprContainer; import io.dapr.testcontainers.DaprLogLevel; +import io.dapr.testcontainers.DaprPlacementContainer; +import io.dapr.testcontainers.DaprSchedulerContainer; import io.dapr.testcontainers.wait.strategy.DaprWait; import org.junit.jupiter.api.AfterAll; import org.testcontainers.Testcontainers; +import org.testcontainers.toxiproxy.ToxiproxyContainer; import java.util.Deque; import java.util.HashMap; @@ -82,9 +86,32 @@ protected static DaprContainer daprBuilder(String appName) { // errors. Without this, the container's stdout is consumed by Testcontainers // and we have no insight when actor registration or component init fails. .withLogConsumer(frame -> System.out.print("[daprd] " + frame.getUtf8String())) - // Reuses the placement sidecar container within this JVM (Testcontainers manages it); - // orthogonal to SharedTestInfra's Redis `withReuse(true)`. - .withReusablePlacement(true); + // Wire every daprd to the ONE shared placement + scheduler on the shared network. + // daprd resolves its control plane by DNS name ("placement"/"scheduler"), so there + // must be exactly one container answering each alias. DaprContainer would otherwise + // auto-create its own placement/scheduler per instance and never stop them (there is + // no stop() override), leaving several containers sharing the "placement"/"scheduler" + // aliases on the shared network -- Docker DNS then round-robins daprd to an arbitrary + // (often empty) one, breaking multi-sidecar failover and sidecar-restart reminder + // recovery. The JVM-singletons make each alias resolve to exactly one container and + // eliminate the per-container control-plane leak. + .withPlacementContainer(SharedTestInfra.placement()) + .withSchedulerContainer(SharedTestInfra.scheduler()); + } + + // ---------- Shared control plane (multi-sidecar ITs) ---------- + + /** The JVM-wide shared placement. Multi-sidecar ITs (e.g. failover) pass this to every + * daprd via {@code withPlacementContainer} so all sidecars share one placement -- the + * same singleton {@link #daprBuilder} wires by default. Not stopped per class; it lives + * for the JVM (see {@link SharedTestInfra#placement()}). */ + protected static DaprPlacementContainer startSharedPlacement() { + return SharedTestInfra.placement(); + } + + /** The JVM-wide shared scheduler (owns actor reminders). See {@link #startSharedPlacement()}. */ + protected static DaprSchedulerContainer startSharedScheduler() { + return SharedTestInfra.scheduler(); } // ---------- App lifecycle ---------- @@ -187,6 +214,51 @@ protected static void waitForActorsReady(DaprContainer dapr) { DaprWait.forActors().waitUntilReady(dapr); } + /** + * Restarts the app subprocess on its same pre-allocated port. The daprd + * container stays up and reconnects to the app via + * {@code host.testcontainers.internal:appPort}. Because {@link #daprBuilder} + * configures NO app-health-check, daprd does not deactivate actors during the + * gap, so in-memory timers survive (matching the legacy + * {@code @DaprRunConfig(enableAppHealthCheck=false)}). There is intentionally + * no sleep between stop and start — {@code ActorTimerRecoveryIT} relies on a + * quick restart. + */ + protected static void restartApp(AppRun app) throws Exception { + app.stop(); + app.start(); + } + + /** + * Restarts the daprd container in place and re-waits for readiness. Placement + * and scheduler are NOT recreated on the second start (their DaprContainer + * fields are non-null), so a persisted actor reminder survives. Pinned host + * ports re-bind, so the app's DAPR_HTTP_PORT/DAPR_GRPC_PORT and any DaprClient + * remain valid. + */ + protected static void restartSidecar(DaprContainer dapr) throws Exception { + dapr.stop(); + dapr.start(); + try (DaprClient client = newDaprClient(dapr)) { + client.waitForSidecar(30_000).block(); + } + waitForActorsReady(dapr); + } + + /** Starts a Toxiproxy container on the shared network and registers it for + * @AfterAll cleanup. Callers create proxies via a ToxiproxyClient against + * {@code getHost()}/{@code getControlPort()} (the control channel), then point + * Dapr clients at {@code getMappedPort()} of a proxy created on a + * fixed listen port (e.g. 8666). Mirrors SdkResiliencyIT. */ + protected static ToxiproxyContainer newToxiproxy() { + ToxiproxyContainer toxiproxy = + new ToxiproxyContainer(ContainerConstants.TOXI_PROXY_IMAGE_TAG) + .withNetwork(SharedTestInfra.network()); + toxiproxy.start(); + deferStop(toxiproxy); + return toxiproxy; + } + // ---------- DaprClient / ActorClient factories ---------- protected static DaprClient newDaprClient(DaprContainer dapr) { @@ -264,6 +336,33 @@ protected static Component mongoStateStore(String name) { )); } + /** + * Kafka-backed input/output binding. Lazily starts the shared Kafka container + * before returning the component. The {@code {appID}} placeholders in + * {@code topics}/{@code publishTopic}/{@code consumerGroup} resolve to the + * owning daprd's app-id, so topic and consumer group are self-consistent + * without needing a {@code scopes} entry. + */ + protected static Component kafkaBinding(String name) { + SharedTestInfra.kafka(); // ensure Kafka is up before DaprContainer needs it + return new Component(name, "bindings.kafka", "v1", Map.of( + "brokers", SharedTestInfra.kafkaInternalBroker(), + "topics", "topic-{appID}", + "publishTopic", "topic-{appID}", + "consumerGroup", "{appID}", + "authRequired", "false", + "initialOffset", "oldest" + )); + } + + /** HTTP output binding pointed at an arbitrary URL. */ + protected static Component httpBinding(String name, String url, boolean errorIfNot2xx) { + return new Component(name, "bindings.http", "v1", Map.of( + "url", url, + "errorIfNot2XX", Boolean.toString(errorIfNot2xx) + )); + } + // ---------- Cleanup ---------- protected static T deferClose(T object) { diff --git a/sdk-tests/src/test/java/io/dapr/it/containers/SharedTestInfra.java b/sdk-tests/src/test/java/io/dapr/it/containers/SharedTestInfra.java index e778d1f7d9..0fa55770b7 100644 --- a/sdk-tests/src/test/java/io/dapr/it/containers/SharedTestInfra.java +++ b/sdk-tests/src/test/java/io/dapr/it/containers/SharedTestInfra.java @@ -13,6 +13,9 @@ package io.dapr.it.containers; +import io.dapr.testcontainers.DaprContainerConstants; +import io.dapr.testcontainers.DaprPlacementContainer; +import io.dapr.testcontainers.DaprSchedulerContainer; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.Network; import org.testcontainers.utility.DockerImageName; @@ -29,11 +32,17 @@ public final class SharedTestInfra { private static final String REDIS_NETWORK_ALIAS = "redis"; private static final String ZIPKIN_NETWORK_ALIAS = "zipkin"; private static final String MONGO_NETWORK_ALIAS = "mongo"; + private static final String KAFKA_NETWORK_ALIAS = "kafka"; + private static final String PLACEMENT_NETWORK_ALIAS = "placement"; + private static final String SCHEDULER_NETWORK_ALIAS = "scheduler"; private static volatile Network network; private static volatile GenericContainer redis; private static volatile GenericContainer zipkin; private static volatile GenericContainer mongo; + private static volatile org.testcontainers.kafka.KafkaContainer kafka; + private static volatile DaprPlacementContainer placement; + private static volatile DaprSchedulerContainer scheduler; private SharedTestInfra() {} @@ -91,4 +100,59 @@ public static synchronized GenericContainer mongo() { public static String mongoInternalHost() { return MONGO_NETWORK_ALIAS + ":27017"; } + + public static synchronized org.testcontainers.kafka.KafkaContainer kafka() { + if (kafka == null) { + kafka = new org.testcontainers.kafka.KafkaContainer("apache/kafka:3.8.0") + .withNetwork(network()) + .withNetworkAliases(KAFKA_NETWORK_ALIAS) + .withListener(KAFKA_NETWORK_ALIAS + ":19092") + .withReuse(true); + kafka.start(); + } + return kafka; + } + + public static String kafkaInternalBroker() { + return KAFKA_NETWORK_ALIAS + ":19092"; + } + + /** + * The single Dapr placement service shared by every daprd in the JVM. + * + *

daprd resolves its placement service by DNS name ({@code placement:50005}), so + * there must be exactly ONE container answering to the {@code "placement"} alias on the + * shared network. If several exist (which happens when each {@code DaprContainer} + * auto-creates its own, never-stopped placement), Docker DNS round-robins daprd to an + * arbitrary one -- often an empty placement that never received this test's actor-host + * registrations -- producing "did not find address for actor" in multi-sidecar failover + * and losing the actor host across a sidecar restart. Sharing this singleton (wired via + * {@code DaprContainer.withPlacementContainer}) guarantees a single DNS answer. + */ + public static synchronized DaprPlacementContainer placement() { + if (placement == null) { + placement = new DaprPlacementContainer(DaprContainerConstants.DAPR_PLACEMENT_IMAGE_TAG) + .withNetwork(network()) + .withNetworkAliases(PLACEMENT_NETWORK_ALIAS) + .withReuse(true); + placement.start(); + } + return placement; + } + + /** + * The single Dapr scheduler service shared by every daprd in the JVM. Owns actor + * reminders in Dapr 1.18+, so it must stay up across a daprd restart for reminder + * recovery to work. Same single-DNS-answer requirement as {@link #placement()}. + */ + public static synchronized DaprSchedulerContainer scheduler() { + if (scheduler == null) { + scheduler = new DaprSchedulerContainer(DaprContainerConstants.DAPR_SCHEDULER_IMAGE_TAG) + .withNetwork(network()) + .withNetworkAliases(SCHEDULER_NETWORK_ALIAS) + .withReuse(true); + scheduler.start(); + } + return scheduler; + } } diff --git a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java b/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java index ac7b157af6..90d3739c82 100644 --- a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java +++ b/sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java @@ -14,7 +14,6 @@ package io.dapr.it.methodinvoke.grpc; import io.dapr.grpc.GrpcHealthCheckService; -import io.dapr.it.DaprRunConfig; import io.dapr.it.MethodInvokeServiceGrpc; import io.grpc.Server; import io.grpc.ServerBuilder; @@ -28,9 +27,6 @@ import static io.dapr.it.MethodInvokeServiceProtos.SleepRequest; import static io.dapr.it.MethodInvokeServiceProtos.SleepResponse; -@DaprRunConfig( - enableAppHealthCheck = true -) public class MethodInvokeService { private static final long STARTUP_DELAY_SECONDS = 10; diff --git a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/http/MethodInvokeService.java b/sdk-tests/src/test/java/io/dapr/it/methodinvoke/http/MethodInvokeService.java index 8b4858fc21..5d0e1cbbca 100644 --- a/sdk-tests/src/test/java/io/dapr/it/methodinvoke/http/MethodInvokeService.java +++ b/sdk-tests/src/test/java/io/dapr/it/methodinvoke/http/MethodInvokeService.java @@ -13,7 +13,6 @@ package io.dapr.it.methodinvoke.http; -import io.dapr.it.DaprRunConfig; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -21,9 +20,6 @@ /** * Service for subscriber. */ -@DaprRunConfig( - enableAppHealthCheck = true -) @SpringBootApplication public class MethodInvokeService { diff --git a/sdk-tests/src/test/java/io/dapr/it/resiliency/WaitForSidecarIT.java b/sdk-tests/src/test/java/io/dapr/it/resiliency/WaitForSidecarIT.java index c7dd5ccc56..00c6bfc64e 100644 --- a/sdk-tests/src/test/java/io/dapr/it/resiliency/WaitForSidecarIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/resiliency/WaitForSidecarIT.java @@ -8,65 +8,91 @@ * 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. + * limitations under the License. */ package io.dapr.it.resiliency; -import io.dapr.it.BaseIT; -import io.dapr.it.DaprRun; -import io.dapr.it.ToxiProxyRun; +import eu.rekawek.toxiproxy.Proxy; +import eu.rekawek.toxiproxy.ToxiproxyClient; +import eu.rekawek.toxiproxy.model.ToxicDirection; +import eu.rekawek.toxiproxy.model.toxic.Latency; +import io.dapr.client.DaprClient; +import io.dapr.client.DaprClientBuilder; +import io.dapr.config.Properties; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.testcontainers.toxiproxy.ToxiproxyContainer; +import java.io.IOException; import java.time.Duration; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; /** - * Test SDK resiliency. + * Test SDK resiliency around {@code waitForSidecar}. */ -public class WaitForSidecarIT extends BaseIT { +public class WaitForSidecarIT extends BaseContainerIT { // Use a number large enough to make sure it will respect the entire timeout. private static final Duration LATENCY = Duration.ofSeconds(5); - private static final Duration JITTER = Duration.ofSeconds(0); + private static DaprContainer dapr; - private static DaprRun daprRun; + private static ToxiproxyContainer toxiproxy; - private static ToxiProxyRun toxiProxyRun; + private static Proxy proxy; - private static DaprRun daprNotRunning; + private static DaprContainer daprNotRunning; + + private static int daprNotRunningHttpPort; @BeforeAll - public static void init() throws Exception { - daprRun = startDaprApp(WaitForSidecarIT.class.getSimpleName(), 5000); - daprNotRunning = startDaprApp(WaitForSidecarIT.class.getSimpleName() + "NotRunning", 5000); + public static void init() throws IOException { + dapr = daprBuilder("wait-for-sidecar-it") + .withNetworkAliases("dapr"); + dapr.start(); + deferStop(dapr); + + toxiproxy = newToxiproxy(); + ToxiproxyClient toxiproxyClient = new ToxiproxyClient(toxiproxy.getHost(), toxiproxy.getControlPort()); + proxy = toxiproxyClient.createProxy("dapr", "0.0.0.0:8666", "dapr:3500"); + + // A second sidecar that is started and then stopped, so a client pointed at its + // (now-dead) mapped port cannot possibly get a response. Deterministic stand-in + // for a "Dapr is not running" target. + daprNotRunning = daprBuilder("wait-for-sidecar-it-not-running") + .withNetworkAliases("dapr-not-running"); + daprNotRunning.start(); + daprNotRunningHttpPort = daprNotRunning.getHttpPort(); daprNotRunning.stop(); - - toxiProxyRun = new ToxiProxyRun(daprRun, LATENCY, JITTER); - toxiProxyRun.start(); } @Test public void waitSucceeds() throws Exception { - try(var client = daprRun.newDaprClient()) { + try (DaprClient client = newDaprClient(dapr)) { client.waitForSidecar(5000).block(); } } @Test - public void waitTimeout() { - int timeoutInMillis = (int)LATENCY.minusMillis(100).toMillis(); + public void waitTimeout() throws IOException { + int timeoutInMillis = (int) LATENCY.minusMillis(100).toMillis(); long started = System.currentTimeMillis(); - assertThrows(RuntimeException.class, () -> { - try(var client = toxiProxyRun.newDaprClientBuilder().build()) { - client.waitForSidecar(timeoutInMillis).block(); - } - }); + Latency latency = proxy.toxics().latency("latency", ToxicDirection.DOWNSTREAM, LATENCY.toMillis()); + try { + assertThrows(RuntimeException.class, () -> { + try (DaprClient client = viaProxyClientBuilder().build()) { + client.waitForSidecar(timeoutInMillis).block(); + } + }); + } finally { + latency.remove(); + } long duration = System.currentTimeMillis() - started; @@ -75,11 +101,16 @@ public void waitTimeout() { @Test public void waitSlow() throws Exception { - int timeoutInMillis = (int)LATENCY.plusMillis(100).toMillis(); + int timeoutInMillis = (int) LATENCY.plusMillis(100).toMillis(); long started = System.currentTimeMillis(); - try(var client = toxiProxyRun.newDaprClientBuilder().build()) { + Latency latency = proxy.toxics().latency("latency", ToxicDirection.DOWNSTREAM, LATENCY.toMillis()); + try { + try (DaprClient client = viaProxyClientBuilder().build()) { client.waitForSidecar(timeoutInMillis).block(); + } + } finally { + latency.remove(); } long duration = System.currentTimeMillis() - started; @@ -95,7 +126,7 @@ public void waitNotRunningTimeout() { long started = System.currentTimeMillis(); assertThrows(RuntimeException.class, () -> { - try(var client = daprNotRunning.newDaprClientBuilder().build()) { + try (DaprClient client = notRunningClientBuilder().build()) { client.waitForSidecar(timeoutMilliseconds).block(); } }); @@ -104,4 +135,16 @@ public void waitNotRunningTimeout() { assertThat(duration).isGreaterThanOrEqualTo(timeoutMilliseconds); } + + private static DaprClientBuilder viaProxyClientBuilder() { + return new DaprClientBuilder() + .withPropertyOverride(Properties.HTTP_ENDPOINT, "http://localhost:" + toxiproxy.getMappedPort(8666)) + .withPropertyOverride(Properties.GRPC_ENDPOINT, "http://localhost:" + toxiproxy.getMappedPort(8666)); + } + + private static DaprClientBuilder notRunningClientBuilder() { + return new DaprClientBuilder() + .withPropertyOverride(Properties.HTTP_ENDPOINT, "http://localhost:" + daprNotRunningHttpPort) + .withPropertyOverride(Properties.GRPC_ENDPOINT, "http://localhost:" + daprNotRunningHttpPort); + } } diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java b/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java index 1aa660d880..fc2e59e050 100644 --- a/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java @@ -15,7 +15,6 @@ import com.google.protobuf.Any; import io.dapr.grpc.GrpcHealthCheckService; -import io.dapr.it.DaprRunConfig; import io.dapr.v1.AppCallbackGrpc; import io.dapr.v1.CommonProtos; import io.grpc.Server; @@ -27,9 +26,6 @@ import static io.dapr.it.MethodInvokeServiceProtos.SleepRequest; import static io.dapr.it.MethodInvokeServiceProtos.SleepResponse; -@DaprRunConfig( - enableAppHealthCheck = true -) public class Service { public static final String SUCCESS_MESSAGE = "application discovered on port "; diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java index c66e795887..5c0447ab2a 100644 --- a/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java @@ -27,8 +27,16 @@ public class OpenTelemetryConfig { public static final String SERVICE_NAME = "integration testing service over http"; @Bean - public OpenTelemetry initOpenTelemetry() throws InterruptedException { - return io.dapr.it.tracing.OpenTelemetry.createOpenTelemetry(SERVICE_NAME); + public OpenTelemetry initOpenTelemetry() { + // Use the explicit-endpoint overload so this bean does NOT block on a Zipkin readiness + // probe during app startup. This app runs as a host subprocess and only needs a propagator + // for context extraction; it exports no spans that the test asserts on (the validated + // "calllocal//sleep" span is emitted by the Dapr sidecar, which exports to Zipkin over + // the container network). The legacy createOpenTelemetry(SERVICE_NAME) overload probed + // 127.0.0.1:9411, but Zipkin now runs as a Testcontainer on a random mapped port, so the + // probe failed and crashed startup -- startAppAndAttach then saw "connection refused". + return io.dapr.it.tracing.OpenTelemetry.createOpenTelemetry( + SERVICE_NAME, "http://localhost:9411/api/v2/spans"); } @Bean diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java index ba43173058..124483d264 100644 --- a/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java @@ -13,7 +13,6 @@ package io.dapr.it.tracing.http; -import io.dapr.it.DaprRunConfig; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -21,9 +20,6 @@ /** * Service for subscriber. */ -@DaprRunConfig( - enableAppHealthCheck = true -) @SpringBootApplication public class Service {