Allow exporters to influence Aggregation

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

@@ -8,9 +8,11 @@
 import io.opentelemetry.exporter.otlp.internal.metrics.MetricsRequestMarshaler;
 import io.opentelemetry.exporter.otlp.internal.okhttp.OkHttpExporter;
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.util.Collection;
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
 /** Exports metrics using OTLP via HTTP, using OpenTelemetry's protobuf model. */
@@ -41,6 +43,14 @@ public static OtlpHttpMetricExporterBuilder builder() {
     return new OtlpHttpMetricExporterBuilder();
   }
 
+  @Nullable
+  @Override
+  public final AggregationTemporality getPreferredTemporality() {
+    // TODO: Lookup based on specification, or constructor
+    // https://github.com/open-telemetry/opentelemetry-java/issues/3790
+    return null;
+  }
+
   /**
    * Submits all the given metrics in a single batch to the OpenTelemetry collector.
    *

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

@@ -8,9 +8,11 @@
 import io.opentelemetry.exporter.otlp.internal.grpc.GrpcExporter;
 import io.opentelemetry.exporter.otlp.internal.metrics.MetricsRequestMarshaler;
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.util.Collection;
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
 /** Exports metrics using OTLP via gRPC, using OpenTelemetry's protobuf model. */
@@ -43,6 +45,14 @@ public static OtlpGrpcMetricExporterBuilder builder() {
     this.delegate = delegate;
   }
 
