[BEAM-6431] Implement Execution Time metrics start,process,finish in the Java SDK

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/SimpleExecutionState.java

@@ -0,0 +1,60 @@
+/*
+ * 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
+ *
+ * https://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 java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker.ExecutionState;
+
+/**
+ * Simple state class which collects the totalMillis spent in the state. Allows storing an arbitrary
+ * set of key value labels in the object which can be retrieved later for reporting purposes via
+ * getLabels().
+ */
+public class SimpleExecutionState extends ExecutionState {
+ private long totalMillis = 0;
+ private HashMap<String, String> labelsMetadata;
+
+ /**
+ * @param urn A string urn for the execution time metric.
+ * @param labelsMetadata arbitrary metadata to use for reporting purposes.
+ */
+ public SimpleExecutionState(String urn, HashMap<String, String> labelsMetadata) {
+ super(urn);
+ this.labelsMetadata = labelsMetadata;
+ }
+
+ public Map<String, String> getLabels() {
+ return Collections.unmodifiableMap(labelsMetadata);
+ }
+
+ @Override
+ public void takeSample(long millisSinceLastSample) {
+ this.totalMillis += millisSinceLastSample;
+ }
+
+ public long getTotalMillis() {
+ return totalMillis;
+ }
+
+ @Override
+ public void reportLull(Thread trackedThread, long millis) {
+ // TOOD(ajamato): Implement lullz detection to log stuck PTransforms.
+ }
+}

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/SimpleMonitoringInfoBuilder.java

