Bound registry work in progress

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

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

@@ -79,4 +79,8 @@ final void recordLong(long value) {
   final void recordDouble(double value) {
     aggregator.recordDouble(value);
   }
+
+  final Aggregator getAggregator() {
+    return aggregator;
+  }
 }

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

@@ -17,11 +17,15 @@
 package io.opentelemetry.sdk.metrics;
 
 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.view.Aggregations;
+import io.opentelemetry.sdk.metrics.view.View;
 import java.util.List;
 import java.util.Map;
 
-abstract class AbstractCounter extends AbstractInstrument {
+abstract class AbstractCounter<B extends AbstractBoundInstrument>
+    extends AbstractInstrumentWithBinding<B> {
   private final boolean monotonic;
   private final InstrumentValueType instrumentValueType;
 
@@ -35,7 +39,20 @@ abstract class AbstractCounter extends AbstractInstrument {
       MeterSharedState meterSharedState,
       InstrumentationLibraryInfo instrumentationLibraryInfo,
       boolean monotonic) {
-    super(name, description, unit, constantLabels, labelKeys);
+    super(
+        name,
+        description,
+        unit,
+        constantLabels,
+        labelKeys,
+        ActiveBatcher.fromViews(
+            getDefaultView(name, description, constantLabels),
+            unit,
+            constantLabels,
+            getCounterInstrumentType(monotonic),
+            instrumentValueType,
+            meterSharedState,
+            instrumentationLibraryInfo));
     this.monotonic = monotonic;
     this.instrumentValueType = instrumentValueType;
   }
@@ -49,14 +66,14 @@ public boolean equals(Object o) {
     if (this == o) {
       return true;
     }
-    if (!(o instanceof AbstractCounter)) {
+    if (!(o instanceof AbstractCounter<?>)) {
       return false;
     }
     if (!super.equals(o)) {
       return false;
     }
 
-    AbstractCounter that = (AbstractCounter) o;
+    AbstractCounter<?> that = (AbstractCounter<?>) o;
 
     return monotonic == that.monotonic && instrumentValueType == that.instrumentValueType;
   }
@@ -68,4 +85,13 @@ public int hashCode() {
     result = 31 * result + instrumentValueType.hashCode();
     return result;
   }
+
+  private static InstrumentType getCounterInstrumentType(boolean monotonic) {
+    return monotonic ? InstrumentType.COUNTER_MONOTONIC : InstrumentType.COUNTER_NON_MONOTONIC;
+  }
+
+  private static View getDefaultView(
+      String name, String description, Map<String, String> constantLabels) {
+    return View.create(name, description, Aggregations.sum(), constantLabels.keySet());
+  }
 }

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

@@ -17,6 +17,7 @@
 package io.opentelemetry.sdk.metrics;
 
 import io.opentelemetry.metrics.Instrument;
+import io.opentelemetry.sdk.metrics.data.MetricData;
 import java.util.List;
 import java.util.Map;
 
@@ -27,19 +28,23 @@ abstract class AbstractInstrument implements Instrument {
   private final String unit;
   private final Map<String, String> constantLabels;
   private final List<String> labelKeys;
+  private final ActiveBatcher activeBatcher;
 
   // All arguments cannot be null because they are checked in the abstract builder classes.
   AbstractInstrument(
       String name,
       String description,
       String unit,
       Map<String, String> constantLabels,
-      List<String> labelKeys) {
+      List<String> labelKeys,
+      ActiveBatcher activeBatcher) {
     this.name = name;
     this.description = description;
     this.unit = unit;
     this.constantLabels = constantLabels;
     this.labelKeys = labelKeys;
+    // TODO: Allow to install views from config instead of always installing the default View.
+    this.activeBatcher = activeBatcher;
   }
 
   final String getName() {
@@ -62,6 +67,12 @@ final List<String> getLabelKeys() {
     return labelKeys;
   }
 
+  final Batcher getActiveBatcher() {
+    return activeBatcher;
+  }
+
+  abstract List<MetricData> collect();
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

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

@@ -0,0 +1,96 @@
+/*
+ * 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 io.opentelemetry.metrics.LabelSet;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+abstract class AbstractInstrumentWithBinding<B extends AbstractBoundInstrument>
+    extends AbstractInstrument {
+  private final ConcurrentHashMap<LabelSet, B> boundLabels;
+  private final ReentrantLock collectLock;
+
+  AbstractInstrumentWithBinding(
+      String name,
+      String description,
+      String unit,
+      Map<String, String> constantLabels,
+      List<String> labelKeys,
+      ActiveBatcher activeBatcher) {
+    super(name, description, unit, constantLabels, labelKeys, activeBatcher);
+    boundLabels = new ConcurrentHashMap<>();
+    collectLock = new ReentrantLock();
+  }
+
+  // Cannot make this "bind" because of a Java problem if we make this class also implement the
+  // InstrumentWithBinding then the subclass will fail to compile because of different "bind"
+  // signature. This is a good trade-off.
+  final B bindInternal(LabelSet labelSet) {
+    B binding = boundLabels.get(labelSet);
+    if (binding != null && binding.bind()) {
+      // At this moment it is guaranteed that the Bound is in the map and will not be removed.
+      return binding;
+    }
+
+    // Missing entry or no longer mapped, try to add a new entry.
+    binding = newBinding(getActiveBatcher());
+    while (true) {
+      B oldBound = boundLabels.putIfAbsent(labelSet, binding);
+      if (oldBound != null) {
+        if (oldBound.bind()) {
+          // At this moment it is guaranteed that the Bound is in the map and will not be removed.
+          return oldBound;
+        }
+        // Try to remove the oldBound. This will race with the collect method, but only one will
+        // succeed.
+        boundLabels.remove(labelSet, oldBound);
+        continue;
+      }
+      return binding;
+    }
+  }
+
+  /**
+   * Collects records from all the entries (labelSet, Bound) that changed since the last collect()
+   * call.
+   */
+  @Override
+  final List<MetricData> collect() {
+    collectLock.lock();
+    try {
+      Batcher batcher = getActiveBatcher();
+      for (Map.Entry<LabelSet, B> entry : boundLabels.entrySet()) {
+        boolean unmappedEntry = entry.getValue().tryUnmap();
+        if (unmappedEntry) {
+          // If able to unmap then remove the record from the current Map. This can race with the
+          // acquire but because we requested a specific value only one will succeed.
+          boundLabels.remove(entry.getKey(), entry.getValue());
+        }
+        batcher.batch(entry.getKey(), entry.getValue().getAggregator(), unmappedEntry);
+      }
+      return batcher.stopCollection();
+    } finally {
+      collectLock.unlock();
+    }
+  }
+
+  abstract B newBinding(Batcher batcher);
+}

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

@@ -17,11 +17,13 @@
 package io.opentelemetry.sdk.metrics;
 
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
+import io.opentelemetry.sdk.metrics.common.InstrumentType;
 import io.opentelemetry.sdk.metrics.common.InstrumentValueType;
 import java.util.List;
 import java.util.Map;
 
-abstract class AbstractMeasure extends AbstractInstrument {
+abstract class AbstractMeasure<B extends AbstractBoundInstrument>
+    extends AbstractInstrumentWithBinding<B> {
   private final boolean absolute;
   private final InstrumentValueType instrumentValueType;
 
@@ -35,7 +37,8 @@ abstract class AbstractMeasure extends AbstractInstrument {
       MeterSharedState meterSharedState,
       InstrumentationLibraryInfo instrumentationLibraryInfo,
       boolean absolute) {
-    super(name, description, unit, constantLabels, labelKeys);
+    super(
+        name, description, unit, constantLabels, labelKeys, new ActiveBatcher(Batchers.getNoop()));
     this.absolute = absolute;
     this.instrumentValueType = instrumentValueType;
   }
@@ -56,7 +59,7 @@ public boolean equals(Object o) {
       return false;
     }
 
-    AbstractMeasure that = (AbstractMeasure) o;
+    AbstractMeasure<?> that = (AbstractMeasure<?>) o;
 
     return absolute == that.absolute && instrumentValueType == that.instrumentValueType;
   }
@@ -68,4 +71,8 @@ public int hashCode() {
     result = 31 * result + instrumentValueType.hashCode();
     return result;
   }
+
+  static InstrumentType getInstrumentType(boolean absolute) {
+    return absolute ? InstrumentType.MEASURE_ABSOLUTE : InstrumentType.MEASURE_NON_ABSOLUTE;
+  }
 }

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

@@ -17,7 +17,10 @@
 package io.opentelemetry.sdk.metrics;
 
 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;
 import java.util.List;
 import java.util.Map;
 
@@ -35,7 +38,7 @@ public class AbstractObserver extends AbstractInstrument {
       MeterSharedState meterSharedState,
       InstrumentationLibraryInfo instrumentationLibraryInfo,
       boolean monotonic) {
-    super(name, description, unit, constantLabels, labelKeys);
+    super(name, description, unit, constantLabels, labelKeys, null);
     this.monotonic = monotonic;
     this.instrumentValueType = instrumentValueType;
   }
@@ -44,6 +47,12 @@ final boolean isMonotonic() {
     return monotonic;
   }
 
+  @Override
+  List<MetricData> collect() {
+    // TODO: Implement this.
+    return Collections.emptyList();
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -68,4 +77,9 @@ public int hashCode() {
     result = 31 * result + instrumentValueType.hashCode();
     return result;
   }
+
+  // TODO: make this private
+  static InstrumentType getInstrumentType(boolean monotonic) {
+    return monotonic ? InstrumentType.OBSERVER_MONOTONIC : InstrumentType.OBSERVER_NON_MONOTONIC;
+  }
 }