Very basic Aggregation-configuration API in the SDK. (#2037)

CHANGELOG: SDK : Enhancement: A basic aggregation configuration API has been added to the SDK's meter provider implementation.

* Create a very basic view API in the SDK.

* fix formatting

* move the ViewRegistry up one package, and clean up the visibility of other classes

* Support matching by instrument name

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

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/src/main/java/io/opentelemetry/sdk/metrics/view/ViewSpecification.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/src/main/java/io/opentelemetry/sdk/metrics/view/ViewSpecification.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/src/main/java/io/opentelemetry/sdk/metrics/view/ViewSpecification.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/src/main/java/io/opentelemetry/sdk/metrics/view/ViewSpecification.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* fix formatting issues from GH

* small renaming to a big name

* small renaming to a big name

* re-order matching check and fix a merge issue

* Update from upstream changes.

* Update from upstream changes.

* Adjust defaults based on the latest behavior

* refactor before writing tests

* tests for the AggregationChooser and a bugfix they uncovered

* tests for the ViewRegistry

* Javadoc for the AggregationConfiguration

* Add more javadoc.

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

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

* fix formatting issues

* Update sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

Co-authored-by: Anuraag Agrawal <anuraaga@gmail.com>

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

@@ -41,4 +41,9 @@ final class ActiveBatcher implements Batcher {
   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 @@ final class Batchers {
     public List<MetricData> completeCollectionCycle() {
       return Collections.emptyList();
     }
+
+    @Override
+    public boolean generatesDeltas() {
+      return false;
+    }
   }
 
   private static final class AllLabels implements Batcher {
@@ -155,6 +160,75 @@ final class Batchers {
               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.ComponentRegistry;
 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;
@@ -35,11 +37,12 @@ public final class MeterSdkProvider implements MeterProvider {
   static final String DEFAULT_METER_NAME = "unknown";
   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);
   }
 
@@ -144,6 +147,34 @@ public final class MeterSdkProvider implements MeterProvider {
     }
   }
 
+  /**
+   * 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,20 +6,22 @@
 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 type √
-//    instrument value type
+//    instrument name  (regex) √
-//    instrument name  (wildcards allowed?)
+//    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
@@ -27,6 +29,21 @@ import io.opentelemetry.sdk.metrics.view.Aggregations;
  */
 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 @@ class ViewRegistry {
       MeterSharedState meterSharedState,
       InstrumentDescriptor descriptor) {
 
-    Aggregation aggregation = getRegisteredAggregation(descriptor);
+    AggregationConfiguration specification = aggregationChooser.chooseAggregation(descriptor);
 
-    // todo: don't just use the defaults!
+    Aggregation aggregation = specification.aggregation();
-    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());
-  }
 
-  private static Aggregation getRegisteredAggregation(InstrumentDescriptor descriptor) {
+    if (Temporality.CUMULATIVE == specification.temporality()) {
-    // todo look up based on fields of the descriptor.
+      return Batchers.getCumulativeAllLabels(
-    switch (descriptor.getType()) {
+          descriptor, meterProviderSharedState, meterSharedState, aggregation);
-      case COUNTER:
+    } else if (Temporality.DELTA == specification.temporality()) {
-      case UP_DOWN_COUNTER:
+      return Batchers.getDeltaAllLabels(
-        return Aggregations.sum();
+          descriptor, meterProviderSharedState, meterSharedState, aggregation);
-      case VALUE_RECORDER:
-        return Aggregations.minMaxSumCount();
-      case VALUE_OBSERVER:
-      case SUM_OBSERVER:
-      case UP_DOWN_SUM_OBSERVER:
-        return Aggregations.lastValue();
     }
-    throw new IllegalArgumentException("Unknown descriptor type: " + descriptor.getType());
+    throw new IllegalStateException("unsupported Temporality: " + specification.temporality());
   }
 }

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics.view;
+
+import com.google.auto.value.AutoValue;
+import io.opentelemetry.api.metrics.Instrument;
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * An AggregationConfiguration describes how an aggregation should be performed. It includes both an
+ * {@link Aggregation} which implements what shape of aggregation is created (i.e. histogram, sum,
+ * minMaxSumCount, etc), and a {@link AggregationConfiguration.Temporality} which describes whether
+ * aggregations should be reset with every collection interval, or continue to accumulate across
+ * collection intervals.
+ */
+@AutoValue
+@Immutable
+public abstract class AggregationConfiguration {
+
+  /** Returns a new configuration with the provided options. */
+  public static AggregationConfiguration create(Aggregation aggregation, Temporality temporality) {
+    return new AutoValue_AggregationConfiguration(aggregation, temporality);
+  }
+
+  /** Returns the {@link Aggregation} that should be used for this View. */
+  public abstract Aggregation aggregation();
+
+  /** Returns the {@link Temporality} that should be used for this View (delta vs. cumulative). */
+  public abstract Temporality temporality();
+
+  /** An enumeration which describes the time period over which metrics should be aggregated. */
+  public enum Temporality {
+    /** Metrics will be aggregated only over the most recent collection interval. */
+    DELTA,
+    /** Metrics will be aggregated over the lifetime of the associated {@link Instrument}. */
+    CUMULATIVE
+  }
+}

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/Aggregations.java

@@ -99,6 +99,11 @@ public class Aggregations {
       return instrumentType == InstrumentType.VALUE_OBSERVER
           || instrumentType == InstrumentType.VALUE_RECORDER;
     }
+
+    @Override
+    public String toString() {
+      return getClass().getSimpleName();
+    }
   }
 
   @Immutable
