You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "clmccart (via GitHub)" <gi...@apache.org> on 2023/12/01 23:17:35 UTC
[PR] Per DoFn latency instrumentation [beam]
clmccart opened a new pull request, #29592:
URL: https://github.com/apache/beam/pull/29592
Modifies and utilizes existing instrumentation infrastructure in order to track the amount of time dofn processing takes for individual messages run on the StreamingDataflowWorker. This information is then sent to Windmill.
For actively processing messages, the name of the DoFn currently executing and the time it has been processing is sent back to Windmill on the heartbeat. For completed messages, the mean, min, max, count, and sum of individual message processing times for each DoFn is sent back to Windmill as part of the commit.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416334980
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -111,15 +120,58 @@ public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
if (duration.equals(Duration.ZERO)) {
continue;
}
- list.add(
- Windmill.LatencyAttribution.newBuilder()
- .setState(state)
- .setTotalDurationMillis(duration.getMillis())
- .build());
+ LatencyAttribution.Builder laBuilder = Windmill.LatencyAttribution.newBuilder();
+ if (state == LatencyAttribution.State.ACTIVE) {
+ laBuilder = addActiveLatencyBreakdownToBuilder(isHeartbeat, laBuilder, workId, sampler);
+ }
+ Windmill.LatencyAttribution la =
+ laBuilder.setState(state).setTotalDurationMillis(duration.getMillis()).build();
+ list.add(la);
}
return list;
}
+ private static LatencyAttribution.Builder addActiveLatencyBreakdownToBuilder(
+ Boolean isHeartbeat,
Review Comment:
1636e774170b95f004943fa7e80175044f06b206
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1422886017
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/ActiveMessageMetadata.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.dataflow.worker;
+
+import com.google.auto.value.AutoValue;
+
+@AutoValue
+public abstract class ActiveMessageMetadata {
+
+ public abstract String userStepName();
+
+ public abstract Long startTime();
Review Comment:
[df04f9e](https://github.com/apache/beam/pull/29592/commits/df04f9e9ed10daaadd1634546aa6dca33b16df27)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on PR #29592:
URL: https://github.com/apache/beam/pull/29592#issuecomment-1839555658
R: @m-trieu
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416334561
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/ActiveMessageMetadata.java:
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.dataflow.worker;
+
+public class ActiveMessageMetadata {
+
+ public String userStepName;
Review Comment:
1636e774170b95f004943fa7e80175044f06b206
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -101,7 +109,8 @@ private void recordGetWorkStreamLatencies(
}
}
- public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
+ public Collection<Windmill.LatencyAttribution> getLatencyAttributions(
+ Boolean isHeartbeat, String workId, DataflowExecutionStateSampler sampler) {
Review Comment:
1636e774170b95f004943fa7e80175044f06b206
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1422888801
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private final ConcurrentHashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Map<String, IntSummaryStatistics>>
+ completedProcessingMetrics = new ConcurrentHashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private static Map<String, IntSummaryStatistics> mergeStepStatsMaps(
+ Map<String, IntSummaryStatistics> map1, Map<String, IntSummaryStatistics> map2) {
+ for (Entry<String, IntSummaryStatistics> steps : map2.entrySet()) {
+ map1.compute(
+ steps.getKey(),
+ (k, v) -> {
+ if (v == null) {
+ return steps.getValue();
+ }
+ v.combine(steps.getValue());
+ return v;
+ });
+ }
+ return map1;
+ }
+
+ @Override
+ public void removeTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionContext.DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
+
+ // Attribute any remaining time since the last sampling while removing the tracker.
+ //
+ // There is a race condition here; if sampling happens in the time between when we remove the
+ // tracker from activeTrackers and read the lastSampleTicks value, the sampling time will
+ // be lost for the tracker being removed. This is acceptable as sampling is already an
+ // approximation of actual execution time.
+ long millisSinceLastSample = clock.getMillis() - this.lastSampleTimeMillis;
+ if (millisSinceLastSample > 0) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Override
+ public void doSampling(long millisSinceLastSample) {
+ for (DataflowExecutionStateTracker tracker : activeTrackersByWorkId.values()) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Nullable
+ public ActiveMessageMetadata getActiveMessageMetadataForWorkId(String workId) {
+ if (activeTrackersByWorkId.containsKey(workId)) {
+ return activeTrackersByWorkId.get(workId).getActiveMessageMetadata().orElse(null);
Review Comment:
i think this constructs an optional of an optional. compiler did not like it :(
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421180847
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationState.java:
##########
@@ -120,8 +121,9 @@ private void forceExecute(Work work) {
}
/** Adds any work started before the refreshDeadline to the GetDataRequest builder. */
- public List<Windmill.KeyedGetDataRequest> getKeysToRefresh(Instant refreshDeadline) {
- return activeWorkState.getKeysToRefresh(refreshDeadline);
+ public List<Windmill.KeyedGetDataRequest> getKeysToRefresh(
Review Comment:
If list shouldn't be mutated by caller, use `ImmutableList` go/java-practices/choosing-collections#immutable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1422887403
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private final ConcurrentHashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Map<String, IntSummaryStatistics>>
+ completedProcessingMetrics = new ConcurrentHashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private static Map<String, IntSummaryStatistics> mergeStepStatsMaps(
+ Map<String, IntSummaryStatistics> map1, Map<String, IntSummaryStatistics> map2) {
+ for (Entry<String, IntSummaryStatistics> steps : map2.entrySet()) {
+ map1.compute(
+ steps.getKey(),
+ (k, v) -> {
+ if (v == null) {
+ return steps.getValue();
+ }
+ v.combine(steps.getValue());
+ return v;
+ });
+ }
+ return map1;
+ }
+
+ @Override
+ public void removeTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionContext.DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
+
+ // Attribute any remaining time since the last sampling while removing the tracker.
+ //
+ // There is a race condition here; if sampling happens in the time between when we remove the
+ // tracker from activeTrackers and read the lastSampleTicks value, the sampling time will
+ // be lost for the tracker being removed. This is acceptable as sampling is already an
+ // approximation of actual execution time.
+ long millisSinceLastSample = clock.getMillis() - this.lastSampleTimeMillis;
+ if (millisSinceLastSample > 0) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Override
+ public void doSampling(long millisSinceLastSample) {
+ for (DataflowExecutionStateTracker tracker : activeTrackersByWorkId.values()) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Nullable
+ public ActiveMessageMetadata getActiveMessageMetadataForWorkId(String workId) {
+ if (activeTrackersByWorkId.containsKey(workId)) {
+ return activeTrackersByWorkId.get(workId).getActiveMessageMetadata().orElse(null);
+ }
+ return null;
+ }
+
+ @Nullable
+ public Map<String, IntSummaryStatistics> getProcessingDistributionsForWorkId(String workId) {
Review Comment:
[9587c95](https://github.com/apache/beam/pull/29592/commits/9587c950826fa66087cf521fa557a66c3b311032)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1424499612
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ExecutionStateTracker.java:
##########
@@ -304,7 +304,7 @@ public long getNextLullReportMs() {
return nextLullReportMs;
}
- void takeSample(long millisSinceLastSample) {
+ public void takeSample(long millisSinceLastSample) {
Review Comment:
This is now public, consider add a javadoc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #29592:
URL: https://github.com/apache/beam/pull/29592
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416335187
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
Review Comment:
6f3b75e75106f002f850c46333c0247c94ab7dd6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1422886895
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private final ConcurrentHashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Map<String, IntSummaryStatistics>>
+ completedProcessingMetrics = new ConcurrentHashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private static Map<String, IntSummaryStatistics> mergeStepStatsMaps(
+ Map<String, IntSummaryStatistics> map1, Map<String, IntSummaryStatistics> map2) {
+ for (Entry<String, IntSummaryStatistics> steps : map2.entrySet()) {
+ map1.compute(
+ steps.getKey(),
+ (k, v) -> {
+ if (v == null) {
+ return steps.getValue();
+ }
+ v.combine(steps.getValue());
+ return v;
+ });
+ }
+ return map1;
+ }
+
+ @Override
+ public void removeTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionContext.DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
+
+ // Attribute any remaining time since the last sampling while removing the tracker.
+ //
+ // There is a race condition here; if sampling happens in the time between when we remove the
+ // tracker from activeTrackers and read the lastSampleTicks value, the sampling time will
+ // be lost for the tracker being removed. This is acceptable as sampling is already an
+ // approximation of actual execution time.
+ long millisSinceLastSample = clock.getMillis() - this.lastSampleTimeMillis;
+ if (millisSinceLastSample > 0) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Override
+ public void doSampling(long millisSinceLastSample) {
+ for (DataflowExecutionStateTracker tracker : activeTrackersByWorkId.values()) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Nullable
+ public ActiveMessageMetadata getActiveMessageMetadataForWorkId(String workId) {
Review Comment:
[075dd57](https://github.com/apache/beam/pull/29592/commits/075dd57cca2409b576222ea6eb014f3dfe368bf0)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421175977
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private final ConcurrentHashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Map<String, IntSummaryStatistics>>
+ completedProcessingMetrics = new ConcurrentHashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private static Map<String, IntSummaryStatistics> mergeStepStatsMaps(
+ Map<String, IntSummaryStatistics> map1, Map<String, IntSummaryStatistics> map2) {
+ for (Entry<String, IntSummaryStatistics> steps : map2.entrySet()) {
+ map1.compute(
+ steps.getKey(),
+ (k, v) -> {
+ if (v == null) {
+ return steps.getValue();
+ }
+ v.combine(steps.getValue());
+ return v;
+ });
+ }
+ return map1;
+ }
+
+ @Override
+ public void removeTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionContext.DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
+
+ // Attribute any remaining time since the last sampling while removing the tracker.
+ //
+ // There is a race condition here; if sampling happens in the time between when we remove the
+ // tracker from activeTrackers and read the lastSampleTicks value, the sampling time will
+ // be lost for the tracker being removed. This is acceptable as sampling is already an
+ // approximation of actual execution time.
+ long millisSinceLastSample = clock.getMillis() - this.lastSampleTimeMillis;
+ if (millisSinceLastSample > 0) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Override
+ public void doSampling(long millisSinceLastSample) {
+ for (DataflowExecutionStateTracker tracker : activeTrackersByWorkId.values()) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Nullable
+ public ActiveMessageMetadata getActiveMessageMetadataForWorkId(String workId) {
+ if (activeTrackersByWorkId.containsKey(workId)) {
+ return activeTrackersByWorkId.get(workId).getActiveMessageMetadata().orElse(null);
Review Comment:
can be simplified by using Optional API `Optional.ofNullable(activeTrackersByWorkId.get(workId)).map(DataflowExecutionTracker::getActiveMessageMetadata)` which will return Optional<ActiveMessageMetadata>.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416336580
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private HashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId = new HashMap<>();
Review Comment:
80a8a5292139ac734f010546acc987e05982907c
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416336176
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -307,5 +335,34 @@ public Closeable enterState(ExecutionState newState) {
public String getWorkItemId() {
return this.workItemId;
}
+
+ public ActiveMessageMetadata getActiveMessageMetadata() {
Review Comment:
ef56c0419a649d246e0ee00e927ee821361be943
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416335879
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -1892,6 +1891,14 @@ private void invalidateStuckCommits() {
}
}
+ public static String constructWorkId(Windmill.WorkItem workItem) {
Review Comment:
cf55f3f7ca1c0db5f22c32f4149fb639b5da9009
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1414692392
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/ActiveMessageMetadata.java:
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.dataflow.worker;
+
+public class ActiveMessageMetadata {
+
+ public String userStepName;
Review Comment:
if this is a a value class it is recommended to use go/autovalue.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -111,15 +120,58 @@ public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
if (duration.equals(Duration.ZERO)) {
continue;
}
- list.add(
- Windmill.LatencyAttribution.newBuilder()
- .setState(state)
- .setTotalDurationMillis(duration.getMillis())
- .build());
+ LatencyAttribution.Builder laBuilder = Windmill.LatencyAttribution.newBuilder();
+ if (state == LatencyAttribution.State.ACTIVE) {
+ laBuilder = addActiveLatencyBreakdownToBuilder(isHeartbeat, laBuilder, workId, sampler);
+ }
+ Windmill.LatencyAttribution la =
+ laBuilder.setState(state).setTotalDurationMillis(duration.getMillis()).build();
+ list.add(la);
}
return list;
}
+ private static LatencyAttribution.Builder addActiveLatencyBreakdownToBuilder(
+ Boolean isHeartbeat,
Review Comment:
unless isHeartbeat is nullable, prefer to use the primitive type (boolean) to the boxed. go/java-practices/primitives#prefer-unboxed
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -1892,6 +1891,14 @@ private void invalidateStuckCommits() {
}
}
+ public static String constructWorkId(Windmill.WorkItem workItem) {
Review Comment:
How about letting this be in `Work.java`? and then it can be called like `Work.getId` or `Work.getLatencyTrackingId` etc.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
Review Comment:
does this class need to be threadsafe?
If yes, then HashMaps are not safe for concurrent writes. Consider using `ConcurrentHashMap` if the operations needed to be threadsafe are independent. If there needs to be synchronization across multiple data structures where the operations need to be atomic, it will needs to be synchronized via a lock (synchronized methods, synchronized blocks, synchronized(someLockObject)).
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
Review Comment:
for example if not atomic, you can make these ConcurrentMaps and then drop the synchornization for `resetForWorkId` since the map will handle it underneath.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -307,5 +335,34 @@ public Closeable enterState(ExecutionState newState) {
public String getWorkItemId() {
return this.workItemId;
}
+
+ public ActiveMessageMetadata getActiveMessageMetadata() {
+ return activeMessageMetadata;
+ }
+
+ public Map<String, IntSummaryStatistics> getProcessingTimesByStep() {
Review Comment:
this will pass the reference and allow callers to mutate the map.
unless that's intentional, consider wrapping in `Collections.unmodifiableMap(processingTimesByStep)` or returning an `ImmutableMap`go/java-practices/immutability#collections
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private HashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId = new HashMap<>();
+ private HashMap<String, Map<String, IntSummaryStatistics>> completedProcessingMetrics =
+ new HashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private Map<String, IntSummaryStatistics> mergeStepStatsMaps(
Review Comment:
nit: can this be static? (not sure if it references any member variables or methods)
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -101,7 +109,8 @@ private void recordGetWorkStreamLatencies(
}
}
- public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
+ public Collection<Windmill.LatencyAttribution> getLatencyAttributions(
+ Boolean isHeartbeat, String workId, DataflowExecutionStateSampler sampler) {
Review Comment:
unless `isHeartbeat` is nullable, prefer to use the primitive type (`boolean`) to the boxed. go/java-practices/primitives#prefer-unboxed
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
Review Comment:
nit: Make final class. go/java-practices/final#classes, unless you need to mock it, but since this is an implementation you can mock the parent class/interface where its used so I think final would be good here.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -307,5 +335,34 @@ public Closeable enterState(ExecutionState newState) {
public String getWorkItemId() {
return this.workItemId;
}
+
+ public ActiveMessageMetadata getActiveMessageMetadata() {
Review Comment:
If this is nullable, recommended to make it explicit by returning `Optional<ActiveMessageMetadata>` [go/java-practices/optional]
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -245,6 +248,16 @@ public static class DataflowExecutionStateTracker extends ExecutionStateTracker
private final ContextActivationObserverRegistry contextActivationObserverRegistry;
private final String workItemId;
+ /**
+ * Metadata on the message whose processing is currently being managed by this tracker. If no
+ * message is actively being processed, activeMessageMetadata will be null.
+ */
+ private ActiveMessageMetadata activeMessageMetadata = null;
Review Comment:
annotate with @Nullable
if threadsafety is required, consider using `AtomicReference<ActiveMessageMetadata>`
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -287,6 +286,8 @@ public class StreamingDataflowWorker {
// Possibly overridden by streaming engine config.
private int maxWorkItemCommitBytes = Integer.MAX_VALUE;
+ private DataflowExecutionStateSampler sampler = DataflowExecutionStateSampler.instance();
Review Comment:
can be final if not reassigned.
also is `DataflowExecutionStateSampler` a singleton?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private HashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId = new HashMap<>();
Review Comment:
Preferred to mark members final go/java-practices/final#fields unless they can be reassigned
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -245,6 +248,16 @@ public static class DataflowExecutionStateTracker extends ExecutionStateTracker
private final ContextActivationObserverRegistry contextActivationObserverRegistry;
private final String workItemId;
+ /**
+ * Metadata on the message whose processing is currently being managed by this tracker. If no
+ * message is actively being processed, activeMessageMetadata will be null.
+ */
+ private ActiveMessageMetadata activeMessageMetadata = null;
+
+ private MillisProvider clock = System::currentTimeMillis;
Review Comment:
make member variables final if not intended to be reassigned
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421177641
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private final ConcurrentHashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Map<String, IntSummaryStatistics>>
+ completedProcessingMetrics = new ConcurrentHashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private static Map<String, IntSummaryStatistics> mergeStepStatsMaps(
+ Map<String, IntSummaryStatistics> map1, Map<String, IntSummaryStatistics> map2) {
+ for (Entry<String, IntSummaryStatistics> steps : map2.entrySet()) {
+ map1.compute(
+ steps.getKey(),
+ (k, v) -> {
+ if (v == null) {
+ return steps.getValue();
+ }
+ v.combine(steps.getValue());
+ return v;
+ });
+ }
+ return map1;
+ }
+
+ @Override
+ public void removeTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionContext.DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
+
+ // Attribute any remaining time since the last sampling while removing the tracker.
+ //
+ // There is a race condition here; if sampling happens in the time between when we remove the
+ // tracker from activeTrackers and read the lastSampleTicks value, the sampling time will
+ // be lost for the tracker being removed. This is acceptable as sampling is already an
+ // approximation of actual execution time.
+ long millisSinceLastSample = clock.getMillis() - this.lastSampleTimeMillis;
+ if (millisSinceLastSample > 0) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Override
+ public void doSampling(long millisSinceLastSample) {
+ for (DataflowExecutionStateTracker tracker : activeTrackersByWorkId.values()) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Nullable
+ public ActiveMessageMetadata getActiveMessageMetadataForWorkId(String workId) {
+ if (activeTrackersByWorkId.containsKey(workId)) {
+ return activeTrackersByWorkId.get(workId).getActiveMessageMetadata().orElse(null);
+ }
+ return null;
+ }
+
+ @Nullable
+ public Map<String, IntSummaryStatistics> getProcessingDistributionsForWorkId(String workId) {
Review Comment:
Can we just return an empty map instead of null? go/java-practices/null
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #29592:
URL: https://github.com/apache/beam/pull/29592#issuecomment-1912229587
The newly added test `testDoFnActiveMessageMetadataReportedOnHeartbeat` is flaky: https://github.com/apache/beam/runs/20887446311
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1424509384
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ExecutionStateTracker.java:
##########
@@ -304,7 +304,7 @@ public long getNextLullReportMs() {
return nextLullReportMs;
}
- void takeSample(long millisSinceLastSample) {
+ public void takeSample(long millisSinceLastSample) {
Review Comment:
[f371c80](https://github.com/apache/beam/pull/29592/commits/f371c80040c44c16e3ffafa0af2c2e9c1c5e9067)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1422887727
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -101,7 +117,8 @@ private void recordGetWorkStreamLatencies(
}
}
- public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
+ public Collection<Windmill.LatencyAttribution> getLatencyAttributions(
Review Comment:
[5dfe51f](https://github.com/apache/beam/pull/29592/commits/5dfe51fe62511b141b0395536ca7698a94362bdc)
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationState.java:
##########
@@ -120,8 +121,9 @@ private void forceExecute(Work work) {
}
/** Adds any work started before the refreshDeadline to the GetDataRequest builder. */
- public List<Windmill.KeyedGetDataRequest> getKeysToRefresh(Instant refreshDeadline) {
- return activeWorkState.getKeysToRefresh(refreshDeadline);
+ public List<Windmill.KeyedGetDataRequest> getKeysToRefresh(
Review Comment:
[5dfe51f](https://github.com/apache/beam/pull/29592/commits/5dfe51fe62511b141b0395536ca7698a94362bdc)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on PR #29592:
URL: https://github.com/apache/beam/pull/29592#issuecomment-1852679481
R: @Abacn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416337988
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -245,6 +248,16 @@ public static class DataflowExecutionStateTracker extends ExecutionStateTracker
private final ContextActivationObserverRegistry contextActivationObserverRegistry;
private final String workItemId;
+ /**
+ * Metadata on the message whose processing is currently being managed by this tracker. If no
+ * message is actively being processed, activeMessageMetadata will be null.
+ */
+ private ActiveMessageMetadata activeMessageMetadata = null;
+
+ private MillisProvider clock = System::currentTimeMillis;
Review Comment:
d6ef04f81684b1f0f8eb3e0f328de88c8a3f8f49
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416337598
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -307,5 +335,34 @@ public Closeable enterState(ExecutionState newState) {
public String getWorkItemId() {
return this.workItemId;
}
+
+ public ActiveMessageMetadata getActiveMessageMetadata() {
+ return activeMessageMetadata;
+ }
+
+ public Map<String, IntSummaryStatistics> getProcessingTimesByStep() {
Review Comment:
64878a1ed775229d2ddcda3ed03586eaa7d3e11d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416455970
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
Review Comment:
d64b73e
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416335675
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private HashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId = new HashMap<>();
+ private HashMap<String, Map<String, IntSummaryStatistics>> completedProcessingMetrics =
+ new HashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private Map<String, IntSummaryStatistics> mergeStepStatsMaps(
Review Comment:
ba58e408fe45269792c997aef6ac0753f9750e8e
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1422888132
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -111,15 +128,58 @@ public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
if (duration.equals(Duration.ZERO)) {
continue;
}
- list.add(
- Windmill.LatencyAttribution.newBuilder()
- .setState(state)
- .setTotalDurationMillis(duration.getMillis())
- .build());
+ LatencyAttribution.Builder laBuilder = Windmill.LatencyAttribution.newBuilder();
+ if (state == LatencyAttribution.State.ACTIVE) {
+ laBuilder = addActiveLatencyBreakdownToBuilder(isHeartbeat, laBuilder, workId, sampler);
+ }
+ Windmill.LatencyAttribution la =
+ laBuilder.setState(state).setTotalDurationMillis(duration.getMillis()).build();
+ list.add(la);
}
return list;
}
+ private static LatencyAttribution.Builder addActiveLatencyBreakdownToBuilder(
+ boolean isHeartbeat,
+ LatencyAttribution.Builder builder,
+ String workId,
+ DataflowExecutionStateSampler sampler) {
+ if (isHeartbeat) {
+ ActiveLatencyBreakdown.Builder stepBuilder = ActiveLatencyBreakdown.newBuilder();
+ ActiveMessageMetadata activeMessage = sampler.getActiveMessageMetadataForWorkId(workId);
Review Comment:
[075dd57](https://github.com/apache/beam/pull/29592/commits/075dd57cca2409b576222ea6eb014f3dfe368bf0)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421172835
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.dataflow.worker;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.HashMap;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.metrics.ExecutionStateSampler;
+import org.apache.beam.runners.core.metrics.ExecutionStateTracker;
+import org.apache.beam.runners.dataflow.worker.DataflowExecutionContext.DataflowExecutionStateTracker;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.DateTimeUtils.MillisProvider;
+
+public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
+
+ private static final MillisProvider SYSTEM_MILLIS_PROVIDER = System::currentTimeMillis;
+ private static final DataflowExecutionStateSampler INSTANCE =
+ new DataflowExecutionStateSampler(SYSTEM_MILLIS_PROVIDER);
+
+ private final ConcurrentHashMap<String, DataflowExecutionStateTracker> activeTrackersByWorkId =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Map<String, IntSummaryStatistics>>
+ completedProcessingMetrics = new ConcurrentHashMap<>();
+
+ public static DataflowExecutionStateSampler instance() {
+ return INSTANCE;
+ }
+
+ @VisibleForTesting
+ public static DataflowExecutionStateSampler newForTest(MillisProvider clock) {
+ return new DataflowExecutionStateSampler(checkNotNull(clock));
+ }
+
+ public DataflowExecutionStateSampler(MillisProvider clock) {
+ super(clock);
+ }
+
+ @Override
+ public void addTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ this.activeTrackersByWorkId.put(dfTracker.getWorkItemId(), dfTracker);
+ }
+
+ private static Map<String, IntSummaryStatistics> mergeStepStatsMaps(
+ Map<String, IntSummaryStatistics> map1, Map<String, IntSummaryStatistics> map2) {
+ for (Entry<String, IntSummaryStatistics> steps : map2.entrySet()) {
+ map1.compute(
+ steps.getKey(),
+ (k, v) -> {
+ if (v == null) {
+ return steps.getValue();
+ }
+ v.combine(steps.getValue());
+ return v;
+ });
+ }
+ return map1;
+ }
+
+ @Override
+ public void removeTracker(ExecutionStateTracker tracker) {
+ if (!(tracker instanceof DataflowExecutionContext.DataflowExecutionStateTracker)) {
+ return;
+ }
+ DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
+ completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
+
+ // Attribute any remaining time since the last sampling while removing the tracker.
+ //
+ // There is a race condition here; if sampling happens in the time between when we remove the
+ // tracker from activeTrackers and read the lastSampleTicks value, the sampling time will
+ // be lost for the tracker being removed. This is acceptable as sampling is already an
+ // approximation of actual execution time.
+ long millisSinceLastSample = clock.getMillis() - this.lastSampleTimeMillis;
+ if (millisSinceLastSample > 0) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Override
+ public void doSampling(long millisSinceLastSample) {
+ for (DataflowExecutionStateTracker tracker : activeTrackersByWorkId.values()) {
+ tracker.takeSample(millisSinceLastSample);
+ }
+ }
+
+ @Nullable
+ public ActiveMessageMetadata getActiveMessageMetadataForWorkId(String workId) {
Review Comment:
return Optional<ActiveMessageMetadata>, we want to remove `null` from the API whereever it makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421166191
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/ActiveMessageMetadata.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.dataflow.worker;
+
+import com.google.auto.value.AutoValue;
+
+@AutoValue
+public abstract class ActiveMessageMetadata {
+
+ public abstract String userStepName();
+
+ public abstract Long startTime();
Review Comment:
use primitive type here (`long`), unless it can be nullable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421182705
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -111,15 +128,58 @@ public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
if (duration.equals(Duration.ZERO)) {
continue;
}
- list.add(
- Windmill.LatencyAttribution.newBuilder()
- .setState(state)
- .setTotalDurationMillis(duration.getMillis())
- .build());
+ LatencyAttribution.Builder laBuilder = Windmill.LatencyAttribution.newBuilder();
+ if (state == LatencyAttribution.State.ACTIVE) {
+ laBuilder = addActiveLatencyBreakdownToBuilder(isHeartbeat, laBuilder, workId, sampler);
+ }
+ Windmill.LatencyAttribution la =
+ laBuilder.setState(state).setTotalDurationMillis(duration.getMillis()).build();
+ list.add(la);
}
return list;
}
+ private static LatencyAttribution.Builder addActiveLatencyBreakdownToBuilder(
+ boolean isHeartbeat,
+ LatencyAttribution.Builder builder,
+ String workId,
+ DataflowExecutionStateSampler sampler) {
+ if (isHeartbeat) {
+ ActiveLatencyBreakdown.Builder stepBuilder = ActiveLatencyBreakdown.newBuilder();
+ ActiveMessageMetadata activeMessage = sampler.getActiveMessageMetadataForWorkId(workId);
Review Comment:
this returns Optional. can use `!activeMessage.isPresent()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on PR #29592:
URL: https://github.com/apache/beam/pull/29592#issuecomment-1850690629
R: @m-trieu
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1421179427
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java:
##########
@@ -101,7 +117,8 @@ private void recordGetWorkStreamLatencies(
}
}
- public Collection<Windmill.LatencyAttribution> getLatencyAttributions() {
+ public Collection<Windmill.LatencyAttribution> getLatencyAttributions(
Review Comment:
if Collection shouldn't be mutated by caller, use `ImmutableCollection`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416338439
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -245,6 +248,16 @@ public static class DataflowExecutionStateTracker extends ExecutionStateTracker
private final ContextActivationObserverRegistry contextActivationObserverRegistry;
private final String workItemId;
+ /**
+ * Metadata on the message whose processing is currently being managed by this tracker. If no
+ * message is actively being processed, activeMessageMetadata will be null.
+ */
+ private ActiveMessageMetadata activeMessageMetadata = null;
Review Comment:
ef56c0419a649d246e0ee00e927ee821361be943
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "clmccart (via GitHub)" <gi...@apache.org>.
clmccart commented on code in PR #29592:
URL: https://github.com/apache/beam/pull/29592#discussion_r1416337324
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -287,6 +286,8 @@ public class StreamingDataflowWorker {
// Possibly overridden by streaming engine config.
private int maxWorkItemCommitBytes = Integer.MAX_VALUE;
+ private DataflowExecutionStateSampler sampler = DataflowExecutionStateSampler.instance();
Review Comment:
3be2aa33f9753b5bad6e50190986ee3a62763967
we're using it as a singleton for the StreamingDataflowWorker but the base class is not necessarily intended to be a singleton
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Per DoFn latency instrumentation [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29592:
URL: https://github.com/apache/beam/pull/29592#issuecomment-1839558852
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org