Merge pull request #8971 from dustin12/ThrottlePiping

[BEAM-7666] Throttle piping
apache · Jul 15, 2019 · 8001ad2 · 8001ad2
2 parents b60aef1 + beb62c1
commit 8001ad2
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 1 deletion.
diff --git a/...a/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowSystemMetrics.java b/...a/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowSystemMetrics.java
@@ -39,7 +39,8 @@ public enum StreamingSystemCounterNames {
     JAVA_HARNESS_USED_MEMORY("dataflow_java_harness_used_memory"),
     JAVA_HARNESS_MAX_MEMORY("dataflow_java_harness_max_memory"),
     JAVA_HARNESS_RESTARTS("dataflow_java_harness_restarts"),
-    WINDMILL_QUOTA_THROTTLING("dataflow_streaming_engine_throttled_msecs");
+    WINDMILL_QUOTA_THROTTLING("dataflow_streaming_engine_throttled_msecs"),
+    MEMORY_THRASHING("dataflow_streaming_engine_user_worker_thrashing");
 
     private final String name;
 

diff --git a/...worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java b/...worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
@@ -415,6 +415,7 @@ public int getSize() {
   private final Counter<Long, Long> javaHarnessUsedMemory;
   private final Counter<Long, Long> javaHarnessMaxMemory;
   private final Counter<Integer, Integer> windmillMaxObservedWorkItemCommitBytes;
+  private final Counter<Integer, Integer> memoryThrashing;
   private Timer refreshActiveWorkTimer;
   private Timer statusPageTimer;
 
@@ -593,6 +594,9 @@ public static StreamingDataflowWorker fromDataflowWorkerHarnessOptions(
     this.windmillMaxObservedWorkItemCommitBytes =
         pendingCumulativeCounters.intMax(
             StreamingSystemCounterNames.WINDMILL_MAX_WORK_ITEM_COMMIT_BYTES.counterName());
+    this.memoryThrashing =
+        pendingCumulativeCounters.intSum(
+            StreamingSystemCounterNames.MEMORY_THRASHING.counterName());
     this.isDoneFuture = new CompletableFuture<>();
 
     this.threadFactory =
@@ -1851,6 +1855,9 @@ private void sendWorkerUpdatesToDataflowService(
 
     // Throttle time is tracked by the windmillServer but is reported to DFE here.
     windmillQuotaThrottling.addValue(windmillServer.getAndResetThrottleTime());
+    if (memoryMonitor.isThrashing()) {
+      memoryThrashing.addValue(1);
+    }
 
     List<CounterUpdate> counterUpdates = new ArrayList<>(128);
 

diff --git a/...java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/MemoryMonitor.java b/...java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/MemoryMonitor.java
@@ -339,6 +339,10 @@ public void stop() {
     }
   }
 
+  public boolean isThrashing() {
+    return isThrashing.get();
+  }
+
   /**
    * Check if we've observed high gc workload in sufficient sample periods to justify classifying
    * the server as in gc thrashing.