Very basic Aggregation-configuration API in the SDK.

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

@@ -41,4 +41,9 @@ public void batch(Labels labelSet, Aggregator aggregator, boolean mappedAggregat
   public List<MetricData> completeCollectionCycle() {
     return batcher.completeCollectionCycle();
   }
+
+  @Override
+  public boolean generatesDeltas() {
+    return batcher.generatesDeltas();
+  }
 }

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

@@ -0,0 +1,104 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics;
+
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration;
+import io.opentelemetry.sdk.metrics.view.Aggregations;
+import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Pattern;
+
+class AggregationChooser {
+  private static final AggregationConfiguration CUMULATIVE_SUM =
+      AggregationConfiguration.create(
+          Aggregations.sum(), AggregationConfiguration.Temporality.CUMULATIVE);
+  private static final AggregationConfiguration DELTA_SUMMARY =
+      AggregationConfiguration.create(
+          Aggregations.minMaxSumCount(), AggregationConfiguration.Temporality.DELTA);
+  private static final AggregationConfiguration CUMULATIVE_LAST_VALUE =
+      AggregationConfiguration.create(
+          Aggregations.lastValue(), AggregationConfiguration.Temporality.CUMULATIVE);
+  private static final AggregationConfiguration DELTA_LAST_VALUE =
+      AggregationConfiguration.create(
+          Aggregations.lastValue(), AggregationConfiguration.Temporality.DELTA);
+
+  private final Map<InstrumentSelector, AggregationConfiguration> configuration =
+      new ConcurrentHashMap<>();
+
+  AggregationConfiguration chooseAggregation(InstrumentDescriptor descriptor) {
+    List<Map.Entry<InstrumentSelector, AggregationConfiguration>> possibleMatches =
+        new ArrayList<>();
+    for (Map.Entry<InstrumentSelector, AggregationConfiguration> entry : configuration.entrySet()) {
+      InstrumentSelector registeredSelector = entry.getKey();
+      // if it matches everything, return it right away...
+      if (matchesOnType(descriptor, registeredSelector)
+          && matchesOnName(descriptor, registeredSelector)) {
+        return entry.getValue();
+      }
+      // otherwise throw it into a bucket of possible matches if it matches one of the criteria
+      if (matchesOne(descriptor, registeredSelector)) {
+        possibleMatches.add(entry);
+      }
+    }
+
+    if (possibleMatches.isEmpty()) {
+      return getDefaultSpecification(descriptor);
+    }
+
+    // If no exact matches found, pick the first one that matches something:
+    return possibleMatches.get(0).getValue();
+  }
+
+  private static boolean matchesOne(InstrumentDescriptor descriptor, InstrumentSelector selector) {
+    if (selector.hasInstrumentNameRegex() && !matchesOnName(descriptor, selector)) {
+      return false;
+    }
+    if (selector.hasInstrumentType() && !matchesOnType(descriptor, selector)) {
+      return false;
+    }
+    return true;
+  }
+
+  private static boolean matchesOnType(
+      InstrumentDescriptor descriptor, InstrumentSelector selector) {
+    if (selector.instrumentType() == null) {
+      return false;
+    }
+    return selector.instrumentType().equals(descriptor.getType());
+  }
+
+  private static boolean matchesOnName(
+      InstrumentDescriptor descriptor, InstrumentSelector registeredSelector) {
+    Pattern pattern = registeredSelector.instrumentNamePattern();
+    if (pattern == null) {
+      return false;
+    }
+    return pattern.matcher(descriptor.getName()).matches();
+  }
+
+  private static AggregationConfiguration getDefaultSpecification(InstrumentDescriptor descriptor) {
+    switch (descriptor.getType()) {
+      case COUNTER:
+      case UP_DOWN_COUNTER:
+        return CUMULATIVE_SUM;
+      case VALUE_RECORDER:
+        return DELTA_SUMMARY;
+      case VALUE_OBSERVER:
+        return DELTA_LAST_VALUE;
+      case SUM_OBSERVER:
+      case UP_DOWN_SUM_OBSERVER:
+        return CUMULATIVE_LAST_VALUE;
+    }
+    throw new IllegalArgumentException("Unknown descriptor type: " + descriptor.getType());
+  }
+
+  void addView(InstrumentSelector selector, AggregationConfiguration specification) {
+    configuration.put(selector, specification);
+  }
+}

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

@@ -51,4 +51,9 @@ interface Batcher {
    * @return the list of metrics batched in this Batcher.
    */
   List<MetricData> completeCollectionCycle();
+
+  /**
+   * Returns whether this batcher generate "delta" style metrics. The alternative is "cumulative".
+   */
+  boolean generatesDeltas();
 }

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