@@ -142,6 +147,11 @@ public class Aggregations {
       // Available for all instruments.
       return true;
     }
+
+    @Override
+    public String toString() {
+      return getClass().getSimpleName();
+    }
   }
 
   @Immutable
@@ -170,6 +180,11 @@ public class Aggregations {
       // Available for all instruments.
       return true;
     }
+
+    @Override
+    public String toString() {
+      return getClass().getSimpleName();
+    }
   }
 
   @Immutable
@@ -201,6 +216,11 @@ public class Aggregations {
     public boolean availableForInstrument(InstrumentType instrumentType) {
       throw new UnsupportedOperationException("Implement this");
     }
+
+    @Override
+    public String toString() {
+      return getClass().getSimpleName();
+    }
   }
 
   @Immutable
@@ -248,6 +268,11 @@ public class Aggregations {
       return instrumentType == InstrumentType.SUM_OBSERVER
           || instrumentType == InstrumentType.UP_DOWN_SUM_OBSERVER;
     }
+
+    @Override
+    public String toString() {
+      return getClass().getSimpleName();
+    }
   }
 
   private Aggregations() {}

sdk/metrics/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics.view;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * Provides means for selecting one ore more {@link io.opentelemetry.api.metrics.Instrument}s. Used
+ * for configuring aggregations for the specified instruments.
+ *
+ * <p>There are two options for selecting instruments: by instrument name and by instrument type.
+ */
+@AutoValue
+@Immutable
+public abstract class InstrumentSelector {
+  public static Builder newBuilder() {
+    return new AutoValue_InstrumentSelector.Builder();
+  }
+
+  /**
+   * Returns {@link InstrumentType} that should be selected. If null, then this specifier will not
+   * be used.
+   */
+  @Nullable
+  public abstract InstrumentType instrumentType();
+
+  /**
+   * Returns which instrument names should be selected. This is a regex. If null, then this
+   * specifier will not be used.
+   */
+  @Nullable
+  public abstract String instrumentNameRegex();
+
+  /**
+   * Returns the {@link Pattern} generated by the provided {@link #instrumentNameRegex()}, or null
+   * if none was specified.
+   */
+  @Nullable
+  @Memoized
+  public Pattern instrumentNamePattern() {
+    return instrumentNameRegex() == null ? null : Pattern.compile(instrumentNameRegex());
+  }
+
+  /** Returns whether the InstrumentType been specified. */
+  public boolean hasInstrumentType() {
+    return instrumentType() != null;
+  }
+
+  /** Returns whether the instrument name regex been specified. */
+  public boolean hasInstrumentNameRegex() {
+    return instrumentNameRegex() != null;
+  }
+
+  /** Builder for {@link InstrumentSelector} instances. */
+  @AutoValue.Builder
+  public interface Builder {
+    /** Sets a specifier for {@link InstrumentType}. */
+    Builder instrumentType(InstrumentType instrumentType);
+
+    /** Sets a specifier for selecting Instruments by name. */
+    Builder instrumentNameRegex(String regex);
+
+    /** Returns an InstrumentSelector instance with the content of this builder. */
+    InstrumentSelector build();
+  }
+}

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/AggregationChooserTest.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
+import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration;
+import io.opentelemetry.sdk.metrics.view.Aggregations;
+import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
+import org.junit.jupiter.api.Test;
+
+class AggregationChooserTest {
+
+  @Test
+  void selection_onType() {
+    AggregationConfiguration configuration =
+        AggregationConfiguration.create(
+            Aggregations.sum(), AggregationConfiguration.Temporality.DELTA);
+
+    AggregationChooser aggregationChooser = new AggregationChooser();
+    aggregationChooser.addView(
+        InstrumentSelector.newBuilder().instrumentType(InstrumentType.COUNTER).build(),
+        configuration);
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(configuration);
+    // this one hasn't been configured, so it gets the default still..
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.UP_DOWN_COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.sum(), AggregationConfiguration.Temporality.CUMULATIVE));
+  }
+
+  @Test
+  void selection_onName() {
+    AggregationConfiguration configuration =
+        AggregationConfiguration.create(
+            Aggregations.sum(), AggregationConfiguration.Temporality.DELTA);
+
+    AggregationChooser aggregationChooser = new AggregationChooser();
+    aggregationChooser.addView(
+        InstrumentSelector.newBuilder().instrumentNameRegex("overridden").build(), configuration);
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "overridden", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(configuration);
+    // this one hasn't been configured, so it gets the default still..
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "default", "", "", InstrumentType.UP_DOWN_COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.sum(), AggregationConfiguration.Temporality.CUMULATIVE));
+  }
+
+  @Test
+  void selection_moreSpecificWins() {
+    AggregationConfiguration configuration1 =
+        AggregationConfiguration.create(
+            Aggregations.sum(), AggregationConfiguration.Temporality.DELTA);
+    AggregationConfiguration configuration2 =
+        AggregationConfiguration.create(
+            Aggregations.count(), AggregationConfiguration.Temporality.DELTA);
+
+    AggregationChooser aggregationChooser = new AggregationChooser();
+    aggregationChooser.addView(
+        InstrumentSelector.newBuilder()
+            .instrumentNameRegex("overridden")
+            .instrumentType(InstrumentType.COUNTER)
+            .build(),
+        configuration2);
+    aggregationChooser.addView(
+        InstrumentSelector.newBuilder().instrumentType(InstrumentType.COUNTER).build(),
+        configuration1);
+
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "overridden", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(configuration2);
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "default", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(configuration1);
+  }
+
+  @Test
+  void selection_regex() {
+    AggregationConfiguration configuration1 =
+        AggregationConfiguration.create(
+            Aggregations.sum(), AggregationConfiguration.Temporality.DELTA);
+
+    AggregationChooser aggregationChooser = new AggregationChooser();
+    aggregationChooser.addView(
+        InstrumentSelector.newBuilder()
+            .instrumentNameRegex("overrid(es|den)")
+            .instrumentType(InstrumentType.COUNTER)
+            .build(),
+        configuration1);
+
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "overridden", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(configuration1);
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "overrides", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(configuration1);
+    // this one hasn't been configured, so it gets the default still..
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "default", "", "", InstrumentType.UP_DOWN_COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.sum(), AggregationConfiguration.Temporality.CUMULATIVE));
+  }
+
+  @Test
+  void defaults() {
+    AggregationChooser aggregationChooser = new AggregationChooser();
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.sum(), AggregationConfiguration.Temporality.CUMULATIVE));
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.UP_DOWN_COUNTER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.sum(), AggregationConfiguration.Temporality.CUMULATIVE));
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.VALUE_RECORDER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.minMaxSumCount(), AggregationConfiguration.Temporality.DELTA));
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.SUM_OBSERVER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.lastValue(), AggregationConfiguration.Temporality.CUMULATIVE));
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.VALUE_OBSERVER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.lastValue(), AggregationConfiguration.Temporality.DELTA));
+    assertThat(
+            aggregationChooser.chooseAggregation(
+                InstrumentDescriptor.create(
+                    "", "", "", InstrumentType.UP_DOWN_SUM_OBSERVER, InstrumentValueType.LONG)))
+        .isEqualTo(
+            AggregationConfiguration.create(
+                Aggregations.lastValue(), AggregationConfiguration.Temporality.CUMULATIVE));
+  }
+}

