MetricReader and Multi-export features

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.flush().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/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 simple 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

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

perf-harness/src/test/java/io/opentelemetry/perf/OtlpPipelineStressTest.java

@@ -21,14 +21,14 @@
 import io.opentelemetry.sdk.metrics.data.LongPointData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.data.PointData;
-import io.opentelemetry.sdk.metrics.export.IntervalMetricReader;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.metrics.testing.InMemoryMetricExporter;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
 import io.opentelemetry.semconv.resource.attributes.ResourceAttributes;
 import java.io.IOException;
-import java.util.Collections;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
@@ -105,7 +105,7 @@ public class OtlpPipelineStressTest {
 
   private SdkTracerProvider sdkTracerProvider;
   private OpenTelemetry openTelemetry;
-  private IntervalMetricReader intervalMetricReader;
+  private SdkMeterProvider meterProvider;
   private Proxy collectorProxy;
   private ToxiproxyClient toxiproxyClient;
 
@@ -131,7 +131,7 @@ void setUp() throws IOException {
 
   @AfterEach
   void tearDown() throws IOException {
-    intervalMetricReader.shutdown();
+    meterProvider.close();
     sdkTracerProvider.shutdown();
 
     toxiproxyClient.reset();
@@ -185,7 +185,7 @@ void oltpExportWithFlakyCollector() throws IOException, InterruptedException {
 
     Thread.sleep(10000);
     List<MetricData> finishedMetricItems = metricExporter.getFinishedMetricItems();
-    intervalMetricReader.shutdown();
+    meterProvider.close();
     Thread.sleep(1000);
     reportMetrics(finishedMetricItems);
     Thread.sleep(10000);
@@ -247,15 +247,12 @@ private void setupSdk() {
                 .build());
 
     // set up the metric exporter and wire it into the SDK and a timed reader.
-    SdkMeterProvider meterProvider =
-        SdkMeterProvider.builder().setResource(resource).buildAndRegisterGlobal();
-
-    intervalMetricReader =
-        IntervalMetricReader.builder()
-            .setMetricExporter(metricExporter)
-            .setMetricProducers(Collections.singleton(meterProvider))
-            .setExportIntervalMillis(1000)
-            .buildAndStart();
+    meterProvider =
+        SdkMeterProvider.builder()
+            .setResource(resource)
+            .registerMetricReader(
+                PeriodicMetricReader.create(metricExporter, Duration.ofSeconds(1)))
+            .buildAndRegisterGlobal();
 
     // set up the span exporter and wire it into the SDK
     OtlpGrpcSpanExporter spanExporter =