You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "jto (via GitHub)" <gi...@apache.org> on 2023/09/22 11:59:39 UTC
[GitHub] [beam] jto opened a new pull request, #28609: Add MetricsContainer support to the Flink sources.
jto opened a new pull request, #28609:
URL: https://github.com/apache/beam/pull/28609
This is an updated version of the abandonned https://github.com/apache/beam/pull/25753.
I simply rebased master in the original PR and reviewed the code.
fixes https://github.com/apache/beam/issues/25741
------------------------
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1747248217
clicked to rerun the tests, maybe a flake?
--
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
[GitHub] [beam] github-actions[bot] commented on pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1736955987
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @johnjcasey added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "jto (via GitHub)" <gi...@apache.org>.
jto commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1748356222
Thanks @tvalentyn. It failed with the same error so I guess it's deterministic...
--
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
[GitHub] [beam] jto commented on a diff in pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "jto (via GitHub)" <gi...@apache.org>.
jto commented on code in PR #28609:
URL: https://github.com/apache/beam/pull/28609#discussion_r1338768446
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * 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.flink.metrics;
+
+import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.sdk.metrics.DistributionResult;
+import org.apache.beam.sdk.metrics.GaugeResult;
+import org.apache.beam.sdk.metrics.MetricKey;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+
+/**
+ * The base helper class for holding a {@link MetricsContainerImpl} and forwarding Beam metrics to
+ * Flink accumulators and metrics. The two subclasses of this base class are {@link
+ * FlinkMetricContainer} and {@link FlinkMetricContainerWithoutAccumulator}. The former is used when
+ * {@link org.apache.flink.api.common.functions.RuntimeContext Flink RuntimeContext} is available.
+ * The latter is used otherwise.
+ */
+abstract class FlinkMetricContainerBase {
+
+ private static final String METRIC_KEY_SEPARATOR =
+ GlobalConfiguration.loadConfiguration().getString(MetricOptions.SCOPE_DELIMITER);
+
+ protected final MetricsContainerStepMap metricsContainers;
+ private final Map<String, Counter> flinkCounterCache;
+ private final Map<String, FlinkDistributionGauge> flinkDistributionGaugeCache;
+ private final Map<String, FlinkGauge> flinkGaugeCache;
+
+ public FlinkMetricContainerBase() {
+ this.flinkCounterCache = new HashMap<>();
+ this.flinkDistributionGaugeCache = new HashMap<>();
+ this.flinkGaugeCache = new HashMap<>();
+ this.metricsContainers = new MetricsContainerStepMap();
+ }
+
+ protected abstract MetricGroup getMetricGroup();
Review Comment:
Hey!
Thanks for reviewing this PR @kennknowles.
I fixed the container implementations :)
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1760184363
@jto do you prefer to merge it or you'd like to work more on https://github.com/apache/beam/commit/4296688a3caaf31910fb3dd804acc8910ffd2f7d ? Perhaps we should move out that commit into a separate PR to make sure we don't make the situation worse in this PR.
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1777090213
Reminder, please take a look at this pr: @johnjcasey
--
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
[GitHub] [beam] github-actions[bot] commented on pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1731386135
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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
[GitHub] [beam] kennknowles commented on a diff in pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #28609:
URL: https://github.com/apache/beam/pull/28609#discussion_r1338659420
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * 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.flink.metrics;
+
+import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.sdk.metrics.DistributionResult;
+import org.apache.beam.sdk.metrics.GaugeResult;
+import org.apache.beam.sdk.metrics.MetricKey;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.configuration.MetricOptions;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+
+/**
+ * The base helper class for holding a {@link MetricsContainerImpl} and forwarding Beam metrics to
+ * Flink accumulators and metrics. The two subclasses of this base class are {@link
+ * FlinkMetricContainer} and {@link FlinkMetricContainerWithoutAccumulator}. The former is used when
+ * {@link org.apache.flink.api.common.functions.RuntimeContext Flink RuntimeContext} is available.
+ * The latter is used otherwise.
+ */
+abstract class FlinkMetricContainerBase {
+
+ private static final String METRIC_KEY_SEPARATOR =
+ GlobalConfiguration.loadConfiguration().getString(MetricOptions.SCOPE_DELIMITER);
+
+ protected final MetricsContainerStepMap metricsContainers;
+ private final Map<String, Counter> flinkCounterCache;
+ private final Map<String, FlinkDistributionGauge> flinkDistributionGaugeCache;
+ private final Map<String, FlinkGauge> flinkGaugeCache;
+
+ public FlinkMetricContainerBase() {
+ this.flinkCounterCache = new HashMap<>();
+ this.flinkDistributionGaugeCache = new HashMap<>();
+ this.flinkGaugeCache = new HashMap<>();
+ this.metricsContainers = new MetricsContainerStepMap();
+ }
+
+ protected abstract MetricGroup getMetricGroup();
Review Comment:
Just a coding style comment that this looks a lot like it should just be a constructor parameter. (I am making the standard "favor composition over inheritance" comment since it looks very straightforward in this case)
--
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
[GitHub] [beam] jto commented on pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "jto (via GitHub)" <gi...@apache.org>.
jto commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1731345136
The failing test is running fine on my machine. I think the test is just flaky. It would probably succeed if someone just re-triggers 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
[GitHub] [beam] jto commented on pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "jto (via GitHub)" <gi...@apache.org>.
jto commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1736952893
assign set of reviewers
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1760171874
waiting on author
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1759488339
Reminder, please take a look at this pr: @johnjcasey
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn merged PR #28609:
URL: https://github.com/apache/beam/pull/28609
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1760181675
Actually, looking at the test execution, it passed:
```
BUILD SUCCESSFUL in 31s
103 actionable tasks: 64 executed, 39 from cache
Publishing build scan...
https://ge.apache.org/s/a7mu5k567st5o
```
and fails later. I am fine with filing a separate issue about test reporting, and merge this.
--
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] Add MetricsContainer support to the Flink sources. [beam]
Posted by "jto (via GitHub)" <gi...@apache.org>.
jto commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1765118575
@tvalentyn thanks!
I don't really know how to fix the build failure. It indeed fails when publishing the build results while all tests are green. I don't know why this PR in particular breaks the build. Seems to be something about the size of an XML file so if you wish to merge this and address the issue in another PR, please feel free to do it :) I'm sure someone more familiar with the build itself will know how to fix this.
--
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
[GitHub] [beam] jto commented on pull request #28609: Add MetricsContainer support to the Flink sources.
Posted by "jto (via GitHub)" <gi...@apache.org>.
jto commented on PR #28609:
URL: https://github.com/apache/beam/pull/28609#issuecomment-1739131758
@kennknowles I made the fix you requested. The test fails when publishing the results with the following error:
```
Error: Error processing result file: CData section too big found, line 92832, column 19 (TEST-org.apache.beam.sdk.transforms.ViewTest.xml, line 92832)
```
It seems unrelated to this change. Would you happen to know if I can do anything to fix this problem ?
--
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