This closes #3053

Author: dhalperi

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricFiltering.java renamed to runners/core-construction-java/src/main/java/org/apache/beam/runners/core/metrics/MetricFiltering.java

@@ -16,10 +16,13 @@
  * limitations under the License.
  */
 
-package org.apache.beam.sdk.metrics;
+package org.apache.beam.runners.core.metrics;
 
 import com.google.common.base.Objects;
 import java.util.Set;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricsFilter;
 
 /**
  * Implements matching for metrics filters. Specifically, matching for metric name,

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricKey.java renamed to runners/core-construction-java/src/main/java/org/apache/beam/runners/core/metrics/MetricKey.java

@@ -15,12 +15,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.metrics;
+
+package org.apache.beam.runners.core.metrics;
 
 import com.google.auto.value.AutoValue;
 import java.io.Serializable;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.metrics.MetricName;
 
 /**
  * Metrics are keyed by the step name they are associated with and the name of the metric.

runners/core-construction-java/src/main/java/org/apache/beam/runners/core/metrics/package-info.java

@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Utilities for runners to implement metrics.
+ */
+package org.apache.beam.runners.core.metrics;

sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricFilteringTest.java renamed to runners/core-construction-java/src/test/java/org/apache/beam/runners/core/metrics/MetricFilteringTest.java

@@ -16,13 +16,16 @@
  * limitations under the License.
  */
 
-package org.apache.beam.sdk.metrics;
+package org.apache.beam.runners.core.metrics;
 
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 import java.util.HashSet;
 import java.util.Set;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricsFilter;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;

runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataUtils.java

@@ -22,7 +22,7 @@
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.Iterables;
 import javax.annotation.Nullable;
-import org.apache.beam.sdk.metrics.CounterCell;
+import org.apache.beam.runners.core.metrics.CounterCell;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.WindowTracing;
 import org.apache.beam.sdk.util.WindowedValue;
