aggregate grpc queued and batch flow controlled latencies

googleapis · Apr 18, 2023 · 2cc55a1 · 2cc55a1
1 parent 5a081a5
commit 2cc55a1
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 19 deletions.
diff --git a/...ud-bigtable-stats/src/main/java/com/google/cloud/bigtable/stats/StatsRecorderWrapper.java b/...ud-bigtable-stats/src/main/java/com/google/cloud/bigtable/stats/StatsRecorderWrapper.java
@@ -115,14 +115,10 @@ public void putGfeMissingHeaders(long connectivityErrors) {
     attemptMeasureMap.put(BuiltinMeasureConstants.CONNECTIVITY_ERROR_COUNT, connectivityErrors);
   }
 
-  public void putBatchRequestThrottled(long throttledTimeMs) {
+  public void putClientBlockingLatencies(long throttledTimeMs) {
     operationMeasureMap.put(BuiltinMeasureConstants.THROTTLING_LATENCIES, throttledTimeMs);
   }
 
-  public void putGrpcChannelQueuedLatencies(long blockedTime) {
-    attemptMeasureMap.put(BuiltinMeasureConstants.THROTTLING_LATENCIES, blockedTime);
-  }
-
   private TagContextBuilder newTagContextBuilder(String tableId, String zone, String cluster) {
     TagContextBuilder tagContextBuilder =
         tagger

diff --git a/...igtable-stats/src/test/java/com/google/cloud/bigtable/stats/StatsRecorderWrapperTest.java b/...igtable-stats/src/test/java/com/google/cloud/bigtable/stats/StatsRecorderWrapperTest.java
@@ -91,8 +91,7 @@ public void testStreamingOperation() throws InterruptedException {
     recorderWrapper.putGfeLatencies(serverLatency);
     recorderWrapper.putGfeMissingHeaders(connectivityErrorCount);
     recorderWrapper.putFirstResponseLatencies(firstResponseLatency);
-    recorderWrapper.putBatchRequestThrottled(throttlingLatency);
-    recorderWrapper.putGrpcChannelQueuedLatencies(throttlingLatency);
+    recorderWrapper.putClientBlockingLatencies(throttlingLatency);
 
     recorderWrapper.recordOperation("OK", TABLE_ID, ZONE, CLUSTER);
     recorderWrapper.recordAttempt("OK", TABLE_ID, ZONE, CLUSTER);
@@ -291,8 +290,7 @@ public void testUnaryOperations() throws InterruptedException {
     recorderWrapper.putGfeLatencies(serverLatency);
     recorderWrapper.putGfeMissingHeaders(connectivityErrorCount);
     recorderWrapper.putFirstResponseLatencies(firstResponseLatency);
-    recorderWrapper.putBatchRequestThrottled(throttlingLatency);
-    recorderWrapper.putGrpcChannelQueuedLatencies(throttlingLatency);
+    recorderWrapper.putClientBlockingLatencies(throttlingLatency);
 
     recorderWrapper.recordOperation("UNAVAILABLE", TABLE_ID, ZONE, CLUSTER);
     recorderWrapper.recordAttempt("UNAVAILABLE", TABLE_ID, ZONE, CLUSTER);

diff --git a/...le/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracer.java b/...le/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracer.java
@@ -71,6 +71,8 @@ class BuiltinMetricsTracer extends BigtableTracer {
   private String zone = "global";
   private String cluster = "unspecified";
 
+  private AtomicLong totalClientBlockingTime = new AtomicLong(0);
+
   @VisibleForTesting
   BuiltinMetricsTracer(
       OperationType operationType, SpanName spanName, StatsRecorderWrapper recorder) {
@@ -219,12 +221,12 @@ public void setLocations(String zone, String cluster) {
 
   @Override
   public void batchRequestThrottled(long throttledTimeMs) {
-    recorder.putBatchRequestThrottled(throttledTimeMs);
+    totalClientBlockingTime.addAndGet(throttledTimeMs);
   }
 
   @Override
   public void grpcChannelQueuedLatencies(long queuedTimeMs) {
-    recorder.putGrpcChannelQueuedLatencies(queuedTimeMs);
+    totalClientBlockingTime.addAndGet(queuedTimeMs);
   }
 
   @Override
@@ -271,6 +273,8 @@ private void recordAttemptCompletion(@Nullable Throwable status) {
       }
     }
 
+    recorder.putClientBlockingLatencies(totalClientBlockingTime.get());
+
     // Patch the status until it's fixed in gax. When an attempt failed,
     // it'll throw a ServerStreamingAttemptException. Unwrap the exception
     // so it could get processed by extractStatus

diff --git a/...rc/test/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracerTest.java b/...rc/test/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracerTest.java
@@ -480,8 +480,8 @@ public void testBatchBlockingLatencies() throws InterruptedException {
 
       int expectedNumRequests = 6 / batchElementCount;
       ArgumentCaptor<Long> throttledTime = ArgumentCaptor.forClass(Long.class);
-      verify(statsRecorderWrapper, times(expectedNumRequests))
-          .putBatchRequestThrottled(throttledTime.capture());
+      verify(statsRecorderWrapper, timeout(1000).times(expectedNumRequests))
+          .putClientBlockingLatencies(throttledTime.capture());
 
       // Adding the first 2 elements should not get throttled since the batch is empty
       assertThat(throttledTime.getAllValues().get(0)).isEqualTo(0);
@@ -493,7 +493,7 @@ public void testBatchBlockingLatencies() throws InterruptedException {
   }
 
   @Test
-  public void testBlockedOnChannelServerStreamLatencies() throws InterruptedException {
+  public void testQueuedOnChannelServerStreamLatencies() throws InterruptedException {
     when(mockFactory.newTracer(any(), any(), any()))
         .thenReturn(
             new BuiltinMetricsTracer(
@@ -505,14 +505,14 @@ public void testBlockedOnChannelServerStreamLatencies() throws InterruptedExcept
 
     ArgumentCaptor<Long> blockedTime = ArgumentCaptor.forClass(Long.class);
 
-    verify(statsRecorderWrapper, times(fakeService.attemptCounter.get()))
-        .putGrpcChannelQueuedLatencies(blockedTime.capture());
+    verify(statsRecorderWrapper, timeout(1000).times(fakeService.attemptCounter.get()))
+        .putClientBlockingLatencies(blockedTime.capture());
 
     assertThat(blockedTime.getAllValues().get(1)).isAtLeast(CHANNEL_BLOCKING_LATENCY);
   }
 
   @Test
-  public void testBlockedOnChannelUnaryLatencies() throws InterruptedException {
+  public void testQueuedOnChannelUnaryLatencies() throws InterruptedException {
     when(mockFactory.newTracer(any(), any(), any()))
         .thenReturn(
             new BuiltinMetricsTracer(
@@ -521,8 +521,8 @@ public void testBlockedOnChannelUnaryLatencies() throws InterruptedException {
 
     ArgumentCaptor<Long> blockedTime = ArgumentCaptor.forClass(Long.class);
 
-    verify(statsRecorderWrapper, times(fakeService.attemptCounter.get()))
-        .putGrpcChannelQueuedLatencies(blockedTime.capture());
+    verify(statsRecorderWrapper, timeout(1000).times(fakeService.attemptCounter.get()))
+        .putClientBlockingLatencies(blockedTime.capture());
 
     assertThat(blockedTime.getAllValues().get(1)).isAtLeast(CHANNEL_BLOCKING_LATENCY);
     assertThat(blockedTime.getAllValues().get(2)).isAtLeast(CHANNEL_BLOCKING_LATENCY);