You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/04/09 03:57:04 UTC
[GitHub] [beam] ajamato opened a new pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
ajamato opened a new pull request #14490:
URL: https://github.com/apache/beam/pull/14490
[BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java SDK.
- Adds InstructionRequest handler for MonitoringInfosMetadataRequest/Response and HarnessMonitoringInfosRequest/Response (ported from Python SDK) See ProcessWideInstructionHandler
- Adds start times to each MonitoringInfo recorded (ported from Python SDK)
- Adds short id assignment and lookup to MonitoringInfos. (ported from Python SDK)
- Enables the InstructionRequest in the Environment
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] 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://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>SDK</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon">
</a>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img src="https://camo.githubusercontent.com/4565d7b7e907114e6c1d12323408bd903aa252fefed5eeab93701b05c9628a84/68747470733a2f2f63692d6265616d2e6170616368652e6f72672f6a6f622f6265616d5f507265436f6d6d69745f507974686f6e446f636b65725f43726f6e2f62616467652f69636f6e" alt="Build Status" data-canonical-src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img src="https://camo.githubusercontent.com/21afb097a5745440598bee5c59a027b140585eec871c3f3b883200610fabf722/68747470733a2f2f63692d6265616d2e6170616368652e6f72672f6a6f622f6265616d5f507265436f6d6d69745f507974686f6e446f63735f43726f6e2f62616467652f69636f6e" alt="Build Status" data-canonical-src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
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)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-819811878
Run Java PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a change in pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r622465920
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/CounterCell.java
##########
@@ -94,6 +99,11 @@ public MetricName getName() {
return name;
}
+ @Override
+ public @Nullable DateTime getStartTime() {
Review comment:
If we always initialize it, when would it be null?
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java
##########
@@ -74,6 +74,8 @@
protected final @Nullable String stepName;
+ private boolean isProcessWide = false;
Review comment:
Let's make this final. We should know at instantiation time whether it's process wide (given that the step name above is final).
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/CounterCell.java
##########
@@ -45,6 +48,8 @@
*/
public CounterCell(MetricName name) {
this.name = name;
+ // The start time of when this cell was first instantied by the container.
+ this.startTime = new DateTime(DateTimeZone.UTC);
Review comment:
Is it a concern that this might be slow?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
##########
@@ -262,6 +265,8 @@ public static void main(
}
});
+ MetricsEnvironment.setProcessWideContainer(new MetricsContainerImpl(null));
Review comment:
Given that we don't hold onto a reference for this, why don't we have MetricsEnvironment create (if possibly lazily) the process-wide container?
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java
##########
@@ -84,7 +86,10 @@
private MetricsMap<KV<MetricName, HistogramData.BucketType>, HistogramCell> histograms =
new MetricsMap<>(HistogramCell::new);
- /** Create a new {@link MetricsContainerImpl} associated with the given {@code stepName}. */
+ /**
+ * Create a new {@link MetricsContainerImpl} associated with the given {@code stepName}. If
+ * stepName is null, this MetricsContainer is not bound to a step.
Review comment:
Alternatively, require this be non-null, and add a no-arg constructor for the no-step-name (presumably is-process-wide) 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on a change in pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r631541991
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/CounterCell.java
##########
@@ -45,6 +48,8 @@
*/
public CounterCell(MetricName name) {
this.name = name;
+ // The start time of when this cell was first instantied by the container.
+ this.startTime = new DateTime(DateTimeZone.UTC);
Review comment:
Moving this to https://github.com/apache/beam/pull/14805
Yes, revised to only make this call once per process in the new 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on a change in pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r631541707
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java
##########
@@ -84,7 +86,10 @@
private MetricsMap<KV<MetricName, HistogramData.BucketType>, HistogramCell> histograms =
new MetricsMap<>(HistogramCell::new);
- /** Create a new {@link MetricsContainerImpl} associated with the given {@code stepName}. */
+ /**
+ * Create a new {@link MetricsContainerImpl} associated with the given {@code stepName}. If
+ * stepName is null, this MetricsContainer is not bound to a step.
Review comment:
Moving this to https://github.com/apache/beam/pull/14805
Done in the new PR
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java
##########
@@ -74,6 +74,8 @@
protected final @Nullable String stepName;
+ private boolean isProcessWide = false;
Review comment:
Moving this to https://github.com/apache/beam/pull/14805
Done in the new 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on a change in pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r615111991
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricCell.java
##########
@@ -36,4 +38,9 @@
/** Reset this metric. */
void reset();
+
+ /** Return the cumulative values for any metrics in this container as MonitoringInfos. */
+ default @Nullable DateTime getStartTime() {
+ return null;
Review comment:
I didn't want to add this to all the metrics, because I didn't want to add unncessary calls to the system to get clock times.
Let's add it to just this place first, and consider adding more if they are needed for another project.
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessWideInstructionHandler.java
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.fn.harness.control;
+
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+
+/**
+ * Processes {@link BeamFnApi.InstructionRequest}'s and {@link
+ * BeamFnApi.MonitoringInfosMetadataRequest} and {@link BeamFnApi.HarnessMonitoringInfosResponse}
+ *
+ * <p>These instructions are not associated with the currently processed bundles. They return
+ * MonitoringInfos payloads and MonitoringInfo metadata for "process-wide" metrics, which return
+ * metric values calculated over the life of the process.
+ */
+public class ProcessWideInstructionHandler {
Review comment:
Renamed to HarnessMonitoringInfosInstructionHandler, which is all this does now.
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java
##########
@@ -302,7 +325,7 @@ public void commitUpdates() {
ImmutableList.Builder<MetricUpdate<UpdateT>> updates = ImmutableList.builder();
for (Map.Entry<MetricName, CellT> cell : cells.entries()) {
UpdateT update = checkNotNull(cell.getValue().getCumulative());
- updates.add(MetricUpdate.create(MetricKey.create(stepName, cell.getKey()), update));
+ updates.add(MetricUpdate.create(MetricKey.create(stepName, cell.getKey()), update, null));
Review comment:
Done
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricCell.java
##########
@@ -36,4 +38,9 @@
/** Reset this metric. */
void reset();
+
+ /** Return the cumulative values for any metrics in this container as MonitoringInfos. */
Review comment:
Update the comment
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java
##########
@@ -54,4 +55,15 @@
default Histogram getHistogram(MetricName metricName, HistogramData.BucketType bucketType) {
throw new RuntimeException("Histogram metric is not supported yet.");
}
+
+ /** Return the cumulative values for any metrics in this container as MonitoringInfos. */
+ default Iterable<MetricsApi.MonitoringInfo> getMonitoringInfos() {
+ throw new RuntimeException("getMonitoringInfos is not implemented on this MetricsContainer.");
+ }
+
+ default boolean isProcessWide() {
+ return false;
+ }
+
+ default void setIsProcessWide(boolean procesWide) {}
Review comment:
Done
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/MonitoringInfoShortIdCache.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.fn.harness.control;
+
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.runners.core.metrics.MonitoringInfoMetricName;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+
+public class MonitoringInfoShortIdCache implements AutoCloseable {
+
+ private int lastShortId = 0;
+
+ private HashMap<MonitoringInfoMetricName, String> infoKeyToShortId = new HashMap<>();
+
+ private HashMap<String, MetricsApi.MonitoringInfo> shortIdToInfo = new HashMap<>();
+
Review comment:
This file was deleted. As @robertwb checked in a different implemetnation in the last few days.
##########
File path: runners/jet/src/main/java/org/apache/beam/runners/jet/metrics/JetMetricsContainer.java
##########
@@ -98,7 +98,8 @@ public void flush(boolean async) {
UpdateT value = cell.getValue();
if (value != null) {
MetricKey key = MetricKey.create(stepName, cell.getName());
- MetricUpdates.MetricUpdate<UpdateT> update = MetricUpdates.MetricUpdate.create(key, value);
+ MetricUpdates.MetricUpdate<UpdateT> update =
+ MetricUpdates.MetricUpdate.create(key, value, null);
Review comment:
No, this is actually now the same MetricCell class. so I can't pull a startTime off it.
its actually a CellT extends AbstractMetric<UpdateT>
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessWideInstructionHandler.java
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.fn.harness.control;
+
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+
+/**
+ * Processes {@link BeamFnApi.InstructionRequest}'s and {@link
+ * BeamFnApi.MonitoringInfosMetadataRequest} and {@link BeamFnApi.HarnessMonitoringInfosResponse}
+ *
+ * <p>These instructions are not associated with the currently processed bundles. They return
+ * MonitoringInfos payloads and MonitoringInfo metadata for "process-wide" metrics, which return
+ * metric values calculated over the life of the process.
+ */
+public class ProcessWideInstructionHandler {
+ public ProcessWideInstructionHandler() {}
+
+ public static BeamFnApi.InstructionResponse.Builder monitoringInfoMetadata(
+ BeamFnApi.InstructionRequest request) {
+ BeamFnApi.MonitoringInfosMetadataResponse.Builder response =
+ BeamFnApi.MonitoringInfosMetadataResponse.newBuilder();
+ response.putAllMonitoringInfo(
+ MonitoringInfoShortIdCache.getShortIdCache()
+ .getInfos(request.getMonitoringInfos().getMonitoringInfoIdList()));
Review comment:
That wasn't implemented when I sent you this PR. Though robert's change did introduce it.
I have updated the PR to merge with his implementation.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on a change in pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r631541888
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/CounterCell.java
##########
@@ -94,6 +99,11 @@ public MetricName getName() {
return name;
}
+ @Override
+ public @Nullable DateTime getStartTime() {
Review comment:
Moving this to https://github.com/apache/beam/pull/14805
I don't want to make the system call to get the time for every metric as you suggested due to performance concerns. So I make only the process wide MetricContainer make this call. So it should be called once for the process.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-840272513
i have split this up into two PRs, rather than 3. As the startTime logic is much simpler now, and only done for the process wide MetricContainer. Overall the whole thing is much smaller. PTAL
(1) Change the way ShortIdMap works,
https://github.com/apache/beam/pull/14804
(2) Add process-wide metrics, and add the notion of startTime to a subset of metrics.
https://github.com/apache/beam/pull/14805
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-816383286
R: @pabloem
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-827118020
Run Java_Examples_Dataflow PreCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r622594151
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ShortIdMap.java
##########
@@ -17,29 +17,46 @@
*/
package org.apache.beam.runners.core.metrics;
+import java.util.HashMap;
+import java.util.Map;
import java.util.NoSuchElementException;
import org.apache.beam.model.pipeline.v1.MetricsApi.MonitoringInfo;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
-import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.BiMap;
-import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.HashBiMap;
/** A Class for registering short ids for MonitoringInfos. */
public class ShortIdMap {
private int counter = 0;
- private BiMap<String, MonitoringInfo> monitoringInfoMap = HashBiMap.create();
+ // We must use two maps instead of a BiMap because the two maps store different MonitoringInfos.
+ // The values in shortIdToInfo contain the MonitoringInfo with only the payload cleared.
+ private Map<String, MonitoringInfo> shortIdToInfo = new HashMap<>();
+ // The keys in cleanedInfoToShortId contain a cleaned MonitoringInfo after clearing
+ // the payload, startTime and type.
+ private Map<MonitoringInfo, String> cleanedInfoToShortId = new HashMap<>();
public synchronized String getOrCreateShortId(MonitoringInfo info) {
Preconditions.checkNotNull(info);
- String shortId = monitoringInfoMap.inverse().get(info);
+ // Remove the payload, startTime and typeUrn before using the MonitoringInfo as a key.
+ // As only the URN+labels uniquely identify the MonitoringInfo
+ MonitoringInfo.Builder cleaner = MonitoringInfo.newBuilder(info);
+ cleaner.clearPayload();
+ cleaner.clearStartTime();
+ cleaner.clearType();
Review comment:
isn't the type also necessary to identify a monitoring info? What identifies it / what goes into the hash? Is it hash consistent?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-819808672
sorry about the delay. taking a look..
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-821422980
R: @robertwb
Looks like you had added the ShortID code in this PR
https://github.com/apache/beam/commit/4dd45e05e27bb7ae8430cd956d96426fc95a082e
I will take a look at your change, and try to merge things together.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato closed pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato closed pull request #14490:
URL: https://github.com/apache/beam/pull/14490
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r613568006
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricCell.java
##########
@@ -36,4 +38,9 @@
/** Reset this metric. */
void reset();
+
+ /** Return the cumulative values for any metrics in this container as MonitoringInfos. */
+ default @Nullable DateTime getStartTime() {
+ return null;
Review comment:
Does it make sense to not add a default, and make this not-nullable? (i.e. force all implementers to define a non-null start time?)
##########
File path: runners/jet/src/main/java/org/apache/beam/runners/jet/metrics/JetMetricsContainer.java
##########
@@ -98,7 +98,8 @@ public void flush(boolean async) {
UpdateT value = cell.getValue();
if (value != null) {
MetricKey key = MetricKey.create(stepName, cell.getName());
- MetricUpdates.MetricUpdate<UpdateT> update = MetricUpdates.MetricUpdate.create(key, value);
+ MetricUpdates.MetricUpdate<UpdateT> update =
+ MetricUpdates.MetricUpdate.create(key, value, null);
Review comment:
does it make sense to get the time from the cell?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/MonitoringInfoShortIdCache.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.fn.harness.control;
+
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.runners.core.metrics.MonitoringInfoMetricName;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+
+public class MonitoringInfoShortIdCache implements AutoCloseable {
+
+ private int lastShortId = 0;
+
+ private HashMap<MonitoringInfoMetricName, String> infoKeyToShortId = new HashMap<>();
+
+ private HashMap<String, MetricsApi.MonitoringInfo> shortIdToInfo = new HashMap<>();
+
+ private static final AtomicReference<MonitoringInfoShortIdCache> SHORT_ID_CACHE =
+ new AtomicReference<>(new MonitoringInfoShortIdCache());
+
+ private MonitoringInfoShortIdCache() {}
+
+ /** Return the {@link MetricsContainer} for the current process. */
+ public static MonitoringInfoShortIdCache getShortIdCache() {
+ return SHORT_ID_CACHE.get();
+ }
+
+ /**
+ * Initialize and assign the {@link MonitoringInfoShortIdCache} to the current process. Usage: try
+ * (MonitoringInfoShortIdCache cache = MonitoringInfoShortIdCache.newShortIdCache()) { ; } catch
+ * (Exception e) { // Handle error when AutoClosure calls close() }
+ *
+ * <p>Instead of setting this statically, we require that it be initialized. This allows tests to
+ * set and clear the ShortIdCache appropriately.
+ *
+ * @return The ShortIdCache for the current process.
+ */
+ public static MonitoringInfoShortIdCache newShortIdCache() {
+ return SHORT_ID_CACHE.getAndSet(new MonitoringInfoShortIdCache());
Review comment:
maybe we should note that this returns the previous value, not the new one, right? perhaps it's worth not returning anything, or renaming the function to convey that it returns the previous cache?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessWideInstructionHandler.java
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.fn.harness.control;
+
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+
+/**
+ * Processes {@link BeamFnApi.InstructionRequest}'s and {@link
+ * BeamFnApi.MonitoringInfosMetadataRequest} and {@link BeamFnApi.HarnessMonitoringInfosResponse}
+ *
+ * <p>These instructions are not associated with the currently processed bundles. They return
+ * MonitoringInfos payloads and MonitoringInfo metadata for "process-wide" metrics, which return
+ * metric values calculated over the life of the process.
+ */
+public class ProcessWideInstructionHandler {
Review comment:
maybe rename this to reference MonitoringInfoMetadataInstructionHandler or something like that? Since it has process-wide and non-process-wide MIs
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricCell.java
##########
@@ -36,4 +38,9 @@
/** Reset this metric. */
void reset();
+
+ /** Return the cumulative values for any metrics in this container as MonitoringInfos. */
Review comment:
is this docstring right?
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java
##########
@@ -54,4 +55,15 @@
default Histogram getHistogram(MetricName metricName, HistogramData.BucketType bucketType) {
throw new RuntimeException("Histogram metric is not supported yet.");
}
+
+ /** Return the cumulative values for any metrics in this container as MonitoringInfos. */
+ default Iterable<MetricsApi.MonitoringInfo> getMonitoringInfos() {
+ throw new RuntimeException("getMonitoringInfos is not implemented on this MetricsContainer.");
+ }
+
+ default boolean isProcessWide() {
+ return false;
+ }
+
+ default void setIsProcessWide(boolean procesWide) {}
Review comment:
if setting does not do anything, maybe we should throw an error instead of ignoring the instruction silently?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/MonitoringInfoShortIdCache.java
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.fn.harness.control;
+
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.runners.core.metrics.MonitoringInfoMetricName;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+
+public class MonitoringInfoShortIdCache implements AutoCloseable {
+
+ private int lastShortId = 0;
+
+ private HashMap<MonitoringInfoMetricName, String> infoKeyToShortId = new HashMap<>();
+
+ private HashMap<String, MetricsApi.MonitoringInfo> shortIdToInfo = new HashMap<>();
+
Review comment:
I wonder if these should use a Guava cache with automatic invalidation and so on?
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessWideInstructionHandler.java
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.fn.harness.control;
+
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+
+/**
+ * Processes {@link BeamFnApi.InstructionRequest}'s and {@link
+ * BeamFnApi.MonitoringInfosMetadataRequest} and {@link BeamFnApi.HarnessMonitoringInfosResponse}
+ *
+ * <p>These instructions are not associated with the currently processed bundles. They return
+ * MonitoringInfos payloads and MonitoringInfo metadata for "process-wide" metrics, which return
+ * metric values calculated over the life of the process.
+ */
+public class ProcessWideInstructionHandler {
+ public ProcessWideInstructionHandler() {}
+
+ public static BeamFnApi.InstructionResponse.Builder monitoringInfoMetadata(
+ BeamFnApi.InstructionRequest request) {
+ BeamFnApi.MonitoringInfosMetadataResponse.Builder response =
+ BeamFnApi.MonitoringInfosMetadataResponse.newBuilder();
+ response.putAllMonitoringInfo(
+ MonitoringInfoShortIdCache.getShortIdCache()
+ .getInfos(request.getMonitoringInfos().getMonitoringInfoIdList()));
Review comment:
It seems that the only moment when a MInfo gets added to the ShortIdCache is when we call `getShortId` on it, right? I don't see `getShortId` getting called for the `monitoringInfoMetadata`, only for the `harnessMonitoringInfos` - so I wonder how do we get the monitoring infos from the other containers in this call?
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerImpl.java
##########
@@ -302,7 +325,7 @@ public void commitUpdates() {
ImmutableList.Builder<MetricUpdate<UpdateT>> updates = ImmutableList.builder();
for (Map.Entry<MetricName, CellT> cell : cells.entries()) {
UpdateT update = checkNotNull(cell.getValue().getCumulative());
- updates.add(MetricUpdate.create(MetricKey.create(stepName, cell.getKey()), update));
+ updates.add(MetricUpdate.create(MetricKey.create(stepName, cell.getKey()), update, null));
Review comment:
maybe the start time should be the current time? or maybe not?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ajamato commented on a change in pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
ajamato commented on a change in pull request #14490:
URL: https://github.com/apache/beam/pull/14490#discussion_r631541650
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
##########
@@ -262,6 +265,8 @@ public static void main(
}
});
+ MetricsEnvironment.setProcessWideContainer(new MetricsContainerImpl(null));
Review comment:
Moving this to https://github.com/apache/beam/pull/14805
Unfortunately I could not make MetricsEnvironment instantiate a MetricsContainerImpl inside itself as this would create a dep cycle between their packages. So keeping this one as is.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-828836493
I'm wondering about the value of attaching startTime to each metric. It seems to add a lot of complexity and is a bit ill-defined in the Beam model. Are these tied somehow to process-wide metrics? If so, could we just use the start time of the process itself when publishing these rather than adding the overhead of tracking it per-metric?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #14490: DO NOT SUBMIT SPLIT INTO OTHER PRsAdd HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-840771624
Thanks.
I'm still not following why we need per-metric start times. And
without start times, it seems the (payload-less) BiMap should just work
which simplifies things even more.
On Wed, May 12, 2021 at 8:47 PM Alex Amato ***@***.***> wrote:
> i have split this up into two PRs, rather than 3. As the startTime logic
> is much simpler now, and only done for the process wide MetricContainer.
> Overall the whole thing is much smaller. PTAL
>
> (1) Change the way ShortIdMap works,
> #14804 <https://github.com/apache/beam/pull/14804>
> (2) Add process-wide metrics, and add the notion of startTime to a subset
> of metrics.
> #14805 <https://github.com/apache/beam/pull/14805>
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/beam/pull/14490#issuecomment-840272513>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AADWVAKMZQS67DO5YM5FCPTTNNDWNANCNFSM42UFLLAQ>
> .
>
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #14490: [BEAM-11994] Add HarnessMonitoringInfosRequest/Response InstructionRequest and MonitoringInfosMetadataRequest/Response support to Java
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #14490:
URL: https://github.com/apache/beam/pull/14490#issuecomment-828712183
Run Java PreCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org