Add config to enable Default Exponential Histogram for Prometheus Exporter

exporters/common/src/main/java/io/opentelemetry/exporter/internal/ExporterBuilderUtil.java

@@ -5,9 +5,15 @@
 
 package io.opentelemetry.exporter.internal;
 
+import static io.opentelemetry.sdk.metrics.Aggregation.explicitBucketHistogram;
+
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException;
 import io.opentelemetry.sdk.common.export.MemoryMode;
+import io.opentelemetry.sdk.metrics.Aggregation;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
+import io.opentelemetry.sdk.metrics.internal.aggregator.AggregationUtil;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Locale;
@@ -54,5 +60,24 @@ public static void configureExporterMemoryMode(
     memoryModeConsumer.accept(memoryMode);
   }
 
+  /**
+   * Invoke the {@code defaultAggregationSelectorConsumer} with the configured {@link
+   * DefaultAggregationSelector}.
+   */
+  public static void configureHistogramDefaultAggregation(
+      String defaultHistogramAggregation,
+      Consumer<DefaultAggregationSelector> defaultAggregationSelectorConsumer) {
+    if (AggregationUtil.aggregationName(Aggregation.base2ExponentialBucketHistogram())
+        .equalsIgnoreCase(defaultHistogramAggregation)) {
+      defaultAggregationSelectorConsumer.accept(
+          DefaultAggregationSelector.getDefault()
+              .with(InstrumentType.HISTOGRAM, Aggregation.base2ExponentialBucketHistogram()));
+    } else if (!AggregationUtil.aggregationName(explicitBucketHistogram())
+        .equalsIgnoreCase(defaultHistogramAggregation)) {
+      throw new ConfigurationException(
+          "Unrecognized default histogram aggregation: " + defaultHistogramAggregation);
+    }
+  }
+
   private ExporterBuilderUtil() {}
 }

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/internal/OtlpConfigUtil.java

@@ -5,19 +5,14 @@
 
 package io.opentelemetry.exporter.otlp.internal;
 
-import static io.opentelemetry.sdk.metrics.Aggregation.explicitBucketHistogram;
-
 import io.opentelemetry.exporter.internal.ExporterBuilderUtil;
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException;
 import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
-import io.opentelemetry.sdk.metrics.Aggregation;
-import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
 import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
