[
https://issues.apache.org/jira/browse/BEAM-3310?focusedWorklogId=94280&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-94280
]
ASF GitHub Bot logged work on BEAM-3310:
----------------------------------------
Author: ASF GitHub Bot
Created on: 23/Apr/18 20:35
Start Date: 23/Apr/18 20:35
Worklog Time Spent: 10m
Work Description: iemejia closed pull request #4548: [BEAM-3310] Metrics
pusher
URL: https://github.com/apache/beam/pull/4548
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/build_rules.gradle b/build_rules.gradle
index ada4c8b5976..13773dc691b 100644
--- a/build_rules.gradle
+++ b/build_rules.gradle
@@ -179,6 +179,7 @@ def apex_malhar_version = "3.4.0"
def postgres_version = "9.4.1212.jre7"
def jaxb_api_version = "2.2.12"
def kafka_version = "1.0.0"
+def jackson_datatype_joda_version = "2.4.0"
// A map of maps containing common libraries used per language. To use:
// dependencies {
@@ -258,6 +259,7 @@ ext.library = [
jackson_databind:
"com.fasterxml.jackson.core:jackson-databind:$jackson_version",
jackson_dataformat_cbor:
"com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:$jackson_version",
jackson_dataformat_yaml:
"com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:$jackson_version",
+ jackson_datatype_joda:
"com.fasterxml.jackson.datatype:jackson-datatype-joda:$jackson_datatype_joda_version",
jackson_module_scala:
"com.fasterxml.jackson.module:jackson-module-scala_2.11:$jackson_version",
jaxb_api: "javax.xml.bind:jaxb-api:$jaxb_api_version",
joda_time: "joda-time:joda-time:2.4",
diff --git a/pom.xml b/pom.xml
index e14a190f2e7..381679b06d6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -185,6 +185,7 @@
<compiler.default.exclude>nothing</compiler.default.exclude>
<gax-grpc.version>0.20.0</gax-grpc.version>
<jaxb-api.version>2.2.12</jaxb-api.version>
+ <jackson-datatype-joda-version>2.4.0</jackson-datatype-joda-version>
<!-- standard binary for kubectl -->
<kubectl>kubectl</kubectl>
@@ -1484,6 +1485,11 @@
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-joda</artifactId>
+ <version>${jackson-datatype-joda-version}</version>
+ </dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
@@ -1491,6 +1497,12 @@
<version>${jaxb-api.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-extensions-java-metrics</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
</dependencies>
</dependencyManagement>
diff --git
a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsPusher.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsPusher.java
new file mode 100644
index 00000000000..3e9147c7e2f
--- /dev/null
+++
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsPusher.java
@@ -0,0 +1,121 @@
+/*
+ * 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.core.metrics;
+
+import static
org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
+
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.Serializable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.apache.beam.sdk.metrics.MetricsSink;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.util.InstanceBuilder;
+
+/** Component that regularly merges metrics and pushes them to a metrics sink.
*/
+@Experimental(Experimental.Kind.METRICS)
+public class MetricsPusher implements Serializable {
+
+ private MetricsSink metricsSink;
+ private long period;
+ @Nullable private transient ScheduledFuture<?> scheduledFuture;
+ private transient PipelineResult pipelineResult;
+ private MetricsContainerStepMap metricsContainerStepMap;
+
+ public MetricsPusher(
+ MetricsContainerStepMap metricsContainerStepMap,
+ PipelineOptions pipelineOptions,
+ PipelineResult pipelineResult) {
+ this.metricsContainerStepMap = metricsContainerStepMap;
+ this.pipelineResult = pipelineResult;
+ period = pipelineOptions.getMetricsPushPeriod();
+ // calls the constructor of MetricsSink implementation specified in
+ // pipelineOptions.getMetricsSink() passing the pipelineOptions
+ metricsSink =
+ InstanceBuilder.ofType(MetricsSink.class)
+ .fromClass(pipelineOptions.getMetricsSink())
+ .withArg(PipelineOptions.class, pipelineOptions)
+ .build();
+ }
+
+ public void start() {
+ if (!(metricsSink instanceof NoOpMetricsSink)) {
+ ScheduledExecutorService scheduler =
+ Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("MetricsPusher-thread")
+ .build());
+ scheduledFuture = scheduler.scheduleAtFixedRate(() -> run(), 0, period,
TimeUnit.SECONDS);
+ }
+ }
+
+ private void tearDown() {
+ pushMetrics();
+ if (!scheduledFuture.isCancelled()) {
+ scheduledFuture.cancel(true);
+ }
+ }
+
+ private void run() {
+ pushMetrics();
+ if (pipelineResult != null) {
+ PipelineResult.State pipelineState = pipelineResult.getState();
+ if (pipelineState.isTerminal()) {
+ tearDown();
+ }
+ }
+ }
+
+ private void pushMetrics() {
+ if (!(metricsSink instanceof NoOpMetricsSink)) {
+ try {
+ // merge metrics
+ MetricResults metricResults =
asAttemptedOnlyMetricResults(metricsContainerStepMap);
+ MetricQueryResults metricQueryResults =
+ metricResults.queryMetrics(MetricsFilter.builder().build());
+ if ((Iterables.size(metricQueryResults.getDistributions()) != 0)
+ || (Iterables.size(metricQueryResults.getGauges()) != 0)
+ || (Iterables.size(metricQueryResults.getCounters()) != 0)) {
+ metricsSink.writeMetrics(metricQueryResults);
+ }
+
+ } catch (Exception e) {
+ MetricsPushException metricsPushException = new
MetricsPushException(e);
+ metricsPushException.printStackTrace();
+ }
+ }
+ }
+
+ /** Exception related to MetricsPusher to wrap technical exceptions. */
+ public static class MetricsPushException extends Exception {
+ MetricsPushException(Throwable cause) {
+ super(cause);
+ }
+ }
+}
diff --git
a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/NoOpMetricsSink.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/NoOpMetricsSink.java
new file mode 100644
index 00000000000..f2dc04a5efa
--- /dev/null
+++
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/NoOpMetricsSink.java
@@ -0,0 +1,34 @@
+/*
+ * 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.core.metrics;
+
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricsSink;
+import org.apache.beam.sdk.options.PipelineOptions;
+
+/**
+ * This is the default metrics sink that does nothing. When it is set,
MetricsPusher does not start
+ * pushing thread.
+ */
+public class NoOpMetricsSink implements MetricsSink {
+
+ public NoOpMetricsSink(PipelineOptions pipelineOptions) {}
+
+ @Override
+ public void writeMetrics(MetricQueryResults metricQueryResults) throws
Exception {}
+}
diff --git
a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsPusherTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsPusherTest.java
new file mode 100644
index 00000000000..a31ef0a253e
--- /dev/null
+++
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsPusherTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.core.metrics;
+
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.UsesAttemptedMetrics;
+import org.apache.beam.sdk.testing.UsesCounterMetrics;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.joda.time.Duration;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/** A test that verifies that metrics push system works. */
+public class MetricsPusherTest {
+
+ private static final long NUM_ELEMENTS = 1000L;
+ @Rule public final TestPipeline pipeline = TestPipeline.create();
+
+ @Before
+ public void init() {
+ TestMetricsSink.clear();
+ PipelineOptions options = pipeline.getOptions();
+ options.setMetricsSink(TestMetricsSink.class);
+ }
+
+ @Category({ValidatesRunner.class, UsesAttemptedMetrics.class,
UsesCounterMetrics.class})
+ @Test
+ public void test() throws Exception {
+ pipeline
+ .apply(
+ // Use maxReadTime to force unbounded mode.
+
GenerateSequence.from(0).to(NUM_ELEMENTS).withMaxReadTime(Duration.standardDays(1)))
+ .apply(ParDo.of(new CountingDoFn()));
+ pipeline.run();
+ // give metrics pusher time to push
+ Thread.sleep((pipeline.getOptions().getMetricsPushPeriod() + 1L) * 1000);
+ assertThat(TestMetricsSink.getCounterValue(), is(NUM_ELEMENTS));
+ }
+
+ private static class CountingDoFn extends DoFn<Long, Long> {
+ private final Counter counter = Metrics.counter(MetricsPusherTest.class,
"counter");
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ try {
+ counter.inc();
+ context.output(context.element());
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+}
diff --git
a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/TestMetricsSink.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/TestMetricsSink.java
new file mode 100644
index 00000000000..f65fc70ba15
--- /dev/null
+++
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/TestMetricsSink.java
@@ -0,0 +1,50 @@
+/*
+ * 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.core.metrics;
+
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricsSink;
+import org.apache.beam.sdk.options.PipelineOptions;
+
+/**
+ * This sink just stores in a static field the first counter (if it exists)
attempted value. This is
+ * usefull for tests.
+ */
+public class TestMetricsSink implements MetricsSink {
+
+ private static long counterValue;
+
+ public TestMetricsSink(PipelineOptions pipelineOptions) {}
+
+ public static long getCounterValue() {
+ return counterValue;
+ }
+
+ public static void clear() {
+ counterValue = 0L;
+ }
+
+ @Override
+ public void writeMetrics(MetricQueryResults metricQueryResults) throws
Exception {
+ counterValue =
+ metricQueryResults.getCounters().iterator().hasNext()
+ ? metricQueryResults.getCounters().iterator().next().getAttempted()
+ : 0L;
+ }
+}
diff --git a/runners/extensions-java/metrics/build.gradle
b/runners/extensions-java/metrics/build.gradle
new file mode 100644
index 00000000000..78a56721459
--- /dev/null
+++ b/runners/extensions-java/metrics/build.gradle
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+apply from: project(":").file("build_rules.gradle")
+applyJavaNature()
+
+description = "Apache Beam :: Runners :: Extensions Java :: Metrics"
+ext.summary = "Beam Runners Extensions Metrics provides implementations of
runners core metrics APIs."
+
+
+dependencies {
+ compile library.java.guava
+ shadow project(path: ":beam-sdks-java-core", configuration: "shadow")
+ shadow library.java.jackson_databind
+ shadow library.java.jackson_datatype_joda
+ shadow library.java.joda_time
+ shadow library.java.findbugs_annotations
+ shadowTest library.java.junit
+}
diff --git a/runners/extensions-java/metrics/pom.xml
b/runners/extensions-java/metrics/pom.xml
new file mode 100644
index 00000000000..286e956863e
--- /dev/null
+++ b/runners/extensions-java/metrics/pom.xml
@@ -0,0 +1,76 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-extensions-java-parent</artifactId>
+ <version>2.5.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>beam-runners-extensions-java-metrics</artifactId>
+ <name>Apache Beam :: Runners :: Extensions Java :: Metrics</name>
+ <description>Beam Runners Extensions Metrics provides implementations of
runners core metrics APIs.</description>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-joda</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>joda-time</groupId>
+ <artifactId>joda-time</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-sdks-java-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.github.stephenc.findbugs</groupId>
+ <artifactId>findbugs-annotations</artifactId>
+ </dependency>
+
+
+ </dependencies>
+</project>
\ No newline at end of file
diff --git
a/runners/extensions-java/metrics/src/main/java/org/apache/beam/runners/extensions/metrics/MetricsHttpSink.java
b/runners/extensions-java/metrics/src/main/java/org/apache/beam/runners/extensions/metrics/MetricsHttpSink.java
new file mode 100644
index 00000000000..0c8718e3981
--- /dev/null
+++
b/runners/extensions-java/metrics/src/main/java/org/apache/beam/runners/extensions/metrics/MetricsHttpSink.java
@@ -0,0 +1,76 @@
+/*
+ * 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.extensions.metrics;
+
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.joda.JodaModule;
+import com.google.common.annotations.VisibleForTesting;
+import java.io.DataOutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import javax.xml.ws.http.HTTPException;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricsSink;
+import org.apache.beam.sdk.options.PipelineOptions;
+
+/** HTTP Sink to push metrics in a POST HTTP request. */
+public class MetricsHttpSink implements MetricsSink {
+ private final String urlString;
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ public MetricsHttpSink(PipelineOptions pipelineOptions) {
+ this.urlString = pipelineOptions.getMetricsHttpSinkUrl();
+ }
+
+ @Experimental(Experimental.Kind.METRICS)
+ @Override
+ public void writeMetrics(MetricQueryResults metricQueryResults) throws
Exception {
+ URL url = new URL(urlString);
+ String metrics = serializeMetrics(metricQueryResults);
+ byte[] postData = metrics.getBytes(StandardCharsets.UTF_8);
+ HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+ connection.setDoOutput(true);
+ connection.setInstanceFollowRedirects(false);
+ connection.setRequestMethod("POST");
+ connection.setRequestProperty("Content-Type", "application/json");
+ connection.setRequestProperty("charset", "utf-8");
+ connection.setRequestProperty("Content-Length",
Integer.toString(postData.length));
+ connection.setUseCaches(false);
+ try (DataOutputStream connectionOuputStream =
+ new DataOutputStream(connection.getOutputStream())) {
+ connectionOuputStream.write(postData);
+ }
+ int responseCode = connection.getResponseCode();
+ if (responseCode != 200) {
+ throw new HTTPException(responseCode);
+ }
+ }
+
+ @VisibleForTesting
+ String serializeMetrics(MetricQueryResults metricQueryResults) throws
Exception {
+ objectMapper.registerModule(new JodaModule());
+ objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+ objectMapper.configure(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY, true);
+ return objectMapper.writeValueAsString(metricQueryResults);
+ }
+}
diff --git
a/runners/extensions-java/metrics/src/main/java/org/apache/beam/runners/extensions/metrics/package-info.java
b/runners/extensions-java/metrics/src/main/java/org/apache/beam/runners/extensions/metrics/package-info.java
new file mode 100644
index 00000000000..c7a558fee6b
--- /dev/null
+++
b/runners/extensions-java/metrics/src/main/java/org/apache/beam/runners/extensions/metrics/package-info.java
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+/**
+ * Provides extensions to be used with the runners such as metrics sinks.
+ *
+ */
+@DefaultAnnotation(NonNull.class)
+package org.apache.beam.runners.extensions.metrics;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
diff --git
a/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/MetricsHttpSinkTest.java
b/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/MetricsHttpSinkTest.java
new file mode 100644
index 00000000000..6a99cd5a3de
--- /dev/null
+++
b/runners/extensions-java/metrics/src/test/java/org/apache/beam/runners/extensions/metrics/MetricsHttpSinkTest.java
@@ -0,0 +1,237 @@
+/*
+ * 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.extensions.metrics;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collections;
+import java.util.List;
+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.options.PipelineOptionsFactory;
+import org.joda.time.Instant;
+import org.junit.Test;
+
+/** Test class for MetricsHttpSink. */
+public class MetricsHttpSinkTest {
+
+ @Test
+ public void testSerializer() throws Exception {
+ MetricQueryResults metricQueryResults =
+ new MetricQueryResults() {
+
+ @Override
+ public List<MetricResult<Long>> getCounters() {
+ return Collections.singletonList(
+ (MetricResult<Long>)
+ new MetricResult<Long>() {
+
+ @Override
+ public MetricName getName() {
+ return new MetricName() {
+
+ @Override
+ public String getNamespace() {
+ return "ns1";
+ }
+
+ @Override
+ public String getName() {
+ return "n1";
+ }
+ };
+ }
+
+ @Override
+ public String getStep() {
+ return "s1";
+ }
+
+ @Override
+ public Long getCommitted() {
+ return 10L;
+ }
+
+ @Override
+ public Long getAttempted() {
+ return 20L;
+ }
+ });
+ }
+
+ @Override
+ public List<MetricResult<DistributionResult>> getDistributions() {
+ return Collections.singletonList(
+ (MetricResult<DistributionResult>)
+ new MetricResult<DistributionResult>() {
+
+ @Override
+ public MetricName getName() {
+ return new MetricName() {
+
+ @Override
+ public String getNamespace() {
+ return "ns1";
+ }
+
+ @Override
+ public String getName() {
+ return "n2";
+ }
+ };
+ }
+
+ @Override
+ public String getStep() {
+ return "s2";
+ }
+
+ @Override
+ public DistributionResult getCommitted() {
+ return new DistributionResult() {
+
+ @Override
+ public long getSum() {
+ return 10L;
+ }
+
+ @Override
+ public long getCount() {
+ return 2L;
+ }
+
+ @Override
+ public long getMin() {
+ return 5L;
+ }
+
+ @Override
+ public long getMax() {
+ return 8L;
+ }
+ };
+ }
+
+ @Override
+ public DistributionResult getAttempted() {
+ return new DistributionResult() {
+
+ @Override
+ public long getSum() {
+ return 25L;
+ }
+
+ @Override
+ public long getCount() {
+ return 4L;
+ }
+
+ @Override
+ public long getMin() {
+ return 3L;
+ }
+
+ @Override
+ public long getMax() {
+ return 9L;
+ }
+ };
+ }
+ });
+ }
+
+ @Override
+ public List<MetricResult<GaugeResult>> getGauges() {
+ return Collections.singletonList(
+ (MetricResult<GaugeResult>)
+ new MetricResult<GaugeResult>() {
+
+ @Override
+ public MetricName getName() {
+ return new MetricName() {
+
+ @Override
+ public String getNamespace() {
+ return "ns1";
+ }
+
+ @Override
+ public String getName() {
+ return "n3";
+ }
+ };
+ }
+
+ @Override
+ public String getStep() {
+ return "s3";
+ }
+
+ @Override
+ public GaugeResult getCommitted() {
+ return new GaugeResult() {
+
+ @Override
+ public long getValue() {
+ return 100L;
+ }
+
+ @Override
+ public Instant getTimestamp() {
+ return new Instant(0L);
+ }
+ };
+ }
+
+ @Override
+ public GaugeResult getAttempted() {
+ return new GaugeResult() {
+
+ @Override
+ public long getValue() {
+ return 120L;
+ }
+
+ @Override
+ public Instant getTimestamp() {
+ return new Instant(0L);
+ }
+ };
+ }
+ });
+ }
+ };
+ MetricsHttpSink metricsHttpSink = new
MetricsHttpSink(PipelineOptionsFactory.create());
+ String serializeMetrics =
metricsHttpSink.serializeMetrics(metricQueryResults);
+ assertEquals(
+ "Errror in serialization",
+
"{\"counters\":[{\"attempted\":20,\"committed\":10,\"name\":{\"name\":\"n1\","
+ +
"\"namespace\":\"ns1\"},\"step\":\"s1\"}],\"distributions\":[{\"attempted\":"
+ +
"{\"count\":4,\"max\":9,\"mean\":6.25,\"min\":3,\"sum\":25},\"committed\":"
+ +
"{\"count\":2,\"max\":8,\"mean\":5.0,\"min\":5,\"sum\":10},\"name\":{\"name\":\"n2\","
+ +
"\"namespace\":\"ns1\"},\"step\":\"s2\"}],\"gauges\":[{\"attempted\":{\"timestamp\":"
+ +
"\"1970-01-01T00:00:00.000Z\",\"value\":120},\"committed\":{\"timestamp\":"
+ +
"\"1970-01-01T00:00:00.000Z\",\"value\":100},\"name\":{\"name\":\"n3\",\"namespace\":"
+ + "\"ns1\"},\"step\":\"s3\"}]}",
+ serializeMetrics);
+ }
+}
diff --git a/runners/extensions-java/pom.xml b/runners/extensions-java/pom.xml
new file mode 100644
index 00000000000..e7d63ed8f58
--- /dev/null
+++ b/runners/extensions-java/pom.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-parent</artifactId>
+ <version>2.5.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>beam-runners-extensions-java-parent</artifactId>
+ <name>Apache Beam :: Runners :: Extensions Java</name>
+
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>metrics</module>
+ </modules>
+
+</project>
\ No newline at end of file
diff --git a/runners/flink/build.gradle b/runners/flink/build.gradle
index 40052c8cf8c..41feccad7c6 100644
--- a/runners/flink/build.gradle
+++ b/runners/flink/build.gradle
@@ -29,6 +29,8 @@ description = "Apache Beam :: Runners :: Flink"
* we are attempting to reference the "sourceSets.test.output" directly.
*/
evaluationDependsOn(":beam-sdks-java-core")
+evaluationDependsOn(":beam-runners-core-java")
+
test {
systemProperty "log4j.configuration", "log4j-test.properties"
@@ -75,6 +77,7 @@ dependencies {
shadowTest "org.apache.flink:flink-streaming-java_2.11:$flink_version:tests"
shadowTest "org.apache.flink:flink-test-utils_2.11:$flink_version"
validatesRunner project(path: ":beam-sdks-java-core", configuration:
"shadowTest")
+ validatesRunner project(path: ":beam-runners-core-java", configuration:
"shadowTest")
validatesRunner project(path: project.path, configuration: "shadow")
}
@@ -92,7 +95,7 @@ def createValidatesRunnerTask(Map m) {
def pipelineOptions = JsonOutput.toJson(["--runner=TestFlinkRunner",
"--streaming=${config.streaming}"])
systemProperty "beamTestPipelineOptions", pipelineOptions
classpath = configurations.validatesRunner
- testClassesDirs =
files(project(":beam-sdks-java-core").sourceSets.test.output.classesDirs)
+ testClassesDirs =
files(project(":beam-sdks-java-core").sourceSets.test.output.classesDirs,
project(":beam-runners-core-java").sourceSets.test.output.classesDirs)
// maxParallelForks decreased from 4 in order to avoid OOM errors
maxParallelForks 2
if (config.streaming) {
diff --git a/runners/flink/pom.xml b/runners/flink/pom.xml
index f65ab911cd4..ed583df8763 100644
--- a/runners/flink/pom.xml
+++ b/runners/flink/pom.xml
@@ -67,6 +67,7 @@
<failIfNoTests>true</failIfNoTests>
<dependenciesToScan>
<dependency>org.apache.beam:beam-sdks-java-core</dependency>
+
<dependency>org.apache.beam:beam-runners-core-java</dependency>
</dependenciesToScan>
<systemPropertyVariables>
<beamTestPipelineOptions>
@@ -100,6 +101,7 @@
<failIfNoTests>true</failIfNoTests>
<dependenciesToScan>
<dependency>org.apache.beam:beam-sdks-java-core</dependency>
+
<dependency>org.apache.beam:beam-runners-core-java</dependency>
</dependenciesToScan>
<systemPropertyVariables>
<beamTestPipelineOptions>
diff --git
a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
index 5fdcdcec121..3776c8b0502 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java
@@ -26,6 +26,7 @@
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import org.apache.beam.runners.core.metrics.MetricsPusher;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.PipelineRunner;
@@ -120,7 +121,9 @@ public PipelineResult run(Pipeline pipeline) {
if (result instanceof DetachedEnvironment.DetachedJobExecutionResult) {
LOG.info("Pipeline submitted in Detached mode");
- return new FlinkDetachedRunnerResult();
+ FlinkDetachedRunnerResult flinkDetachedRunnerResult = new
FlinkDetachedRunnerResult();
+ // no metricsPusher because metrics are not supported in detached mode
+ return flinkDetachedRunnerResult;
} else {
LOG.info("Execution finished in {} msecs", result.getNetRuntime());
Map<String, Object> accumulators = result.getAllAccumulatorResults();
@@ -130,8 +133,13 @@ public PipelineResult run(Pipeline pipeline) {
LOG.info("{} : {}", entry.getKey(), entry.getValue());
}
}
-
- return new FlinkRunnerResult(accumulators, result.getNetRuntime());
+ FlinkRunnerResult flinkRunnerResult = new FlinkRunnerResult(accumulators,
+ result.getNetRuntime());
+ MetricsPusher metricsPusher =
+ new MetricsPusher(
+ flinkRunnerResult.getMetricsContainerStepMap(), options,
flinkRunnerResult);
+ metricsPusher.start();
+ return flinkRunnerResult;
}
}
diff --git
a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerResult.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerResult.java
index d54c4a5cab1..741e000e90e 100644
---
a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerResult.java
+++
b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerResult.java
@@ -76,7 +76,10 @@ public State waitUntilFinish(Duration duration) {
@Override
public MetricResults metrics() {
- return asAttemptedOnlyMetricResults(
- (MetricsContainerStepMap)
accumulators.get(FlinkMetricContainer.ACCUMULATOR_NAME));
+ return asAttemptedOnlyMetricResults(getMetricsContainerStepMap());
+ }
+
+ MetricsContainerStepMap getMetricsContainerStepMap() {
+ return (MetricsContainerStepMap)
accumulators.get(FlinkMetricContainer.ACCUMULATOR_NAME);
}
}
diff --git a/runners/pom.xml b/runners/pom.xml
index 4ff5cc7534b..1a339b1328e 100644
--- a/runners/pom.xml
+++ b/runners/pom.xml
@@ -46,6 +46,7 @@
<module>spark</module>
<module>apex</module>
<module>gcp</module>
+ <module>extensions-java</module>
</modules>
<profiles>
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
index 32972e7e9a0..467162a863f 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
@@ -29,6 +29,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.apache.beam.runners.core.construction.TransformInputs;
+import org.apache.beam.runners.core.metrics.MetricsPusher;
import org.apache.beam.runners.spark.aggregators.AggregatorsAccumulator;
import org.apache.beam.runners.spark.io.CreateStream;
import org.apache.beam.runners.spark.metrics.AggregatorMetricSource;
@@ -212,7 +213,6 @@ public SparkPipelineResult run(final Pipeline pipeline) {
updateCacheCandidates(pipeline, translator, evaluationContext);
initAccumulators(mOptions, jsc);
-
startPipeline =
executorService.submit(
() -> {
@@ -229,6 +229,12 @@ public SparkPipelineResult run(final Pipeline pipeline) {
registerMetricsSource(mOptions.getAppName());
}
+ // it would have been better to create MetricsPusher from runner-core but
we need
+ // runner-specific
+ // MetricsContainerStepMap
+ MetricsPusher metricsPusher =
+ new MetricsPusher(MetricsAccumulator.getInstance().value(), mOptions,
result);
+ metricsPusher.start();
return result;
}
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
index 6a153ffd6d4..53f2bbecb35 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
@@ -81,6 +81,7 @@ public JavaStreamingContext call() throws Exception {
// We must first init accumulators since translators expect them to be
instantiated.
SparkRunner.initAccumulators(options, jsc);
+ //do not need to create a MetricsPusher instance here because if is called
in SparkRunner.run()
EvaluationContext ctxt = new EvaluationContext(jsc, pipeline, options,
jssc);
// update cache candidates
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkMetricsPusherTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkMetricsPusherTest.java
new file mode 100644
index 00000000000..04f1319ccaa
--- /dev/null
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkMetricsPusherTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.spark.metrics;
+
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.runners.core.metrics.TestMetricsSink;
+import org.apache.beam.runners.spark.ReuseSparkContextRule;
+import org.apache.beam.runners.spark.SparkPipelineOptions;
+import org.apache.beam.runners.spark.StreamingTest;
+import org.apache.beam.runners.spark.io.CreateStream;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.ValidatesRunner;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/**
+ * A test that verifies that metrics push system works in spark runner.
+ */
+public class SparkMetricsPusherTest {
+
+ @Rule
+ public final transient ReuseSparkContextRule noContextResue =
ReuseSparkContextRule.no();
+
+ @Rule
+ public final TestPipeline pipeline = TestPipeline.create();
+
+ private Duration batchDuration() {
+ return Duration.millis(
+
(pipeline.getOptions().as(SparkPipelineOptions.class)).getBatchIntervalMillis());
+ }
+
+ @Before
+ public void init(){
+ TestMetricsSink.clear();
+ PipelineOptions options = pipeline.getOptions();
+ options.setMetricsSink(TestMetricsSink.class);
+ }
+
+ @Category(StreamingTest.class)
+ @Test
+ public void testInStreamingMode() throws Exception {
+ Instant instant = new Instant(0);
+ CreateStream<Integer> source =
+ CreateStream.of(VarIntCoder.of(), batchDuration())
+ .emptyBatch()
+ .advanceWatermarkForNextBatch(instant)
+ .nextBatch(
+ TimestampedValue.of(1, instant),
+ TimestampedValue.of(2, instant),
+ TimestampedValue.of(3, instant))
+
.advanceWatermarkForNextBatch(instant.plus(Duration.standardSeconds(1L)))
+ .nextBatch(
+ TimestampedValue.of(4,
instant.plus(Duration.standardSeconds(1L))),
+ TimestampedValue.of(5,
instant.plus(Duration.standardSeconds(1L))),
+ TimestampedValue.of(6,
instant.plus(Duration.standardSeconds(1L))))
+ .advanceNextBatchWatermarkToInfinity();
+ pipeline
+ .apply(source)
+ .apply(
+ Window.<Integer>into(FixedWindows.of(Duration.standardSeconds(3L)))
+ .withAllowedLateness(Duration.ZERO))
+ .apply(ParDo.of(new CountingDoFn()));
+
+ pipeline.run();
+ // give metrics pusher time to push
+ Thread.sleep((pipeline.getOptions().getMetricsPushPeriod() + 1L) * 1000);
+ assertThat(TestMetricsSink.getCounterValue(), is(6L));
+ }
+
+ private static class CountingDoFn extends DoFn<Integer, Integer>{
+ private final Counter counter = Metrics
+ .counter(SparkMetricsPusherTest.class, "counter");
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ try {
+ counter.inc();
+ context.output(context.element());
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ @Category(ValidatesRunner.class)
+ @Test
+ public void testInSBatchMode() throws Exception {
+ pipeline.apply(Create.of(1, 2, 3, 4, 5, 6))
+ .apply(ParDo.of(new CountingDoFn()));
+
+ pipeline.run();
+ // give metrics pusher time to push
+ Thread.sleep((pipeline.getOptions().getMetricsPushPeriod() + 1L) * 1000);
+ assertThat(TestMetricsSink.getCounterValue(), is(6L));
+ }
+
+}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsSink.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsSink.java
new file mode 100644
index 00000000000..4b6154780c0
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsSink.java
@@ -0,0 +1,28 @@
+/*
+ * 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.sdk.metrics;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.annotations.Experimental;
+
+/** Interface for all metric sinks. */
+@Experimental(Experimental.Kind.METRICS)
+public interface MetricsSink extends Serializable {
+ void writeMetrics(MetricQueryResults metricQueryResults) throws Exception;
+}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
index 4eb461efe8a..8e9cf7b638f 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
@@ -31,6 +31,7 @@
import javax.annotation.concurrent.ThreadSafe;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineRunner;
+import org.apache.beam.sdk.metrics.MetricsSink;
import org.apache.beam.sdk.options.ProxyInvocationHandler.Deserializer;
import org.apache.beam.sdk.options.ProxyInvocationHandler.Serializer;
import org.apache.beam.sdk.transforms.DoFn;
@@ -392,4 +393,45 @@ public String create(PipelineOptions options) {
return String.format("%s/%s", info.getName(),
info.getVersion()).replace(" ", "_");
}
}
+
+ @Description("The beam sink class to which the metrics will be pushed")
+ @Default.InstanceFactory(NoOpMetricsSink.class)
+ Class<? extends MetricsSink> getMetricsSink();
+ void setMetricsSink(Class<? extends MetricsSink> metricsSink);
+
+ /**
+ * A {@link DefaultValueFactory} that obtains the class of the {@code
NoOpMetricsSink}
+ * if it exists on the classpath, and throws an exception otherwise.
+ *
+ * <p>As the {@code NoOpMetricsSink} is in an independent module, it cannot
be directly referenced
+ * as the {@link Default}. However, it should still be used if available.
+ */
+ class NoOpMetricsSink implements DefaultValueFactory<Class<? extends
MetricsSink>> {
+ @Override
+ public Class<? extends MetricsSink> create(PipelineOptions options) {
+ try {
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ Class<? extends MetricsSink> noOpMetricsSinkClass =
+ (Class<? extends MetricsSink>)
+ Class.forName(
+ "org.apache.beam.runners.core.metrics.NoOpMetricsSink",
true,
+ ReflectHelpers.findClassLoader());
+ return noOpMetricsSinkClass;
+ } catch (ClassNotFoundException e) {
+ throw new IllegalArgumentException(String.format(
+ "NoOpMetricsSink was not found on classpath"));
+ }
+ }
+ }
+
+ @Description("The metrics push period in seconds")
+ @Default.Long(5)
+ Long getMetricsPushPeriod();
+ void setMetricsPushPeriod(Long period);
+
+ @Description("MetricsHttpSink url")
+ String getMetricsHttpSinkUrl();
+ void setMetricsHttpSinkUrl(String metricsSink);
+
+
}
diff --git a/sdks/java/javadoc/pom.xml b/sdks/java/javadoc/pom.xml
index 463e67ed8ee..ffd15464353 100644
--- a/sdks/java/javadoc/pom.xml
+++ b/sdks/java/javadoc/pom.xml
@@ -243,6 +243,11 @@
<artifactId>beam-sdks-java-io-xml</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-extensions-java-metrics</artifactId>
+ </dependency>
+
<!-- provided and optional dependencies.-->
<dependency>
<groupId>com.google.auto.service</groupId>
diff --git a/settings.gradle b/settings.gradle
index c81471245de..f0703a32adb 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -36,6 +36,8 @@ include "beam-runners-core-java"
project(":beam-runners-core-java").dir = file("runners/core-java")
include "beam-runners-direct-java"
project(":beam-runners-direct-java").dir = file("runners/direct-java")
+include "beam-runners-extensions-java-metrics"
+project(":beam-runners-extensions-java-metrics").dir =
file("runners/extensions-java/metrics")
include "beam-runners-flink_2.11"
project(":beam-runners-flink_2.11").dir = file("runners/flink")
include "beam-runners-gcp-gcemd"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 94280)
Time Spent: 12.5h (was: 12h 20m)
> Push metrics to a backend in an runner agnostic way
> ---------------------------------------------------
>
> Key: BEAM-3310
> URL: https://issues.apache.org/jira/browse/BEAM-3310
> Project: Beam
> Issue Type: New Feature
> Components: sdk-java-core
> Reporter: Etienne Chauchot
> Assignee: Etienne Chauchot
> Priority: Major
> Time Spent: 12.5h
> Remaining Estimate: 0h
>
> The idea is to avoid relying on the runners to provide access to the metrics
> (either at the end of the pipeline or while it runs) because they don't have
> all the same capabilities towards metrics (e.g. spark runner configures sinks
> like csv, graphite or in memory sinks using the spark engine conf). The
> target is to push the metrics in the common runner code so that no matter the
> chosen runner, a user can get his metrics out of beam.
> Here is the link to the discussion thread on the dev ML:
> https://lists.apache.org/thread.html/01a80d62f2df6b84bfa41f05e15fda900178f882877c294fed8be91e@%3Cdev.beam.apache.org%3E
> And the design doc:
> https://s.apache.org/runner_independent_metrics_extraction
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