@@ -80,6 +80,11 @@ public void batch(Labels labelSet, Aggregator aggregator, boolean mappedAggregat
     public List<MetricData> completeCollectionCycle() {
       return Collections.emptyList();
     }
+
+    @Override
+    public boolean generatesDeltas() {
+      return false;
+    }
   }
 
   private static final class AllLabels implements Batcher {
@@ -155,6 +160,75 @@ public final List<MetricData> completeCollectionCycle() {
               aggregation.getDescriptorType(descriptor.getType(), descriptor.getValueType()),
               points));
     }
+
+    @Override
+    public boolean generatesDeltas() {
+      return delta;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+
+      AllLabels allLabels = (AllLabels) o;
+
+      if (startEpochNanos != allLabels.startEpochNanos) {
+        return false;
+      }
+      if (delta != allLabels.delta) {
+        return false;
+      }
+      if (descriptor != null
+          ? !descriptor.equals(allLabels.descriptor)
+          : allLabels.descriptor != null) {
+        return false;
+      }
+      if (aggregation != null
+          ? !aggregation.equals(allLabels.aggregation)
+          : allLabels.aggregation != null) {
+        return false;
+      }
+      if (resource != null ? !resource.equals(allLabels.resource) : allLabels.resource != null) {
+        return false;
+      }
+      if (instrumentationLibraryInfo != null
+          ? !instrumentationLibraryInfo.equals(allLabels.instrumentationLibraryInfo)
+          : allLabels.instrumentationLibraryInfo != null) {
+        return false;
+      }
+      if (clock != null ? !clock.equals(allLabels.clock) : allLabels.clock != null) {
+        return false;
+      }
+      if (aggregatorFactory != null
+          ? !aggregatorFactory.equals(allLabels.aggregatorFactory)
+          : allLabels.aggregatorFactory != null) {
+        return false;
+      }
+      return aggregatorMap != null
+          ? aggregatorMap.equals(allLabels.aggregatorMap)
+          : allLabels.aggregatorMap == null;
+    }
+
+    @Override
+    public int hashCode() {
+      int result = descriptor != null ? descriptor.hashCode() : 0;
+      result = 31 * result + (aggregation != null ? aggregation.hashCode() : 0);
+      result = 31 * result + (resource != null ? resource.hashCode() : 0);
+      result =
+          31 * result
+              + (instrumentationLibraryInfo != null ? instrumentationLibraryInfo.hashCode() : 0);
+      result = 31 * result + (clock != null ? clock.hashCode() : 0);
+      result = 31 * result + (aggregatorFactory != null ? aggregatorFactory.hashCode() : 0);
+      result = 31 * result + (aggregatorMap != null ? aggregatorMap.hashCode() : 0);
+      result = 31 * result + (int) (startEpochNanos ^ (startEpochNanos >>> 32));
+      result = 31 * result + (delta ? 1 : 0);
+      return result;
+    }
   }
 
   private Batchers() {}

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

@@ -13,6 +13,8 @@
 import io.opentelemetry.sdk.internal.MillisClock;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricProducer;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration;
+import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -31,11 +33,12 @@ public final class MeterSdkProvider implements MeterProvider {
 
   private final MeterSdkComponentRegistry registry;
   private final MetricProducer metricProducer;
+  private final ViewRegistry viewRegistry = new ViewRegistry();
 
   private MeterSdkProvider(Clock clock, Resource resource) {
     this.registry =
         new MeterSdkComponentRegistry(
-            MeterProviderSharedState.create(clock, resource), new ViewRegistry());
+            MeterProviderSharedState.create(clock, resource), viewRegistry);
     this.metricProducer = new MetricProducerSdk(this.registry);
   }
 
@@ -135,6 +138,34 @@ public MeterSdk newComponent(InstrumentationLibraryInfo instrumentationLibraryIn
     }
   }
 
