rls: add counter metrics (#11138)

temawi · web-flow · commit a9fb272b7841 · 2024-05-01T11:24:38.000-07:00
Adds the following metrics to the RlsLoadBalancer:
- grpc.lb.rls.default_target_picks
- grpc.lb.rls.target_picks
- grpc.lb.rls.failed_picks
diff --git a/api/src/main/java/io/grpc/LoadBalancer.java b/api/src/main/java/io/grpc/LoadBalancer.java
@@ -716,6 +716,13 @@ public boolean isDrop() {
       return drop;
     }
 
+    /**
+     * Returns {@code true} if the pick was not created with {@link #withNoResult()}.
+     */
+    public boolean hasResult() {
+      return !(subchannel == null && status.isOk());
+    }
+
     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this)
diff --git a/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java b/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java
@@ -24,6 +24,7 @@
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
 import com.google.common.base.Ticker;
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -36,9 +37,11 @@
 import io.grpc.LoadBalancer.PickSubchannelArgs;
 import io.grpc.LoadBalancer.ResolvedAddresses;
 import io.grpc.LoadBalancer.SubchannelPicker;
+import io.grpc.LongCounterMetricInstrument;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
+import io.grpc.MetricInstrumentRegistry;
 import io.grpc.Status;
 import io.grpc.internal.BackoffPolicy;
 import io.grpc.internal.ExponentialBackoffPolicy;
