xds: Add counter and gauge metrics

api/src/main/java/io/grpc/NameResolver.java

@@ -288,6 +288,7 @@
     @Nullable private final ChannelLogger channelLogger;
     @Nullable private final Executor executor;
     @Nullable private final String overrideAuthority;
+    @Nullable private final MetricRecorder metricRecorder;
 
     private Args(
         Integer defaultPort,
@@ -297,7 +298,8 @@
         @Nullable ScheduledExecutorService scheduledExecutorService,
         @Nullable ChannelLogger channelLogger,
         @Nullable Executor executor,
-        @Nullable String overrideAuthority) {
+        @Nullable String overrideAuthority,
+        @Nullable MetricRecorder metricRecorder) {
       this.defaultPort = checkNotNull(defaultPort, "defaultPort not set");
       this.proxyDetector = checkNotNull(proxyDetector, "proxyDetector not set");
       this.syncContext = checkNotNull(syncContext, "syncContext not set");
@@ -306,6 +308,7 @@
       this.channelLogger = channelLogger;
       this.executor = executor;
       this.overrideAuthority = overrideAuthority;
+      this.metricRecorder = metricRecorder;
     }
 
     /**
@@ -403,6 +406,15 @@
       return overrideAuthority;
     }
 
+    /**
+     * Returns the {@link MetricRecorder} that the channel uses to record metrics.
+     */
+    @Nullable
+    @ExperimentalApi("Insert GitHub issue")
+    public MetricRecorder getMetricRecorder() {
+      return metricRecorder;
+    }
+
 
     @Override
     public String toString() {
@@ -415,6 +427,7 @@
           .add("channelLogger", channelLogger)
           .add("executor", executor)
           .add("overrideAuthority", overrideAuthority)
+          .add("metricRecorder", metricRecorder)
           .toString();
     }
 
@@ -433,6 +446,7 @@
       builder.setChannelLogger(channelLogger);
       builder.setOffloadExecutor(executor);
       builder.setOverrideAuthority(overrideAuthority);
+      builder.setMetricRecorder(metricRecorder);
       return builder;
     }
 
@@ -459,6 +473,7 @@
       private ChannelLogger channelLogger;
       private Executor executor;
       private String overrideAuthority;
+      private MetricRecorder metricRecorder;
 
       Builder() {
       }
@@ -545,6 +560,15 @@
         return this;
       }
 
