You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by th...@apache.org on 2019/01/28 03:42:08 UTC
[beam] branch master updated: [BEAM-4776] AutoValues for
MetricResults, MetricQueryResults (#7621)
This is an automated email from the ASF dual-hosted git repository.
thw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 11baa1e [BEAM-4776] AutoValues for MetricResults, MetricQueryResults (#7621)
11baa1e is described below
commit 11baa1ea69d37935be147e89c8ff1b3ddac7731c
Author: Ryan Williams <ry...@gmail.com>
AuthorDate: Sun Jan 27 22:41:58 2019 -0500
[BEAM-4776] AutoValues for MetricResults, MetricQueryResults (#7621)
---
.../core/metrics/MetricsContainerStepMap.java | 2 +-
.../apache/beam/runners/direct/DirectMetrics.java | 38 ++--------------------
.../runners/direct/portable/DirectMetrics.java | 38 ++--------------------
.../metrics/CustomMetricQueryResults.java | 2 +-
.../beam/runners/dataflow/DataflowMetrics.java | 15 ++-------
...cQueryResults.java => DefaultMetricResult.java} | 30 +++++++++++------
.../beam/sdk/metrics/MetricQueryResults.java | 23 +++++++++----
.../org/apache/beam/sdk/metrics/MetricResult.java | 5 +++
.../org/apache/beam/sdk/metrics/MetricResults.java | 2 +-
9 files changed, 50 insertions(+), 105 deletions(-)
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
index e285493..f2c6f03 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
@@ -217,7 +217,7 @@ public class MetricsContainerStepMap implements Serializable {
return new QueryResults(filter);
}
- private class QueryResults implements MetricQueryResults {
+ private class QueryResults extends MetricQueryResults {
private final MetricsFilter filter;
private QueryResults(MetricsFilter filter) {
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
index d28b297..e3f34da 100644
--- 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
@@ -19,7 +19,6 @@ package org.apache.beam.runners.direct;
import static java.util.Arrays.asList;
-import com.google.auto.value.AutoValue;
import java.util.ArrayList;
import java.util.Map;
import java.util.Map.Entry;
@@ -39,7 +38,6 @@ import org.apache.beam.runners.core.metrics.MetricUpdates.MetricUpdate;
import org.apache.beam.runners.core.metrics.MetricsMap;
import org.apache.beam.sdk.metrics.DistributionResult;
import org.apache.beam.sdk.metrics.GaugeResult;
-import org.apache.beam.sdk.metrics.MetricName;
import org.apache.beam.sdk.metrics.MetricQueryResults;
import org.apache.beam.sdk.metrics.MetricResult;
import org.apache.beam.sdk.metrics.MetricResults;
@@ -224,38 +222,6 @@ class DirectMetrics extends MetricResults {
private final MetricsMap<MetricKey, DirectMetric<GaugeData, GaugeResult>> gauges;
- @AutoValue
- abstract static class DirectMetricQueryResults implements MetricQueryResults {
- public static MetricQueryResults create(
- Iterable<MetricResult<Long>> counters,
- Iterable<MetricResult<DistributionResult>> distributions,
- Iterable<MetricResult<GaugeResult>> gauges) {
- return new AutoValue_DirectMetrics_DirectMetricQueryResults(counters, distributions, gauges);
- }
- }
-
- @AutoValue
- abstract static class DirectMetricResult<T> implements MetricResult<T> {
- // need to define these here so they appear in the correct order
- // and the generated constructor is usable and consistent
- @Override
- public abstract MetricName getName();
-
- @Override
- public abstract String getStep();
-
- @Override
- public abstract T getCommitted();
-
- @Override
- public abstract T getAttempted();
-
- public static <T> MetricResult<T> create(
- MetricName name, String scope, T committed, T attempted) {
- return new AutoValue_DirectMetrics_DirectMetricResult<>(name, scope, committed, attempted);
- }
- }
-
DirectMetrics(ExecutorService executorService) {
this.counters = new MetricsMap<>(unusedKey -> new DirectMetric<>(COUNTER, executorService));
this.distributions =
@@ -280,7 +246,7 @@ class DirectMetrics extends MetricResults {
maybeExtractResult(filter, gaugeResults, gauge);
}
- return DirectMetricQueryResults.create(
+ return MetricQueryResults.create(
counterResults.build(), distributionResults.build(), gaugeResults.build());
}
@@ -290,7 +256,7 @@ class DirectMetrics extends MetricResults {
Map.Entry<MetricKey, ? extends DirectMetric<?, ResultT>> entry) {
if (MetricFiltering.matches(filter, entry.getKey())) {
resultsBuilder.add(
- DirectMetricResult.create(
+ MetricResult.create(
entry.getKey().metricName(),
entry.getKey().stepName(),
entry.getValue().extractCommitted(),
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectMetrics.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectMetrics.java
index b025bbb..93d75d0 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectMetrics.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectMetrics.java
@@ -19,7 +19,6 @@ package org.apache.beam.runners.direct.portable;
import static java.util.Arrays.asList;
-import com.google.auto.value.AutoValue;
import java.util.ArrayList;
import java.util.Map;
import java.util.Map.Entry;
@@ -39,7 +38,6 @@ import org.apache.beam.runners.core.metrics.MetricUpdates.MetricUpdate;
import org.apache.beam.runners.core.metrics.MetricsMap;
import org.apache.beam.sdk.metrics.DistributionResult;
import org.apache.beam.sdk.metrics.GaugeResult;
-import org.apache.beam.sdk.metrics.MetricName;
import org.apache.beam.sdk.metrics.MetricQueryResults;
import org.apache.beam.sdk.metrics.MetricResult;
import org.apache.beam.sdk.metrics.MetricResults;
@@ -235,38 +233,6 @@ class DirectMetrics extends MetricResults {
private MetricsMap<MetricKey, DirectMetric<GaugeData, GaugeResult>> gauges =
new MetricsMap<>(unusedKey -> new DirectMetric<>(GAUGE));
- @AutoValue
- abstract static class DirectMetricQueryResults implements MetricQueryResults {
- public static MetricQueryResults create(
- Iterable<MetricResult<Long>> counters,
- Iterable<MetricResult<DistributionResult>> distributions,
- Iterable<MetricResult<GaugeResult>> gauges) {
- return new AutoValue_DirectMetrics_DirectMetricQueryResults(counters, distributions, gauges);
- }
- }
-
- @AutoValue
- abstract static class DirectMetricResult<T> implements MetricResult<T> {
- // need to define these here so they appear in the correct order
- // and the generated constructor is usable and consistent
- @Override
- public abstract MetricName getName();
-
- @Override
- public abstract String getStep();
-
- @Override
- public abstract T getCommitted();
-
- @Override
- public abstract T getAttempted();
-
- public static <T> MetricResult<T> create(
- MetricName name, String scope, T committed, T attempted) {
- return new AutoValue_DirectMetrics_DirectMetricResult<>(name, scope, committed, attempted);
- }
- }
-
@Override
public MetricQueryResults queryMetrics(@Nullable MetricsFilter filter) {
ImmutableList.Builder<MetricResult<Long>> counterResults = ImmutableList.builder();
@@ -284,7 +250,7 @@ class DirectMetrics extends MetricResults {
maybeExtractResult(filter, gaugeResults, gauge);
}
- return DirectMetricQueryResults.create(
+ return MetricQueryResults.create(
counterResults.build(), distributionResults.build(), gaugeResults.build());
}
@@ -294,7 +260,7 @@ class DirectMetrics extends MetricResults {
Map.Entry<MetricKey, ? extends DirectMetric<?, ResultT>> entry) {
if (MetricFiltering.matches(filter, entry.getKey())) {
resultsBuilder.add(
- DirectMetricResult.create(
+ MetricResult.create(
entry.getKey().metricName(),
entry.getKey().stepName(),
entry.getValue().extractCommitted(),
diff --git a/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/CustomMetricQueryResults.java b/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/CustomMetricQueryResults.java
index fd46170..44db87b 100644
--- a/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/CustomMetricQueryResults.java
+++ b/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/CustomMetricQueryResults.java
@@ -28,7 +28,7 @@ import org.apache.beam.sdk.metrics.MetricsSink;
import org.joda.time.Instant;
/** Test class to be used as a input to {@link MetricsSink} implementations tests. */
-class CustomMetricQueryResults implements MetricQueryResults {
+class CustomMetricQueryResults extends MetricQueryResults {
private final boolean isCommittedSupported;
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
index 8a1901d..d5311c5 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
@@ -98,7 +98,7 @@ class DataflowMetrics extends MetricResults {
jobMetrics = getJobMetrics();
} catch (IOException e) {
LOG.warn("Unable to query job metrics.\n");
- return DataflowMetricQueryResults.create(counters, distributions, gauges);
+ return MetricQueryResults.create(counters, distributions, gauges);
}
metricUpdates = firstNonNull(jobMetrics.getMetrics(), Collections.<MetricUpdate>emptyList());
return populateMetricQueryResults(metricUpdates, filter);
@@ -322,7 +322,7 @@ class DataflowMetrics extends MetricResults {
extractor.addMetricResult(
metricKey, committedByName.get(metricKey), tentativeByName.get(metricKey));
}
- return DataflowMetricQueryResults.create(
+ return MetricQueryResults.create(
extractor.getCounterResults(),
extractor.getDistributionResults(),
extractor.getGaugeResults());
@@ -330,17 +330,6 @@ class DataflowMetrics extends MetricResults {
}
@AutoValue
- abstract static class DataflowMetricQueryResults implements MetricQueryResults {
- public static MetricQueryResults create(
- Iterable<MetricResult<Long>> counters,
- Iterable<MetricResult<DistributionResult>> distributions,
- Iterable<MetricResult<GaugeResult>> gauges) {
- return new AutoValue_DataflowMetrics_DataflowMetricQueryResults(
- counters, distributions, gauges);
- }
- }
-
- @AutoValue
abstract static class DataflowMetricResult<T> implements MetricResult<T> {
// need to define these here so they appear in the correct order
// and the generated constructor is usable and consistent
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DefaultMetricResult.java
similarity index 60%
copy from sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
copy to sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DefaultMetricResult.java
index f43a936..28e0a5e 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DefaultMetricResult.java
@@ -17,20 +17,30 @@
*/
package org.apache.beam.sdk.metrics;
+import com.fasterxml.jackson.annotation.JsonFilter;
+import com.google.auto.value.AutoValue;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
-/**
- * The results of a query for metrics. Allows accessing all of the metrics that matched the filter.
- */
+/** The results of a single current metric. */
@Experimental(Kind.METRICS)
-public interface MetricQueryResults {
- /** Return the metric results for the counters that matched the filter. */
- Iterable<MetricResult<Long>> getCounters();
+@JsonFilter("committedMetrics")
+@AutoValue
+public abstract class DefaultMetricResult<T> implements MetricResult<T> {
+ @Override
+ public abstract MetricName getName();
+
+ @Override
+ public abstract String getStep();
+
+ @Override
+ public abstract T getCommitted();
- /** Return the metric results for the distributions that matched the filter. */
- Iterable<MetricResult<DistributionResult>> getDistributions();
+ @Override
+ public abstract T getAttempted();
- /** Return the metric results for the gauges that matched the filter. */
- Iterable<MetricResult<GaugeResult>> getGauges();
+ public static <T> DefaultMetricResult<T> create(
+ MetricName name, String step, T committed, T attempted) {
+ return new AutoValue_DefaultMetricResult<>(name, step, committed, attempted);
+ }
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
index f43a936..baae446 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
@@ -17,20 +17,29 @@
*/
package org.apache.beam.sdk.metrics;
+import com.google.auto.value.AutoValue;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
/**
* The results of a query for metrics. Allows accessing all of the metrics that matched the filter.
*/
+@AutoValue
@Experimental(Kind.METRICS)
-public interface MetricQueryResults {
- /** Return the metric results for the counters that matched the filter. */
- Iterable<MetricResult<Long>> getCounters();
+public abstract class MetricQueryResults {
+ /** Return the metric results for the getCounters that matched the filter. */
+ public abstract Iterable<MetricResult<Long>> getCounters();
- /** Return the metric results for the distributions that matched the filter. */
- Iterable<MetricResult<DistributionResult>> getDistributions();
+ /** Return the metric results for the getDistributions that matched the filter. */
+ public abstract Iterable<MetricResult<DistributionResult>> getDistributions();
- /** Return the metric results for the gauges that matched the filter. */
- Iterable<MetricResult<GaugeResult>> getGauges();
+ /** Return the metric results for the getGauges that matched the filter. */
+ public abstract Iterable<MetricResult<GaugeResult>> getGauges();
+
+ public static MetricQueryResults create(
+ Iterable<MetricResult<Long>> counters,
+ Iterable<MetricResult<DistributionResult>> distributions,
+ Iterable<MetricResult<GaugeResult>> gauges) {
+ return new AutoValue_MetricQueryResults(counters, distributions, gauges);
+ }
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
index d8e0875..da6f6d1 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
@@ -30,6 +30,7 @@ import org.apache.beam.sdk.annotations.Experimental.Kind;
public interface MetricResult<T> {
/** Return the name of the metric. */
MetricName getName();
+
/** Return the step context to which this metric result applies. */
String getStep();
@@ -43,4 +44,8 @@ public interface MetricResult<T> {
/** Return the value of this metric across all attempts of executing all parts of the pipeline. */
T getAttempted();
+
+ static <T> DefaultMetricResult<T> create(MetricName name, String step, T committed, T attempted) {
+ return new AutoValue_DefaultMetricResult<>(name, step, committed, attempted);
+ }
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
index c781727..585d18b 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
@@ -54,7 +54,7 @@ public abstract class MetricResults {
* .addNameFilter("my-counter")
* .addStepFilter("myStepName1").addStepFilter("myStepName2")
* .build());
- * Iterable<MetricResult<Long>> counters = metricResults.counters();
+ * Iterable<MetricResult<Long>> counters = metricResults.getCounters();
* // counters should contain the value of my-counter reported from each of the ParDo
* // applications.
* }</pre>