Rebase grpc#2

Author: DNVindhya

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

@@ -564,8 +564,6 @@ public Builder setOverrideAuthority(String authority) {
 
       /**
        * See {@link Args#getMetricRecorder()}. This is an optional field.
-       *
-       * @since 1.67.0
        */
       @ExperimentalApi("Insert github issue")
       public Builder setMetricRecorder(MetricRecorder 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/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/SharedXdsClientPoolProvider.java

@@ -124,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();

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

@@ -53,6 +53,8 @@ public class XdsClientMetricReporterImpl implements XdsClientMetricReporter {
   private Registration gaugeRegistration = null;
   @Nullable
   private XdsClient xdsClient = null;
+  @Nullable
+  private CallbackMetricReporter callbackMetricReporter = null;
 
   static {
     MetricInstrumentRegistry metricInstrumentRegistry
@@ -129,12 +131,18 @@ public void accept(BatchRecorder recorder) {
 
   @VisibleForTesting
   void reportCallbackMetrics(BatchRecorder recorder) {
-    CallbackMetricReporter callbackMetricReporter = new CallbackMetricReporterImpl(recorder);
+    if (callbackMetricReporter == null) {
+      // Instantiate only if not injected
+      callbackMetricReporter = new CallbackMetricReporterImpl(recorder);
+    }
     try {
       reportResourceCounts(callbackMetricReporter);
       reportServerConnections(callbackMetricReporter);
     } catch (Exception e) {
-      logger.log(Level.WARNING, "Reporting gauge metrics failed", e);
+      if (e instanceof InterruptedException) {
+        Thread.currentThread().interrupt(); // re-set the current thread's interruption state
+      }
+      logger.log(Level.WARNING, "Failed to report gauge metrics", e);
     }
   }
 
@@ -147,7 +155,7 @@ void reportResourceCounts(CallbackMetricReporter callbackMetricReporter) throws
     SettableFuture<Void> ret = this.xdsClient.reportResourceCounts(
         callbackMetricReporter);
     // Normally this shouldn't take long, but adding a timeout to avoid indefinite blocking
-    ret.get(5, TimeUnit.SECONDS);
+    Void unused = ret.get(5, TimeUnit.SECONDS);
   }
 
   /**
@@ -158,7 +166,15 @@ void reportResourceCounts(CallbackMetricReporter callbackMetricReporter) throws
   void reportServerConnections(CallbackMetricReporter callbackMetricReporter) throws Exception {
     SettableFuture<Void> ret = this.xdsClient.reportServerConnections(callbackMetricReporter);
     // Normally this shouldn't take long, but adding a timeout to avoid indefinite blocking
-    ret.get(5, TimeUnit.SECONDS);
+    Void unused = ret.get(5, TimeUnit.SECONDS);
+  }
+
+  /**
+   * Allows injecting a custom {@link CallbackMetricReporter} for testing purposes.
+   */
+  @VisibleForTesting
+  void injectCallbackMetricReporter(CallbackMetricReporter reporter) {
+    this.callbackMetricReporter = reporter;
   }
 
   @VisibleForTesting

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

@@ -143,7 +143,7 @@ final class XdsNameResolver extends NameResolver {
       ServiceConfigParser serviceConfigParser,
       SynchronizationContext syncContext, ScheduledExecutorService scheduler,
       @Nullable Map<String, ?> bootstrapOverride,
-      @Nullable MetricRecorder metricRecorder) {
+      MetricRecorder metricRecorder) {
     this(targetUri, targetUri.getAuthority(), name, overrideAuthority, serviceConfigParser,
         syncContext, scheduler, SharedXdsClientPoolProvider.getDefaultProvider(),
         ThreadSafeRandomImpl.instance, FilterRegistry.getDefaultRegistry(), bootstrapOverride,
@@ -157,7 +157,7 @@ final class XdsNameResolver extends NameResolver {
       SynchronizationContext syncContext, ScheduledExecutorService scheduler,
       XdsClientPoolFactory xdsClientPoolFactory, ThreadSafeRandom random,
       FilterRegistry filterRegistry, @Nullable Map<String, ?> bootstrapOverride,
-      @Nullable MetricRecorder metricRecorder) {
+      MetricRecorder metricRecorder) {
     this.targetAuthority = targetAuthority;
     target = targetUri.toString();
 

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -547,8 +547,8 @@ public SettableFuture<Void> reportServerConnections(
               controlPlaneClient.hasWorkingAdsStream() ? 1 : 0,
               target,
               serverInfo.target()));
-      future.set(null);
     });
+    future.set(null);
     return future;
   }
 
@@ -559,8 +559,8 @@ public SettableFuture<Void> reportResourceCounts(CallbackMetricReporter callback
       Map<XdsResourceType<?>, Map<String, Long>> resourceCountsByType =
           getResourceCountsByType();
       reportResourceCountsToCallback(callbackMetricReporter, resourceCountsByType);
-      future.set(null);
     });
+    future.set(null);
     return future;
   }
 
@@ -571,7 +571,7 @@ private String cacheStateFromResourceStatus(ResourceMetadata metadata, boolean i
   }
 
   /**
-   * Calculates resource counts by ResourceType and ResourceSubscriber.metadata.status
+   * Calculates number of resources by ResourceType and ResourceSubscriber.metadata.status
    */
   Map<XdsResourceType<?>, Map<String, Long>> getResourceCountsByType() {
     Map<XdsResourceType<?>, Map<String, Long>> resourceCountsByType = new HashMap<>();