feat: create the backbone of counting errors per connection each minute.

.gitignore

@@ -42,3 +42,4 @@ api_key
 artman-genfiles
 
 .flattened-pom.xml
+dependency-reduced-pom.xml

google-cloud-bigtable-stats/src/main/java/com/google/cloud/bigtable/stats/StatsRecorderWrapper.java

@@ -32,16 +32,17 @@
 @InternalApi("For internal use only")
 public class StatsRecorderWrapper {
 
-  private final OperationType operationType;
+  private OperationType operationType;
 
   private final Tagger tagger;
   private final StatsRecorder statsRecorder;
   private final TagContext parentContext;
-  private final SpanName spanName;
+  private SpanName spanName;
   private final Map<String, String> statsAttributes;
 
   private MeasureMap attemptMeasureMap;
   private MeasureMap operationMeasureMap;
+  private MeasureMap perConnectionErrorCountMeasureMap;
 
   public StatsRecorderWrapper(
       OperationType operationType,
@@ -59,6 +60,14 @@ public StatsRecorderWrapper(
     this.operationMeasureMap = statsRecorder.newMeasureMap();
   }
 
+  public StatsRecorderWrapper(Map<String, String> statsAttributes, StatsRecorder statsRecorder) {
+    this.tagger = Tags.getTagger();
+    this.statsRecorder = statsRecorder;
+    this.parentContext = tagger.getCurrentTagContext();
+    this.statsAttributes = statsAttributes;
+    this.perConnectionErrorCountMeasureMap = statsRecorder.newMeasureMap();
+  }
+
   public void recordOperation(String status, String tableId, String zone, String cluster) {
     TagContextBuilder tagCtx =
         newTagContextBuilder(tableId, zone, cluster)
@@ -87,6 +96,15 @@ public void recordAttempt(String status, String tableId, String zone, String clu
     attemptMeasureMap = statsRecorder.newMeasureMap();
   }
 
+  public void putAndRecordPerConnectionErrorCount(long errorCount) {
+    perConnectionErrorCountMeasureMap.put(
+        BuiltinMeasureConstants.PER_CONNECTION_ERROR_COUNT, errorCount);
+
+    perConnectionErrorCountMeasureMap.record(
+        newTagContextBuilderForPerConnectionErrorCount().build());
+    perConnectionErrorCountMeasureMap = statsRecorder.newMeasureMap();
+  }
+
   public void putOperationLatencies(long operationLatency) {
     operationMeasureMap.put(BuiltinMeasureConstants.OPERATION_LATENCIES, operationLatency);
   }
@@ -132,4 +150,12 @@ private TagContextBuilder newTagContextBuilder(String tableId, String zone, Stri
     }
     return tagContextBuilder;
   }
+
+  private TagContextBuilder newTagContextBuilderForPerConnectionErrorCount() {
+    TagContextBuilder tagContextBuilder = tagger.toBuilder(parentContext);
+    for (Map.Entry<String, String> entry : statsAttributes.entrySet()) {
+      tagContextBuilder.putLocal(TagKey.create(entry.getKey()), TagValue.create(entry.getValue()));
+    }
+    return tagContextBuilder;
+  }
 }

google-cloud-bigtable-stats/src/main/java/com/google/cloud/bigtable/stats/StatsWrapper.java

@@ -40,6 +40,10 @@ public static StatsRecorderWrapper createRecorder(
         operationType, spanName, statsAttributes, Stats.getStatsRecorder());
   }
 