+  /**
+   * Register a view with the given {@link InstrumentSelector}.
+   *
+   * <p>Example on how to register a view:
+   *
+   * <pre>{@code
+   * // get a handle to the MeterSdkProvider
+   * MeterSdkProvider meterProvider = OpenTelemetrySdk.getMeterProvider();
+   *
+   * // create a selector to select which instruments to customize:
+   * InstrumentSelector instrumentSelector = InstrumentSelector.newBuilder()
+   *   .instrumentType(InstrumentType.COUNTER)
+   *   .build();
+   *
+   * // create a specification of how you want the metrics aggregated:
+   * AggregationConfiguration viewSpecification =
+   *   AggregationConfiguration.create(Aggregations.minMaxSumCount(), Temporality.DELTA);
+   *
+   * //register the view with the MeterSdkProvider
+   * meterProvider.registerView(instrumentSelector, viewSpecification);
+   * }</pre>
+   *
+   * @see AggregationConfiguration
+   */
+  public void registerView(InstrumentSelector selector, AggregationConfiguration specification) {
+    viewRegistry.registerView(selector, specification);
+  }
+
   private static final class MetricProducerSdk implements MetricProducer {
     private final MeterSdkComponentRegistry registry;
 

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

@@ -6,27 +6,44 @@
 package io.opentelemetry.sdk.metrics;
 
 import io.opentelemetry.sdk.metrics.view.Aggregation;
-import io.opentelemetry.sdk.metrics.view.Aggregations;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration.Temporality;
+import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
 
 // notes:
 //  specify by pieces of the descriptor.
-//    instrument type
-//    instrument value type
-//    instrument name  (wildcards allowed?)
+//    instrument type √
+//    instrument name  (regex) √
+//    instrument value type (?)
 //    constant labels (?)
 //    units (?)
 
 // what you can choose:
-//   aggregation
+//   aggregation √
+//   delta vs. cumulative √
 //   all labels vs. a list of labels
-//   delta vs. cumulative
 
 /**
  * Central location for Views to be registered. Registration of a view should eventually be done via
  * the {@link io.opentelemetry.sdk.metrics.MeterSdkProvider}.
  */
 class ViewRegistry {
 
+  private final AggregationChooser aggregationChooser;
+
+  ViewRegistry() {
+    this(new AggregationChooser());
+  }
+
+  // VisibleForTesting
+  ViewRegistry(AggregationChooser aggregationChooser) {
+    this.aggregationChooser = aggregationChooser;
+  }
+
+  void registerView(InstrumentSelector selector, AggregationConfiguration specification) {
+    aggregationChooser.addView(selector, specification);
+  }
+
   /**
    * Create a new {@link io.opentelemetry.sdk.metrics.Batcher} for use in metric recording
    * aggregation.
@@ -36,39 +53,17 @@ Batcher createBatcher(
       MeterSharedState meterSharedState,
       InstrumentDescriptor descriptor) {
 
-    Aggregation aggregation = getRegisteredAggregation(descriptor);
+    AggregationConfiguration specification = aggregationChooser.chooseAggregation(descriptor);
 
-    // todo: don't just use the defaults!
-    switch (descriptor.getType()) {
-      case COUNTER:
-      case UP_DOWN_COUNTER:
-      case SUM_OBSERVER:
-      case UP_DOWN_SUM_OBSERVER:
-        return Batchers.getCumulativeAllLabels(
-            descriptor, meterProviderSharedState, meterSharedState, aggregation);
-      case VALUE_RECORDER:
-        // TODO: Revisit the batcher used here for value observers,
-        // currently this does not remove duplicate records in the same cycle.
-      case VALUE_OBSERVER:
-        return Batchers.getDeltaAllLabels(
-            descriptor, meterProviderSharedState, meterSharedState, aggregation);
-    }
-    throw new IllegalArgumentException("Unknown descriptor type: " + descriptor.getType());
-  }
+    Aggregation aggregation = specification.aggregation();
 
-  private static Aggregation getRegisteredAggregation(InstrumentDescriptor descriptor) {
-    // todo look up based on fields of the descriptor.
-    switch (descriptor.getType()) {
-      case COUNTER:
-      case UP_DOWN_COUNTER:
-        return Aggregations.sum();
-      case VALUE_RECORDER:
-        return Aggregations.minMaxSumCount();
-      case VALUE_OBSERVER:
-      case SUM_OBSERVER:
-      case UP_DOWN_SUM_OBSERVER:
-        return Aggregations.lastValue();
+    if (Temporality.CUMULATIVE == specification.temporality()) {
+      return Batchers.getCumulativeAllLabels(
+          descriptor, meterProviderSharedState, meterSharedState, aggregation);
+    } else if (Temporality.DELTA == specification.temporality()) {
+      return Batchers.getDeltaAllLabels(
+          descriptor, meterProviderSharedState, meterSharedState, aggregation);
     }
-    throw new IllegalArgumentException("Unknown descriptor type: " + descriptor.getType());
+    throw new IllegalStateException("unsupported Temporality: " + specification.temporality());
   }
 }