Add MetricSink implementation for gRPC OpenTelemetry

This adds the following components that are required for gRPC A79
non-per-call metrics architecture.

- MetricSink implementation for gRPC OpenTelemetry
- Configurator for plumbing per call metrics ClientInterceptor and
  ServerStreamTracer.Factory via unified OpenTelemetryModule.

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

@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+/**
+ * Provides hooks for modifying gRPC channels and servers during their construction.
+ */
+@Internal
+public interface Configurator {
+  /**
+   * Allows implementations to modify the channel builder.
+   *
+   * @param channelBuilder the channel builder being constructed
+   */
+  default void configureChannelBuilder(ManagedChannelBuilder<?> channelBuilder) {}
+
+  /**
+   * Allows implementations to modify the server builder.
+   *
+   * @param serverBuilder the server builder being constructed
+   */
+  default void configureServerBuilder(ServerBuilder<?> serverBuilder) {}
+}

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

@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A registry for {@link Configurator} instances.
+ *
+ * <p>This class is responsible for maintaining a list of configurators and providing access to
+ * them. The default registry can be obtained using {@link #getDefaultRegistry()}.
+ */
+@Internal
+public final class ConfiguratorRegistry {
+  private static ConfiguratorRegistry instance;
+  private static boolean isConfiguratorsSet;
+  private List<Configurator> configurators = Collections.emptyList();
+
+  ConfiguratorRegistry() {}
+
+  /**
+   * Returns the default global instance of the configurator registry.
+   */
+  public static synchronized ConfiguratorRegistry getDefaultRegistry() {
+    if (instance == null) {
+      instance = new ConfiguratorRegistry();
+    }
+    return instance;
+  }
+
+  /**
+   * Sets the configurators in this registry. This method can only be called once.
+   *
+   * @param configurators the configurators to set
+   * @throws IllegalStateException if this method is called more than once
+   */
+  public synchronized void setConfigurators(List<Configurator> configurators) {
+    if (isConfiguratorsSet) {
+      throw new IllegalStateException("Configurators are already set");
+    }
+    configurators = Collections.unmodifiableList(new ArrayList<>(configurators));
+    isConfiguratorsSet = true;
+  }
+
+  /**
+   * Returns a list of the configurators in this registry.
+   */
+  public synchronized List<Configurator> getConfigurators() {
+    return configurators;
+  }
+}

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

@@ -23,7 +23,7 @@
  */
 @Internal
 public final class DoubleCounterMetricInstrument extends PartialMetricInstrument {
-  DoubleCounterMetricInstrument(int index, String name, String description, String unit,
+  public DoubleCounterMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
   }

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

@@ -25,7 +25,7 @@
 public final class DoubleHistogramMetricInstrument extends PartialMetricInstrument {
   private final List<Double> bucketBoundaries;
 
-  DoubleHistogramMetricInstrument(int index, String name, String description, String unit,
+  public DoubleHistogramMetricInstrument(int index, String name, String description, String unit,
       List<Double> bucketBoundaries, List<String> requiredLabelKeys, List<String> optionalLabelKeys,
       boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);

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

@@ -23,7 +23,7 @@
  */
 @Internal
 public final class LongCounterMetricInstrument extends PartialMetricInstrument {
-  LongCounterMetricInstrument(int index, String name, String description, String unit,
+  public LongCounterMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
   }

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

@@ -23,7 +23,7 @@
  */
 @Internal
 public final class LongGaugeMetricInstrument extends PartialMetricInstrument {
-  LongGaugeMetricInstrument(int index, String name, String description, String unit,
+  public LongGaugeMetricInstrument(int index, String name, String description, String unit,
       List<String> requiredLabelKeys, List<String> optionalLabelKeys, boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);
   }

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

@@ -25,7 +25,7 @@
 public final class LongHistogramMetricInstrument extends PartialMetricInstrument {
   private final List<Long> bucketBoundaries;
 
-  LongHistogramMetricInstrument(int index, String name, String description, String unit,
+  public LongHistogramMetricInstrument(int index, String name, String description, String unit,
       List<Long> bucketBoundaries, List<String> requiredLabelKeys, List<String> optionalLabelKeys,
       boolean enableByDefault) {
     super(index, name, description, unit, requiredLabelKeys, optionalLabelKeys, enableByDefault);

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

@@ -95,6 +95,7 @@ public DoubleCounterMetricInstrument registerDoubleCounter(String name,
       if (index + 1 == metricInstruments.length) {
         resizeMetricInstruments();
       }
+      // TODO(dnvindhya): add limit for number of optional labels allowed
       DoubleCounterMetricInstrument instrument = new DoubleCounterMetricInstrument(
           index, name, description, unit, requiredLabelKeys, optionalLabelKeys,
           enableByDefault);

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

@@ -17,6 +17,7 @@
 package io.grpc;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -26,50 +27,49 @@
 public interface MetricSink {
 
   /**
-   * Returns a set of names for the metrics that are currently enabled for this `MetricSink`.
+   * Returns a set of names for the metrics that are currently enabled or disabled.
    *
    * @return A set of enabled metric names.
    */
-  Set<String> getEnabledMetrics();
+  Map<String, Boolean> getEnabledMetrics();
 
   /**
-   * Returns a list of label names that are considered optional for metrics collected by this
-   * `MetricSink`.
+   * Returns a set of optional label names for metrics that the sink actually wants.
    *
-   * @return A list of optional label names.
+   * @return A set of optional label names.
    */
-  List<String> getOptionalLabels();
+  Set<String> getOptionalLabels();
 
   /**
-   * Returns a list of metric measures used to record metric values. These measures are created
+   * Returns size of metric measures used to record metric values. These measures are created
    * based on registered metrics (via MetricInstrumentRegistry) and are ordered according to their
    * registration sequence.
    *
-   * @return A list of metric measures.
+   * @return Size of metric measures.
    */
-  List<Object> getMetricsMeasures();
+  int getMeasuresSize();
 
   /**
-   * Records a value for a double-precision counter associated with specified metric instrument.
+   * Adds a value for a double-precision counter associated with specified metric instrument.
    *
-   * @param metricInstrument The counter metric instrument identifies metric measure to record.
+   * @param metricInstrument The counter metric instrument identifies metric measure to add.
    * @param value The value to record.
    * @param requiredLabelValues A list of required label values for the metric.
    * @param optionalLabelValues A list of additional, optional label values for the metric.
    */
-  default void recordDoubleCounter(DoubleCounterMetricInstrument metricInstrument, double value,
+  default void addDoubleCounter(DoubleCounterMetricInstrument metricInstrument, double value,
       List<String> requiredLabelValues, List<String> optionalLabelValues) {
   }
 
   /**
-   * Records a value for a long valued counter metric associated with specified metric instrument.
+   * Adds a value for a long valued counter metric associated with specified metric instrument.
    *
-   * @param metricInstrument The counter metric instrument identifies metric measure to record.
+   * @param metricInstrument The counter metric instrument identifies metric measure to add.
    * @param value The value to record.
    * @param requiredLabelValues A list of required label values for the metric.
    * @param optionalLabelValues A list of additional, optional label values for the metric.
    */
-  default void recordLongCounter(LongCounterMetricInstrument metricInstrument, long value,
+  default void addLongCounter(LongCounterMetricInstrument metricInstrument, long value,
       List<String> requiredLabelValues, List<String> optionalLabelValues) {
   }
 
@@ -99,5 +99,5 @@ default void recordLongHistogram(LongHistogramMetricInstrument metricInstrument,
       List<String> requiredLabelValues, List<String> optionalLabelValues) {
   }
 
-  default void updateMeasures(List<MetricInstrument> instruments) {}
+  void updateMeasures(List<MetricInstrument> instruments);
 }

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

@@ -72,6 +72,8 @@
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
+import io.grpc.MetricInstrumentRegistry;
+import io.grpc.MetricRecorder;
 import io.grpc.NameResolver;
 import io.grpc.NameResolver.ConfigOrError;
 import io.grpc.NameResolver.ResolutionResult;
@@ -588,6 +590,7 @@ ClientStream newSubstream(
   private final ChannelStreamProvider transportProvider = new ChannelStreamProvider();
 
   private final Rescheduler idleTimer;
+  private final MetricRecorder metricRecorder;
 
   ManagedChannelImpl(
       ManagedChannelImplBuilder builder,
@@ -715,6 +718,8 @@ public CallTracer create() {
       }
       serviceConfigUpdated = true;
     }
+    this.metricRecorder = new MetricRecorderImpl(builder.metricSinks,
+        MetricInstrumentRegistry.getDefaultRegistry());
   }
 
   @VisibleForTesting
@@ -1743,6 +1748,11 @@ public NameResolverRegistry getNameResolverRegistry() {
       return nameResolverRegistry;
     }
 
+    @Override
+    public MetricRecorder getMetricRecorder() {
+      return metricRecorder;
+    }
+
     /**
      * A placeholder for channel creds if user did not specify channel creds for the channel.
      */

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

@@ -30,12 +30,15 @@
 import io.grpc.ClientInterceptor;
 import io.grpc.ClientTransportFilter;
 import io.grpc.CompressorRegistry;
+import io.grpc.Configurator;
+import io.grpc.ConfiguratorRegistry;
 import io.grpc.DecompressorRegistry;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.InternalChannelz;
 import io.grpc.InternalGlobalInterceptors;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.MetricSink;
 import io.grpc.NameResolver;
 import io.grpc.NameResolverProvider;
 import io.grpc.NameResolverRegistry;
@@ -192,6 +195,7 @@ public static ManagedChannelBuilder<?> forTarget(String target) {
   private boolean recordRealTimeMetrics = false;
   private boolean recordRetryMetrics = true;
   private boolean tracingEnabled = true;
+  List<MetricSink> metricSinks = new ArrayList<>();
 
   /**
    * An interface for Transport implementors to provide the {@link ClientTransportFactory}
@@ -340,6 +344,10 @@ public ManagedChannelImplBuilder(SocketAddress directServerAddress, String autho
     } else {
       this.channelBuilderDefaultPortProvider = new ManagedChannelDefaultPortProvider();
     }
+    // TODO(dnvindhya): Move configurator to all the individual builders
+    for (Configurator configurator : ConfiguratorRegistry.getDefaultRegistry().getConfigurators()) {
+      configurator.configureChannelBuilder(this);
+    }
   }
 
   @Override
@@ -661,6 +669,12 @@ public ManagedChannelImplBuilder enableCheckAuthority() {
     return this;
   }
 
+  @Override
+  public ManagedChannelImplBuilder addMetricSink(MetricSink metricSink) {
+    metricSinks.add(checkNotNull(metricSink, "metric sink"));
+    return this;
+  }
+
   @Override
   public ManagedChannel build() {
     return new ManagedChannelOrphanWrapper(new ManagedChannelImpl(
@@ -680,6 +694,7 @@ public ManagedChannel build() {
   List<ClientInterceptor> getEffectiveInterceptors() {
     List<ClientInterceptor> effectiveInterceptors = new ArrayList<>(this.interceptors);
     boolean isGlobalInterceptorsSet = false;
+    // TODO(dnvindhya) : Convert to Configurator
     List<ClientInterceptor> globalClientInterceptors =
         InternalGlobalInterceptors.getClientInterceptors();
     if (globalClientInterceptors != null) {

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

@@ -68,14 +68,14 @@ public void addDoubleCounter(DoubleCounterMetricInstrument metricInstrument, dou
             + metricInstrument.getOptionalLabelKeys().size());
     for (MetricSink sink : metricSinks) {
       // TODO(dnvindhya): Move updating measures logic from sink to here
-      List<Object> measures = sink.getMetricsMeasures();
-      if (measures.size() <= metricInstrument.getIndex()) {
+      int measuresSize = sink.getMeasuresSize();
+      if (measuresSize <= metricInstrument.getIndex()) {
         // Measures may need updating in two cases:
         // 1. When the sink is initially created with an empty list of measures.
         // 2. When new metric instruments are registered, requiring the sink to accommodate them.
         sink.updateMeasures(registry.getMetricInstruments());
       }
-      sink.recordDoubleCounter(metricInstrument, value, requiredLabelValues, optionalLabelValues);
+      sink.addDoubleCounter(metricInstrument, value, requiredLabelValues, optionalLabelValues);
     }
   }
 
@@ -99,14 +99,14 @@ public void addLongCounter(LongCounterMetricInstrument metricInstrument, long va
         "Incorrect number of optional labels provided. Expected: "
             + metricInstrument.getOptionalLabelKeys().size());
     for (MetricSink sink : metricSinks) {
-      List<Object> measures = sink.getMetricsMeasures();
-      if (measures.size() <= metricInstrument.getIndex()) {
+      int measuresSize = sink.getMeasuresSize();
+      if (measuresSize <= metricInstrument.getIndex()) {
         // Measures may need updating in two cases:
         // 1. When the sink is initially created with an empty list of measures.
         // 2. When new metric instruments are registered, requiring the sink to accommodate them.
         sink.updateMeasures(registry.getMetricInstruments());
       }
-      sink.recordLongCounter(metricInstrument, value, requiredLabelValues, optionalLabelValues);
+      sink.addLongCounter(metricInstrument, value, requiredLabelValues, optionalLabelValues);
     }
   }
 
@@ -130,8 +130,8 @@ public void recordDoubleHistogram(DoubleHistogramMetricInstrument metricInstrume
         "Incorrect number of optional labels provided. Expected: "
             + metricInstrument.getOptionalLabelKeys().size());
     for (MetricSink sink : metricSinks) {
-      List<Object> measures = sink.getMetricsMeasures();
-      if (measures.size() <= metricInstrument.getIndex()) {
+      int measuresSize = sink.getMeasuresSize();
+      if (measuresSize <= metricInstrument.getIndex()) {
         // Measures may need updating in two cases:
         // 1. When the sink is initially created with an empty list of measures.
         // 2. When new metric instruments are registered, requiring the sink to accommodate them.
@@ -161,8 +161,8 @@ public void recordLongHistogram(LongHistogramMetricInstrument metricInstrument,
         "Incorrect number of optional labels provided. Expected: "
             + metricInstrument.getOptionalLabelKeys().size());
     for (MetricSink sink : metricSinks) {
-      List<Object> measures = sink.getMetricsMeasures();
-      if (measures.size() <= metricInstrument.getIndex()) {
+      int measuresSize = sink.getMeasuresSize();
+      if (measuresSize <= metricInstrument.getIndex()) {
         // Measures may need updating in two cases:
         // 1. When the sink is initially created with an empty list of measures.
         // 2. When new metric instruments are registered, requiring the sink to accommodate them.