diff --git a/sdk-workflows/pom.xml b/sdk-workflows/pom.xml index 0d52f5d9b1..274c989424 100644 --- a/sdk-workflows/pom.xml +++ b/sdk-workflows/pom.xml @@ -46,6 +46,11 @@ io.opentelemetry opentelemetry-api + + io.grpc + grpc-testing + test + diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java b/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java new file mode 100644 index 0000000000..352164fe34 --- /dev/null +++ b/sdk-workflows/src/main/java/io/dapr/workflows/internal/GrpcChannelKeepalive.java @@ -0,0 +1,95 @@ +/* + * 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.workflows.internal; + +import com.google.protobuf.Empty; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * Application-level keepalive for a gRPC channel to the Dapr sidecar. + * + *

Periodically invokes the sidecar's {@code hello} RPC so that intermediaries + * (e.g. AWS ALBs) that do not treat HTTP/2 PING frames as connection activity + * never see the connection as idle and close it.

+ */ +public final class GrpcChannelKeepalive implements AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(GrpcChannelKeepalive.class); + private static final long KEEPALIVE_DEADLINE_SECONDS = 5; + + private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub stub; + private final String threadName; + private final Duration interval; + private ScheduledExecutorService scheduler; + + /** + * Creates a keepalive for the given channel. No pings are sent until {@link #start()}. + * + * @param channel channel to the Dapr sidecar to keep alive. + * @param threadName name of the keepalive thread, to tell instances apart in thread dumps. + * @param interval delay between pings. + */ + public GrpcChannelKeepalive(Channel channel, String threadName, Duration interval) { + this.stub = TaskHubSidecarServiceGrpc.newBlockingStub(channel); + this.threadName = threadName; + this.interval = interval; + } + + /** + * Starts the keepalive loop. Does nothing if already started. + */ + public synchronized void start() { + if (this.scheduler != null) { + return; + } + ScheduledExecutorService newScheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, this.threadName); + thread.setDaemon(true); + return thread; + }); + // The deadline must be applied per ping: Deadline.after snapshots the clock when + // withDeadlineAfter is invoked, so hoisting it onto the cached stub would make + // every ping after the first fail immediately with DEADLINE_EXCEEDED. + // Catch Throwable, not just RuntimeException: any throwable escaping a periodic + // task silently cancels all future runs. + newScheduler.scheduleWithFixedDelay(() -> { + try { + this.stub.withDeadlineAfter(KEEPALIVE_DEADLINE_SECONDS, TimeUnit.SECONDS) + .hello(Empty.getDefaultInstance()); + } catch (Throwable e) { + LOGGER.debug("Sidecar keepalive ping failed", e); + } + }, this.interval.toMillis(), this.interval.toMillis(), TimeUnit.MILLISECONDS); + this.scheduler = newScheduler; + } + + /** + * Stops the keepalive loop. Does nothing if not started. + */ + @Override + public synchronized void close() { + if (this.scheduler != null) { + this.scheduler.shutdownNow(); + this.scheduler = null; + } + } +} diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java index f5d47f3e79..5f7221f6ff 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java @@ -14,8 +14,11 @@ package io.dapr.workflows.runtime; import io.dapr.durabletask.DurableTaskGrpcWorker; +import io.dapr.workflows.internal.GrpcChannelKeepalive; import io.grpc.ManagedChannel; +import javax.annotation.Nullable; + import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; @@ -27,6 +30,7 @@ public class WorkflowRuntime implements AutoCloseable { private final DurableTaskGrpcWorker worker; private final ManagedChannel managedChannel; private final ExecutorService executorService; + private final GrpcChannelKeepalive keepalive; /** * Constructor. @@ -38,9 +42,27 @@ public class WorkflowRuntime implements AutoCloseable { public WorkflowRuntime(DurableTaskGrpcWorker worker, ManagedChannel managedChannel, ExecutorService executorService) { + this(worker, managedChannel, executorService, null); + } + + /** + * Constructor. + * + * @param worker grpcWorker processing activities. + * @param managedChannel grpc channel. + * @param executorService executor service responsible for running the threads. + * @param keepalive application-level keepalive on the worker's channel, started with + * {@link #start()} and stopped on {@link #close()}. May be null when + * no keepalive is wanted. + */ + public WorkflowRuntime(DurableTaskGrpcWorker worker, + ManagedChannel managedChannel, + ExecutorService executorService, + @Nullable GrpcChannelKeepalive keepalive) { this.worker = worker; this.managedChannel = managedChannel; this.executorService = executorService; + this.keepalive = keepalive; } /** @@ -57,6 +79,9 @@ public void start() { * @param block block the thread if true */ public void start(boolean block) { + if (this.keepalive != null) { + this.keepalive.start(); + } if (block) { this.worker.startAndBlock(); } else { @@ -68,6 +93,9 @@ public void start(boolean block) { * {@inheritDoc} */ public void close() { + if (this.keepalive != null) { + this.keepalive.close(); + } this.shutDownWorkerPool(); this.closeSideCarChannel(); this.worker.close(); diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java index 3f901e7b0e..e812e55aad 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java @@ -21,6 +21,7 @@ import io.dapr.workflows.Workflow; import io.dapr.workflows.WorkflowActivity; import io.dapr.workflows.internal.ApiTokenClientInterceptor; +import io.dapr.workflows.internal.GrpcChannelKeepalive; import io.grpc.ClientInterceptor; import io.grpc.ManagedChannel; import org.apache.commons.lang3.StringUtils; @@ -43,6 +44,7 @@ public class WorkflowRuntimeBuilder { private final Set workflowSet = Collections.synchronizedSet(new HashSet<>()); private final DurableTaskGrpcWorkerBuilder builder; private final ManagedChannel managedChannel; + private final Properties properties; private ExecutorService executorService; /** @@ -69,6 +71,7 @@ private WorkflowRuntimeBuilder(Properties properties, Logger logger) { this.workflowApiTokenInterceptor = new ApiTokenClientInterceptor(properties); this.managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, workflowApiTokenInterceptor); this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(this.managedChannel); + this.properties = properties; this.logger = logger; } @@ -82,9 +85,14 @@ public WorkflowRuntime build() { synchronized (WorkflowRuntime.class) { this.executorService = this.executorService == null ? Executors.newCachedThreadPool() : this.executorService; if (instance == null) { + GrpcChannelKeepalive keepalive = null; + if (this.properties.getValue(Properties.WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED)) { + keepalive = new GrpcChannelKeepalive(this.managedChannel, "dapr-workflow-runtime-keepalive", + this.properties.getValue(Properties.WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS)); + } instance = new WorkflowRuntime( this.builder.withExecutorService(this.executorService).build(), - this.managedChannel, this.executorService); + this.managedChannel, this.executorService, keepalive); } } } diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java new file mode 100644 index 0000000000..534d434a7d --- /dev/null +++ b/sdk-workflows/src/test/java/io/dapr/workflows/internal/GrpcChannelKeepaliveTest.java @@ -0,0 +1,148 @@ +/* + * 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.workflows.internal; + +import com.google.protobuf.Empty; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class GrpcChannelKeepaliveTest { + + private static final Duration TEST_INTERVAL = Duration.ofMillis(50); + + private final AtomicInteger helloCount = new AtomicInteger(); + private final AtomicBoolean failPings = new AtomicBoolean(false); + + private Server server; + private ManagedChannel channel; + + @BeforeEach + public void setUp() throws IOException { + String serverName = InProcessServerBuilder.generateName(); + this.server = InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void hello(Empty request, StreamObserver responseObserver) { + helloCount.incrementAndGet(); + if (failPings.get()) { + responseObserver.onError(Status.UNAVAILABLE.asRuntimeException()); + return; + } + responseObserver.onNext(Empty.getDefaultInstance()); + responseObserver.onCompleted(); + } + }) + .build() + .start(); + this.channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + } + + @AfterEach + public void tearDown() { + this.channel.shutdownNow(); + this.server.shutdownNow(); + } + + @Test + public void noPingsBeforeStart() throws InterruptedException { + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + Thread.sleep(TEST_INTERVAL.toMillis() * 4); + assertEquals(0, helloCount.get()); + } + } + + @Test + public void pingsPeriodically() throws InterruptedException { + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + keepalive.start(); + // A second start must not schedule a second ping loop. + keepalive.start(); + awaitHelloCountAtLeast(2); + } + } + + @Test + public void continuesPingingAfterFailures() throws InterruptedException { + failPings.set(true); + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL)) { + keepalive.start(); + awaitHelloCountAtLeast(2); + failPings.set(false); + awaitHelloCountAtLeast(helloCount.get() + 2); + } + } + + @Test + public void closeStopsPinging() throws InterruptedException { + GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, "test-keepalive", TEST_INTERVAL); + keepalive.start(); + awaitHelloCountAtLeast(1); + keepalive.close(); + // Allow an already in-flight ping to finish before snapshotting. + Thread.sleep(TEST_INTERVAL.toMillis() * 2); + int countAfterClose = helloCount.get(); + Thread.sleep(TEST_INTERVAL.toMillis() * 4); + assertEquals(countAfterClose, helloCount.get()); + } + + @Test + public void keepaliveThreadIsDaemonAndStopsOnClose() throws InterruptedException { + String threadName = "test-keepalive-lifecycle"; + try (GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, TEST_INTERVAL)) { + keepalive.start(); + awaitHelloCountAtLeast(1); + Thread thread = findThread(threadName); + assertTrue(thread != null && thread.isDaemon(), "expected a live daemon keepalive thread"); + } + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (findThread(threadName) != null && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertTrue(findThread(threadName) == null, "keepalive thread should terminate after close()"); + } + + private void awaitHelloCountAtLeast(int expected) throws InterruptedException { + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (helloCount.get() < expected && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertTrue(helloCount.get() >= expected, + "expected at least " + expected + " hello pings, got " + helloCount.get()); + } + + private static Thread findThread(String name) { + return Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().equals(name) && t.isAlive()) + .findFirst() + .orElse(null); + } +} diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java index 9ee2710e79..af534b836e 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeTest.java @@ -18,12 +18,18 @@ import io.dapr.durabletask.DurableTaskGrpcWorkerBuilder; import io.dapr.config.Properties; import io.dapr.utils.NetworkUtils; +import io.dapr.workflows.internal.GrpcChannelKeepalive; +import io.grpc.ManagedChannel; import org.junit.jupiter.api.Test; +import java.time.Duration; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; public class WorkflowRuntimeTest { @@ -35,6 +41,29 @@ public void startTest() { assertDoesNotThrow(() -> runtime.start(false)); } + @Test + public void startStartsKeepaliveAndCloseStopsIt() throws InterruptedException { + String threadName = "dapr-workflow-runtime-keepalive"; + DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build(); + ManagedChannel channel = NetworkUtils.buildGrpcManagedChannel(new Properties()); + GrpcChannelKeepalive keepalive = new GrpcChannelKeepalive(channel, threadName, Duration.ofSeconds(30)); + WorkflowRuntime runtime = new WorkflowRuntime(worker, channel, Executors.newCachedThreadPool(), keepalive); + assertFalse(keepaliveThreadAlive(threadName), "keepalive must stay inert until the runtime starts"); + runtime.start(false); + assertTrue(keepaliveThreadAlive(threadName), "keepalive thread expected after runtime start"); + assertDoesNotThrow(runtime::close); + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(5); + while (keepaliveThreadAlive(threadName) && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertFalse(keepaliveThreadAlive(threadName), "keepalive thread should terminate after close()"); + } + + private static boolean keepaliveThreadAlive(String threadName) { + return Thread.getAllStackTraces().keySet().stream() + .anyMatch(t -> t.getName().equals(threadName) && t.isAlive()); + } + @Test public void closeWithoutStarting() { DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build(); diff --git a/sdk/src/main/java/io/dapr/config/Properties.java b/sdk/src/main/java/io/dapr/config/Properties.java index 85110c1922..eb99fd2671 100644 --- a/sdk/src/main/java/io/dapr/config/Properties.java +++ b/sdk/src/main/java/io/dapr/config/Properties.java @@ -189,6 +189,31 @@ public class Properties { "DAPR_GRPC_KEEP_ALIVE_WITHOUT_CALLS", true); + /** + * Enables the application-level keepalive on the workflow runtime's gRPC channel. + * While the runtime is started, the SDK periodically invokes the sidecar's hello RPC + * so that intermediaries that do not treat HTTP/2 PING frames as connection activity + * (e.g. AWS ALBs) never see the connection as idle and close it. + * Environment variable: DAPR_WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED + * System property: dapr.workflows.runtime.app.keep.alive.enabled + * Default: true + */ + public static final Property WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED = new BooleanProperty( + "dapr.workflows.runtime.app.keep.alive.enabled", + "DAPR_WORKFLOWS_RUNTIME_APP_KEEP_ALIVE_ENABLED", + true); + + /** + * Interval between application-level keepalive pings on the workflow runtime's gRPC channel. + * Environment variable: DAPR_WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS + * System property: dapr.workflows.app.keep.alive.interval.seconds + * Default: 30 seconds + */ + public static final Property WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS = new SecondsDurationProperty( + "dapr.workflows.app.keep.alive.interval.seconds", + "DAPR_WORKFLOWS_APP_KEEP_ALIVE_INTERVAL_SECONDS", + Duration.ofSeconds(30)); + /** * GRPC endpoint for remote sidecar connectivity. */