Integrate direct path/fan out logic #31504

m-trieu · 2024-06-04T23:32:36Z

Integrate direct path/fan out logic

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2024-06-05T02:18:09Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

m-trieu · 2024-06-05T06:05:20Z

R: @scwhittle

github-actions · 2024-06-05T06:06:28Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

...src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java

scwhittle · 2024-06-05T12:54:35Z

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

@@ -49,6 +56,8 @@
  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
 })
 public class MetricTrackingWindmillServerStub {
+  private static final Logger LOG = LoggerFactory.getLogger(MetricTrackingWindmillServerStub.class);
+  private static final String FAN_OUT_REFRESH_WORK_EXECUTOR = "FanOutActiveWorkRefreshExecutor";


append NAME to variable name

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

scwhittle · 2024-06-07T10:48:45Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+      worker.start();
+    } catch (Throwable e) {
+      LOG.error("Harness shutting down due to uncaught exception.", e);
+      worker.stop();


just throw the exception instead of trying to tear-down cleanly?
As is the process exit status will be ok since the exception is being caught, and stop() could get stuck somehow and we'd rather just teardown possibly uncleanly

scwhittle · 2024-06-07T10:50:12Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+
+  private static boolean isDirectPathPipeline(DataflowWorkerHarnessOptions options) {
+    boolean isIpV6Enabled = options.getDataflowServiceOptions().contains(ENABLE_IPV6_EXPERIMENT);
+    if (options.isEnableWindmillServiceDirectPath() && !isIpV6Enabled) {


log only if isEnableStreamingEngine is also set?

scwhittle · 2024-06-07T11:26:15Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

+      long secondsWaited = 0;
+      long waitFor = 10;
+      while (!batchSent) {
+        if (isClosed()) {


by polling we could add latency in cases where the threads are tied up waiting for reads and thus new work is not processed.

Do we need some finally (batch.countDown()) above to handle exceptions in sendBatch if the stream is closed? In that case it seems like it woudl unspool without needing this isClosed check here.

messaged offline

scwhittle · 2024-06-07T11:29:08Z

.../org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcWindmillStreamFactory.java

+   * @implNote Used to create stream observers for direct path streams that do not share any
+   *     underlying resources between threads.
+   */
+  private StreamObserverFactory newSimpleStreamObserverFactory() {


I think that we may still want the isReady checking, as otherwise if outgoing sending is falling behind it just buffers more and more in grpc output stream buffer and can oom.

gotcha we may want to bump the deadline then

"Exceeded timeout waiting for the outboundObserver to become ready meaning " + "that the stream deadline was not respected."

was getting a lot of these errors in testing.
reverted

scwhittle · 2024-06-07T11:32:53Z

.../org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcWindmillStreamFactory.java

@@ -209,11 +261,29 @@ public GetWorkerMetadataStream createGetWorkerMetadataStream(
        onNewWindmillEndpoints);
  }

-  private StreamObserverFactory newStreamObserverFactory() {
+  private StreamObserverFactory newBufferringStreamObserverFactory() {


this is confusing name since it doesn't buffer beyond what grpc stream internally does. It just respects grpc flow control which it seems like we'd want to do in the direct path case as well.

There is a different BufferingStreamObserver for fnapi which does do buffering with a queue.

scwhittle · 2024-06-07T11:38:02Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/StreamingEngineClient.java

            .setGlobalDataStreams(
                createNewGlobalDataStreams(newWindmillEndpoints.globalDataEndpoints()))
            .build();

+    closeStaleStreams.join();


does building the connection state take any time? Or is there a reasno we can't move this join to after updating connections and the budget refresh below so we start using hte new streams earlier?

Otherwise it seems simpler to just keep the closeStaleStreamsAndCreateNewStreams which internally does them in parallel.

good point moved

scwhittle · 2024-06-07T11:41:18Z

...g/apache/beam/runners/dataflow/worker/windmill/client/grpc/stubs/WindmillChannelFactory.java

-                .overrideAuthority(authenticatedGcpServiceAddress.authenticatingService()),
-            windmillServiceRpcChannelTimeoutSec)
+      AuthenticatedGcpServiceAddress authenticatedGcpServiceAddress) {
+    return AltsChannelBuilder.forAddress(


I think we still want the withDefaultChannelOptions
Probably want the flow control window and ssl context from createRemoteChannel unless we've tested otherwise.

scwhittle

Sorry meant to send these yesterday but they were still pending

scwhittle · 2024-06-13T10:57:19Z

...a/org/apache/beam/runners/dataflow/worker/windmill/client/WindmillStreamClosedException.java

+  }
+
+  /** Returns whether an exception was caused by a {@link WindmillStreamClosedException}. */
+  public static boolean isWindmillStreamCancelledException(Throwable t) {


update method name, actually since this is static and nested under class name maybe the method should just be wasCausedOf so it looks like:

WindmillStreamClosedException.wasCauseOf(e);

scwhittle · 2024-06-13T11:02:30Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

    int stuckCommitDurationMillis =
        windmillServiceEnabled && options.getStuckCommitDurationMillis() > 0
            ? options.getStuckCommitDurationMillis()
            : 0;
+    if (isDirectPathPipeline(options)) {
+      this.streamingEngineClient =


For SE non-direct path, can we still use StreamingEngineClient but just hard-code the single dispatcher as the worker endpoint? Seems like it would allow us to remove streamingDispatchLoop and other duplication and also seems like we want that to work if we need to fallback from non direct path.

I think we can

We can probably just inject the endpoints manually

might be better to do in a separate PR since we would have to change the way we test in StreamingDataflowWorker as that uses FakeWindmillServer and StreamingEngineClient does not.

scwhittle · 2024-06-13T11:15:20Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

+                          keyedRequest ->
+                              "KeyedGetState=["
+                                  + "key="
+                                  + keyedRequest.getKey()


avoid logging customer keys, they might be PII and they can also be large

But also not sure it's worth having all the information in the exception, each key can lot itself when it's read finishes with the error instead.

scwhittle · 2024-06-13T11:18:50Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

+    // Stream has been explicitly closed. Drain pending input streams and request batches.
+    // Future calls to send RPCs will fail.
+    pending.values().forEach(AppendableInputStream::cancel);
+    pending.clear();


clearing pending here differs from cancellation in onNewStream and I think doing so could lead to verify errors in onResponse.

when we receive a response after the stream is explictly closed what should we do? just drop it correct as that data/state would be invalid (except maybe side input state?)

scwhittle · 2024-06-13T11:22:53Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

@@ -308,7 +364,12 @@ private <ResponseT> ResponseT issueRequest(QueuedRequest request, ParseFn<Respon
    }
  }

-  private void queueRequestAndWait(QueuedRequest request) throws InterruptedException {
+  private void tryQueueRequestAndWaitForSend(QueuedRequest request) throws InterruptedException {
+    if (isClosed()) {


move beneath syncrhonized blcok?

reverted this code

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

scwhittle · 2024-06-13T12:08:55Z

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

+        sender.sendHeartbeats(heartbeat.getValue());
+      }
+
+      // There are multiple destinations to send heartbeat requests. Fan out requests in parallel.


nit: I think
else {
// comment

is easier to read

scwhittle · 2024-06-13T12:22:38Z

...main/java/org/apache/beam/runners/dataflow/worker/windmill/work/refresh/HeartbeatSender.java

+public interface HeartbeatSender {
+  void sendHeartbeats(Map<String, List<Windmill.HeartbeatRequest>> heartbeats);
+
+  default boolean isInvalid() {


can we remove this and just handle it internally to the stream?
In either case it's racy checking this and then sending on stream so the stream will have to handle it anyway

wanted to find a way to mark the work as failed for the stream since in this case we will skip, then still try to process the work

will think about another way to propagate this information

added late binding of an onStreamClosed handler for DirectHeartbeatSender

…DataflowWorker

fix get data stream

…observer

…ce the stream is shutdown, no more interactions should be performed on the stream.

scwhittle

Sending before looking at tests

scwhittle · 2024-07-01T08:10:40Z

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

+  }
+
+  /**
+   * Attempts to refresh active work, fanning out to each {@link GetDataStream} in parallel.


update comment based upon heartbeat sender

scwhittle · 2024-07-01T08:12:45Z

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

-          heartbeatStreamPool.releaseStream(stream);
-        }
+      if (heartbeats.size() == 1) {
+        // There is 1 destination to send heartbeat requests.


not setting active heartbeats to one here

scwhittle · 2024-07-01T08:14:29Z

.../src/main/java/org/apache/beam/runners/dataflow/worker/MetricTrackingWindmillServerStub.java

+          activeHeartbeats.getAndUpdate(existing -> existing + heartbeat.getValue().size());
+          HeartbeatSender sender = heartbeat.getKey();
+          Map<String, List<HeartbeatRequest>> heartbeatRequests = heartbeat.getValue();
+          sender.sendHeartbeats(heartbeatRequests);


as follow up might be nice if this returned a completable future instead of blocking, then we wouldn't have to have a thread blocked per-destination for heartbeats

Don't do in this PR, it's big enough. Could put todo for now

scwhittle · 2024-07-01T08:21:16Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+            .orElse(false);
+    if (options.isEnableStreamingEngine()
+        && options.getIsWindmillServiceDirectPathEnabled()
+        && !isIpV6Enabled) {


how about structuring to avoid duplication of conditions. Seems less brittle since prevents accidents like just modifying one of the conditions.

if (options.isEnableStreamingEngine()
&& options.getIsWindmillServiceDirectPathEnabled()) {
if (isIpV6Enabled) {
return true;
}
LOG.warn(...);
}
return false;

scwhittle · 2024-07-01T08:23:07Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+        dispatchThread.join();
+        workCommitter.stop();
+      }
+


can't add comment at the line below, but can you rename dispatchLoop to applianceDispatchLoop?

scwhittle · 2024-07-01T14:03:28Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

  }

  private <ResponseT> ResponseT issueRequest(QueuedRequest request, ParseFn<ResponseT> parseFn) {
    while (true) {
+      // Handle stream closure during loop.
+      if (isShutdown() || isClosed()) {


similar to other comments, I'd remove this since queueReuqstAndWait already needs to handle it and having another check here is just more possible codepaths.

scwhittle · 2024-07-01T14:06:38Z

...ava/org/apache/beam/runners/dataflow/worker/windmill/work/refresh/DirectHeartbeatSender.java

+
+  @Override
+  public void sendHeartbeats(Map<String, List<HeartbeatRequest>> heartbeats) {
+    if (getDataStream.isShutdown() || getDataStream.isClosed()) {


whe have shutdown||closed in the other file too, can we just always close if we shutdown and simplify these?

scwhittle · 2024-07-01T14:07:56Z

...ava/org/apache/beam/runners/dataflow/worker/windmill/work/refresh/DirectHeartbeatSender.java

+  }
+
+  public HeartbeatSender withStreamClosedHandler(Runnable onStreamClosed) {
+    return new DirectHeartbeatSender(getDataStream, onStreamClosed);


think this shoudl fail if one was already set (but see other comment, I think that we can have this noticed more from the top to avoid instance of checks)

scwhittle · 2024-07-01T14:09:31Z

...main/java/org/apache/beam/runners/dataflow/worker/windmill/work/refresh/HeartbeatSender.java

+/** Interface for sending heartbeats. */
+@FunctionalInterface
+public interface HeartbeatSender {
+  void sendHeartbeats(Map<String, List<Windmill.HeartbeatRequest>> heartbeats);


comment on string

scwhittle · 2024-07-01T14:10:01Z

...org/apache/beam/runners/dataflow/worker/windmill/work/refresh/PoolBackedHeartbeatSender.java

+
+/** StreamingEngine stream pool based implementation of {@link HeartbeatSender}. */
+@Internal
+public final class PoolBackedHeartbeatSender implements HeartbeatSender {


StreamPoolHeartbeatSender?

github-actions · 2024-08-31T12:42:35Z

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

github-actions bot added runners dataflow labels Jun 4, 2024

m-trieu force-pushed the mt-integrate-fan-out branch 2 times, most recently from f6cf353 to e41988f Compare June 5, 2024 00:58

scwhittle requested changes Jun 7, 2024

View reviewed changes

m-trieu force-pushed the mt-integrate-fan-out branch 4 times, most recently from de5266c to 080a511 Compare June 12, 2024 20:21

scwhittle requested changes Jun 14, 2024

View reviewed changes

m-trieu force-pushed the mt-integrate-fan-out branch 6 times, most recently from 2ba55f7 to 5d7cfc2 Compare June 18, 2024 21:44

m-trieu force-pushed the mt-integrate-fan-out branch 6 times, most recently from 95a7f2e to bca34e1 Compare June 27, 2024 23:23

m-trieu added 5 commits June 28, 2024 11:34

integrate direct path/fanout worker harness components with Streaming…

e990fb3

…DataflowWorker

address PR comments

4022f20

address PR comments

ab51953

fix get data stream

add short circuit to phaser waiting when we explicitly close request …

e6d61e1

…observer

add readonly refreshable view of Work

ff09734

m-trieu added 7 commits June 28, 2024 15:51

address PR comments

e7b544c

fix dangling commits, add budget freeing when failing work

5c31658

fix stream observer not returning after calling onNext

af44fde

add stream id

3205784

add stream id, change synchronization in AbstractWindmillStream

e480e82

add more visibility via debug capture pages

ba1d134

add shutdown boolean to handle the stream being explicitly closed. on…

50b5da6

…ce the stream is shutdown, no more interactions should be performed on the stream.

m-trieu force-pushed the mt-integrate-fan-out branch from bca34e1 to 50b5da6 Compare June 30, 2024 06:14

scwhittle requested changes Jul 1, 2024

View reviewed changes

address PR comments

c3ff2c3

m-trieu mentioned this pull request Jul 5, 2024

Fix/Refactor GetData interfaces for Direct Path integration #31784

Merged

3 tasks

github-actions bot added the stale label Aug 31, 2024

m-trieu closed this Sep 3, 2024

Integrate direct path/fan out logic #31504

Integrate direct path/fan out logic #31504

Conversation

m-trieu commented Jun 4, 2024

GitHub Actions Tests Status (on master branch)

github-actions bot commented Jun 5, 2024

m-trieu commented Jun 5, 2024

github-actions bot commented Jun 5, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scwhittle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scwhittle left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Aug 31, 2024