wip

Author: sushanb

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

@@ -106,7 +106,7 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
         errorCountPerConnectionMetricTracker =
             setupPerConnectionErrorTracer(builder, transportProvider, internalOtel);
 
-        outstandingRpcsMetricTracker = new OutstandingRpcsMetricTracker(internalOtel, "LB_POLICY");
+        outstandingRpcsMetricTracker = new OutstandingRpcsMetricTracker(internalOtel);
 
         // Configure grpc metrics
         configureGrpcOtel(transportProvider, internalOtel);

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

@@ -18,8 +18,8 @@
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.METER_NAME;
 import static com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsConstants.OUTSTANDING_RPCS_PER_CHANNEL_NAME;
 
-import com.google.cloud.bigtable.gaxx.grpc.BigtableChannelInsight;
-import com.google.cloud.bigtable.gaxx.grpc.BigtableChannelInsightsProvider;
+import com.google.cloud.bigtable.gaxx.grpc.BigtableChannelObserver;
+import com.google.cloud.bigtable.gaxx.grpc.BigtableChannelPoolObserver;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.LongHistogram;
@@ -29,15 +29,24 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
 
 public class OutstandingRpcsMetricTracker implements Runnable {
   private static final int SAMPLING_PERIOD_SECONDS = 60;
   private final LongHistogram outstandingRpcsHistogram;
-  private final Attributes baseAttributes;
-  private final AtomicReference<BigtableChannelInsightsProvider>
+  private final AtomicReference<BigtableChannelPoolObserver>
       bigtableChannelInsightsProviderRef = new AtomicReference<>();
+  private final AtomicReference<String> lbPolicyRef = new AtomicReference<>();
 
-  public OutstandingRpcsMetricTracker(OpenTelemetry openTelemetry, String lbPolicy) {
+  // Base attributes common to all recordings
+  private static final Attributes TRANSPORT_GRPC_ATTR =
+      Attributes.builder().put("transport_type", "grpc").build();
+
+  // Attributes for unary and streaming RPCs, built on demand in run()
+  @Nullable private Attributes unaryAttributes;
+  @Nullable private Attributes streamingAttributes;
+
+  public OutstandingRpcsMetricTracker(OpenTelemetry openTelemetry) {
     Meter meter = openTelemetry.getMeter(METER_NAME);
     this.outstandingRpcsHistogram =
         meter
@@ -47,20 +56,22 @@ public OutstandingRpcsMetricTracker(OpenTelemetry openTelemetry, String lbPolicy
                 "A distribution of the number of outstanding RPCs per connection in the client pool, sampled periodically.")
             .setUnit("1")
             .build();
-
-    this.baseAttributes =
-        Attributes.builder().put("transport_type", "grpc").put("lb_policy", lbPolicy).build();
   }
 
   /**
    * Registers the provider for the channel pool entries. This should be called by the component
    * that creates the BigtableChannelPool.
    */
   public void registerChannelInsightsProvider(
-      BigtableChannelInsightsProvider channelInsightsProvider) {
+      BigtableChannelPoolObserver channelInsightsProvider) {
     this.bigtableChannelInsightsProviderRef.set(channelInsightsProvider);
   }
 
+  /** Register the current lb policy * */
+  public void registerLoadBalancingStrategy(String lbPolicy) {
+    this.lbPolicyRef.set(lbPolicy);
+  }
+
   /** Starts the periodic collection. */
   public ScheduledFuture<?> start(ScheduledExecutorService scheduler) {
     return scheduler.scheduleAtFixedRate(
@@ -69,25 +80,33 @@ public ScheduledFuture<?> start(ScheduledExecutorService scheduler) {
 
   @Override
   public void run() {
-    BigtableChannelInsightsProvider channelInsightsProvider =
+    BigtableChannelPoolObserver channelInsightsProvider =
         bigtableChannelInsightsProviderRef.get();
     if (channelInsightsProvider == null) {
       return; // Not registered yet
     }
-    List<? extends BigtableChannelInsight> channelInsights =
+    String lbPolicy = lbPolicyRef.get();
+    if (lbPolicy == null) {
+      lbPolicy = "ROUND_ROBIN";
+    }
+
+    // Build attributes if they haven't been built yet or were invalidated
+    if (unaryAttributes == null || streamingAttributes == null) {
+      Attributes baseAttrs = TRANSPORT_GRPC_ATTR.toBuilder().put("lb_policy", lbPolicy).build();
+      this.unaryAttributes = baseAttrs.toBuilder().put("streaming", false).build();
+      this.streamingAttributes = baseAttrs.toBuilder().put("streaming", true).build();
+    }
+    List<? extends BigtableChannelObserver> channelInsights =
         channelInsightsProvider.getChannelInfos();
     if (channelInsights == null || channelInsights.isEmpty()) {
       return;
     }
-    for (BigtableChannelInsight info : channelInsights) {
-      long currentOutstandingUnaryRpcs = info.getOutstandingStreamingRpcs();
+    for (BigtableChannelObserver info : channelInsights) {
+      long currentOutstandingUnaryRpcs = info.getOutstandingUnaryRpcs();
       long currentOutstandingStreamingRpcs = info.getOutstandingStreamingRpcs();
       // Record outstanding unary RPCs with streaming=false
-      Attributes unaryAttributes = baseAttributes.toBuilder().put("streaming", false).build();
       outstandingRpcsHistogram.record(currentOutstandingUnaryRpcs, unaryAttributes);
-
       // Record outstanding streaming RPCs with streaming=true
-      Attributes streamingAttributes = baseAttributes.toBuilder().put("streaming", true).build();
       outstandingRpcsHistogram.record(currentOutstandingStreamingRpcs, streamingAttributes);
     }
   }

…le/gaxx/grpc/BigtableChannelInsight.java …e/gaxx/grpc/BigtableChannelObserver.javagoogle-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelInsight.java renamed to google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelObserver.java google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelInsight.java renamed to google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelObserver.java

@@ -17,9 +17,9 @@
 
 import com.google.api.core.InternalApi;
 
-/** Provides insights about a single channel in the channel pool. */
+/** Provides observability about a single channel in the channel pool. */
 @InternalApi
-public interface BigtableChannelInsight {
+public interface BigtableChannelObserver {
   /** Gets the current number of outstanding Unary RPCs on this channel. */
   int getOutstandingUnaryRpcs();
 

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java

@@ -57,7 +57,7 @@
  * <p>Internal API
  */
 @InternalApi
-public class BigtableChannelPool extends ManagedChannel implements BigtableChannelInsightsProvider {
+public class BigtableChannelPool extends ManagedChannel implements BigtableChannelPoolObserver {
   @VisibleForTesting
   static final Logger LOG = Logger.getLogger(BigtableChannelPool.class.getName());
 
@@ -509,12 +509,12 @@ private Entry getEntry(int affinity) {
 
   /** Gets the current list of BigtableChannelInsight objects. */
   @Override
-  public List<? extends BigtableChannelInsight> getChannelInfos() {
+  public List<? extends BigtableChannelObserver> getChannelInfos() {
     return entries.get();
   }
 
   /** Bundles a gRPC {@link ManagedChannel} with some usage accounting. */
-  static class Entry implements BigtableChannelInsight {
+  static class Entry implements BigtableChannelObserver {
     private final ManagedChannel channel;
 
     /**

…rpc/BigtableChannelInsightsProvider.java …xx/grpc/BigtableChannelPoolObserver.javagoogle-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelInsightsProvider.java renamed to google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolObserver.java google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelInsightsProvider.java renamed to google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolObserver.java

@@ -20,7 +20,7 @@
 
 @InternalApi
 @FunctionalInterface
-public interface BigtableChannelInsightsProvider {
+public interface BigtableChannelPoolObserver {
   /** Gets the current list of BigtableChannelInsight objects. */
-  List<? extends BigtableChannelInsight> getChannelInfos();
+  List<? extends BigtableChannelObserver> getChannelInfos();
 }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java

@@ -147,6 +147,8 @@ public TransportChannel getTransportChannel() throws IOException {
 
     if (outstandingRpcsMetricTracker != null) {
       outstandingRpcsMetricTracker.registerChannelInsightsProvider(btChannelPool::getChannelInfos);
+      outstandingRpcsMetricTracker.registerLoadBalancingStrategy(
+          btPoolSettings.getLoadBalancingStrategy().name());
     }
 
     return GrpcTransportChannel.create(btChannelPool);