beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bchamb...@apache.org
Subject [1/5] incubator-beam git commit: Implement Metrics in the DirectRunner
Date Thu, 13 Oct 2016 22:39:34 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master e969f3d38 -> 3c731707b


Implement Metrics in the DirectRunner


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/834933c5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/834933c5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/834933c5

Branch: refs/heads/master
Commit: 834933c520997b4f83cf8b04219c2c63dac61e61
Parents: 51fee39
Author: bchambers <bchambers@google.com>
Authored: Wed Oct 12 10:55:53 2016 -0700
Committer: bchambers <bchambers@google.com>
Committed: Thu Oct 13 15:29:29 2016 -0700

----------------------------------------------------------------------
 .../beam/runners/direct/DirectMetrics.java      | 331 +++++++++++++++++++
 .../beam/runners/direct/DirectRunner.java       |   8 +-
 .../beam/runners/direct/EvaluationContext.java  |  10 +
 .../direct/ExecutorServiceParallelExecutor.java |   1 +
 .../direct/ImmutableListBundleFactory.java      |  10 +
 .../runners/direct/StepTransformResult.java     |  49 ++-
 .../beam/runners/direct/TransformExecutor.java  |  35 +-
 .../beam/runners/direct/TransformResult.java    |  12 +
 .../beam/runners/direct/DirectMetricsTest.java  | 133 ++++++++
 .../beam/runners/direct/DirectRunnerTest.java   |  36 ++
 .../runners/direct/TransformExecutorTest.java   |  12 +
 11 files changed, 602 insertions(+), 35 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