@@ -60,6 +60,12 @@
 public class SimpleMonitoringInfoBuilder {
  public static final String ELEMENT_COUNT_URN =
  BeamUrns.getUrn(MonitoringInfoUrns.Enum.ELEMENT_COUNT);
+ public static final String START_BUNDLE_MSECS_URN =
+ BeamUrns.getUrn(MonitoringInfoUrns.Enum.START_BUNDLE_MSECS);
+ public static final String PROCESS_BUNDLE_MSECS_URN =
+ BeamUrns.getUrn(MonitoringInfoUrns.Enum.PROCESS_BUNDLE_MSECS);
+ public static final String FINISH_BUNDLE_MSECS_URN =
+ BeamUrns.getUrn(MonitoringInfoUrns.Enum.FINISH_BUNDLE_MSECS);
  public static final String USER_COUNTER_URN_PREFIX =
  BeamUrns.getUrn(MonitoringInfoUrns.Enum.USER_COUNTER_URN_PREFIX);
  public static final String SUM_INT64_TYPE_URN =
@@ -147,9 +153,15 @@ public SimpleMonitoringInfoBuilder setTimestampToNow() {
  return this;
  }
 
- /** Sets the int64Value of the CounterData in the MonitoringInfo, and the appropraite type URN. */
+ /** Sets the int64Value of the CounterData in the MonitoringInfo, and the appropriate type URN. */
  public SimpleMonitoringInfoBuilder setInt64Value(long value) {
  this.builder.getMetricBuilder().getCounterDataBuilder().setInt64Value(value);
+ this.setInt64TypeUrn();
+ return this;
+ }
+
+ /** Sets the the appropriate type URN for sum int64 counters. */
+ public SimpleMonitoringInfoBuilder setInt64TypeUrn() {
  this.builder.setType(SUM_INT64_TYPE_URN);
  return this;
  }

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/SimpleStateRegistry.java

@@ -0,0 +1,49 @@
+/*
+ * 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
+ *
+ * https://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 java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.MonitoringInfo;
+
+/**
+ * A Class for registering SimpleExecutionStates with and extracting execution time MonitoringInfos.
+ */
+public class SimpleStateRegistry {
+ private List<SimpleExecutionState> executionStates = new ArrayList<SimpleExecutionState>();
+
+ public void register(SimpleExecutionState state) {
+ this.executionStates.add(state);
+ }
+
+ /** @return Execution Time MonitoringInfos based on the tracked start or finish function. */
+ public List<MonitoringInfo> getExecutionTimeMonitoringInfos() {
+ List<MonitoringInfo> monitoringInfos = new ArrayList<MonitoringInfo>();
+ for (SimpleExecutionState state : executionStates) {
+ SimpleMonitoringInfoBuilder builder = new SimpleMonitoringInfoBuilder();
+ builder.setUrn(state.getStateName());
+ for (Map.Entry<String, String> entry : state.getLabels().entrySet()) {
+ builder.setLabel(entry.getKey(), entry.getValue());
+ }
+ builder.setInt64Value(state.getTotalMillis());
+ monitoringInfos.add(builder.build());
+ }
+ return monitoringInfos;
+ }
+}

runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MonitoringInfoMatchers.java

@@ -0,0 +1,98 @@
+/*
+ * 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
+ *
+ * https://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.model.fnexecution.v1.BeamFnApi.MonitoringInfo;
+import org.hamcrest.Description;
+import org.hamcrest.TypeSafeMatcher;
+
+/** Matchers for {@link MonitoringInfo}. */
+public class MonitoringInfoMatchers {
+
+ /**
+ * Matches a {@link MonitoringInfo} with that has the set fields in the provide MonitoringInfo.
+ *
+ * <p>This is useful for tests which do not want to match the specific value (execution times).
+ * Currently this will only check for URNs, labels, type URNs and int64Values.
+ */
+ public static TypeSafeMatcher<MonitoringInfo> matchSetFields(final MonitoringInfo mi) {
+ return new TypeSafeMatcher<MonitoringInfo>() {
+
+ @Override
+ protected boolean matchesSafely(MonitoringInfo item) {
+ if (!item.getUrn().equals(mi.getUrn())) {
+ return false;
+ }
+ if (!item.getLabels().equals(mi.getLabels())) {
+ return false;
+ }
+ if (!item.getType().equals(mi.getType())) {
+ return false;
+ }
+
+ if (mi.getMetric().hasCounterData()) {
+ long valueToMatch = mi.getMetric().getCounterData().getInt64Value();
+ if (valueToMatch != item.getMetric().getCounterData().getInt64Value()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description
+ .appendText("URN=")
+ .appendValue(mi.getUrn())
+ .appendText(", labels=")
+ .appendValue(mi.getLabels())
+ .appendText(", type=")
+ .appendValue(mi.getType());
+ if (mi.getMetric().hasCounterData()) {
+ description
+ .appendText(", value=")
+ .appendValue(mi.getMetric().getCounterData().getInt64Value());
+ }
+ }
+ };
+ }
+
+ /**
+ * Matches a {@link MonitoringInfo} with that has the set fields in the provide MonitoringInfo.
+ *
+ * <p>This is useful for tests which do not want to match the specific value (execution times).
+ * Currently this will only check for URNs, labels, type URNs and int64Values.
+ */
+ public static TypeSafeMatcher<MonitoringInfo> valueGreaterThan(final long value) {
+ return new TypeSafeMatcher<MonitoringInfo>() {
+
+ @Override
+ protected boolean matchesSafely(MonitoringInfo item) {
+ if (item.getMetric().getCounterData().getInt64Value() < value) {
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("value=").appendValue(value);
+ }
+ };
+ }
+}

runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/SimpleExecutionStateTest.java

@@ -0,0 +1,56 @@
+/*
+ * 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
+ *
+ * https://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.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.junit.Assert.assertEquals;
+
+import java.util.HashMap;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link SimpleExecutionState}. */
+@RunWith(JUnit4.class)
+public class SimpleExecutionStateTest {
+
+ @Test
+ public void testLabelsAndNameAreExtracted() {
+ String stateName = "myState";
+ HashMap<String, String> labelsMetadata = new HashMap<String, String>();
+ labelsMetadata.put("k1", "v1");
+ labelsMetadata.put("k2", "v2");
+ SimpleExecutionState testObject = new SimpleExecutionState(stateName, labelsMetadata);
+
+ assertEquals(testObject.getStateName(), stateName);
+ assertEquals(2, testObject.getLabels().size());
+ assertThat(testObject.getLabels(), hasEntry("k1", "v1"));
+ assertThat(testObject.getLabels(), hasEntry("k2", "v2"));
+ }
+
+ @Test
+ public void testtakeSampleIncrementsTotal() {
+ SimpleExecutionState testObject = new SimpleExecutionState("myState", null);
+ assertEquals(0, testObject.getTotalMillis());
+ testObject.takeSample(10);
+ assertEquals(10, testObject.getTotalMillis());
+ testObject.takeSample(5);
+ assertEquals(15, testObject.getTotalMillis());
+ }
+}

runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/SimpleStateRegistryTest.java

@@ -0,0 +1,78 @@
+/*
+ * 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
+ *
+ * https://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.junit.Assert.assertThat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.MonitoringInfo;
+import org.hamcrest.Matcher;
+import org.hamcrest.Matchers;
+import org.junit.Test;
+
+/** Tests for {@link SimpleStateRegistryTest}. */
+public class SimpleStateRegistryTest {
+
+ @Test
+ public void testExecutionTimeUrnsBuildMonitoringInfos() throws Exception {
+ String testPTransformId = "pTransformId";
+ HashMap<String, String> labelsMetadata = new HashMap<String, String>();
+ labelsMetadata.put(SimpleMonitoringInfoBuilder.PTRANSFORM_LABEL, testPTransformId);
+ SimpleExecutionState startState =
+ new SimpleExecutionState(
+ SimpleMonitoringInfoBuilder.START_BUNDLE_MSECS_URN, labelsMetadata);
+ SimpleExecutionState processState =
+ new SimpleExecutionState(
+ SimpleMonitoringInfoBuilder.PROCESS_BUNDLE_MSECS_URN, labelsMetadata);
+ SimpleExecutionState finishState =
+ new SimpleExecutionState(
+ SimpleMonitoringInfoBuilder.FINISH_BUNDLE_MSECS_URN, labelsMetadata);
+
+ SimpleStateRegistry testObject = new SimpleStateRegistry();
+ testObject.register(startState);
+ testObject.register(processState);
+ testObject.register(finishState);
+ List<MonitoringInfo> testOutput = testObject.getExecutionTimeMonitoringInfos();
+
+ List<Matcher<MonitoringInfo>> matchers = new ArrayList<Matcher<MonitoringInfo>>();
+ SimpleMonitoringInfoBuilder builder = new SimpleMonitoringInfoBuilder();
+ builder.setUrn(SimpleMonitoringInfoBuilder.START_BUNDLE_MSECS_URN);
+ builder.setInt64Value(0);
+ builder.setPTransformLabel(testPTransformId);
+ matchers.add(MonitoringInfoMatchers.matchSetFields(builder.build()));
+
+ // Check for execution time metrics for the testPTransformId
+ builder = new SimpleMonitoringInfoBuilder();
+ builder.setUrn(SimpleMonitoringInfoBuilder.PROCESS_BUNDLE_MSECS_URN);
+ builder.setInt64Value(0);
+ builder.setPTransformLabel(testPTransformId);
+ matchers.add(MonitoringInfoMatchers.matchSetFields(builder.build()));
+
+ builder = new SimpleMonitoringInfoBuilder();
+ builder.setUrn(SimpleMonitoringInfoBuilder.FINISH_BUNDLE_MSECS_URN);
+ builder.setInt64Value(0);
+ builder.setPTransformLabel(testPTransformId);
+ matchers.add(MonitoringInfoMatchers.matchSetFields(builder.build()));
+
+ for (Matcher<MonitoringInfo> matcher : matchers) {
+ assertThat(testOutput, Matchers.hasItem(matcher));
+ }
+ }
+}

runners/java-fn-execution/build.gradle

@@ -34,6 +34,7 @@ dependencies {
  shadow library.java.slf4j_api
  testCompile project(":beam-sdks-java-harness")
  testCompile project(path: ":beam-runners-core-construction-java", configuration: "shadow")
+ testCompile project(path: ":beam-runners-core-java", configuration: "shadowTest")
  testCompile project(path: ":beam-sdks-java-core", configuration: "shadowTest")
  testCompile library.java.junit
  testCompile library.java.hamcrest_core