+  public static StatsRecorderWrapper createRecorder(Map<String, String> statsAttributes) {
+    return new StatsRecorderWrapper(statsAttributes, Stats.getStatsRecorder());
+  }
+
   // This is used in integration tests to get the tag value strings from view manager because Stats
   // is relocated to com.google.bigtable.veneer.repackaged.io.opencensus.
   @InternalApi("Visible for testing")

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/ConnectionErrorCountInterceptor.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * 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
+ *
+ *     https://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 com.google.cloud.bigtable.data.v2.stub;
+
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall;
+import io.grpc.ForwardingClientCallListener;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/** An interceptor which counts the number of failed responses for a channel. */
+class ConnectionErrorCountInterceptor implements ClientInterceptor {
+  private final AtomicInteger numOfErrors;
+  private final AtomicInteger numOfSuccesses;
+
+  public ConnectionErrorCountInterceptor(Set<ConnectionErrorCountInterceptor> interceptors) {
+    numOfErrors = new AtomicInteger(0);
+    numOfSuccesses = new AtomicInteger(0);
+    interceptors.add(this);
+  }
+
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+      MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions, Channel channel) {
+    return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
+        channel.newCall(methodDescriptor, callOptions)) {
+      @Override
+      public void start(Listener<RespT> responseListener, Metadata headers) {
+        super.start(
+            new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(
+                responseListener) {
+              @Override
+              public void onClose(Status status, Metadata trailers) {
+                if (status.isOk()) {
+                  numOfSuccesses.getAndIncrement();
+                } else {
+                  numOfErrors.getAndIncrement();
+                }
+                super.onClose(status, trailers);
+              }
+            },
+            headers);
+      }
+    };
+  }
+
+  public int getAndResetNumOfErrors() {
+    return numOfErrors.getAndSet(0);
+  }
+
+  public int getAndResetNumOfSuccesses() {
+    return numOfSuccesses.getAndSet(0);
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/CountErrorsPerInterceptorTask.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * 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
+ *
+ *     https://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 com.google.cloud.bigtable.data.v2.stub;
+
+import com.google.cloud.bigtable.stats.StatsRecorderWrapper;
+import com.google.cloud.bigtable.stats.StatsWrapper;
+import com.google.common.collect.ImmutableMap;
+import java.util.Set;
+
+class CountErrorsPerInterceptorTask implements Runnable {
+  private final Set<ConnectionErrorCountInterceptor> interceptors;
+  private final StatsRecorderWrapper statsRecorderWrapper;
+
+  public CountErrorsPerInterceptorTask(
+      Set<ConnectionErrorCountInterceptor> interceptors,
+      ImmutableMap<String, String> builtinAttributes) {
+    this.interceptors = interceptors;
+    this.statsRecorderWrapper = StatsWrapper.createRecorder(builtinAttributes);
+  }
+
+  @Override
+  public void run() {
+    for (ConnectionErrorCountInterceptor interceptor : interceptors) {
+      int errors = interceptor.getAndResetNumOfErrors();
+      int successes = interceptor.getAndResetNumOfSuccesses();
+      // We avoid keeping track of inactive connections (i.e., without any failed or successful
+      // requests).
+      if (errors > 0 || successes > 0) {
+        // TODO: add a metric to also keep track of the number of successful requests per each
+        // connection.
+        this.statsRecorderWrapper.putAndRecordPerConnectionErrorCount(errors);
+      }
+    }
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java

@@ -126,9 +126,9 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -149,6 +149,7 @@
 public class EnhancedBigtableStub implements AutoCloseable {
   private static final String CLIENT_NAME = "Bigtable";
   private static final long FLOW_CONTROL_ADJUSTING_INTERVAL_MS = TimeUnit.SECONDS.toMillis(20);
+  private static final Integer PER_CONNECTION_ERROR_COUNT_PERIOD_SECONDS = 60;
 
   private final EnhancedBigtableStubSettings settings;
   private final ClientContext clientContext;
@@ -204,11 +205,7 @@ public static ClientContext createClientContext(EnhancedBigtableStubSettings set
             ? ((InstantiatingGrpcChannelProvider) builder.getTransportChannelProvider()).toBuilder()
             : null;
 
-    if (builder.getEnableRoutingCookie() && transportProvider != null) {
-      // TODO: this also need to be added to BigtableClientFactory
-      // patch cookies interceptor
-      transportProvider.setInterceptorProvider(() -> ImmutableList.of(new CookiesInterceptor()));
-    }
+    setInterceptors(transportProvider, builder);
 
     // Inject channel priming
     if (settings.isRefreshingChannel()) {
@@ -236,6 +233,39 @@ public static ClientContext createClientContext(EnhancedBigtableStubSettings set
     return ClientContext.create(builder.build());
   }
 
+  private static void setInterceptors(
+      InstantiatingGrpcChannelProvider.Builder transportProvider,
+      EnhancedBigtableStubSettings.Builder settings) {
+    Set<ConnectionErrorCountInterceptor> interceptors =
+        setupConnectionErrorCountInterceptors(settings);
+    if (transportProvider != null) {
+      if (settings.getEnableRoutingCookie()) {
+        // TODO: this also need to be added to BigtableClientFactory
+        transportProvider.setInterceptorProvider(
+            () ->
+                ImmutableList.of(
+                    new CookiesInterceptor(), new ConnectionErrorCountInterceptor(interceptors)));
+      } else {
+        transportProvider.setInterceptorProvider(
+            () -> ImmutableList.of(new ConnectionErrorCountInterceptor(interceptors)));
+      }
+    }
+  }
+
+  private static Set<ConnectionErrorCountInterceptor> setupConnectionErrorCountInterceptors(
+      EnhancedBigtableStubSettings.Builder settings) {
+    Set<ConnectionErrorCountInterceptor> interceptors =
+        Collections.newSetFromMap(new WeakHashMap<>());
+    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    ImmutableMap<String, String> builtinAttributes = createBuiltinAttributes(settings);
+    scheduler.scheduleAtFixedRate(
+        new CountErrorsPerInterceptorTask(interceptors, builtinAttributes),
+        0,
+        PER_CONNECTION_ERROR_COUNT_PERIOD_SECONDS,
+        TimeUnit.SECONDS);
+    return interceptors;
+  }
+
   public static ApiTracerFactory createBigtableTracerFactory(
       EnhancedBigtableStubSettings settings) {
     return createBigtableTracerFactory(settings, Tags.getTagger(), Stats.getStatsRecorder());
@@ -254,13 +284,7 @@ public static ApiTracerFactory createBigtableTracerFactory(
             .put(RpcMeasureConstants.BIGTABLE_INSTANCE_ID, TagValue.create(instanceId))
             .put(RpcMeasureConstants.BIGTABLE_APP_PROFILE_ID, TagValue.create(appProfileId))
             .build();
-    ImmutableMap<String, String> builtinAttributes =
-        ImmutableMap.<String, String>builder()
-            .put("project_id", projectId)
-            .put("instance", instanceId)
-            .put("app_profile", appProfileId)
-            .put("client_name", "bigtable-java/" + Version.VERSION)
-            .build();
+    ImmutableMap<String, String> builtinAttributes = createBuiltinAttributes(settings.toBuilder());
 
     return new CompositeTracerFactory(
         ImmutableList.of(
@@ -283,6 +307,16 @@ public static ApiTracerFactory createBigtableTracerFactory(
             settings.getTracerFactory()));
   }
 
+  private static ImmutableMap<String, String> createBuiltinAttributes(
+      EnhancedBigtableStubSettings.Builder settings) {
+    return ImmutableMap.<String, String>builder()
+        .put("project", settings.getProjectId())
+        .put("instance", settings.getInstanceId())
+        .put("app_profile", settings.getAppProfileId())
+        .put("client_name", "bigtable-java/" + Version.VERSION)
+        .build();
+  }
+
   private static void patchCredentials(EnhancedBigtableStubSettings.Builder settings)
       throws IOException {
     int i = settings.getEndpoint().lastIndexOf(":");