Add AllLabels Batcher and install default Barcher for Counters

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

@@ -18,7 +18,11 @@
 
 import io.opentelemetry.metrics.Counter;
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
 import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
+import io.opentelemetry.sdk.metrics.data.MetricData.Descriptor;
+import io.opentelemetry.sdk.metrics.view.Aggregation;
+import io.opentelemetry.sdk.metrics.view.Aggregations;
 
 abstract class AbstractCounter<B extends AbstractBoundInstrument>
     extends AbstractInstrumentWithBinding<B> {
@@ -35,7 +39,13 @@ abstract class AbstractCounter<B extends AbstractBoundInstrument>
         descriptor,
         meterProviderSharedState,
         instrumentationLibraryInfo,
-        new ActiveBatcher(Batchers.getNoop()));
+        new ActiveBatcher(
+            getDefaultBatcher(
+                descriptor,
+                getCounterInstrumentType(monotonic),
+                instrumentValueType,
+                meterProviderSharedState,
+                instrumentationLibraryInfo)));
     this.monotonic = monotonic;
     this.instrumentValueType = instrumentValueType;
   }
@@ -90,4 +100,37 @@ final boolean isMonotonic() {
       return this.monotonic;
     }
   }
+
+  private static InstrumentType getCounterInstrumentType(boolean monotonic) {
+    return monotonic ? InstrumentType.COUNTER_MONOTONIC : InstrumentType.COUNTER_NON_MONOTONIC;
+  }
+
+  private static Batcher getDefaultBatcher(
+      InstrumentDescriptor descriptor,
+      InstrumentType instrumentType,
+      InstrumentValueType instrumentValueType,
+      MeterProviderSharedState meterProviderSharedState,
+      InstrumentationLibraryInfo instrumentationLibraryInfo) {
+    Aggregation defaultAggregation = Aggregations.sum();
+    return Batchers.getCumulativeAllLabels(
+        getDefaultMetricDescriptor(
+            descriptor, instrumentType, instrumentValueType, defaultAggregation),
+        meterProviderSharedState.getResource(),
+        instrumentationLibraryInfo,
+        defaultAggregation.getAggregatorFactory(instrumentValueType),
+        meterProviderSharedState.getClock());
+  }
+
+  private static Descriptor getDefaultMetricDescriptor(
+      InstrumentDescriptor descriptor,
+      InstrumentType instrumentType,
+      InstrumentValueType instrumentValueType,
+      Aggregation aggregation) {
+    return Descriptor.create(
+        descriptor.getName(),
+        descriptor.getDescription(),
+        aggregation.getUnit(descriptor.getUnit()),
+        aggregation.getDescriptorType(instrumentType, instrumentValueType),
+        descriptor.getConstantLabels());
+  }
 }

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

@@ -18,6 +18,7 @@
 
 import io.opentelemetry.metrics.Measure;
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
 import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
 
 abstract class AbstractMeasure<B extends AbstractBoundInstrument>
@@ -90,4 +91,8 @@ final boolean isAbsolute() {
       return this.absolute;
     }
   }
+
+  static InstrumentType getInstrumentType(boolean absolute) {
+    return absolute ? InstrumentType.MEASURE_ABSOLUTE : InstrumentType.MEASURE_NON_ABSOLUTE;
+  }
 }

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

@@ -18,6 +18,7 @@
 
 import io.opentelemetry.metrics.Observer;
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
 import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import java.util.Collections;
@@ -98,4 +99,9 @@ final boolean isMonotonic() {
       return this.monotonic;
     }
   }
+
+  // TODO: make this private
+  static InstrumentType getInstrumentType(boolean monotonic) {
+    return monotonic ? InstrumentType.OBSERVER_MONOTONIC : InstrumentType.OBSERVER_NON_MONOTONIC;
+  }
 }

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

@@ -25,6 +25,10 @@
  * A {@code Batcher} represents an internal representation of an {code Instrument} aggregation
  * process. It records individual measurements (via the {@code Aggregator}). It batches together
  * {@code Aggregator}s for the similar {@code LabelSet}.