+      /**
+       * See {@link Args#getMetricRecorder()}. This is an optional field.
+       */
+      @ExperimentalApi("Insert github issue")
+      public Builder setMetricRecorder(MetricRecorder metricRecorder) {
+        this.metricRecorder = metricRecorder;
+        return this;
+      }
+
       /**
        * Builds an {@link Args}.
        *
@@ -554,7 +578,8 @@
         return
             new Args(
                 defaultPort, proxyDetector, syncContext, serviceConfigParser,
-                scheduledExecutorService, channelLogger, executor, overrideAuthority);
+                scheduledExecutorService, channelLogger, executor, overrideAuthority,
+                metricRecorder);
       }
     }
   }

api/src/test/java/io/grpc/NameResolverTest.java

@@ -64,6 +64,7 @@ public class NameResolverTest {
   private final ChannelLogger channelLogger = mock(ChannelLogger.class);
   private final Executor executor = Executors.newSingleThreadExecutor();
   private final String overrideAuthority = "grpc.io";
+  private final MetricRecorder metricRecorder = new MetricRecorder() {};
   @Mock NameResolver.Listener mockListener;
 
   @Test
@@ -77,6 +78,7 @@ public void args() {
     assertThat(args.getChannelLogger()).isSameInstanceAs(channelLogger);
     assertThat(args.getOffloadExecutor()).isSameInstanceAs(executor);
     assertThat(args.getOverrideAuthority()).isSameInstanceAs(overrideAuthority);
+    assertThat(args.getMetricRecorder()).isSameInstanceAs(metricRecorder);
 
     NameResolver.Args args2 = args.toBuilder().build();
     assertThat(args2.getDefaultPort()).isEqualTo(defaultPort);
@@ -87,6 +89,7 @@ public void args() {
     assertThat(args2.getChannelLogger()).isSameInstanceAs(channelLogger);
     assertThat(args2.getOffloadExecutor()).isSameInstanceAs(executor);
     assertThat(args2.getOverrideAuthority()).isSameInstanceAs(overrideAuthority);
+    assertThat(args.getMetricRecorder()).isSameInstanceAs(metricRecorder);
 
     assertThat(args2).isNotSameInstanceAs(args);
     assertThat(args2).isNotEqualTo(args);
@@ -102,6 +105,7 @@ private NameResolver.Args createArgs() {
         .setChannelLogger(channelLogger)
         .setOffloadExecutor(executor)
         .setOverrideAuthority(overrideAuthority)
+        .setMetricRecorder(metricRecorder)
         .build();
   }
 

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -589,6 +589,8 @@ ClientStream newSubstream(
             builder.maxHedgedAttempts,
             loadBalancerFactory);
     this.authorityOverride = builder.authorityOverride;
+    this.metricRecorder = new MetricRecorderImpl(builder.metricSinks,
+        MetricInstrumentRegistry.getDefaultRegistry());
     this.nameResolverArgs =
         NameResolver.Args.newBuilder()
             .setDefaultPort(builder.getDefaultPort())
@@ -599,6 +601,7 @@ ClientStream newSubstream(
             .setChannelLogger(channelLogger)
             .setOffloadExecutor(this.offloadExecutorHolder)
             .setOverrideAuthority(this.authorityOverride)
+            .setMetricRecorder(this.metricRecorder)
             .build();
     this.nameResolver = getNameResolver(
         targetUri, authorityOverride, nameResolverProvider, nameResolverArgs);
@@ -671,8 +674,6 @@ public CallTracer create() {
       }
       serviceConfigUpdated = true;
     }
-    this.metricRecorder = new MetricRecorderImpl(builder.metricSinks,
-        MetricInstrumentRegistry.getDefaultRegistry());
   }
 
   @VisibleForTesting

core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java

@@ -687,6 +687,30 @@ public void metricRecorder_recordsToMetricSink() {
         eq(optionalLabelValues));
   }
 
+  @Test
+  public void metricRecorder_fromNameResolverArgs_recordsToMetricSink() {
+    MetricSink mockSink1 = mock(MetricSink.class);
+    MetricSink mockSink2 = mock(MetricSink.class);
+    channelBuilder.addMetricSink(mockSink1);
+    channelBuilder.addMetricSink(mockSink2);
+    createChannel();
+
+    LongCounterMetricInstrument counter = metricInstrumentRegistry.registerLongCounter(
+        "test_counter", "Time taken by metric recorder", "s",
+        ImmutableList.of("grpc.method"), Collections.emptyList(), false);
+    List<String> requiredLabelValues = ImmutableList.of("testMethod");
+    List<String> optionalLabelValues = Collections.emptyList();
+
+    NameResolver.Args args = helper.getNameResolverArgs();
+    assertThat(args.getMetricRecorder()).isNotNull();
+    args.getMetricRecorder()
+        .addLongCounter(counter, 10, requiredLabelValues, optionalLabelValues);
+    verify(mockSink1).addLongCounter(eq(counter), eq(10L), eq(requiredLabelValues),
+        eq(optionalLabelValues));
+    verify(mockSink2).addLongCounter(eq(counter), eq(10L), eq(requiredLabelValues),
+        eq(optionalLabelValues));
+  }
+
   @Test
   public void shutdownWithNoTransportsEverCreated() {
     channelBuilder.nameResolverFactory(
@@ -2240,6 +2264,7 @@ public void lbHelper_getNameResolverArgs() {
     assertThat(args.getSynchronizationContext())
         .isSameInstanceAs(helper.getSynchronizationContext());
     assertThat(args.getServiceConfigParser()).isNotNull();
+    assertThat(args.getMetricRecorder()).isNotNull();
   }
 
   @Test

xds/src/main/java/io/grpc/xds/InternalSharedXdsClientPoolProvider.java

@@ -17,6 +17,7 @@
 package io.grpc.xds;
 
 import io.grpc.Internal;
+import io.grpc.MetricRecorder;
 import io.grpc.internal.ObjectPool;
 import io.grpc.xds.client.XdsClient;
 import io.grpc.xds.client.XdsInitializationException;
@@ -36,6 +37,11 @@
 
   public static ObjectPool<XdsClient> getOrCreate(String target)
       throws XdsInitializationException {
-    return SharedXdsClientPoolProvider.getDefaultProvider().getOrCreate(target);
+    return getOrCreate(target, new MetricRecorder() {});
+  }
+
+  public static ObjectPool<XdsClient> getOrCreate(String target, MetricRecorder metricRecorder)
+      throws XdsInitializationException {
+    return SharedXdsClientPoolProvider.getDefaultProvider().getOrCreate(target, metricRecorder);
   }
 }

xds/src/main/java/io/grpc/xds/SharedXdsClientPoolProvider.java

@@ -21,6 +21,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import io.grpc.MetricRecorder;
 import io.grpc.internal.ExponentialBackoffPolicy;
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.ObjectPool;
@@ -56,6 +57,7 @@ final class SharedXdsClientPoolProvider implements XdsClientPoolFactory {
   private final Object lock = new Object();
   private final AtomicReference<Map<String, ?>> bootstrapOverride = new AtomicReference<>();
   private final Map<String, ObjectPool<XdsClient>> targetToXdsClientMap = new ConcurrentHashMap<>();
+  private final Map<String, MetricRecorder> targetToMetricRecorderMap = new ConcurrentHashMap<>();
 
   SharedXdsClientPoolProvider() {
     this(new GrpcBootstrapperImpl());
@@ -82,7 +84,14 @@ public ObjectPool<XdsClient> get(String target) {
   }
 
   @Override
+  @Nullable
   public ObjectPool<XdsClient> getOrCreate(String target) throws XdsInitializationException {
+    return this.getOrCreate(target, new MetricRecorder() {});
+  }
+
+  @Override
+  public ObjectPool<XdsClient> getOrCreate(String target, MetricRecorder metricRecorder)
+      throws XdsInitializationException {
     ObjectPool<XdsClient> ref = targetToXdsClientMap.get(target);
     if (ref == null) {
       synchronized (lock) {
@@ -98,8 +107,12 @@ public ObjectPool<XdsClient> getOrCreate(String target) throws XdsInitialization
           if (bootstrapInfo.servers().isEmpty()) {
             throw new XdsInitializationException("No xDS server provided");
           }
-          ref = new RefCountedXdsClientObjectPool(bootstrapInfo, target);
+          MetricRecorder metricRecorderForTarget = targetToMetricRecorderMap.get(target);
+          metricRecorder =
+              metricRecorderForTarget != null ? metricRecorderForTarget : metricRecorder;
+          ref = new RefCountedXdsClientObjectPool(bootstrapInfo, target, metricRecorder);
           targetToXdsClientMap.put(target, ref);
+          targetToMetricRecorderMap.putIfAbsent(target, metricRecorder);
         }
       }
     }
@@ -111,6 +124,11 @@ public ImmutableList<String> getTargets() {
     return ImmutableList.copyOf(targetToXdsClientMap.keySet());
   }
 
+  @VisibleForTesting
+  MetricRecorder getMetricRecorder(String target) {
+    return targetToMetricRecorderMap.get(target);
+  }
+
 
   private static class SharedXdsClientPoolProviderHolder {
     private static final SharedXdsClientPoolProvider instance = new SharedXdsClientPoolProvider();
@@ -124,6 +142,7 @@ static class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
         new ExponentialBackoffPolicy.Provider();
     private final BootstrapInfo bootstrapInfo;
     private final String target; // The target associated with the xDS client.
+    private final MetricRecorder metricRecorder;
     private final Object lock = new Object();
     @GuardedBy("lock")
     private ScheduledExecutorService scheduler;
@@ -133,9 +152,11 @@ static class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
     private int refCount;
 
     @VisibleForTesting
-    RefCountedXdsClientObjectPool(BootstrapInfo bootstrapInfo, String target) {
+    RefCountedXdsClientObjectPool(BootstrapInfo bootstrapInfo, String target,
+        MetricRecorder metricRecorder) {
       this.bootstrapInfo = checkNotNull(bootstrapInfo);
       this.target = target;
+      this.metricRecorder = metricRecorder;
     }
 
     @Override
@@ -154,7 +175,9 @@ public XdsClient getObject() {
               GrpcUtil.STOPWATCH_SUPPLIER,
               TimeProvider.SYSTEM_TIME_PROVIDER,
               MessagePrinter.INSTANCE,
-              new TlsContextManagerImpl(bootstrapInfo));
+              new TlsContextManagerImpl(bootstrapInfo),
+              getTarget(),
+              new XdsClientMetricReporterImpl(metricRecorder));
         }
         refCount++;
         return xdsClient;

xds/src/main/java/io/grpc/xds/XdsClientMetricReporter.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * 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
+ *
+ *     http://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 io.grpc.xds;
+
+import io.grpc.Internal;
+import io.grpc.xds.client.XdsClient;
+
+/**
+ * Interface for reporting metrics from the xDS client.
+ */
+@Internal
+public interface XdsClientMetricReporter {
+
+  /**
+   * Reports number of valid and invalid resources.
+   *
+   * @param validResourceCount Number of resources that were valid.
+   * @param invalidResourceCount Number of resources that were invalid.
+   * @param target Target of the gRPC channel.
+   * @param xdsServer Target URI of the xDS server with which the XdsClient is communicating.
+   * @param resourceType Type of XDS resource (e.g., "envoy.config.listener.v3.Listener").
+   */
+  default void reportResourceUpdates(long validResourceCount, long invalidResourceCount,
+      String target, String xdsServer, String resourceType) {
+  }
+
+  /**
+   * Reports number of xDS servers going from healthy to unhealthy.
+   *
+   * @param serverFailure Number of xDS server failures.
+   * @param target Target of the gRPC channel.
+   * @param xdsServer Target URI of the xDS server with which the XdsClient is communicating.
+   */
+  default void reportServerFailure(long serverFailure, String target, String xdsServer) {
+  }
+
+  /**
+   * Sets the {@link XdsClient} instance.
+   */
+  default void setXdsClient(XdsClient xdsClient) {
+  }
+
+  /**
+   * Closes the metric reporter.
+   */
+  default void close() {
+  }
+
+  /**
+   * Interface for reporting metrics through callback.
+   *
+   */
+  interface CallbackMetricReporter {
+
+    /**
+     * Reports number of resources in each cache state.
+     *
+     * @param resourceCount Number of resources.
+     * @param cacheState Status of the resource metadata
+     *     {@link io.grpc.xds.client.XdsClient.ResourceMetadata.ResourceMetadataStatus}.
+     * @param resourceType Type of XDS resource (e.g., "envoy.config.listener.v3.Listener").
+     * @param target Target of the gRPC channel.
+     */
+    // TODO(@dnvindhya): include the "authority" label once xds.authority is available.
+    default void reportResourceCounts(long resourceCount, String cacheState, String resourceType,
+        String target) {
+    }
+
+    /**
+     * Reports whether xDS client has a working ADS stream to the xDS server.
+     *
+     * @param isConnected 1 if the client is connected to the xDS server, 0 otherwise.
+     * @param target Target of the gRPC channel.
+     * @param xdsServer Target URI of the xDS server with which the XdsClient is communicating.
+     */
+    default void reportServerConnections(int isConnected, String target, String xdsServer) {
+    }
+  }
+}