new file mode 100644
index 0000000..a749a76
--- /dev/null
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
@@ -0,0 +1,331 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.direct;
+
+import static java.util.Arrays.asList;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.base.Objects;
+import com.google.common.collect.ImmutableList;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.concurrent.GuardedBy;
+import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
+import org.apache.beam.sdk.metrics.DistributionData;
+import org.apache.beam.sdk.metrics.DistributionResult;
+import org.apache.beam.sdk.metrics.MetricKey;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricUpdates;
+import org.apache.beam.sdk.metrics.MetricUpdates.MetricUpdate;
+import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.apache.beam.sdk.metrics.MetricsMap;
+
+/**
+ * Implementation of {@link MetricResults} for the Direct Runner.
+ */
+class DirectMetrics extends MetricResults {
+
+  // TODO: (BEAM-723) Create a shared ExecutorService for maintenance tasks in the DirectRunner.
+  private static final ExecutorService COUNTER_COMMITTER = Executors.newCachedThreadPool();
+
+  private interface MetricAggregation<UpdateT, ResultT> {
+    UpdateT zero();
+    UpdateT combine(Iterable<UpdateT> updates);
+    ResultT extract(UpdateT data);
+  }
+
+  /**
+   * Implementation of a metric in the direct runner.
+   *
+   * @param <UpdateT> The type of raw data received and aggregated across updates.
+   * @param <ResultT> The type of result extracted from the data.
+   */
+  private static class DirectMetric<UpdateT, ResultT> {
+    private final MetricAggregation<UpdateT, ResultT> aggregation;
+
+    private final AtomicReference<UpdateT> finishedCommitted;
+
+    private final Object attemptedLock = new Object();
+    @GuardedBy("attemptedLock")
+    private volatile UpdateT finishedAttempted;
+    @GuardedBy("attemptedLock")
+    private final ConcurrentMap<CommittedBundle<?>, UpdateT> inflightAttempted
=
+        new ConcurrentHashMap<>();
+
+    public DirectMetric(MetricAggregation<UpdateT, ResultT> aggregation) {
+      this.aggregation = aggregation;
+      finishedCommitted = new AtomicReference<>(aggregation.zero());
+      finishedAttempted = aggregation.zero();
+    }
+
+    /**
+     * Add the given {@code tentativeCumulative} update to the physical aggregate.
+     *
+     * @param bundle The bundle receiving an update.
+     * @param tentativeCumulative The new cumulative value for the given bundle.
+     */
+    public void updatePhysical(CommittedBundle<?> bundle, UpdateT tentativeCumulative)
{
+      // Add (or update) the cumulatiev value for the given bundle.
+      inflightAttempted.put(bundle, tentativeCumulative);
+    }
+
+    /**
+     * Commit a physical value for the given {@code bundle}.
+     *
+     * @param bundle The bundle being committed.
+     * @param finalCumulative The final cumulative value for the given bundle.
+     */
+    public void commitPhysical(final CommittedBundle<?> bundle, final UpdateT finalCumulative)
{
+      // To prevent a query from blocking the commit, we perform the commit in two steps.
+      // 1. We perform a non-blocking write to the uncommitted table to make the new vaule
+      //    available immediately.
+      // 2. We submit a runnable that will commit the update and remove the tentative value
in
+      //    a synchronized block.
+      inflightAttempted.put(bundle, finalCumulative);
+      COUNTER_COMMITTER.submit(new Runnable() {
+        @Override
+        public void run() {
+          synchronized (attemptedLock) {
+            finishedAttempted = aggregation.combine(asList(finishedAttempted, finalCumulative));
+            inflightAttempted.remove(bundle);
+          }
+        }
+      });
+    }
+
+    /** Extract the latest values from all attempted and in-progress bundles. */
+    public ResultT extractLatestAttempted() {
+      ArrayList<UpdateT> updates = new ArrayList<>(inflightAttempted.size() +
1);
+      // Within this block we know that will be consistent. Specifically, the only change
that can
+      // happen concurrently is the addition of new (larger) values to inflightAttempted.
+      synchronized (attemptedLock) {
+        updates.add(finishedAttempted);
+        updates.addAll(inflightAttempted.values());
+      }
+      return aggregation.extract(aggregation.combine(updates));
+    }
+
+    /**
+     * Commit a logical value for the given {@code bundle}.
+     *
+     * @param bundle The bundle being committed.
+     * @param finalCumulative The final cumulative value for the given bundle.
+     */
+    public void commitLogical(final CommittedBundle<?> bundle, final UpdateT finalCumulative)
{
+      UpdateT current;
+      do {
+        current = finishedCommitted.get();
+      } while (!finishedCommitted.compareAndSet(current,
+          aggregation.combine(asList(current, finalCumulative))));
+    }
+
+    /** Extract the value from all successfully committed bundles. */
+    public ResultT extractCommitted() {
+      return aggregation.extract(finishedCommitted.get());
+    }
+  }
+
+  private static final MetricAggregation<Long, Long> COUNTER =
+      new MetricAggregation<Long, Long>() {
+    @Override
+    public Long zero() {
+      return 0L;
+    }
+
+    @Override
+    public Long combine(Iterable<Long> updates) {
+      long value = 0;
+      for (long update : updates) {
+        value += update;
+      }
+      return value;
+    }
+
+    @Override
+    public Long extract(Long data) {
+      return data;
+    }
+  };
+
+  private static final MetricAggregation<DistributionData, DistributionResult> DISTRIBUTION
=
+      new MetricAggregation<DistributionData, DistributionResult>() {
+        @Override
+        public DistributionData zero() {
+          return DistributionData.EMPTY;
+        }
+
+        @Override
+        public DistributionData combine(Iterable<DistributionData> updates) {
+          DistributionData result = DistributionData.EMPTY;
+          for (DistributionData update : updates) {
+            result = result.combine(update);
+          }
+          return result;
+        }
+
+        @Override
+        public DistributionResult extract(DistributionData data) {
+          return data.extractResult();
+        }
+      };
+
+  /** The current values of counters in memory. */
+  private MetricsMap<MetricKey, DirectMetric<Long, Long>> counters =
+      new MetricsMap<>(new MetricsMap.Factory<MetricKey, DirectMetric<Long, Long>>()
{
+        @Override
+        public DirectMetric<Long, Long> createInstance(MetricKey unusedKey) {
+          return new DirectMetric<>(COUNTER);
+        }
+      });
+  private MetricsMap<MetricKey, DirectMetric<DistributionData, DistributionResult>>
distributions =
+      new MetricsMap<>(
+          new MetricsMap.Factory<MetricKey, DirectMetric<DistributionData, DistributionResult>>()
{
+        @Override
+        public DirectMetric<DistributionData, DistributionResult> createInstance(
+            MetricKey unusedKey) {
+          return new DirectMetric<>(DISTRIBUTION);
+        }
+      });
+
+  @AutoValue
+  abstract static class DirectMetricQueryResults implements MetricQueryResults {
+    public static MetricQueryResults create(
+        Iterable<MetricResult<Long>> counters,
+        Iterable<MetricResult<DistributionResult>> distributions) {
+      return new AutoValue_DirectMetrics_DirectMetricQueryResults(counters, distributions);
+    }
+  }
+
+  @AutoValue
+  abstract static class DirectMetricResult<T> implements MetricResult<T> {
+    public static <T> MetricResult<T> create(MetricName name, String scope,
+        T committed, T attempted) {
+      return new AutoValue_DirectMetrics_DirectMetricResult<T>(
+          name, scope, committed, attempted);
+    }
+  }
+
+  @Override
+  public MetricQueryResults queryMetrics(MetricsFilter filter) {
+    ImmutableList.Builder<MetricResult<Long>> counterResults = ImmutableList.builder();
+    for (Entry<MetricKey, DirectMetric<Long, Long>> counter : counters.entries())
{
+      maybeExtractResult(filter, counterResults, counter);
+    }
+    ImmutableList.Builder<MetricResult<DistributionResult>> distributionResults
=
+        ImmutableList.builder();
+    for (Entry<MetricKey, DirectMetric<DistributionData, DistributionResult>>
distribution
+        : distributions.entries()) {
+      maybeExtractResult(filter, distributionResults, distribution);
+    }
+
+    return DirectMetricQueryResults.create(counterResults.build(), distributionResults.build());
+  }
+
+  private <ResultT> void maybeExtractResult(
+      MetricsFilter filter,
+      ImmutableList.Builder<MetricResult<ResultT>> resultsBuilder,
+      Map.Entry<MetricKey, ? extends DirectMetric<?, ResultT>> entry) {
+    if (matches(filter, entry.getKey())) {
+      resultsBuilder.add(DirectMetricResult.create(
+          entry.getKey().metricName(),
+          entry.getKey().stepName(),
+          entry.getValue().extractCommitted(),
+          entry.getValue().extractLatestAttempted()));
+    }
+  }
+
+  // Matching logic is implemented here rather than in MetricsFilter because we would like
+  // MetricsFilter to act as a "dumb" value-object, with the possibility of replacing it
with
+  // a Proto/JSON/etc. schema object.
+  private boolean matches(MetricsFilter filter, MetricKey key) {
+    return matchesName(key.metricName(), filter.names())
+        && matchesScope(key.stepName(), filter.steps());
+  }
+
+  private boolean matchesScope(String actualScope, Set<String> scopes) {
+    if (scopes.isEmpty() || scopes.contains(actualScope)) {
+      return true;
+    }
+
+    for (String scope : scopes) {
+      if (actualScope.startsWith(scope)) {
+        return true;
+      }
+    }
+
+    return false;
+  }
+
+  private boolean matchesName(MetricName metricName, Set<MetricNameFilter> nameFilters)
{
+    if (nameFilters.isEmpty()) {
+      return true;
+    }
+
+    for (MetricNameFilter nameFilter : nameFilters) {
+      if ((nameFilter.getName() == null || nameFilter.getName().equals(metricName.name()))
+          && Objects.equal(metricName.namespace(), nameFilter.getNamespace())) {
+        return true;
+      }
+    }
+
+    return false;
+  }
+
+  /** Apply metric updates that represent physical counter deltas to the current metric values.
*/
+  public void updatePhysical(CommittedBundle<?> bundle, MetricUpdates updates) {
+    for (MetricUpdate<Long> counter : updates.counterUpdates()) {
+      counters.get(counter.getKey()).updatePhysical(bundle, counter.getUpdate());
+    }
+    for (MetricUpdate<DistributionData> distribution : updates.distributionUpdates())
{
+      distributions.get(distribution.getKey())
+          .updatePhysical(bundle, distribution.getUpdate());
+    }
+  }
+
+  public void commitPhysical(CommittedBundle<?> bundle, MetricUpdates updates) {
+    for (MetricUpdate<Long> counter : updates.counterUpdates()) {
+      counters.get(counter.getKey()).commitPhysical(bundle, counter.getUpdate());
+    }
+    for (MetricUpdate<DistributionData> distribution : updates.distributionUpdates())
{
+      distributions.get(distribution.getKey())
+          .commitPhysical(bundle, distribution.getUpdate());
+    }
+  }
+
+  /** Apply metric updates that represent new logical values from a bundle being committed.
*/
+  public void commitLogical(CommittedBundle<?> bundle, MetricUpdates updates) {
+    for (MetricUpdate<Long> counter : updates.counterUpdates()) {
+      counters.get(counter.getKey()).commitLogical(bundle, counter.getUpdate());
+    }
+    for (MetricUpdate<DistributionData> distribution : updates.distributionUpdates())
{
+      distributions.get(distribution.getKey())
+          .commitLogical(bundle, distribution.getUpdate());
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
index e13046d..8941093 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
@@ -43,6 +43,7 @@ import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.io.Read;
 import org.apache.beam.sdk.io.Write;
 import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.runners.PipelineRunner;
 import org.apache.beam.sdk.testing.TestStream;
@@ -226,6 +227,7 @@ public class DirectRunner
 
   @Override
   public DirectPipelineResult run(Pipeline pipeline) {
+    MetricsEnvironment.setMetricsSupported(true);
     ConsumerTrackingPipelineVisitor consumerTrackingVisitor = new ConsumerTrackingPipelineVisitor();
     pipeline.traverseTopologically(consumerTrackingVisitor);
     for (PValue unfinalized : consumerTrackingVisitor.getUnfinalizedPValues()) {
@@ -268,8 +270,7 @@ public class DirectRunner
 
     Map<Aggregator<?, ?>, Collection<PTransform<?, ?>>> aggregatorSteps
=
         pipeline.getAggregatorSteps();
-    DirectPipelineResult result =
-        new DirectPipelineResult(executor, context, aggregatorSteps);
+    DirectPipelineResult result = new DirectPipelineResult(executor, context, aggregatorSteps);
     if (options.isBlockOnRun()) {
       try {
         result.awaitCompletion();
@@ -383,8 +384,7 @@ public class DirectRunner
 
     @Override
     public MetricResults metrics() {
-      throw new UnsupportedOperationException(
-          "The DirectRunner does not currently support metrics.");
+      return evaluationContext.getMetrics();
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
index 2901254..e5a30d4 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
@@ -96,6 +96,8 @@ class EvaluationContext {
 
   private final AggregatorContainer mergedAggregators;
 
+  private final DirectMetrics metrics;
+
   public static EvaluationContext create(
       DirectOptions options,
       Clock clock,
@@ -130,6 +132,7 @@ class EvaluationContext {
 
     this.applicationStateInternals = new ConcurrentHashMap<>();
     this.mergedAggregators = AggregatorContainer.create();
+    this.metrics = new DirectMetrics();
 
     this.callbackExecutor =
         WatermarkCallbackExecutor.create(MoreExecutors.directExecutor());
@@ -161,6 +164,8 @@ class EvaluationContext {
       TransformResult result) {
     Iterable<? extends CommittedBundle<?>> committedBundles =
         commitBundles(result.getOutputBundles());
+    metrics.commitLogical(completedBundle, result.getLogicalMetricUpdates());
+
     // Update watermarks and timers
     EnumSet<OutputType> outputTypes = EnumSet.copyOf(result.getOutputTypes());
     if (Iterables.isEmpty(committedBundles)) {
@@ -367,6 +372,11 @@ class EvaluationContext {
     return mergedAggregators;
   }
 
+  /** Returns the metrics container for this pipeline. */
+  public DirectMetrics getMetrics() {
+    return metrics;
+  }
+
   @VisibleForTesting
   void forceRefresh() {
     watermarkManager.refreshAll();

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
index fab6a33..3761574 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
@@ -212,6 +212,7 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
 
     TransformExecutor<T> callable =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             enforcements,
             bundle,

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
index 4972340..db92542 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
@@ -123,5 +123,15 @@ class ImmutableListBundleFactory implements BundleFactory {
           ImmutableList.copyOf(elements),
           getSynchronizedProcessingOutputWatermark());
     }
+
+    @Override
+    public int hashCode() {
+      return System.identityHashCode(this);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return this == obj;
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java
index 1829e4a..989109f 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java
@@ -22,10 +22,10 @@ import com.google.common.collect.ImmutableList;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.Set;
-import javax.annotation.Nullable;
 import org.apache.beam.runners.direct.CommittedResult.OutputType;
 import org.apache.beam.runners.direct.DirectRunner.UncommittedBundle;
 import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.sdk.metrics.MetricUpdates;
 import org.apache.beam.sdk.transforms.AppliedPTransform;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.WindowedValue;
@@ -37,31 +37,6 @@ import org.joda.time.Instant;
  */
 @AutoValue
 public abstract class StepTransformResult implements TransformResult {
-  @Override
-  public abstract AppliedPTransform<?, ?, ?> getTransform();
-
-  @Override
-  public abstract Iterable<? extends UncommittedBundle<?>> getOutputBundles();
-
-  @Override
-  public abstract Iterable<? extends WindowedValue<?>> getUnprocessedElements();
-
-  @Override
-  @Nullable
-  public abstract AggregatorContainer.Mutator getAggregatorChanges();
-
-  @Override
-  public abstract Instant getWatermarkHold();
-
-  @Nullable
-  @Override
-  public abstract CopyOnAccessInMemoryStateInternals<?> getState();
-
-  @Override
-  public abstract TimerUpdate getTimerUpdate();
-
-  @Override
-  public abstract Set<OutputType> getOutputTypes();
 
   public static Builder withHold(AppliedPTransform<?, ?, ?> transform, Instant watermarkHold)
{
     return new Builder(transform, watermarkHold);
@@ -71,6 +46,20 @@ public abstract class StepTransformResult implements TransformResult {
     return new Builder(transform, BoundedWindow.TIMESTAMP_MAX_VALUE);
   }
 
+  @Override
+  public TransformResult withLogicalMetricUpdates(MetricUpdates metricUpdates) {
+    return new AutoValue_StepTransformResult(
+        getTransform(),
+        getOutputBundles(),
+        getUnprocessedElements(),
+        getAggregatorChanges(),
+        metricUpdates,
+        getWatermarkHold(),
+        getState(),
+        getTimerUpdate(),
+        getOutputTypes());
+  }
+
   /**
    * A builder for creating instances of {@link StepTransformResult}.
    */
@@ -78,6 +67,7 @@ public abstract class StepTransformResult implements TransformResult {
     private final AppliedPTransform<?, ?, ?> transform;
     private final ImmutableList.Builder<UncommittedBundle<?>> bundlesBuilder;
     private final ImmutableList.Builder<WindowedValue<?>> unprocessedElementsBuilder;
+    private MetricUpdates metricUpdates;
     private CopyOnAccessInMemoryStateInternals<?> state;
     private TimerUpdate timerUpdate;
     private AggregatorContainer.Mutator aggregatorChanges;
@@ -91,6 +81,7 @@ public abstract class StepTransformResult implements TransformResult {
       this.producedOutputs = EnumSet.noneOf(OutputType.class);
       this.unprocessedElementsBuilder = ImmutableList.builder();
       this.timerUpdate = TimerUpdate.builder(null).build();
+      this.metricUpdates = MetricUpdates.EMPTY;
     }
 
     public StepTransformResult build() {
@@ -99,6 +90,7 @@ public abstract class StepTransformResult implements TransformResult {
           bundlesBuilder.build(),
           unprocessedElementsBuilder.build(),
           aggregatorChanges,
+          metricUpdates,
           watermarkHold,
           state,
           timerUpdate,
@@ -110,6 +102,11 @@ public abstract class StepTransformResult implements TransformResult
{
       return this;
     }
 
+    public Builder withMetricUpdates(MetricUpdates metricUpdates) {
+      this.metricUpdates = metricUpdates;
+      return this;
+    }
+
     public Builder withState(CopyOnAccessInMemoryStateInternals<?> state) {
       this.state = state;
       return this;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
index aaee9a5..03f615b 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
@@ -25,6 +25,9 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.annotation.Nullable;
 import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
+import org.apache.beam.sdk.metrics.MetricUpdates;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
 import org.apache.beam.sdk.transforms.AppliedPTransform;
 import org.apache.beam.sdk.util.WindowedValue;
 
@@ -38,6 +41,7 @@ import org.apache.beam.sdk.util.WindowedValue;
  */
 class TransformExecutor<T> implements Runnable {
   public static <T> TransformExecutor<T> create(
+      EvaluationContext context,
       TransformEvaluatorFactory factory,
       Iterable<? extends ModelEnforcementFactory> modelEnforcements,
       CommittedBundle<T> inputBundle,
@@ -45,6 +49,7 @@ class TransformExecutor<T> implements Runnable {
       CompletionCallback completionCallback,
       TransformExecutorService transformEvaluationState) {
     return new TransformExecutor<>(
+        context,
         factory,
         modelEnforcements,
         inputBundle,
@@ -63,10 +68,12 @@ class TransformExecutor<T> implements Runnable {
 
   private final CompletionCallback onComplete;
   private final TransformExecutorService transformEvaluationState;
+  private final EvaluationContext context;
 
   private final AtomicReference<Thread> thread;
 
   private TransformExecutor(
+      EvaluationContext context,
       TransformEvaluatorFactory factory,
       Iterable<? extends ModelEnforcementFactory> modelEnforcements,
       CommittedBundle<T> inputBundle,
@@ -82,11 +89,14 @@ class TransformExecutor<T> implements Runnable {
     this.onComplete = completionCallback;
 
     this.transformEvaluationState = transformEvaluationState;
+    this.context = context;
     this.thread = new AtomicReference<>();
   }
 
   @Override
   public void run() {
+    MetricsContainer metricsContainer = new MetricsContainer(transform.getFullName());
+    MetricsEnvironment.setMetricsContainer(metricsContainer);
     checkState(
         thread.compareAndSet(null, Thread.currentThread()),
         "Tried to execute %s for %s on thread %s, but is already executing on thread %s",
@@ -108,9 +118,9 @@ class TransformExecutor<T> implements Runnable {
         return;
       }
 
-      processElements(evaluator, enforcements);
+      processElements(evaluator, metricsContainer, enforcements);
 
-      finishBundle(evaluator, enforcements);
+      finishBundle(evaluator, metricsContainer, enforcements);
     } catch (Throwable t) {
       onComplete.handleThrowable(inputBundle, t);
       if (t instanceof RuntimeException) {
@@ -118,6 +128,10 @@ class TransformExecutor<T> implements Runnable {
       }
       throw new RuntimeException(t);
     } finally {
+      // Report the physical metrics from the end of this step.
+      context.getMetrics().commitPhysical(inputBundle, metricsContainer.getCumulative());
+
+      MetricsEnvironment.unsetMetricsContainer();
       transformEvaluationState.complete(this);
     }
   }
@@ -127,7 +141,9 @@ class TransformExecutor<T> implements Runnable {
    * necessary {@link ModelEnforcement ModelEnforcements}.
    */
   private void processElements(
-      TransformEvaluator<T> evaluator, Collection<ModelEnforcement<T>>
enforcements)
+      TransformEvaluator<T> evaluator,
+      MetricsContainer metricsContainer,
+      Collection<ModelEnforcement<T>> enforcements)
       throws Exception {
     if (inputBundle != null) {
       for (WindowedValue<T> value : inputBundle.getElements()) {
@@ -137,6 +153,13 @@ class TransformExecutor<T> implements Runnable {
 
         evaluator.processElement(value);
 
+        // Report the physical metrics after each element
+        MetricUpdates deltas = metricsContainer.getUpdates();
+        if (deltas != null) {
+          context.getMetrics().updatePhysical(inputBundle, deltas);
+          metricsContainer.commitUpdates();
+        }
+
         for (ModelEnforcement<T> enforcement : enforcements) {
           enforcement.afterElement(value);
         }
@@ -152,9 +175,11 @@ class TransformExecutor<T> implements Runnable {
    *         {@link TransformEvaluator#finishBundle()}
    */
   private TransformResult finishBundle(
-      TransformEvaluator<T> evaluator, Collection<ModelEnforcement<T>>
enforcements)
+      TransformEvaluator<T> evaluator, MetricsContainer metricsContainer,
+      Collection<ModelEnforcement<T>> enforcements)
       throws Exception {
-    TransformResult result = evaluator.finishBundle();
+    TransformResult result = evaluator.finishBundle()
+        .withLogicalMetricUpdates(metricsContainer.getCumulative());
     CommittedResult outputs = onComplete.handleResult(inputBundle, result);
     for (ModelEnforcement<T> enforcement : enforcements) {
       enforcement.afterFinish(inputBundle, result, outputs.getOutputs());

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java
index ba2d48e..ac1e395 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java
@@ -22,6 +22,7 @@ import javax.annotation.Nullable;
 import org.apache.beam.runners.direct.CommittedResult.OutputType;
 import org.apache.beam.runners.direct.DirectRunner.UncommittedBundle;
 import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.sdk.metrics.MetricUpdates;
 import org.apache.beam.sdk.transforms.AppliedPTransform;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -57,6 +58,11 @@ public interface TransformResult {
   @Nullable AggregatorContainer.Mutator getAggregatorChanges();
 
   /**
+   * Returns the logical metric updates.
+   */
+  MetricUpdates getLogicalMetricUpdates();
+
+  /**
    * Returns the Watermark Hold for the transform at the time this result was produced.
    *
    * <p>If the transform does not set any watermark hold, returns
@@ -86,4 +92,10 @@ public interface TransformResult {
    * {@link OutputType#BUNDLE}, as empty bundles may be dropped when the transform is committed.
    */
   Set<OutputType> getOutputTypes();
+
+  /**
+   * Returns a new TransformResult based on this one but overwriting any existing logical
metric
+   * updates with {@code metricUpdates}.
+   */
+  TransformResult withLogicalMetricUpdates(MetricUpdates metricUpdates);
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
new file mode 100644
index 0000000..df01244
--- /dev/null
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.
+ */
+
+package org.apache.beam.runners.direct;
+
+import static org.apache.beam.sdk.metrics.MetricMatchers.metricResult;
+import static org.apache.beam.sdk.metrics.MetricNameFilter.inNamespace;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.junit.Assert.assertThat;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
+import org.apache.beam.sdk.metrics.DistributionData;
+import org.apache.beam.sdk.metrics.DistributionResult;
+import org.apache.beam.sdk.metrics.MetricKey;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricUpdates;
+import org.apache.beam.sdk.metrics.MetricUpdates.MetricUpdate;
+import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/**
+ * Tests for {@link DirectMetrics}.
+ */
+@RunWith(JUnit4.class)
+public class DirectMetricsTest {
+
+  @Mock
+  private CommittedBundle<Object> bundle1;
+  @Mock
+  private CommittedBundle<Object> bundle2;
+
+  private static final MetricName NAME1 = MetricName.named("ns1", "name1");
+  private static final MetricName NAME2 = MetricName.named("ns1", "name2");
+  private static final MetricName NAME3 = MetricName.named("ns2", "name1");
+
+  private DirectMetrics metrics = new DirectMetrics();
+
+  @Before
+  public void setUp() {
+    MockitoAnnotations.initMocks(this);
+  }
+
+  @Test
+  public void testApplyLogicalQueryNoFilter() {
+    metrics.commitLogical(bundle1, MetricUpdates.create(
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("step1", NAME1), 5L),
+            MetricUpdate.create(MetricKey.create("step1", NAME2), 8L)),
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("step1", NAME1),
+                DistributionData.create(8, 2, 3, 5)))));
+    metrics.commitLogical(bundle1, MetricUpdates.create(
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("step2", NAME1), 7L),
+            MetricUpdate.create(MetricKey.create("step1", NAME2), 4L)),
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("step1", NAME1),
+                DistributionData.create(4, 1, 4, 4)))));
+
+    MetricQueryResults results = metrics.queryMetrics(MetricsFilter.builder().build());
+    assertThat(results.counters(), containsInAnyOrder(
+        metricResult("ns1", "name1", "step1", 5L, 0L),
+        metricResult("ns1", "name2", "step1", 12L, 0L),
+        metricResult("ns1", "name1", "step2", 7L, 0L)));
+    assertThat(results.distributions(), contains(
+        metricResult("ns1", "name1", "step1",
+            DistributionResult.create(12, 3, 3, 5),
+            DistributionResult.ZERO)));
+  }
+
+  @Test
+  public void testApplyPhysicalCountersQueryOneNamespace() {
+    metrics.updatePhysical(bundle1, MetricUpdates.create(
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("step1", NAME1), 5L),
+            MetricUpdate.create(MetricKey.create("step1", NAME3), 8L)),
+        ImmutableList.<MetricUpdate<DistributionData>>of()));
+    metrics.updatePhysical(bundle1, MetricUpdates.create(
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("step2", NAME1), 7L),
+            MetricUpdate.create(MetricKey.create("step1", NAME3), 4L)),
+        ImmutableList.<MetricUpdate<DistributionData>>of()));
+
+    assertThat(metrics.queryMetrics(
+        MetricsFilter.builder().addNameFilter(inNamespace("ns1")).build()).counters(),
+        containsInAnyOrder(
+            metricResult("ns1", "name1", "step1", 0L, 5L),
+            metricResult("ns1", "name1", "step2", 0L, 7L)));
+  }
+
+  @Test
+  public void testApplyPhysicalQueryCompositeScope() {
+    metrics.updatePhysical(bundle1, MetricUpdates.create(
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("Outer1/Inner1", NAME1), 5L),
+            MetricUpdate.create(MetricKey.create("Outer1/Inner2", NAME1), 8L)),
+        ImmutableList.<MetricUpdate<DistributionData>>of()));
+    metrics.updatePhysical(bundle1, MetricUpdates.create(
+        ImmutableList.of(
+            MetricUpdate.create(MetricKey.create("Outer1/Inner1", NAME1), 12L),
+            MetricUpdate.create(MetricKey.create("Outer2/Inner2", NAME1), 18L)),
+        ImmutableList.<MetricUpdate<DistributionData>>of()));
+
+    assertThat(metrics.queryMetrics(
+        MetricsFilter.builder().addStep("Outer1").build()).counters(),
+        containsInAnyOrder(
+            metricResult("ns1", "name1", "Outer1/Inner1", 0L, 12L),
+            metricResult("ns1", "name1", "Outer1/Inner2", 0L, 8L)));
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index 4768fb0..d93dd7a 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.runners.direct;
 
+import static org.apache.beam.sdk.metrics.MetricMatchers.metricResult;
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.isA;
@@ -35,12 +37,20 @@ import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.beam.runners.direct.DirectRunner.DirectPipelineResult;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.coders.AtomicCoder;
 import org.apache.beam.sdk.coders.CoderException;
 import org.apache.beam.sdk.coders.ListCoder;
 import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.beam.sdk.coders.VarLongCoder;
 import org.apache.beam.sdk.io.CountingInput;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.DistributionResult;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.metrics.MetricsFilter;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.runners.PipelineRunner;
@@ -444,4 +454,30 @@ public class DirectRunnerTest implements Serializable {
       throw new CoderException("Cannot decode a long");
     }
   }
+
+  public void testMetrics() throws Exception {
+    Pipeline pipeline = getPipeline();
+    pipeline
+        .apply(Create.of(5, 8, 13))
+        .apply("MyStep", ParDo.of(new DoFn<Integer, Void>() {
+          @ProcessElement
+          public void processElement(ProcessContext c) {
+            Counter count = Metrics.counter(DirectRunnerTest.class, "count");
+            Distribution values = Metrics.distribution(DirectRunnerTest.class, "input");
+
+            count.inc();
+            values.update(c.element());
+          }
+        }));
+    PipelineResult result = pipeline.run();
+    MetricQueryResults metrics = result.metrics().queryMetrics(MetricsFilter.builder()
+        .addNameFilter(MetricNameFilter.inNamespace(DirectRunnerTest.class))
+        .build());
+    assertThat(metrics.counters(), contains(
+        metricResult(DirectRunnerTest.class.getName(), "count", "MyStep", 3L, 3L)));
+    assertThat(metrics.distributions(), contains(
+        metricResult(DirectRunnerTest.class.getName(), "input", "MyStep",
+            DistributionResult.create(26L, 3L, 5L, 13L),
+            DistributionResult.create(26L, 3L, 5L, 13L))));
+  }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/834933c5/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
index c63e9bd..5015e5a 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
@@ -72,6 +72,7 @@ public class TransformExecutorTest {
   private RegisteringCompletionCallback completionCallback;
   private TransformExecutorService transformEvaluationState;
   private BundleFactory bundleFactory;
+  @Mock private DirectMetrics metrics;
   @Mock private EvaluationContext evaluationContext;
   @Mock private TransformEvaluatorRegistry registry;
 
@@ -90,6 +91,8 @@ public class TransformExecutorTest {
     TestPipeline p = TestPipeline.create();
     created = p.apply(Create.of("foo", "spam", "third"));
     downstream = created.apply(WithKeys.<Integer, String>of(3));
+
+    when(evaluationContext.getMetrics()).thenReturn(metrics);
   }
 
   @Test
@@ -116,6 +119,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<Object> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>emptyList(),
             null,
@@ -135,6 +139,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<Object> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>emptyList(),
             null,
@@ -177,6 +182,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<String> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>emptyList(),
             inputBundle,
@@ -219,6 +225,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<String> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>emptyList(),
             inputBundle,
@@ -254,6 +261,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<String> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>emptyList(),
             inputBundle,
@@ -294,6 +302,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<String> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>emptyList(),
             null,
@@ -335,6 +344,7 @@ public class TransformExecutorTest {
     TestEnforcementFactory enforcement = new TestEnforcementFactory();
     TransformExecutor<String> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>singleton(enforcement),
             inputBundle,
@@ -392,6 +402,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<byte[]> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>singleton(ImmutabilityEnforcementFactory.create()),
             inputBundle,
@@ -448,6 +459,7 @@ public class TransformExecutorTest {
 
     TransformExecutor<byte[]> executor =
         TransformExecutor.create(
+            evaluationContext,
             registry,
             Collections.<ModelEnforcementFactory>singleton(ImmutabilityEnforcementFactory.create()),
             inputBundle,



Mime
View raw message