+ *
+ * <p>The only thread safe method in this class is {@link #getAggregator()}. An entire collection
+ * cycle must be protected by a lock. A collection cycle is defined by multiple calls to {@link
+ * #batch(LabelSet, Aggregator, boolean)} followed by one {@link #completeCollectionCycle()};
  */
 interface Batcher {
 

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

@@ -17,11 +17,20 @@
 package io.opentelemetry.sdk.metrics;
 
 import io.opentelemetry.metrics.LabelSet;
+import io.opentelemetry.sdk.common.Clock;
+import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
 import io.opentelemetry.sdk.metrics.aggregator.Aggregator;
+import io.opentelemetry.sdk.metrics.aggregator.AggregatorFactory;
 import io.opentelemetry.sdk.metrics.aggregator.NoopAggregator;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.data.MetricData.Descriptor;
+import io.opentelemetry.sdk.metrics.data.MetricData.Point;
+import io.opentelemetry.sdk.resources.Resource;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /** A collection of available Batchers. */
 final class Batchers {
@@ -30,6 +39,21 @@ static Batcher getNoop() {
     return Noop.INSTANCE;
   }
 
+  static Batcher getCumulativeAllLabels(
+      Descriptor descriptor,
+      Resource resource,
+      InstrumentationLibraryInfo instrumentationLibraryInfo,
+      AggregatorFactory aggregatorFactory,
+      Clock clock) {
+    return new AllLabels(
+        descriptor,
+        resource,
+        instrumentationLibraryInfo,
+        aggregatorFactory,
+        clock,
+        /* delta= */ false);
+  }
+
   private static final class Noop implements Batcher {
     private static final Noop INSTANCE = new Noop();
 
@@ -47,5 +71,69 @@ public List<MetricData> completeCollectionCycle() {
     }
   }
 
+  private static final class AllLabels implements Batcher {
+    private final Descriptor descriptor;
+    private final Resource resource;
+    private final InstrumentationLibraryInfo instrumentationLibraryInfo;
+    private final Clock clock;
+    private final AggregatorFactory aggregatorFactory;
+    private Map<Map<String, String>, Aggregator> aggregatorMap;
+    private long startEpochNanos;
+    private final boolean delta;
+
+    private AllLabels(
+        Descriptor descriptor,
+        Resource resource,
+        InstrumentationLibraryInfo instrumentationLibraryInfo,
+        AggregatorFactory aggregatorFactory,
+        Clock clock,
+        boolean delta) {
+      this.descriptor = descriptor;
+      this.resource = resource;
+      this.instrumentationLibraryInfo = instrumentationLibraryInfo;
+      this.clock = clock;
+      this.aggregatorFactory = aggregatorFactory;
+      this.delta = delta;
+      this.aggregatorMap = new HashMap<>();
+      startEpochNanos = clock.now();
+    }
+
+    @Override
+    public final Aggregator getAggregator() {
+      return aggregatorFactory.getAggregator();
+    }
+
+    @Override
+    public final void batch(LabelSet labelSet, Aggregator aggregator, boolean unmappedAggregator) {
+      Map<String, String> labels = ((LabelSetSdk) labelSet).getLabels();
+      Aggregator currentAggregator = aggregatorMap.get(labels);
+      if (currentAggregator == null) {
+        // This aggregator is not mapped, we can use this instance.
+        if (unmappedAggregator) {
+          aggregatorMap.put(labels, aggregator);
+          return;
+        }
+        currentAggregator = aggregatorFactory.getAggregator();
+        aggregatorMap.put(labels, currentAggregator);
+      }
+      aggregator.mergeToAndReset(currentAggregator);
+    }
+
+    @Override
+    public final List<MetricData> completeCollectionCycle() {
+      List<Point> points = new ArrayList<>(aggregatorMap.size());
+      long epochNanos = clock.now();
+      for (Map.Entry<Map<String, String>, Aggregator> entry : aggregatorMap.entrySet()) {
+        points.add(entry.getValue().toPoint(startEpochNanos, epochNanos, entry.getKey()));
+      }
+      if (delta) {
+        startEpochNanos = epochNanos;
+        aggregatorMap = new HashMap<>();
+      }
+      return Collections.singletonList(
+          MetricData.create(descriptor, resource, instrumentationLibraryInfo, points));
+    }
+  }
+
   private Batchers() {}
 }

sdk/src/main/java/io/opentelemetry/sdk/metrics/data/MetricData.java

@@ -63,8 +63,7 @@ public abstract class MetricData {
    * <p>Only one type of points are available at any moment for a {@link MetricData}, and the type
    * is determined by the {@link Descriptor.Type}.
    *
-   * @return the data {@link Point}s for this metric, or {@code null} if this type of points are not
-   *     accepted.
+   * @return the data {@link Point}s for this metric, or empty {@code Collection} if no points.
    * @since 0.3.0
    */
   public abstract Collection<Point> getPoints();

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

@@ -51,6 +51,14 @@ public interface Aggregation {
   MetricData.Descriptor.Type getDescriptorType(
       InstrumentType instrumentType, InstrumentValueType instrumentValueType);
 
+  /**
+   * Returns the unit that this {@code Aggregation} will produce.
+   *
+   * @param initialUnit the initial unit for the {@code Instrument}'s measurements.
+   * @return the unit that this {@code Aggregation} will produce.
+   */
+  String getUnit(String initialUnit);
+
   /**
    * Returns {@code true} if this {@code Aggregation} can be applied to the given {@code
    * InstrumentType}.

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

@@ -104,6 +104,11 @@ public Type getDescriptorType(
       throw new IllegalArgumentException("Unsupported instrument/value types");
     }
 
+    @Override
+    public String getUnit(String initialUnit) {
+      return initialUnit;
+    }
+
     @Override
     public boolean availableForInstrument(InstrumentType instrumentType) {
       // Available for all instruments.
@@ -127,6 +132,11 @@ public Type getDescriptorType(
       return Type.MONOTONIC_LONG;
     }
 
+    @Override
+    public String getUnit(String initialUnit) {
+      return "1";
+    }
+
     @Override
     public boolean availableForInstrument(InstrumentType instrumentType) {
       // Available for all instruments.
@@ -154,6 +164,11 @@ public Type getDescriptorType(
       throw new UnsupportedOperationException("Implement this");
     }
 
+    @Override
+    public String getUnit(String initialUnit) {
+      return initialUnit;
+    }
+
     @Override
     public boolean availableForInstrument(InstrumentType instrumentType) {
       throw new UnsupportedOperationException("Implement this");
@@ -176,6 +191,11 @@ public Type getDescriptorType(
       throw new UnsupportedOperationException("Implement this");
     }
 
+    @Override
+    public String getUnit(String initialUnit) {
+      return initialUnit;
+    }
+
     @Override
     public boolean availableForInstrument(InstrumentType instrumentType) {
       throw new UnsupportedOperationException("Implement this");