Add config to enable Default Exponential Histogram for Prometheus Exp…

…orter (#6541)

Co-authored-by: Jack Berg <[email protected]>

dependencyManagement/build.gradle.kts

@@ -50,6 +50,7 @@ val DEPENDENCIES = listOf(
   "org.mockito:mockito-junit-jupiter:${mockitoVersion}",
   "org.slf4j:slf4j-simple:${slf4jVersion}",
   "org.slf4j:jul-to-slf4j:${slf4jVersion}",
+  "io.prometheus:prometheus-metrics-shaded-protobuf:1.3.1",
   "io.prometheus:simpleclient:${prometheusClientVersion}",
   "io.prometheus:simpleclient_common:${prometheusClientVersion}",
   "io.prometheus:simpleclient_httpserver:${prometheusClientVersion}",

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

@@ -5,10 +5,16 @@
 
 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.autoconfigure.spi.internal.StructuredConfigProperties;
 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;
@@ -71,5 +77,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

@@ -310,18 +310,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/build.gradle.kts

@@ -19,6 +19,7 @@ dependencies {
 
   testImplementation(project(":sdk:testing"))
   testImplementation("io.opentelemetry.proto:opentelemetry-proto")
+  testImplementation("io.prometheus:prometheus-metrics-shaded-protobuf")
   testImplementation("com.sun.net.httpserver:http")
   testImplementation("com.google.guava:guava")
   testImplementation("com.linecorp.armeria:armeria")

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/PrometheusHttpServerTest.java

@@ -16,18 +16,24 @@
 import com.linecorp.armeria.client.retry.RetryRule;
 import com.linecorp.armeria.client.retry.RetryingClient;
 import com.linecorp.armeria.common.AggregatedHttpResponse;
+import com.linecorp.armeria.common.HttpData;
 import com.linecorp.armeria.common.HttpHeaderNames;
 import com.linecorp.armeria.common.HttpMethod;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.common.RequestHeaders;
 import io.github.netmikey.logunit.api.LogCapturer;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.internal.testing.slf4j.SuppressLogger;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.common.export.MemoryMode;
+import io.opentelemetry.sdk.metrics.Aggregation;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.CollectionRegistration;
+import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableDoublePointData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableGaugeData;
 import io.opentelemetry.sdk.metrics.internal.data.ImmutableLongPointData;
@@ -36,7 +42,9 @@
 import io.opentelemetry.sdk.resources.Resource;
 import io.prometheus.metrics.exporter.httpserver.HTTPServer;
 import io.prometheus.metrics.exporter.httpserver.MetricsHandler;
+import io.prometheus.metrics.expositionformats.generated.com_google_protobuf_3_25_3.Metrics;
 import io.prometheus.metrics.model.registry.PrometheusRegistry;
+import io.prometheus.metrics.shaded.com_google_protobuf_3_25_3.TextFormat;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.ServerSocket;
@@ -113,6 +121,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
@@ -526,4 +537,75 @@ void toBuilder() {
         .hasFieldOrPropertyWithValue("executor", executor)
         .hasFieldOrPropertyWithValue("prometheusRegistry", prometheusRegistry);
   }
+
+  /**
+   * Set the default histogram aggregation to be {@link
+   * Aggregation#base2ExponentialBucketHistogram()}. In order to validate that exponential
+   * histograms are produced, we request protobuf encoded metrics when scraping since the prometheus
+   * text format does not support native histograms. We parse the binary content protobuf payload to
+   * the protobuf java bindings, and assert against the string representation.
+   */
+  @Test
+  void histogramDefaultBase2ExponentialHistogram() throws IOException {
+    PrometheusHttpServer prometheusServer =
+        PrometheusHttpServer.builder()
+            .setHost("localhost")
+            .setPort(0)
+            .setDefaultAggregationSelector(
+                DefaultAggregationSelector.getDefault()
+                    .with(InstrumentType.HISTOGRAM, Aggregation.base2ExponentialBucketHistogram()))
+            .build();
+    try (SdkMeterProvider meterProvider =
+        SdkMeterProvider.builder().registerMetricReader(prometheusServer).build()) {
+      DoubleHistogram histogram = meterProvider.get("meter").histogramBuilder("histogram").build();
+      histogram.record(1.0);
+
+      WebClient client =
+          WebClient.builder("http://localhost:" + prometheusServer.getAddress().getPort())
+              .decorator(RetryingClient.newDecorator(RetryRule.failsafe()))
+              // Request protobuf binary encoding, which is required for the prometheus native
+              // histogram format
+              .addHeader(
+                  "Accept",
+                  "application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily")
+              .build();
+      AggregatedHttpResponse response = client.get("/metrics").aggregate().join();
+      assertThat(response.status()).isEqualTo(HttpStatus.OK);
+      assertThat(response.headers().get(HttpHeaderNames.CONTENT_TYPE))
+          .isEqualTo(
+              "application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily; encoding=delimited");
+      // Parse the data to Metrics.MetricFamily protobuf java binding and assert against the string
+      // representation
+      try (HttpData data = response.content()) {
+        Metrics.MetricFamily metricFamily =
+            Metrics.MetricFamily.parseDelimitedFrom(data.toInputStream());
+        String s = TextFormat.printer().printToString(metricFamily);
+        assertThat(s)
+            .isEqualTo(
+                "name: \"histogram\"\n"
+                    + "help: \"\"\n"
+                    + "type: HISTOGRAM\n"
+                    + "metric {\n"
+                    + "  label {\n"
+                    + "    name: \"otel_scope_name\"\n"
+                    + "    value: \"meter\"\n"
+                    + "  }\n"
+                    + "  histogram {\n"
+                    + "    sample_count: 1\n"
+                    + "    sample_sum: 1.0\n"
+                    + "    schema: 8\n"
+                    + "    zero_threshold: 0.0\n"
+                    + "    zero_count: 0\n"
+                    + "    positive_span {\n"
+                    + "      offset: 0\n"
+                    + "      length: 1\n"
+                    + "    }\n"
+                    + "    positive_delta: 1\n"
+                    + "  }\n"
+                    + "}\n");
+      }
+    } finally {
+      prometheusServer.shutdown();
+    }
+  }
 }

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,20 @@ 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");
+
+    assertThatThrownBy(
+            () -> provider.createMetricReader(DefaultConfigProperties.createFromMap(config)))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("Unrecognized default histogram aggregation:");
+  }
 }