sdk/metrics/src/test/java/io/opentelemetry/sdk/metrics/ViewRegistryTest.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
+import io.opentelemetry.sdk.internal.TestClock;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
+import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration;
+import io.opentelemetry.sdk.metrics.view.Aggregations;
+import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
+import io.opentelemetry.sdk.resources.Resource;
+import org.junit.jupiter.api.Test;
+
+class ViewRegistryTest {
+
+  @Test
+  void registerView() {
+    AggregationChooser chooser = mock(AggregationChooser.class);
+
+    ViewRegistry viewRegistry = new ViewRegistry(chooser);
+    InstrumentSelector selector =
+        InstrumentSelector.newBuilder().instrumentType(InstrumentType.COUNTER).build();
+    AggregationConfiguration specification =
+        AggregationConfiguration.create(
+            Aggregations.count(), AggregationConfiguration.Temporality.CUMULATIVE);
+
+    viewRegistry.registerView(selector, specification);
+
+    verify(chooser).addView(selector, specification);
+  }
+
+  @Test
+  void createBatcher_cumulative() {
+    AggregationChooser chooser = mock(AggregationChooser.class);
+
+    ViewRegistry viewRegistry = new ViewRegistry(chooser);
+
+    InstrumentDescriptor descriptor =
+        InstrumentDescriptor.create(
+            "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.DOUBLE);
+    MeterProviderSharedState providerSharedState =
+        MeterProviderSharedState.create(TestClock.create(), Resource.getEmpty());
+    MeterSharedState meterSharedState =
+        MeterSharedState.create(InstrumentationLibraryInfo.create("test", "1.0"));
+
+    AggregationConfiguration specification =
+        AggregationConfiguration.create(
+            Aggregations.count(), AggregationConfiguration.Temporality.CUMULATIVE);
+    Batcher expectedBatcher =
+        Batchers.getCumulativeAllLabels(
+            descriptor, providerSharedState, meterSharedState, Aggregations.count());
+
+    when(chooser.chooseAggregation(descriptor)).thenReturn(specification);
+
+    Batcher result = viewRegistry.createBatcher(providerSharedState, meterSharedState, descriptor);
+
+    assertThat(result.generatesDeltas()).isFalse();
+    assertThat(result).isEqualTo(expectedBatcher);
+
+    assertThat(result).isNotNull();
+  }
+
+  @Test
+  void createBatcher_delta() {
+    AggregationChooser chooser = mock(AggregationChooser.class);
+
+    ViewRegistry viewRegistry = new ViewRegistry(chooser);
+
+    InstrumentDescriptor descriptor =
+        InstrumentDescriptor.create(
+            "name", "description", "unit", InstrumentType.COUNTER, InstrumentValueType.DOUBLE);
+    MeterProviderSharedState providerSharedState =
+        MeterProviderSharedState.create(TestClock.create(), Resource.getEmpty());
+    MeterSharedState meterSharedState =
+        MeterSharedState.create(InstrumentationLibraryInfo.create("test", "1.0"));
+
+    AggregationConfiguration specification =
+        AggregationConfiguration.create(
+            Aggregations.count(), AggregationConfiguration.Temporality.DELTA);
+    Batcher expectedBatcher =
+        Batchers.getDeltaAllLabels(
+            descriptor, providerSharedState, meterSharedState, Aggregations.count());
+
+    when(chooser.chooseAggregation(descriptor)).thenReturn(specification);
+
+    Batcher result = viewRegistry.createBatcher(providerSharedState, meterSharedState, descriptor);
+
+    assertThat(result.generatesDeltas()).isTrue();
+    assertThat(result).isEqualTo(expectedBatcher);
+
+    assertThat(result).isNotNull();
+  }
+}

sdk/src/main/java/io/opentelemetry/sdk/metrics/view/AggregationConfiguration.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2020, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opentelemetry.sdk.metrics.view;
+
+import com.google.auto.value.AutoValue;
+import io.opentelemetry.metrics.Instrument;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+@AutoValue
+@Immutable
+public abstract class AggregationConfiguration {
+
+  public static AggregationConfiguration create(Aggregation aggregation, Temporality temporality) {
+    return new AutoValue_AggregationConfiguration(aggregation, temporality);
+  }
+
+  /** Returns the {@link Aggregation} that should be used for this View. */
+  @Nullable
+  public abstract Aggregation aggregation();
+
+  /** Returns the {@link Temporality} that should be used for this View (delta vs. cumulative). */
+  @Nullable
+  public abstract Temporality temporality();
+
+  /** An enumeration which describes the time period over which metrics should be aggregated. */
+  public enum Temporality {
+    /** Metrics will be aggregated only over the most recent collection interval. */
+    DELTA,
+    /** Metrics will be aggregated over the lifetime of the associated {@link Instrument}. */
+    CUMULATIVE
+  }
+}

sdk/src/main/java/io/opentelemetry/sdk/metrics/view/InstrumentSelector.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2020, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opentelemetry.sdk.metrics.view;
+
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+@AutoValue
+@Immutable
+public abstract class InstrumentSelector {
+
+  public static Builder newBuilder() {
+    return new AutoValue_InstrumentSelector.Builder();
+  }
+
+  @Nullable
+  public abstract InstrumentType instrumentType();
+
+  @Nullable
+  public abstract String instrumentNameRegex();
+
+  @Memoized
+  @Nullable
+  public Pattern instrumentNamePattern() {
+    return instrumentNameRegex() == null ? null : Pattern.compile(instrumentNameRegex());
+  }
+
+  @AutoValue.Builder
+  public interface Builder {
+    Builder instrumentType(InstrumentType instrumentType);
+
+    Builder instrumentNameRegex(String regex);
+
+    InstrumentSelector build();
+  }
+}

sdk/src/test/java/io/opentelemetry/sdk/metrics/ViewRegistryTest.java

@@ -0,0 +1,251 @@
+/*
+ * Copyright 2020, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opentelemetry.sdk.metrics;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.when;
+
+import io.opentelemetry.common.Labels;
+import io.opentelemetry.sdk.internal.TestClock;
+import io.opentelemetry.sdk.metrics.aggregator.DoubleLastValueAggregator;
+import io.opentelemetry.sdk.metrics.aggregator.DoubleMinMaxSumCount;
+import io.opentelemetry.sdk.metrics.aggregator.DoubleSumAggregator;
+import io.opentelemetry.sdk.metrics.aggregator.LongLastValueAggregator;
+import io.opentelemetry.sdk.metrics.aggregator.LongMinMaxSumCount;
+import io.opentelemetry.sdk.metrics.aggregator.LongSumAggregator;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
+import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration;
+import io.opentelemetry.sdk.metrics.view.AggregationConfiguration.Temporality;
+import io.opentelemetry.sdk.metrics.view.Aggregations;
+import io.opentelemetry.sdk.metrics.view.InstrumentSelector;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ViewRegistryTest {
+
+  @Mock private MeterSharedState meterSharedState;
+  @Mock private MeterProviderSharedState meterProviderSharedState;
+
+  @Before
+  public void setUp() {
+    when(meterProviderSharedState.getClock()).thenReturn(TestClock.create());
+  }
+
+  @Test
+  public void defaultAggregations() {
+    ViewRegistry viewRegistry = new ViewRegistry();
+
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.VALUE_RECORDER,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ true,
+        DoubleMinMaxSumCount.class);
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.VALUE_RECORDER,
+        InstrumentValueType.LONG,
+        /* expectedDeltas=*/ true,
+        LongMinMaxSumCount.class);
+
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.VALUE_OBSERVER,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ true,
+        DoubleMinMaxSumCount.class);
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.VALUE_OBSERVER,
+        InstrumentValueType.LONG,
+        /* expectedDeltas=*/ true,
+        LongMinMaxSumCount.class);
+
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.COUNTER,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ false,
+        DoubleSumAggregator.class);
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.COUNTER,
+        InstrumentValueType.LONG,
+        /* expectedDeltas=*/ false,
+        LongSumAggregator.class);
+
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.UP_DOWN_COUNTER,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ false,
+        DoubleSumAggregator.class);
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.UP_DOWN_COUNTER,
+        InstrumentValueType.LONG,
+        /* expectedDeltas=*/ false,
+        LongSumAggregator.class);
+
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.SUM_OBSERVER,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ false,
+        DoubleLastValueAggregator.class);
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.SUM_OBSERVER,
+        InstrumentValueType.LONG,
+        /* expectedDeltas=*/ false,
+        LongLastValueAggregator.class);
+
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.UP_DOWN_SUM_OBSERVER,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ false,
+        DoubleLastValueAggregator.class);
+    verifyCorrect(
+        viewRegistry,
+        InstrumentType.UP_DOWN_SUM_OBSERVER,
+        InstrumentValueType.LONG,
+        /* expectedDeltas=*/ false,
+        LongLastValueAggregator.class);
+  }
+
+  @Test
+  public void selectByInstrumentType() {
+    ViewRegistry viewRegistry = new ViewRegistry();
+
+    InstrumentType instrumentType = InstrumentType.VALUE_RECORDER;
+
+    InstrumentSelector selector =
+        InstrumentSelector.newBuilder().instrumentType(instrumentType).build();
+    AggregationConfiguration view =
+        AggregationConfiguration.create(Aggregations.sum(), Temporality.CUMULATIVE);
+    viewRegistry.registerView(selector, view);
+
+    verifyCorrect(
+        viewRegistry,
+        instrumentType,
+        InstrumentValueType.DOUBLE,
+        /* expectedDeltas=*/ false,
+        DoubleSumAggregator.class);
+  }
+
+  @Test
+  public void selectByInstrumentName() {
+    ViewRegistry viewRegistry = new ViewRegistry();
+
+    InstrumentSelector selector =
+        InstrumentSelector.newBuilder().instrumentNameRegex("http.*duration").build();
+    AggregationConfiguration view =
+        AggregationConfiguration.create(Aggregations.sum(), Temporality.CUMULATIVE);
+
+    viewRegistry.registerView(selector, view);
+
+    InstrumentType instrumentType = InstrumentType.VALUE_RECORDER;
+    // this one matches on name
+    verifyCorrect(
+        viewRegistry,
+        createDescriptor(instrumentType, InstrumentValueType.DOUBLE, "http.server.duration"),
+        /* expectedDeltas= */ false,
+        DoubleSumAggregator.class);
+    // this one does not match on name
+    verifyCorrect(
+        viewRegistry,
+        createDescriptor(instrumentType, InstrumentValueType.DOUBLE, "foo.bar.duration"),
+        /* expectedDeltas=*/ true,
+        DoubleMinMaxSumCount.class);
+  }
+
+  @Test
+  public void selectByInstrumentNameAndType() {
+    ViewRegistry viewRegistry = new ViewRegistry();
+
+    InstrumentSelector selector =
+        InstrumentSelector.newBuilder()
+            .instrumentType(InstrumentType.VALUE_RECORDER)
+            .instrumentNameRegex("http.*duration")
+            .build();
+    AggregationConfiguration view =
+        AggregationConfiguration.create(Aggregations.sum(), Temporality.CUMULATIVE);
+
+    viewRegistry.registerView(selector, view);
+
+    // this one matches on name
+    verifyCorrect(
+        viewRegistry,
+        createDescriptor(
+            InstrumentType.VALUE_RECORDER, InstrumentValueType.DOUBLE, "http.server.duration"),
+        /* expectedDeltas= */ false,
+        DoubleSumAggregator.class);
+    // this one does not match on name, but does on type, so should get the default
+    verifyCorrect(
+        viewRegistry,
+        createDescriptor(
+            InstrumentType.VALUE_RECORDER, InstrumentValueType.DOUBLE, "foo.bar.duration"),
+        /* expectedDeltas=*/ true,
+        DoubleMinMaxSumCount.class);
+    // this one does not match on type, but does on name, so should get the default
+    verifyCorrect(
+        viewRegistry,
+        createDescriptor(
+            InstrumentType.SUM_OBSERVER, InstrumentValueType.DOUBLE, "http.bar.duration"),
+        /* expectedDeltas=*/ false,
+        DoubleLastValueAggregator.class);
+  }
+
+  private void verifyCorrect(
+      ViewRegistry viewRegistry,
+      InstrumentType instrumentType,
+      InstrumentValueType valueType,
+      boolean expectedDeltas,
+      Class<?> expectedAggregator) {
+    verifyCorrect(
+        viewRegistry,
+        createDescriptor(instrumentType, valueType, "foo"),
+        expectedDeltas,
+        expectedAggregator);
+  }
+
+  private void verifyCorrect(
+      ViewRegistry viewRegistry,
+      InstrumentDescriptor descriptor,
+      boolean expectedDeltas,
+      Class<?> expectedAggregator) {
+    Batcher batcher =
+        viewRegistry.createBatcher(meterProviderSharedState, meterSharedState, descriptor);
+
+    assertThat(batcher.generatesDeltas()).isEqualTo(expectedDeltas);
+    assertThat(batcher.getAggregator()).isInstanceOf(expectedAggregator);
+  }
+
+  private static InstrumentDescriptor createDescriptor(
+      InstrumentType instrumentType,
+      InstrumentValueType instrumentValueType,
+      String instrumentName) {
+    return InstrumentDescriptor.create(
+        instrumentName, "foo desc", "ms", Labels.empty(), instrumentType, instrumentValueType);
+  }
+}