@@ -71,7 +71,7 @@ public boolean apply(@Nullable WindowedValue<V> input) {
                         .isBefore(timerInternals.currentInputWatermarkTime());
                 if (expired) {
                   // The element is too late for this window.
-                  droppedDueToLateness.update(1L);
+                  droppedDueToLateness.inc();
                   WindowTracing.debug(
                       "GroupAlsoByWindow: Dropping element at {} for key: {}; "
                           + "window: {} since it is too far behind inputWatermark: {}",

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/CounterCell.java renamed to runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/CounterCell.java

@@ -15,11 +15,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.metrics;
+
+package org.apache.beam.runners.core.metrics;
 
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.MetricName;
 
 /**
  * Tracks the current value (and delta) for a Counter metric for a specific context and bundle.
@@ -30,34 +33,40 @@
  * indirection.
  */
 @Experimental(Kind.METRICS)
-public class CounterCell implements MetricCell<Counter, Long> {
+public class CounterCell implements Counter, MetricCell<Long> {
 
   private final DirtyState dirty = new DirtyState();
   private final AtomicLong value = new AtomicLong();
+  private final MetricName name;
 
   /**
    * Package-visibility because all {@link CounterCell CounterCells} should be created by
-   * {@link MetricsContainer#getCounter(MetricName)}.
+   * {@link MetricsContainerImpl#getCounter(MetricName)}.
    */
-  CounterCell() {}
+  CounterCell(MetricName name) {
+    this.name = name;
+  }
 
   /**
    * Increment the counter by the given amount.
    * @param n value to increment by. Can be negative to decrement.
    */
-  public void update(long n) {
+  @Override
+  public void inc(long n) {
     value.addAndGet(n);
     dirty.afterModification();
   }
 
-  @Override
-  public void update(Long n) {
-    throw new UnsupportedOperationException("CounterCell.update(Long n) should not be used"
-    + " as it performs unnecessary boxing/unboxing. Use CounterCell.update(long n) instead.");
+  public void inc() {
+    inc(1);
   }
 
-  @Override public void update(MetricCell<Counter, Long> other) {
-    update((long) other.getCumulative());
+  public void dec() {
+    inc(-1);
+  }
+
+  public void dec(long n) {
+    inc(-1 * n);
   }
 
   @Override
@@ -69,4 +78,9 @@ public DirtyState getDirty() {
   public Long getCumulative() {
     return value.get();
   }
+
+  @Override
+  public MetricName getName() {
+    return name;
+  }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DirtyState.java renamed to runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/DirtyState.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.beam.sdk.metrics;
+package org.apache.beam.runners.core.metrics;
 
 import java.io.Serializable;
 import java.util.concurrent.atomic.AtomicReference;

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionCell.java renamed to runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/DistributionCell.java

@@ -15,11 +15,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.metrics;
+
+package org.apache.beam.runners.core.metrics;
 
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.MetricName;
 
 /**
  * Tracks the current value (and delta) for a Distribution metric.
@@ -30,37 +33,35 @@
  * of indirection.
  */
 @Experimental(Kind.METRICS)
-public class DistributionCell implements MetricCell<Distribution, DistributionData> {
+public class DistributionCell implements Distribution, MetricCell<DistributionData> {
 
   private final DirtyState dirty = new DirtyState();
   private final AtomicReference<DistributionData> value =
       new AtomicReference<>(DistributionData.EMPTY);
+  private final MetricName name;
 
   /**
    * Package-visibility because all {@link DistributionCell DistributionCells} should be created by
-   * {@link MetricsContainer#getDistribution(MetricName)}.
+   * {@link MetricsContainerImpl#getDistribution(MetricName)}.
    */
-  DistributionCell() {}
+  DistributionCell(MetricName name) {
+    this.name = name;
+  }
 
   /** Increment the distribution by the given amount. */
+  @Override
   public void update(long n) {
     update(DistributionData.singleton(n));
   }
 
-  @Override
-  public void update(DistributionData data) {
+  void update(DistributionData data) {
     DistributionData original;
     do {
       original = value.get();
     } while (!value.compareAndSet(original, original.combine(data)));
     dirty.afterModification();
   }
 
-  @Override
-  public void update(MetricCell<Distribution, DistributionData> other) {
-    update(other.getCumulative());
-  }
-
   @Override
   public DirtyState getDirty() {
     return dirty;
@@ -70,5 +71,10 @@ public DirtyState getDirty() {
   public DistributionData getCumulative() {
     return value.get();
   }
+
+  @Override
+  public MetricName getName() {
+    return name;
+  }
 }
 

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionData.java renamed to runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/DistributionData.java

@@ -15,10 +15,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.metrics;
+
+package org.apache.beam.runners.core.metrics;
 
 import com.google.auto.value.AutoValue;
 import java.io.Serializable;
+import org.apache.beam.sdk.metrics.DistributionResult;
 
 /**
  * Data describing the the distribution. This should retain enough detail that it can be combined

sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeCell.java renamed to runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/GaugeCell.java

@@ -15,10 +15,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.metrics;
+
+package org.apache.beam.runners.core.metrics;
 
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.metrics.Gauge;
+import org.apache.beam.sdk.metrics.MetricName;
 
 /**
  * Tracks the current value (and delta) for a {@link Gauge} metric.
@@ -29,35 +32,34 @@
  * of indirection.
  */
 @Experimental(Experimental.Kind.METRICS)
-public class GaugeCell implements MetricCell<Gauge, GaugeData> {
+public class GaugeCell implements Gauge, MetricCell<GaugeData> {
 
   private final DirtyState dirty = new DirtyState();
   private final AtomicReference<GaugeData> gaugeValue = new AtomicReference<>(GaugeData.empty());
+  private final MetricName name;
+
+  /**
+   * Package-visibility because all {@link GaugeCell GaugeCells} should be created by
+   * {@link MetricsContainerImpl#getGauge(MetricName)}.
+   */
+  GaugeCell(MetricName name) {
+    this.name = name;
+  }
 
   /** Set the gauge to the given value. */
+  @Override
   public void set(long value) {
     update(GaugeData.create(value));
   }
 
-  @Override
-  public void update(GaugeData data) {
+  void update(GaugeData data) {
     GaugeData original;
     do {
       original = gaugeValue.get();
     } while (!gaugeValue.compareAndSet(original, original.combine(data)));
     dirty.afterModification();
   }
 
-  @Override
-  public void update(MetricCell<Gauge, GaugeData> other) {
-    GaugeData original;
-    do {
-      original = gaugeValue.get();
-    } while (!gaugeValue.compareAndSet(original, original.combine(other.getCumulative())));
-    dirty.afterModification();
-    update(other.getCumulative());
-  }
-
   @Override
   public DirtyState getDirty() {
     return dirty;
@@ -67,4 +69,10 @@ public DirtyState getDirty() {
   public GaugeData getCumulative() {
     return gaugeValue.get();
   }
+
+
+  @Override
+  public MetricName getName() {
+    return name;
+  }
 }