-import io.opentelemetry.sdk.metrics.internal.aggregator.AggregationUtil;
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
@@ -197,18 +192,9 @@ public static void configureOtlpHistogramDefaultAggregation(
       Consumer<DefaultAggregationSelector> defaultAggregationSelectorConsumer) {
     String defaultHistogramAggregation =
         config.getString("otel.exporter.otlp.metrics.default.histogram.aggregation");
-    if (defaultHistogramAggregation == null) {
-      return;
-    }
-    if (AggregationUtil.aggregationName(Aggregation.base2ExponentialBucketHistogram())
-        .equalsIgnoreCase(defaultHistogramAggregation)) {
-      defaultAggregationSelectorConsumer.accept(
-          DefaultAggregationSelector.getDefault()
-              .with(InstrumentType.HISTOGRAM, Aggregation.base2ExponentialBucketHistogram()));
-    } else if (!AggregationUtil.aggregationName(explicitBucketHistogram())
-        .equalsIgnoreCase(defaultHistogramAggregation)) {
-      throw new ConfigurationException(
-          "Unrecognized default histogram aggregation: " + defaultHistogramAggregation);
+    if (defaultHistogramAggregation != null) {
+      ExporterBuilderUtil.configureHistogramDefaultAggregation(
+          defaultHistogramAggregation, defaultAggregationSelectorConsumer);
     }
   }
 

exporters/prometheus/src/main/java/io/opentelemetry/exporter/prometheus/PrometheusHttpServer.java

@@ -14,9 +14,11 @@
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.MemoryMode;
 import io.opentelemetry.sdk.internal.DaemonThreadFactory;
+import io.opentelemetry.sdk.metrics.Aggregation;
 import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.export.CollectionRegistration;
+import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
 import io.prometheus.metrics.exporter.httpserver.HTTPServer;
 import io.prometheus.metrics.model.registry.PrometheusRegistry;
@@ -41,6 +43,7 @@ public final class PrometheusHttpServer implements MetricReader {
   private final PrometheusRegistry prometheusRegistry;
   private final String host;
   private final MemoryMode memoryMode;
+  private final DefaultAggregationSelector defaultAggregationSelector;
 
   /**
    * Returns a new {@link PrometheusHttpServer} which can be registered to an {@link
@@ -65,7 +68,8 @@ public static PrometheusHttpServerBuilder builder() {
       boolean otelScopeEnabled,
       @Nullable Predicate<String> allowedResourceAttributesFilter,
       MemoryMode memoryMode,
-      @Nullable HttpHandler defaultHandler) {
+      @Nullable HttpHandler defaultHandler,
+      DefaultAggregationSelector defaultAggregationSelector) {
     this.builder = builder;
     this.prometheusMetricReader =
         new PrometheusMetricReader(otelScopeEnabled, allowedResourceAttributesFilter);
@@ -92,13 +96,19 @@ public static PrometheusHttpServerBuilder builder() {
     } catch (IOException e) {
       throw new UncheckedIOException("Could not create Prometheus HTTP server", e);
     }
+    this.defaultAggregationSelector = defaultAggregationSelector;
   }
 
   @Override
   public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
     return prometheusMetricReader.getAggregationTemporality(instrumentType);
   }
 
+  @Override
+  public Aggregation getDefaultAggregation(InstrumentType instrumentType) {
+    return defaultAggregationSelector.getDefaultAggregation(instrumentType);
+  }
+
   @Override
   public MemoryMode getMemoryMode() {
     return memoryMode;

exporters/prometheus/src/main/java/io/opentelemetry/exporter/prometheus/PrometheusHttpServerBuilder.java

@@ -10,6 +10,9 @@
 
 import com.sun.net.httpserver.HttpHandler;
 import io.opentelemetry.sdk.common.export.MemoryMode;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
+import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import io.prometheus.metrics.model.registry.PrometheusRegistry;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -31,6 +34,8 @@ public final class PrometheusHttpServerBuilder {
   @Nullable private ExecutorService executor;
   private MemoryMode memoryMode = DEFAULT_MEMORY_MODE;
   @Nullable private HttpHandler defaultHandler;
+  private DefaultAggregationSelector defaultAggregationSelector =
+      DefaultAggregationSelector.getDefault();
 
   PrometheusHttpServerBuilder() {}
 
@@ -41,6 +46,7 @@ public final class PrometheusHttpServerBuilder {
     this.otelScopeEnabled = builder.otelScopeEnabled;
     this.allowedResourceAttributesFilter = builder.allowedResourceAttributesFilter;
     this.executor = builder.executor;
+    this.defaultAggregationSelector = builder.defaultAggregationSelector;
   }
 
   /** Sets the host to bind to. If unset, defaults to {@value #DEFAULT_HOST}. */
@@ -126,6 +132,19 @@ public PrometheusHttpServerBuilder setDefaultHandler(HttpHandler defaultHandler)
     return this;
   }
 
+  /**
+   * Set the {@link DefaultAggregationSelector} used for {@link
+   * MetricExporter#getDefaultAggregation(InstrumentType)}.
+   *
+   * <p>If unset, defaults to {@link DefaultAggregationSelector#getDefault()}.
+   */
+  public PrometheusHttpServerBuilder setDefaultAggregationSelector(
+      DefaultAggregationSelector defaultAggregationSelector) {
+    requireNonNull(defaultAggregationSelector, "defaultAggregationSelector");
+    this.defaultAggregationSelector = defaultAggregationSelector;
+    return this;
+  }
+
   /**
    * Returns a new {@link PrometheusHttpServer} with the configuration of this builder which can be
    * registered with a {@link io.opentelemetry.sdk.metrics.SdkMeterProvider}.
@@ -140,6 +159,7 @@ public PrometheusHttpServer build() {
         otelScopeEnabled,
         allowedResourceAttributesFilter,
         memoryMode,
-        defaultHandler);
+        defaultHandler,
+        defaultAggregationSelector);
   }
 }

exporters/prometheus/src/main/java/io/opentelemetry/exporter/prometheus/internal/PrometheusMetricReaderProvider.java

@@ -35,6 +35,14 @@ public MetricReader createMetricReader(ConfigProperties config) {
 
     ExporterBuilderUtil.configureExporterMemoryMode(config, prometheusBuilder::setMemoryMode);
 
+    String defaultHistogramAggregation =
+        config.getString(
+            "otel.java.experimental.exporter.prometheus.metrics.default.histogram.aggregation");
+    if (defaultHistogramAggregation != null) {
+      ExporterBuilderUtil.configureHistogramDefaultAggregation(
+          defaultHistogramAggregation, prometheusBuilder::setDefaultAggregationSelector);
+    }
+
     return prometheusBuilder.build();
   }
 

exporters/prometheus/src/test/java/io/opentelemetry/exporter/prometheus/CollectorIntegrationTest.java

@@ -17,18 +17,24 @@
 import com.linecorp.armeria.server.grpc.protocol.AbstractUnaryGrpcService;
 import com.linecorp.armeria.testing.junit5.server.ServerExtension;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceResponse;
 import io.opentelemetry.proto.common.v1.AnyValue;
 import io.opentelemetry.proto.common.v1.KeyValue;
 import io.opentelemetry.proto.metrics.v1.AggregationTemporality;
+import io.opentelemetry.proto.metrics.v1.Histogram;
+import io.opentelemetry.proto.metrics.v1.HistogramDataPoint;
 import io.opentelemetry.proto.metrics.v1.Metric;
 import io.opentelemetry.proto.metrics.v1.NumberDataPoint;
 import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
 import io.opentelemetry.proto.metrics.v1.ScopeMetrics;
 import io.opentelemetry.proto.metrics.v1.Sum;
+import io.opentelemetry.sdk.metrics.Aggregation;
+import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.UncheckedIOException;
 import java.time.Duration;
@@ -71,7 +77,14 @@ class CollectorIntegrationTest {
 
   @BeforeAll
   static void beforeAll() {
-    PrometheusHttpServer prometheusHttpServer = PrometheusHttpServer.builder().setPort(0).build();
+    PrometheusHttpServer prometheusHttpServer =
+        PrometheusHttpServer.builder()
+            .setPort(0)
+            .setDefaultAggregationSelector(
+                DefaultAggregationSelector.getDefault()
+                    .with(InstrumentType.HISTOGRAM, Aggregation.base2ExponentialBucketHistogram()))
+            .build();
+
     prometheusPort = prometheusHttpServer.getAddress().getPort();
     resource = Resource.getDefault();
     meterProvider =
@@ -115,12 +128,16 @@ void afterEach() {
   @Test
   void endToEnd() {
     Meter meter = meterProvider.meterBuilder("test").setInstrumentationVersion("1.0.0").build();
-
     meter
         .counterBuilder("requests")
         .build()
         .add(3, Attributes.builder().put("animal", "bear").build());
 
+    DoubleHistogram histogram = meter.histogramBuilder("latency").build();
+
+    histogram.record(3, Attributes.builder().put("animal", "bear").build());
+    histogram.record(5, Attributes.builder().put("animal", "bear").build());
+
     await()
         .atMost(Duration.ofSeconds(30))
         .untilAsserted(() -> assertThat(grpcServer.metricRequests.size()).isGreaterThan(0));
@@ -169,6 +186,11 @@ void endToEnd() {
             .orElseThrow(() -> new IllegalStateException("missing scope with name \"test\""));
     assertThat(testScopeMetrics.getScope().getVersion()).isEqualTo("1.0.0");
 
+    verifyCounterMetric(testScopeMetrics);
+    verifyHistogramMetric(testScopeMetrics);
+  }
+
+  private static void verifyCounterMetric(ScopeMetrics testScopeMetrics) {
     Optional<Metric> optRequestTotal =
         testScopeMetrics.getMetricsList().stream()
             .filter(metric -> metric.getName().equals("requests_total"))
@@ -194,6 +216,32 @@ void endToEnd() {
             stringKeyValue("otel_scope_version", "1.0.0"));
   }
 
+  private static void verifyHistogramMetric(ScopeMetrics testScopeMetrics) {
+    Optional<Metric> optRequestTotal =
+        testScopeMetrics.getMetricsList().stream()
+            .filter(metric -> metric.getName().equals("latency"))
+            .findFirst();
+    assertThat(optRequestTotal).isPresent();
+    Metric requestTotal = optRequestTotal.get();
+    assertThat(requestTotal.getDataCase()).isEqualTo(Metric.DataCase.HISTOGRAM);
+
+    Histogram requestHistogram = requestTotal.getHistogram();
+    assertThat(requestHistogram.getAggregationTemporality())
+        .isEqualTo(AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE);
+    assertThat(requestHistogram.getDataPointsCount()).isEqualTo(1);
+
+    HistogramDataPoint requestHistogramDataPoints = requestHistogram.getDataPoints(0);
+    assertThat(requestHistogramDataPoints.getCount()).isEqualTo(2);
+    assertThat(requestHistogramDataPoints.getBucketCountsCount()).isEqualTo(1);
+    assertThat(requestHistogramDataPoints.getAttributesList())
+        .containsExactlyInAnyOrder(
+            stringKeyValue("animal", "bear"),
+            // Scope name and version are serialized as attributes to disambiguate metrics with the
+            // same name in different scopes
+            stringKeyValue("otel_scope_name", "test"),
+            stringKeyValue("otel_scope_version", "1.0.0"));
+  }
+
   private static KeyValue stringKeyValue(String key, String value) {
     return KeyValue.newBuilder()
         .setKey(key)

exporters/prometheus/src/test/java/io/opentelemetry/exporter/prometheus/PrometheusHttpServerTest.java

@@ -113,6 +113,9 @@ void invalidConfig() {
     assertThatThrownBy(() -> PrometheusHttpServer.builder().setHost(""))
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("host must not be empty");
+    assertThatThrownBy(() -> PrometheusHttpServer.builder().setDefaultAggregationSelector(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("defaultAggregationSelector");
   }
 
   @Test

exporters/prometheus/src/test/java/io/opentelemetry/exporter/prometheus/internal/PrometheusMetricReaderProviderTest.java

@@ -7,14 +7,18 @@
 
 import static org.assertj.core.api.Assertions.as;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
 import com.sun.net.httpserver.HttpServer;
 import io.opentelemetry.exporter.prometheus.PrometheusHttpServer;
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException;
 import io.opentelemetry.sdk.autoconfigure.spi.internal.DefaultConfigProperties;
 import io.opentelemetry.sdk.common.export.MemoryMode;
+import io.opentelemetry.sdk.metrics.Aggregation;
+import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
 import io.prometheus.metrics.exporter.httpserver.HTTPServer;
 import java.io.IOException;
@@ -59,6 +63,8 @@ void createMetricReader_Default() throws IOException {
                 assertThat(server.getAddress().getPort()).isEqualTo(9464);
               });
       assertThat(metricReader.getMemoryMode()).isEqualTo(MemoryMode.IMMUTABLE_DATA);
+      assertThat(metricReader.getDefaultAggregation(InstrumentType.HISTOGRAM))
+          .isEqualTo(Aggregation.defaultAggregation());
     }
   }
 
@@ -76,6 +82,9 @@ void createMetricReader_WithConfiguration() throws IOException {
     config.put("otel.exporter.prometheus.host", "localhost");
     config.put("otel.exporter.prometheus.port", String.valueOf(port));
     config.put("otel.java.experimental.exporter.memory_mode", "reusable_data");
+    config.put(
+        "otel.java.experimental.exporter.prometheus.metrics.default.histogram.aggregation",
+        "BASE2_EXPONENTIAL_BUCKET_HISTOGRAM");
 
     when(configProperties.getInt(any())).thenReturn(null);
     when(configProperties.getString(any())).thenReturn(null);
@@ -91,6 +100,23 @@ void createMetricReader_WithConfiguration() throws IOException {
                 assertThat(server.getAddress().getPort()).isEqualTo(port);
               });
       assertThat(metricReader.getMemoryMode()).isEqualTo(MemoryMode.REUSABLE_DATA);
+      assertThat(metricReader.getDefaultAggregation(InstrumentType.HISTOGRAM))
+          .isEqualTo(Aggregation.base2ExponentialBucketHistogram());
     }
   }
+
+  @Test
+  void createMetricReader_WithWrongConfiguration() {
+    Map<String, String> config = new HashMap<>();
+    config.put(
+        "otel.java.experimental.exporter.prometheus.metrics.default.histogram.aggregation", "foo");
+
+    when(configProperties.getInt(any())).thenReturn(null);
+    when(configProperties.getString(any())).thenReturn(null);
+
+    assertThatThrownBy(
+            () -> provider.createMetricReader(DefaultConfigProperties.createFromMap(config)))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("Unrecognized default histogram aggregation:");
+  }
 }