Change to metrics to keep track of when the metric value was first set

This CL also adds IncrementableMetric#reset() methods to allow resetting the
value and start timestamp of IncrementableMetrics.

This is necessary because some backends, like Stackdriver, use non-monotonic
changes in cumulative metric values to detect timeseries restarts. Tracking and
re-setting the start timestamp allows users to track mostly monotonic metrics
which may have non-monotonic discontinuities.

See https://cloud.google.com/monitoring/api/ref_v3/rest/v3/TimeSeries#Point for
more details.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=130795229

java/google/registry/monitoring/metrics/Counter.java

@@ -20,8 +20,12 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.AtomicLongMap;
+import com.google.common.util.concurrent.Striped;
 import google.registry.monitoring.metrics.MetricSchema.Kind;
 import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
 import javax.annotation.concurrent.ThreadSafe;
 import org.joda.time.Instant;
 
@@ -29,17 +33,53 @@ import org.joda.time.Instant;
  * A metric which stores Long values. It is stateful and can be changed in increments.
  *
  * <p>This metric is generally suitable for counters, such as requests served or errors generated.
+ *
+ * <p>The start of the {@link MetricPoint#interval()} of values of instances of this metric will be
+ * set to the time that the metric was first set or last {@link #reset()}.
  */
 @ThreadSafe
 public final class Counter extends AbstractMetric<Long>
     implements SettableMetric<Long>, IncrementableMetric {
 
+  /**
+   * The below constants replicate the default initial capacity, load factor, and concurrency level
+   * for {@link ConcurrentHashMap} as of Java SE 7. They are hardcoded here so that the concurrency
+   * level in {@code valueLocks} below can be set identically.
+   */
+  private static final int HASHMAP_INITIAL_CAPACITY = 16;
+  private static final float HASHMAP_LOAD_FACTOR = 0.75f;
+  private static final int HASHMAP_CONCURRENCY_LEVEL = 16;
+
   private static final String LABEL_COUNT_ERROR =
       "The count of labelValues must be equal to the underlying "
           + "MetricDescriptor's count of labels.";
 
+  /**
+   * A map of the {@link Counter} values, with a list of label values as the keys.
+   *
+   * <p>This should be modified in a critical section with {@code valueStartTimestamps} so that the
+   * values are in sync.
+   */
   private final AtomicLongMap<ImmutableList<String>> values = AtomicLongMap.create();
 
+  /**
+   * A map of the {@link Instant} that each value was created, with a list of label values as the
+   * keys. The start timestamp (as part of the {@link MetricPoint#interval()} can be used by
+   * implementations of {@link MetricWriter} to encode resets of monotonic counters.
+   */
+  private final ConcurrentHashMap<ImmutableList<String>, Instant> valueStartTimestamps =
+      new ConcurrentHashMap<>(
+          HASHMAP_INITIAL_CAPACITY, HASHMAP_LOAD_FACTOR, HASHMAP_CONCURRENCY_LEVEL);
+
+  /**
+   * A fine-grained lock to ensure that {@code values} and {@code valueStartTimestamps} are modified
+   * and read in a critical section. The initialization parameter is the concurrency level, set to
+   * match the default concurrency level of {@link ConcurrentHashMap}.
+   *
+   * @see Striped
+   */
+  private final Striped<Lock> valueLocks = Striped.lock(HASHMAP_CONCURRENCY_LEVEL);
+
   Counter(
       String name,
       String description,
@@ -49,8 +89,16 @@ public final class Counter extends AbstractMetric<Long>
   }
 
   @VisibleForTesting
-  void incrementBy(long offset, ImmutableList<String> labelValues) {
-    values.addAndGet(labelValues, offset);
+  void incrementBy(long offset, Instant startTime, ImmutableList<String> labelValues) {
+    Lock lock = valueLocks.get(labelValues);
+    lock.lock();
+
+    try {
+      values.addAndGet(labelValues, offset);
+      valueStartTimestamps.putIfAbsent(labelValues, startTime);
+    } finally {
+      lock.unlock();
+    }
   }
 
   @Override
@@ -58,14 +106,14 @@ public final class Counter extends AbstractMetric<Long>
     checkArgument(labelValues.length == this.getMetricSchema().labels().size(), LABEL_COUNT_ERROR);
     checkArgument(offset >= 0, "The offset provided must be non-negative");
 
-    incrementBy(offset, ImmutableList.copyOf(labelValues));
+    incrementBy(offset, Instant.now(), ImmutableList.copyOf(labelValues));
   }
 
   @Override
   public final void increment(String... labelValues) {
     checkArgument(labelValues.length == this.getMetricSchema().labels().size(), LABEL_COUNT_ERROR);
 
-    incrementBy(1L, ImmutableList.copyOf(labelValues));
+    incrementBy(1L, Instant.now(), ImmutableList.copyOf(labelValues));
   }
 
   /**
@@ -83,23 +131,87 @@ public final class Counter extends AbstractMetric<Long>
   }
 
   @VisibleForTesting
-  final ImmutableList<MetricPoint<Long>> getTimestampedValues(Instant timestamp) {
+  final ImmutableList<MetricPoint<Long>> getTimestampedValues(Instant endTimestamp) {
     ImmutableList.Builder<MetricPoint<Long>> timestampedValues = new ImmutableList.Builder<>();
     for (Entry<ImmutableList<String>, Long> entry : values.asMap().entrySet()) {
-      timestampedValues.add(MetricPoint.create(this, entry.getKey(), timestamp, entry.getValue()));
+      ImmutableList<String> labelValues = entry.getKey();
+      valueLocks.get(labelValues).lock();
+
+      Instant startTimestamp;
+      try {
+        startTimestamp = valueStartTimestamps.get(labelValues);
+      } finally {
+        valueLocks.get(labelValues).unlock();
+      }
+
+      timestampedValues.add(
+          MetricPoint.create(this, labelValues, startTimestamp, endTimestamp, entry.getValue()));
+
     }
     return timestampedValues.build();
   }
 
   @VisibleForTesting
-  final void set(Long value, ImmutableList<String> labelValues) {
-    this.values.put(labelValues, value);
+  final void set(Long value, Instant startTime, ImmutableList<String> labelValues) {
+    Lock lock = valueLocks.get(labelValues);
+    lock.lock();
+
+    try {
+      this.values.put(labelValues, value);
+      valueStartTimestamps.putIfAbsent(labelValues, startTime);
+    } finally {
+      lock.unlock();
+    }
   }
 
   @Override
   public final void set(Long value, String... labelValues) {
     checkArgument(labelValues.length == this.getMetricSchema().labels().size(), LABEL_COUNT_ERROR);
 
-    set(value, ImmutableList.copyOf(labelValues));
+    set(value, Instant.now(), ImmutableList.copyOf(labelValues));
+  }
+
+  @VisibleForTesting
+  final void reset(Instant startTime) {
+    // Lock the entire set of values so that all existing values will have a consistent timestamp
+    // after this call, without the possibility of interleaving with another reset() call.
+    Set<ImmutableList<String>> keys = values.asMap().keySet();
+    for (int i = 0; i < valueLocks.size(); i++) {
+      valueLocks.getAt(i).lock();
+    }
+
+    for (ImmutableList<String> labelValues : keys) {
+      this.values.put(labelValues, 0);
+      this.valueStartTimestamps.put(labelValues, startTime);
+    }
+
+    for (int i = 0; i < valueLocks.size(); i++) {
+      valueLocks.getAt(i).unlock();
+    }
+  }
+
+  @Override
+  public final void reset() {
+    reset(Instant.now());
+  }
+
+  @VisibleForTesting
+  final void reset(Instant startTime, ImmutableList<String> labelValues) {
+    Lock lock = valueLocks.get(labelValues);
+    lock.lock();
+
+    try {
+      this.values.put(labelValues, 0);
+      this.valueStartTimestamps.put(labelValues, startTime);
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  @Override
+  public final void reset(String... labelValues) {
+    checkArgument(labelValues.length == this.getMetricSchema().labels().size(), LABEL_COUNT_ERROR);
+
+    reset(Instant.now(), ImmutableList.copyOf(labelValues));
   }
 }