[Dataflow Java Streaming] Reduce log spam for direct path mode and improve status page. (#35880)

Author: scwhittle

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java

@@ -76,6 +76,7 @@ private static class StreamAndRequest {
   private final AtomicLong idGenerator;
   private final JobHeader jobHeader;
   private final int streamingRpcBatchLimit;
+  private volatile boolean logMissingResponse = true;
 
   private GrpcCommitWorkStream(
       String backendWorkerToken,
@@ -200,7 +201,9 @@ public void onResponse(StreamingCommitResponse response) {
 
         @Nullable StreamAndRequest entry = pending.remove(requestId);
         if (entry == null) {
-          LOG.error("Got unknown commit request ID: {}", requestId);
+          if (logMissingResponse) {
+            LOG.error("Got unknown commit request ID: {}", requestId);
+          }
           continue;
         }
         if (entry.handler != this) {
@@ -250,6 +253,7 @@ protected PhysicalStreamHandler newResponseHandler() {
 
   @Override
   protected synchronized void shutdownInternal() {
+    logMissingResponse = false;
     Iterator<StreamAndRequest> pendingRequests = pending.values().iterator();
     while (pendingRequests.hasNext()) {
       PendingRequest pendingRequest = pendingRequests.next().request;

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcDispatcherClient.java

@@ -21,6 +21,8 @@
 import static org.apache.beam.runners.dataflow.worker.windmill.client.grpc.stubs.WindmillChannels.localhostChannel;
 
 import com.google.auto.value.AutoValue;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.List;
 import java.util.Random;
 import java.util.Set;
@@ -132,25 +134,26 @@ ImmutableSet<HostAndPort> getDispatcherEndpoints() {
 
   /** Will block the calling thread until the initial endpoints are present. */
   public CloudWindmillMetadataServiceV1Alpha1Stub getWindmillMetadataServiceStubBlocking() {
-    boolean initialized = false;
-    long secondsWaited = 0;
-    while (!initialized) {
-      LOG.info(
-          "Blocking until Windmill Service endpoint has been set. "
-              + "Currently waited for [{}] seconds.",
-          secondsWaited);
+    Instant startTime = Instant.now();
+    while (true) {
       try {
-        initialized = onInitializedEndpoints.await(10, TimeUnit.SECONDS);
-        secondsWaited += 10;
+        if (onInitializedEndpoints.await(10, TimeUnit.SECONDS)) {
+          break;
+        }
+        LOG.info(
+            "Blocking until Windmill Service endpoint has been set. " + "Currently waited for {}.",
+            Duration.between(startTime, Instant.now()));
       } catch (InterruptedException e) {
         LOG.error(
             "Interrupted while waiting for initial Windmill Service endpoints. "
                 + "These endpoints are required to do any pipeline processing.",
             e);
       }
     }
-
-    LOG.info("Windmill Service endpoint initialized after {} seconds.", secondsWaited);
+    Duration elapsed = Duration.between(startTime, Instant.now());
+    if (elapsed.getSeconds() >= 5) {
+      LOG.info("Windmill Service endpoint initialized after {}.", elapsed);
+    }
 
     ImmutableList<CloudWindmillMetadataServiceV1Alpha1Stub> windmillMetadataServiceStubs =
         dispatcherStubs.get().windmillMetadataServiceStubs();

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetWorkerMetadataStream.java

@@ -20,6 +20,7 @@
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import java.io.PrintWriter;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
@@ -52,6 +53,9 @@ public final class GrpcGetWorkerMetadataStream
   @GuardedBy("metadataLock")
   private WorkerMetadataResponse latestResponse;
 
+  @GuardedBy("metadataLock")
+  private Instant latestResponseReceived = Instant.EPOCH;
+
   private GrpcGetWorkerMetadataStream(
       Function<StreamObserver<WorkerMetadataResponse>, StreamObserver<WorkerMetadataRequest>>
           startGetWorkerMetadataRpcFn,
@@ -112,6 +116,7 @@ private Optional<WindmillEndpoints> extractWindmillEndpointsFrom(
     synchronized (metadataLock) {
       if (response.getMetadataVersion() > latestResponse.getMetadataVersion()) {
         this.latestResponse = response;
+        this.latestResponseReceived = Instant.now();
         return Optional.of(WindmillEndpoints.from(response));
       } else {
         // If the currentMetadataVersion is greater than or equal to one in the response, the
@@ -165,8 +170,8 @@ protected void sendHealthCheck() throws WindmillStreamShutdownException {
   protected void appendSpecificHtml(PrintWriter writer) {
     synchronized (metadataLock) {
       writer.format(
-          "GetWorkerMetadataStream:  job_header=[%s], current_metadata=[%s] ",
-          workerMetadataRequest.getHeader(), latestResponse);
+          "GetWorkerMetadataStream:  job_header=[%s], current_metadata=[%s] received_at=[%s]",
+          workerMetadataRequest.getHeader(), latestResponse, latestResponseReceived);
     }
   }
 }

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java

@@ -425,7 +425,7 @@ private ExecuteWorkResult executeWork(
       // If processing failed due to a thrown exception, close the executionState. Do not
       // return/release the executionState back to computationState as that will lead to this
       // executionState instance being reused.
-      LOG.info("Invalidating executor after work item {} failed with Exception:", key, t);
+      LOG.debug("Invalidating executor after work item {} failed", workItem.getWorkToken(), t);
       computationWorkExecutor.invalidate();
 
       // Re-throw the exception, it will be caught and handled by workFailureProcessor downstream.