+  @Nullable
+  @Override
+  public AggregationTemporality getPreferredTemporality() {
+    // TODO: Lookup based on specification, or constructor
+    // https://github.com/open-telemetry/opentelemetry-java/issues/3790
+    return null;
+  }
+
   /**
    * Submits all the given metrics in a single batch to the OpenTelemetry collector.
    *

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

@@ -6,6 +6,7 @@
 package io.opentelemetry.exporter.prometheus;
 
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricProducer;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
@@ -15,6 +16,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.List;
 
 /**
@@ -47,6 +49,16 @@ public static MetricReaderFactory create() {
     return new Factory();
   }
 
+  @Override
+  public EnumSet<AggregationTemporality> getSupportedTemporality() {
+    return EnumSet.of(AggregationTemporality.CUMULATIVE);
+  }
+
+  @Override
+  public AggregationTemporality getPreferredTemporality() {
+    return AggregationTemporality.CUMULATIVE;
+  }
+
   // Prometheus cannot flush.
   @Override
   public CompletableResultCode flush() {

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

@@ -15,6 +15,7 @@
 import com.sun.net.httpserver.HttpServer;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.internal.DaemonThreadFactory;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import io.opentelemetry.sdk.metrics.export.MetricProducer;
@@ -35,6 +36,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -104,6 +106,16 @@ private void start() {
     }
   }
 
+  @Override
+  public EnumSet<AggregationTemporality> getSupportedTemporality() {
+    return EnumSet.of(AggregationTemporality.CUMULATIVE);
+  }
+
+  @Override
+  public AggregationTemporality getPreferredTemporality() {
+    return AggregationTemporality.CUMULATIVE;
+  }
+
   @Override
   public CompletableResultCode flush() {
     return CompletableResultCode.ofSuccess();

sdk/metrics-testing/src/main/java/io/opentelemetry/sdk/testing/assertj/metrics/AbstractSumDataAssert.java

@@ -34,8 +34,7 @@ public SumAssertT isNotMonotonic() {
       failWithActualExpectedAndMessage(
           actual,
           "montonic: fail",
-          "Exepcted Sum to be non-monotonic",
-          false,
+          "Expected Sum to be non-monotonic, found: %s",
           actual.isMonotonic());
     }
     return myself;
@@ -48,8 +47,7 @@ public SumAssertT isCumulative() {
       failWithActualExpectedAndMessage(
           actual,
           "aggregationTemporality: CUMULATIVE",
-          "Exepcted Sum to have cumulative aggregation but found <%s>",
-          AggregationTemporality.CUMULATIVE,
+          "Expected Sum to have cumulative aggregation but found <%s>",
           actual.getAggregationTemporality());
     }
     return myself;
@@ -62,8 +60,7 @@ public SumAssertT isDelta() {
       failWithActualExpectedAndMessage(
           actual,
           "aggregationTemporality: DELTA",
-          "Exepcted Sum to have cumulative aggregation but found <%s>",
-          AggregationTemporality.DELTA,
+          "Expected Sum to have delta aggregation but found <%s>",
           actual.getAggregationTemporality());
     }
     return myself;

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/DoubleHistogramBenchmark.java

@@ -5,14 +5,11 @@
 
 package io.opentelemetry.sdk.metrics.internal.aggregator;
 
-import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
 import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor;
 import io.opentelemetry.sdk.metrics.common.InstrumentType;
 import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
-import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
-import io.opentelemetry.sdk.metrics.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
-import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.metrics.exemplar.ExemplarFilter;
+import io.opentelemetry.sdk.metrics.view.Aggregation;
 import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -31,14 +28,11 @@
 @State(Scope.Benchmark)
 public class DoubleHistogramBenchmark {
   private static final Aggregator<HistogramAccumulation> aggregator =
-      AggregatorFactory.histogram(Arrays.asList(10.0, 100.0, 1_000.0), AggregationTemporality.DELTA)
-          .create(
-              Resource.getDefault(),
-              InstrumentationLibraryInfo.empty(),
+      Aggregation.explicitBucketHistogram(Arrays.asList(10.0, 100.0, 1_000.0))
+          .createAggregator(
               InstrumentDescriptor.create(
                   "name", "description", "1", InstrumentType.HISTOGRAM, InstrumentValueType.DOUBLE),
-              MetricDescriptor.create("name", "description", "1"),
-              ExemplarReservoir::noSamples);
+              ExemplarFilter.neverSample());
   private AggregatorHandle<HistogramAccumulation> aggregatorHandle;
 
   @Setup(Level.Trial)

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/DoubleMinMaxSumCountBenchmark.java

@@ -5,13 +5,7 @@
 
 package io.opentelemetry.sdk.metrics.internal.aggregator;
 
-import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
-import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor;
-import io.opentelemetry.sdk.metrics.common.InstrumentType;
-import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
 import io.opentelemetry.sdk.metrics.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
-import io.opentelemetry.sdk.resources.Resource;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -29,14 +23,7 @@
 @State(Scope.Benchmark)
 public class DoubleMinMaxSumCountBenchmark {
   private static final Aggregator<MinMaxSumCountAccumulation> aggregator =
-      AggregatorFactory.minMaxSumCount()
-          .create(
-              Resource.getDefault(),
-              InstrumentationLibraryInfo.empty(),
-              InstrumentDescriptor.create(
-                  "name", "description", "1", InstrumentType.HISTOGRAM, InstrumentValueType.DOUBLE),
-              MetricDescriptor.create("name", "description", "1"),
-              ExemplarReservoir::noSamples);
+      new DoubleMinMaxSumCountAggregator(ExemplarReservoir::noSamples);
   private AggregatorHandle<MinMaxSumCountAccumulation> aggregatorHandle;
 
   @Setup(Level.Trial)

sdk/metrics/src/jmh/java/io/opentelemetry/sdk/metrics/internal/aggregator/LongMinMaxSumCountBenchmark.java

@@ -5,13 +5,7 @@
 
 package io.opentelemetry.sdk.metrics.internal.aggregator;
 
-import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
-import io.opentelemetry.sdk.metrics.common.InstrumentDescriptor;
-import io.opentelemetry.sdk.metrics.common.InstrumentType;
-import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
 import io.opentelemetry.sdk.metrics.exemplar.ExemplarReservoir;
-import io.opentelemetry.sdk.metrics.internal.descriptor.MetricDescriptor;
-import io.opentelemetry.sdk.resources.Resource;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -29,14 +23,7 @@
 @State(Scope.Benchmark)
 public class LongMinMaxSumCountBenchmark {
   private static final Aggregator<MinMaxSumCountAccumulation> aggregator =
-      AggregatorFactory.minMaxSumCount()
-          .create(
-              Resource.getDefault(),
-              InstrumentationLibraryInfo.empty(),
-              InstrumentDescriptor.create(
-                  "name", "description", "1", InstrumentType.HISTOGRAM, InstrumentValueType.LONG),
-              MetricDescriptor.create("name", "description", "1"),
-              ExemplarReservoir::noSamples);
+      new LongMinMaxSumCountAggregator(ExemplarReservoir::noSamples);
   private AggregatorHandle<MinMaxSumCountAccumulation> aggregatorHandle;
 
   @Setup(Level.Trial)

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/DefaultSdkMeterProvider.java

@@ -16,13 +16,16 @@
 import io.opentelemetry.sdk.metrics.export.MetricReader;
 import io.opentelemetry.sdk.metrics.export.MetricReaderFactory;
 import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle;
+import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo;
 import io.opentelemetry.sdk.metrics.internal.state.MeterProviderSharedState;
 import io.opentelemetry.sdk.metrics.internal.view.ViewRegistry;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -45,7 +48,7 @@ final class DefaultSdkMeterProvider implements SdkMeterProvider {
   private final ComponentRegistry<SdkMeter> registry;
   private final MeterProviderSharedState sharedState;
   private final Set<CollectionHandle> collectors;
-  private final List<MetricReader> readers;
+  private final Map<CollectionHandle, CollectionInfo> collectionInfoMap;
   private final AtomicBoolean isClosed = new AtomicBoolean(false);
   private final AtomicLong lastCollectionTimestamp;
 
@@ -72,14 +75,14 @@ final class DefaultSdkMeterProvider implements SdkMeterProvider {
     // These are guaranteed to be unique per-reader for this SDK, and only this SDK.
     // These are *only* mutated in our constructor, and safe to use concurrently after construction.
     collectors = CollectionHandle.mutableSet();
-    readers = new ArrayList<>();
+    collectionInfoMap = new HashMap<>();
     Supplier<CollectionHandle> handleSupplier = CollectionHandle.createSupplier();
     for (MetricReaderFactory readerFactory : readerFactories) {
       CollectionHandle handle = handleSupplier.get();
       // TODO: handle failure in creation or just crash?
       MetricReader reader = readerFactory.apply(new LeasedMetricProducer(handle));
+      collectionInfoMap.put(handle, CollectionInfo.create(handle, collectors, reader));
       collectors.add(handle);
-      readers.add(reader);
     }
   }
 
@@ -95,8 +98,8 @@ public MeterBuilder meterBuilder(@Nullable String instrumentationName) {
   @Override
   public CompletableResultCode forceFlush() {
     List<CompletableResultCode> results = new ArrayList<>();
-    for (MetricReader reader : readers) {
-      results.add(reader.flush());
+    for (CollectionInfo collectionInfo : collectionInfoMap.values()) {
+      results.add(collectionInfo.getReader().shutdown());
     }
     return CompletableResultCode.ofAll(results);
   }
@@ -108,8 +111,8 @@ public CompletableResultCode close() {
       return CompletableResultCode.ofSuccess();
     }
     List<CompletableResultCode> results = new ArrayList<>();
-    for (MetricReader reader : readers) {
-      results.add(reader.shutdown());
+    for (CollectionInfo info : collectionInfoMap.values()) {
+      results.add(info.getReader().shutdown());
     }
     return CompletableResultCode.ofAll(results);
   }
@@ -147,7 +150,9 @@ public Collection<MetricData> collectAllMetrics() {
       for (SdkMeter meter : meters) {
         result.addAll(
             meter.collectAll(
-                handle, collectors, sharedState.getClock().now(), disableSynchronousCollection));
+                collectionInfoMap.get(handle),
+                sharedState.getClock().now(),
+                disableSynchronousCollection));
       }
       return Collections.unmodifiableCollection(result);
     }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/SdkMeter.java

@@ -12,11 +12,10 @@
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
 import io.opentelemetry.sdk.metrics.data.MetricData;
-import io.opentelemetry.sdk.metrics.internal.export.CollectionHandle;
+import io.opentelemetry.sdk.metrics.internal.export.CollectionInfo;
 import io.opentelemetry.sdk.metrics.internal.state.MeterProviderSharedState;
 import io.opentelemetry.sdk.metrics.internal.state.MeterSharedState;
 import java.util.Collection;
-import java.util.Set;
 
 /** {@link SdkMeter} is SDK implementation of {@link Meter}. */
 final class SdkMeter implements Meter {
@@ -37,16 +36,9 @@ InstrumentationLibraryInfo getInstrumentationLibraryInfo() {
 
   /** Collects all the metric recordings that changed since the previous call. */
   Collection<MetricData> collectAll(
-      CollectionHandle collector,
-      Set<CollectionHandle> allCollectors,
-      long epochNanos,
-      boolean suppressSynchronousCollection) {
+      CollectionInfo collectionInfo, long epochNanos, boolean suppressSynchronousCollection) {
     return meterSharedState.collectAll(
-        collector,
-        allCollectors,
-        meterProviderSharedState,
-        epochNanos,
-        suppressSynchronousCollection);
+        collectionInfo, meterProviderSharedState, epochNanos, suppressSynchronousCollection);
   }
 
   @Override

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/export/MetricExporter.java

@@ -6,8 +6,11 @@
 package io.opentelemetry.sdk.metrics.export;
 
 import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import java.util.Collection;
+import java.util.EnumSet;
+import javax.annotation.Nullable;
 
 /**
  * {@code MetricExporter} is the interface that all "push based" metric libraries should use to
@@ -17,6 +20,17 @@
  */
 public interface MetricExporter {
 
+  /** Returns the set of all supported temporalities for this exporter. */
+  default EnumSet<AggregationTemporality> getSupportedTemporality() {
+    return EnumSet.allOf(AggregationTemporality.class);
+  }
+
+  /** Returns the preferred temporality for metrics. */
+  @Nullable
+  default AggregationTemporality getPreferredTemporality() {
+    return null;
+  }
+
   /**
    * Exports the collection of given {@link MetricData}. Note that export operations can be
    * performed simultaneously depending on the type of metric reader being used. However, the caller