@@ -87,6 +90,10 @@ final class CachingRlsLbClient {
   /** Minimum bytes for a Java Object. */
   public static final int OBJ_OVERHEAD_B = 16;
 
+  private static final LongCounterMetricInstrument DEFAULT_TARGET_PICKS_COUNTER;
+  private static final LongCounterMetricInstrument TARGET_PICKS_COUNTER;
+  private static final LongCounterMetricInstrument FAILED_PICKS_COUNTER;
+
   // All cache status changes (pending, backoff, success) must be under this lock
   private final Object lock = new Object();
   // LRU cache based on access order (BACKOFF and actual data will be here)
@@ -115,6 +122,23 @@ final class CachingRlsLbClient {
   private final RefCountedChildPolicyWrapperFactory refCountedChildPolicyWrapperFactory;
   private final ChannelLogger logger;
 
+  static {
+    MetricInstrumentRegistry metricInstrumentRegistry
+        = MetricInstrumentRegistry.getDefaultRegistry();
+    DEFAULT_TARGET_PICKS_COUNTER = metricInstrumentRegistry.registerLongCounter(
+        "grpc.lb.rls.default_target_picks", "Number of LB picks sent to the default target", "pick",
+        Lists.newArrayList("grpc.target", "grpc.lb.rls.server_target",
+            "grpc.lb.rls.data_plane_target", "grpc.lb.pick_result"), Lists.newArrayList(), true);
+    TARGET_PICKS_COUNTER = metricInstrumentRegistry.registerLongCounter("grpc.lb.rls.target_picks",
+        "Number of LB picks sent to each RLS target", "pick",
+        Lists.newArrayList("grpc.target", "grpc.lb.rls.server_target",
+            "grpc.lb.rls.data_plane_target", "grpc.lb.pick_result"), Lists.newArrayList(), true);
+    FAILED_PICKS_COUNTER = metricInstrumentRegistry.registerLongCounter("grpc.lb.rls.failed_picks",
+        "Number of LB picks failed due to either a failed RLS request or the RLS channel being "
+            + "throttled", "pick", Lists.newArrayList("grpc.target", "grpc.lb.rls.server_target"),
+        Lists.newArrayList(), true);
+  }
+
   private CachingRlsLbClient(Builder builder) {
     helper = new RlsLbHelper(checkNotNull(builder.helper, "helper"));
     scheduledExecutorService = helper.getScheduledExecutorService();
@@ -147,7 +171,7 @@ private CachingRlsLbClient(Builder builder) {
     }
     RlsRequestFactory requestFactory = new RlsRequestFactory(
         lbPolicyConfig.getRouteLookupConfig(), serverHost);
-    rlsPicker = new RlsPicker(requestFactory);
+    rlsPicker = new RlsPicker(requestFactory, rlsConfig.lookupService());
     // It is safe to use helper.getUnsafeChannelCredentials() because the client authenticates the
     // RLS server using the same authority as the backends, even though the RLS server’s addresses
     // will be looked up differently than the backends; overrideAuthority(helper.getAuthority()) is
@@ -904,9 +928,11 @@ public void onStatusChanged(ConnectivityState newState) {
   final class RlsPicker extends SubchannelPicker {
 
     private final RlsRequestFactory requestFactory;
+    private final String lookupService;
 
-    RlsPicker(RlsRequestFactory requestFactory) {
+    RlsPicker(RlsRequestFactory requestFactory, String lookupService) {
       this.requestFactory = checkNotNull(requestFactory, "requestFactory");
+      this.lookupService = checkNotNull(lookupService, "rlsConfig");
     }
 
     @Override
@@ -941,14 +967,24 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {
         }
         // Happy path
         logger.log(ChannelLogLevel.DEBUG, "Returning PickResult");
-        return picker.pickSubchannel(args);
+        PickResult pickResult = picker.pickSubchannel(args);
+        // TODO: include the "grpc.target" label once target is available here.
+        if (pickResult.hasResult()) {
+          helper.getMetricRecorder().addLongCounter(TARGET_PICKS_COUNTER, 1,
+              Lists.newArrayList("", lookupService, childPolicyWrapper.getTarget(),
+                  determineMetricsPickResult(pickResult)), Lists.newArrayList());
+        }
+        return pickResult;
       } else if (response.hasError()) {
         logger.log(ChannelLogLevel.DEBUG, "RLS response has errors");
         if (hasFallback) {
           logger.log(ChannelLogLevel.DEBUG, "Using RLS fallback");
           return useFallback(args);
         }
         logger.log(ChannelLogLevel.DEBUG, "No RLS fallback, returning PickResult with an error");
+        // TODO: include the "grpc.target" label once target is available here.
+        helper.getMetricRecorder().addLongCounter(FAILED_PICKS_COUNTER, 1,
+            Lists.newArrayList("", lookupService), Lists.newArrayList());
         return PickResult.withError(
             convertRlsServerStatus(response.getStatus(),
                 lbPolicyConfig.getRouteLookupConfig().lookupService()));
@@ -969,7 +1005,24 @@ private PickResult useFallback(PickSubchannelArgs args) {
       if (picker == null) {
         return PickResult.withNoResult();
       }
-      return picker.pickSubchannel(args);
+      PickResult pickResult = picker.pickSubchannel(args);
+      if (pickResult.hasResult()) {
+        // TODO: include the grpc.target label once target is available here.
+        helper.getMetricRecorder().addLongCounter(DEFAULT_TARGET_PICKS_COUNTER, 1,
+            Lists.newArrayList("", lookupService, fallbackChildPolicyWrapper.getTarget(),
+                determineMetricsPickResult(pickResult)), Lists.newArrayList());
+      }
+      return pickResult;
+    }
+
+    private String determineMetricsPickResult(PickResult pickResult) {
+      if (pickResult.getStatus().isOk()) {
+        return "complete";
+      } else if (pickResult.isDrop()) {
+        return "drop";
+      } else {
+        return "fail";
+      }
     }
 
     private void startFallbackChildPolicy() {
diff --git a/rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java b/rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java
@@ -50,6 +50,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
+import io.grpc.MetricRecorder;
 import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.Status;
 import io.grpc.Status.Code;
@@ -121,6 +122,8 @@ public class CachingRlsLbClientTest {
   private EvictionListener<RouteLookupRequest, CacheEntry> evictionListener;
   @Mock
   private SocketAddress socketAddress;
+  @Mock
+  private MetricRecorder mockMetricRecorder;
 
   private final SynchronizationContext syncContext =
       new SynchronizationContext(new UncaughtExceptionHandler() {
@@ -892,6 +895,11 @@ public SynchronizationContext getSynchronizationContext() {
     public ChannelLogger getChannelLogger() {
       return mock(ChannelLogger.class);
     }
+
+    @Override
+    public MetricRecorder getMetricRecorder() {
+      return mockMetricRecorder;
+    }
   }
 
   private static final class FakeThrottler implements Throttler {
diff --git a/rls/src/test/java/io/grpc/rls/RlsLoadBalancerTest.java b/rls/src/test/java/io/grpc/rls/RlsLoadBalancerTest.java
