MetricReader and Multi-export features

docs/apidiffs/current_vs_latest/opentelemetry-exporter-logging-otlp.txt

@@ -1,2 +1,5 @@
 Comparing source compatibility of  against 
-No changes.
+***! MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.logging.otlp.OtlpJsonLoggingMetricExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	---! REMOVED METHOD: PUBLIC(-) io.opentelemetry.sdk.common.CompletableResultCode flush()
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.common.CompletableResultCode forceFlush()

docs/apidiffs/current_vs_latest/opentelemetry-exporter-logging.txt

@@ -1,2 +1,5 @@
 Comparing source compatibility of  against 
-No changes.
+***! MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.logging.LoggingMetricExporter  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	---! REMOVED METHOD: PUBLIC(-) io.opentelemetry.sdk.common.CompletableResultCode flush()
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.sdk.common.CompletableResultCode forceFlush()

exporters/logging-otlp/src/main/java/io/opentelemetry/exporter/logging/otlp/OtlpJsonLoggingMetricExporter.java

@@ -51,7 +51,7 @@ public CompletableResultCode export(Collection<MetricData> metrics) {
   }
 
   @Override
-  public CompletableResultCode flush() {
+  public CompletableResultCode forceFlush() {
     return CompletableResultCode.ofSuccess();
   }
 

exporters/logging-otlp/src/test/java/io/opentelemetry/exporter/logging/otlp/OtlpJsonLoggingMetricExporterTest.java

@@ -137,7 +137,7 @@ void log() throws Exception {
 
   @Test
   void flush() {
-    assertThat(exporter.flush().isSuccess()).isTrue();
+    assertThat(exporter.forceFlush().isSuccess()).isTrue();
   }
 
   @Test

exporters/logging/src/main/java/io/opentelemetry/exporter/logging/LoggingMetricExporter.java

@@ -31,7 +31,7 @@ public CompletableResultCode export(Collection<MetricData> metrics) {
    * @return the result of the operation
    */
   @Override
-  public CompletableResultCode flush() {
+  public CompletableResultCode forceFlush() {
     CompletableResultCode resultCode = new CompletableResultCode();
     for (Handler handler : logger.getHandlers()) {
       try {
@@ -46,7 +46,7 @@ public CompletableResultCode flush() {
   @Override
   public CompletableResultCode shutdown() {
     // no-op
-    this.flush();
+    this.forceFlush();
     return CompletableResultCode.ofSuccess();
   }
 }

exporters/logging/src/test/java/io/opentelemetry/exporter/logging/LoggingMetricExporterTest.java

@@ -115,7 +115,7 @@ public synchronized void flush() {
                 flushed.set(true);
               }
             });
-    exporter.flush();
+    exporter.forceFlush();
     assertThat(flushed.get()).isTrue();
   }
 }

exporters/otlp-http/metrics/src/main/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporter.java

@@ -153,7 +153,7 @@ private static String extractErrorStatus(Response response) {
    * @return always Success
    */
   @Override
-  public CompletableResultCode flush() {
+  public CompletableResultCode forceFlush() {
     return CompletableResultCode.ofSuccess();
   }
 

exporters/otlp-http/metrics/src/test/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterTest.java

@@ -153,6 +153,16 @@ void testExportGzipCompressed() {
     assertThat(parseRequestBody(gzipDecompress(request.content().array()))).isEqualTo(payload);
   }
 
+  @Test
+  void testExport_flush() {
+    OtlpHttpMetricExporter exporter = OtlpHttpMetricExporter.builder().build();
+    try {
+      assertThat(exporter.forceFlush().isSuccess()).isTrue();
+    } finally {
+      exporter.shutdown();
+    }
+  }
+
   private static void assertRequestCommon(AggregatedHttpRequest request) {
     assertThat(request.method()).isEqualTo(HttpMethod.POST);
     assertThat(request.path()).isEqualTo("/v1/metrics");

exporters/otlp/common/src/jmh/java/io/opentelemetry/exporter/otlp/internal/GrpcGzipBenchmark.java

@@ -19,6 +19,7 @@
 import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.testing.InMemoryMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -49,8 +50,10 @@ public class GrpcGzipBenchmark {
   private static final Codec IDENTITY_CODEC = Codec.Identity.NONE;
 
   static {
+    InMemoryMetricReader metricReader = new InMemoryMetricReader();
     SdkMeterProvider meterProvider =
         SdkMeterProvider.builder()
+            .registerMetricReader(metricReader)
             .setResource(
                 Resource.create(
                     Attributes.builder()
@@ -118,7 +121,7 @@ public class GrpcGzipBenchmark {
     histogram.record(3.0);
     histogram.record(4.0);
     histogram.record(5.0);
-    Collection<MetricData> metricData = meterProvider.collectAllMetrics();
+    Collection<MetricData> metricData = metricReader.collectAllMetrics();
 
     List<ResourceMetrics> resourceMetrics =
         Arrays.stream(ResourceMetricsMarshaler.create(metricData))

exporters/otlp/common/src/jmh/java/io/opentelemetry/exporter/otlp/internal/MetricsRequestMarshalerBenchmark.java

@@ -16,6 +16,7 @@
 import io.opentelemetry.exporter.otlp.internal.metrics.MetricsRequestMarshaler;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.testing.InMemoryMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -40,8 +41,10 @@ public class MetricsRequestMarshalerBenchmark {
   private static final Collection<MetricData> METRICS;
 
   static {
+    InMemoryMetricReader metricReader = new InMemoryMetricReader();
     SdkMeterProvider meterProvider =
         SdkMeterProvider.builder()
+            .registerMetricReader(metricReader)
             .setResource(
                 Resource.create(
                     Attributes.builder()
@@ -110,7 +113,7 @@ public class MetricsRequestMarshalerBenchmark {
     histogram.record(4.0);
     histogram.record(5.0);
 
-    METRICS = meterProvider.collectAllMetrics();
+    METRICS = metricReader.collectAllMetrics();
   }
 
   @Benchmark

exporters/otlp/metrics/src/main/java/io/opentelemetry/exporter/otlp/metrics/OtlpGrpcMetricExporter.java

@@ -119,7 +119,7 @@ public void onFailure(Throwable t) {
    * @return always Success
    */
   @Override
-  public CompletableResultCode flush() {
+  public CompletableResultCode forceFlush() {
     return CompletableResultCode.ofSuccess();
   }
 

exporters/otlp/metrics/src/test/java/io/opentelemetry/exporter/otlp/metrics/OtlpGrpcMetricExporterTest.java

@@ -307,7 +307,7 @@ void testExport_flush() {
     OtlpGrpcMetricExporter exporter =
         OtlpGrpcMetricExporter.builder().setChannel(inProcessChannel).build();
     try {
-      assertThat(exporter.flush().isSuccess()).isTrue();
+      assertThat(exporter.forceFlush().isSuccess()).isTrue();
     } finally {
       exporter.shutdown();
     }

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

@@ -5,14 +5,24 @@
 
 package io.opentelemetry.exporter.prometheus;
 
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricProducer;
+import io.opentelemetry.sdk.metrics.export.MetricReader;
+import io.opentelemetry.sdk.metrics.export.MetricReaderFactory;
 import io.prometheus.client.Collector;
+import io.prometheus.client.CollectorRegistry;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 
-public final class PrometheusCollector extends Collector {
+/**
+ * A reader of OpenTelemetry metrics that exports into Prometheus as a Collector.
+ *
+ * <p>Usage: <code>sdkMeterProvider.registerMetricReader(PrometheusCollector.create());</code>
+ */
+public final class PrometheusCollector extends Collector implements MetricReader {
   private final MetricProducer metricProducer;
 
   PrometheusCollector(MetricProducer metricProducer) {
@@ -26,15 +36,39 @@ public List<MetricFamilySamples> collect() {
     for (MetricData metricData : allMetrics) {
       allSamples.add(MetricAdapter.toMetricFamilySamples(metricData));
     }
-    return allSamples;
+    return Collections.unmodifiableList(allSamples);
   }
 
   /**
-   * Returns a new builder instance for this exporter.
-   *
-   * @return a new builder instance for this exporter.
+   * Returns a new collector to be registered with a {@link
+   * io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder}.
    */
-  public static PrometheusCollectorBuilder builder() {
-    return new PrometheusCollectorBuilder();
+  public static MetricReaderFactory create() {
+    return new Factory();
+  }
+
+  // Prometheus cannot flush.
+  @Override
+  public CompletableResultCode flush() {
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    CollectorRegistry.defaultRegistry.unregister(this);
+    return CompletableResultCode.ofSuccess();
+  }
+
+  /** Our implementation of the metric reader factory. */
+  // NOTE: This should be updated to (optionally) start the simpel Http server exposing the metrics
+  // path.
+  private static class Factory implements MetricReaderFactory {
+    @Override
+    public MetricReader apply(MetricProducer producer) {
+      PrometheusCollector collector = new PrometheusCollector(producer);
+      // When SdkMeterProvider constructs us, we register with prometheus.
+      collector.register();
+      return collector;
+    }
   }
 }

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

@@ -38,8 +38,9 @@ class PrometheusCollectorTest {
 
   @BeforeEach
   void setUp() {
-    prometheusCollector =
-        PrometheusCollector.builder().setMetricProducer(metricProducer).buildAndRegister();
+    // Apply the SDK metric producer registers with prometheus.
+    prometheusCollector = new PrometheusCollector(metricProducer);
+    prometheusCollector.register();
   }
 
   @Test

integration-tests/src/test/java/io/opentelemetry/OtlpExporterIntegrationTest.java

@@ -42,8 +42,9 @@
 import io.opentelemetry.proto.trace.v1.ResourceSpans;
 import io.opentelemetry.proto.trace.v1.Span.Link;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
-import io.opentelemetry.sdk.metrics.export.IntervalMetricReader;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
+import io.opentelemetry.sdk.metrics.export.MetricReaderFactory;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.IdGenerator;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
@@ -131,9 +132,7 @@ void beforeEach() {
   }
 
   @AfterEach
-  void afterEach() {
-    IntervalMetricReader.resetGlobalForTest();
-  }
+  void afterEach() {}
 
   @Test
   void testOtlpGrpcTraceExport() {
@@ -263,23 +262,22 @@ void testOtlpHttpMetricExport() {
   }
 
   private static void testMetricExport(MetricExporter metricExporter) {
-    SdkMeterProvider meterProvider = SdkMeterProvider.builder().setResource(RESOURCE).build();
-    IntervalMetricReader intervalMetricReader =
-        IntervalMetricReader.builder()
-            .setMetricExporter(metricExporter)
-            .setMetricProducers(Collections.singletonList(meterProvider))
-            .setExportIntervalMillis(5000)
-            .build();
-    intervalMetricReader.startAndRegisterGlobal();
+    MetricReaderFactory reader = PeriodicMetricReader.create(metricExporter, Duration.ofSeconds(5));
+    SdkMeterProvider meterProvider =
+        SdkMeterProvider.builder().setResource(RESOURCE).registerMetricReader(reader).build();
 
     Meter meter = meterProvider.meterBuilder(OtlpExporterIntegrationTest.class.getName()).build();
 
     LongCounter longCounter = meter.counterBuilder("my-counter").build();
     longCounter.add(100, Attributes.builder().put("key", "value").build());
 
-    Awaitility.await()
-        .atMost(Duration.ofSeconds(30))
-        .until(() -> grpcServer.metricRequests.size() == 1);
+    try {
+      Awaitility.await()
+          .atMost(Duration.ofSeconds(30))
+          .until(() -> grpcServer.metricRequests.size() == 1);
+    } finally {
+      meterProvider.close();
+    }
 
     ExportMetricsServiceRequest request = grpcServer.metricRequests.get(0);
     assertThat(request.getResourceMetricsCount()).isEqualTo(1);

opencensus-shim/src/test/java/io/opentelemetry/opencensusshim/FakeMetricExporter.java

@@ -71,8 +71,8 @@ public CompletableResultCode export(Collection<MetricData> metrics) {
   }
 
   @Override
-  public CompletableResultCode flush() {
-    return null;
+  public CompletableResultCode forceFlush() {
+    return CompletableResultCode.ofSuccess();
   }
 
   @Override