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/10/04 23:14:00 UTC
[GitHub] [beam] riteshghorse opened a new pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
riteshghorse opened a new pull request #15657:
URL: https://github.com/apache/beam/pull/15657
DoFn metrics implementation in Go SDK.
------------------------
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).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</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>---</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>
<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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</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_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?subject=V1">
</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?subject=V1+Streaming">
</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?subject=V1+Java+11">
</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?subject=V2">
</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?subject=V2+Streaming">
</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?subject=Java+8">
</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?subject=Java+11">
</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?subject=Portable">
</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?subject=Portable+Streaming">
</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><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</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?subject=Portable">
</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?subject=Structured+Streaming">
</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>---</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?subject=V1">
</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?subject=V2">
</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?subject=ValCont">
</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?subject=Portable">
</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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</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>
<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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</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>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</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>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<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?subject=V1">
</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?subject=V1+Java11">
</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?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<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_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_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</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?subject=3.7">
</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?subject=3.8">
</a>
</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?subject=Tests">
</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?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731375722
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -174,36 +174,35 @@ type executionTracker struct {
numberOfTransitions int64
millisSinceLastTransition time.Duration
transitionsAtLastSample int64
-}
-
-// executionState is used to store as atomic.Value in Store.
-type pTransformState struct {
- pid string
- state bundleProcState
-}
-
-func newPTransformState(pid string, state bundleProcState) *pTransformState {
- return &pTransformState{pid: pid, state: state}
+ currentState bundleProcState
+ pid string
}
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
- ps := newPTransformState(bctx.ptransformID, state)
- bctx.pStore.Store(ps)
- atomic.AddInt64(&bctx.transitions, 1)
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
Review comment:
`allocs/op` changed significantly (from 2 to 12) when I used atomic.Value to store and load BundleState into it. `MillisSinceLastTransition` and `TransitionsAtLastSample` are stored in `StateSampler`. Also, the time per loop changed from 2566 ns/op to 3376 ns/op.
Did this benchmarking without pprof. Now I'll do it with pprof.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738462955
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -136,6 +136,10 @@ func SetPTransformID(ctx context.Context, id string) context.Context {
// Checking for *beamCtx is an optimization, so we don't dig deeply
// for ids if not necessary.
if bctx, ok := ctx.(*beamCtx); ok {
+ if _, ok := bctx.store.stateRegistry[id]; !ok {
+ bctx.store.stateRegistry[id] = [4]*ExecutionState{&ExecutionState{}, &ExecutionState{}, &ExecutionState{}, &ExecutionState{}}
Review comment:
Yes. I mean what if we changed the type?
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954093792
Made the required changes. PTAL.
--
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] lostluck commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-942779836
Sgtm. I'll focus my next review pass on the metrics package and see what's going on with the error you're seeing.
--
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] lostluck commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954209874
Run Go Flink ValidatesRunner
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954178296
PTAL. Missed the UrnToString last time.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731374488
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -30,60 +30,24 @@ func NewSampler(ctx context.Context, store *Store) StateSampler {
return StateSampler{store: store}
}
-func initialize() [4]*ExecutionState {
- var v [4]*ExecutionState
- for i := 0; i < 4; i++ {
- v[i] = &ExecutionState{}
- }
- return v
-}
-
func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadPTransformState(ctx)
- pid := PTransformLabels(ps.pid)
-
+ ps := loadCurrentState(ctx)
s.store.mu.Lock()
defer s.store.mu.Unlock()
- if _, ok := s.store.stateRegistry[pid]; !ok {
- s.store.stateRegistry[pid] = initialize()
- }
- if v, ok := s.store.stateRegistry[pid]; ok {
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
v[ps.state].TotalTime += t
v[TotalBundle].TotalTime += t
- }
- if _, ok := s.store.executionStore[pid]; !ok {
- s.store.executionStore[pid] = &executionTracker{}
- }
- if v, ok := s.store.executionStore[pid]; ok {
+ e := s.store.executionStore
- if v.transitionsAtLastSample != ps.transitions {
- // state change
- v.millisSinceLastTransition = 0
- v.numberOfTransitions = ps.transitions
- v.transitionsAtLastSample = ps.transitions
+ if e.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ e.millisSinceLastTransition = 0
+ e.numberOfTransitions = ps.transitions
Review comment:
`loadCurrentState(ctx)` reads the `numberOfTransitions ` itself. So we need `transitionsAtLastSample ` somewhere to track the transitions change. After restructuring, the sampler can have this field on its side instead of storing it in the `metrics.Store`
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r728629054
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,63 @@
+// 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 metrics
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSampler(t *testing.T) {
+ ctx := context.Background()
+ pctx := SetPTransformID(ctx, "transform")
+ st := GetStore(pctx)
+
+ s := NewSampler(pctx, st)
+
+ SetPTransformState(pctx, "transform", StartBundle)
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ SetPTransformState(pctx, "transform", ProcessBundle)
+ IncTransition(pctx)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ SetPTransformState(pctx, "transform", FinishBundle)
+ IncTransition(pctx)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Stop(pctx, 200*time.Millisecond)
Review comment:
Why is stop taking a increment value? We don't know what the time difference has been since the last sample call. It's OK if we lose a bit of time at the end. It's not a precise sampler.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,63 @@
+// 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 metrics
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSampler(t *testing.T) {
+ ctx := context.Background()
+ pctx := SetPTransformID(ctx, "transform")
+ st := GetStore(pctx)
+
+ s := NewSampler(pctx, st)
+
+ SetPTransformState(pctx, "transform", StartBundle)
Review comment:
Since `pctx` has had the TransformID set, then it knows that it's the Ptransform's "transform"'s context, and SetPTransformState doesn't need to have the transform ID passed in on every single call. It's already being passed, in the context.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,63 @@
+// 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 metrics
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSampler(t *testing.T) {
Review comment:
Here's some pseudo code for how a basic test should look. The goal is a simple straightforward test case. Make it as simple as possible, without sleeps or starting go routines. Don't validate correctness with concurrency if you don't have to. You do not have to in this case.
The point is the test is being all 3 involved threads. The one setting the state (the SetState calls), the one sampling the state (the Sample calls), and the one reading out the metrics for export (or whatever other internal details we can look at for the purpose of this test).
Copy paste this, make it real code, make it pass.
```
1. Create a initialized "bundle" context.
2. Create a sampler object s, from the bundle context.
3. Create a "PTransform" context for pid "transform", based on the bundle context.
// Here the ptransform start bundle call happened, and it's taken a bit of time, so a sample is taken during StartBundle
4. SetState(pctx, StartBundle)
5. s.Sample(200ms)
// Validate that that works properly.
6.a Validate that the StartBundle state for pid "transform" has 200 ms.
6.b Validate that the ProcessBundle state for pid "transform" has 0 ms.
6.c Validate that the FinishBundle state for pid "transform" has 0 ms.
6.d Validate that the TotalProcessingTime state for "pid" transform" has 200 ms.
6.e Validate that the number of transitions so far is 1.
// Here the PTransform is processing a few elements gets sampled during it a couple of them.
7. SetState(pctx, Process)
8. s.Sample(200ms)
9. SetState(pctx, Process)
10. SetState(pctx, Process)
11. s.Sample(200ms)
12.a Validate that the StartBundle state for pid "transform" has 200 ms.
12.b Validate that the ProcessBundle state for pid "transform" has 400 ms.
12.c Validate that the FinishBundle state for pid "transform" has 0 ms.
12.d Validate that the TotalProcessingTime state for "pid" transform" has 600 ms.
12.e Validate that the number of transitions so far is 4.
12.f Validate that the the "time since last transition" 0.
// Pretend the ptransform is busy with a single element.
13. s.Sample(200ms)
14. s.Sample(200ms)
15. s.Sample(200ms)
// Check that additional samples have accumulated while the PTransform was "busy" with one element.
16.a Validate that the StartBundle state for pid "transform" has 200 ms.
16.b Validate that the ProcessBundle state for pid "transform" has 1000 ms.
16.c Validate that the FinishBundle state for pid "transform" has 0 ms.
16.d Validate that the TotalProcessingTime state for "pid" transform" has 1200 ms.
16.e Validate that the number of transitions so far is 4.
16.f Validate that the the "time since last transition" 600ms.
17. SetState(pctx, FinishBundle)
// Check that the final metric readout matches what we expect.
18.a Validate that the StartBundle state for pid "transform" has 200 ms.
18.b Validate that the ProcessBundle state for pid "transform" has 1000 ms.
18.c Validate that the FinishBundle state for pid "transform" has 0 ms.
18.d Validate that the TotalProcessingTime state for "pid" transform" has 1200 ms.
18.e Validate that the number of transitions so far is 5.
18.f Validate that the the "time since last transition" 0ms.
// Nothing needs to be done here since sampling has "stopped" and the final metrics have been read out.
```
As always, if you aren't sure about something, ask.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,63 @@
+// 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 metrics
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSampler(t *testing.T) {
+ ctx := context.Background()
+ pctx := SetPTransformID(ctx, "transform")
+ st := GetStore(pctx)
+
+ s := NewSampler(pctx, st)
+
+ SetPTransformState(pctx, "transform", StartBundle)
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
Review comment:
In this test, the point is to never actually sleep, and never actually spin off new goroutines. Do everything in a single linear thread. No asynchrony. Do not call time.Sleep in this test. If you are, something has gone wrong.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,63 @@
+// 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 metrics
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSampler(t *testing.T) {
+ ctx := context.Background()
+ pctx := SetPTransformID(ctx, "transform")
+ st := GetStore(pctx)
+
+ s := NewSampler(pctx, st)
+
+ SetPTransformState(pctx, "transform", StartBundle)
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ time.Sleep(200 * time.Millisecond)
+ go s.Start(pctx, 200*time.Millisecond)
+
+ SetPTransformState(pctx, "transform", ProcessBundle)
+ IncTransition(pctx)
Review comment:
I've mentioned this before. Setting the state should increment the transitions at the same time. One call should do both thing. Every time the state is set, *that* is a transition and must be counted.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738573767
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -136,6 +136,10 @@ func SetPTransformID(ctx context.Context, id string) context.Context {
// Checking for *beamCtx is an optimization, so we don't dig deeply
// for ids if not necessary.
if bctx, ok := ctx.(*beamCtx); ok {
+ if _, ok := bctx.store.stateRegistry[id]; !ok {
+ bctx.store.stateRegistry[id] = [4]*ExecutionState{&ExecutionState{}, &ExecutionState{}, &ExecutionState{}, &ExecutionState{}}
Review comment:
Ah, I see what you mean. Then make the stateRegistry field type `map[string]*[4]ExecutionState` Making it a pointer to the array, rather than an array of pointers.
Then this line can become `bctx.store.stateRegistry[id] = &[4]ExecutionState{}`
It's a map, keyed by strings, to pointers of arrays of length 4.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738654358
##########
File path: sdks/go/pkg/beam/core/runtime/exec/pardo.go
##########
@@ -197,6 +202,7 @@ func mustExplodeWindows(fn *funcx.Fn, elm *FullValue, usesSideInput bool) bool {
// FinishBundle does post-bundle processing operations for the DoFn.
// Note: This is not a "FinalizeBundle" operation. Data is not yet durably
// persisted at this point.
+
Review comment:
rm spare line. It disconnects the Go Doc comment from the method.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/sampler.go
##########
@@ -0,0 +1,48 @@
+// 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 harness
+
+import (
+ "context"
+ "time"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics"
+)
+
+type stateSampler struct {
+ done chan (int)
+ sampler metrics.StateSampler
+}
+
+func newSampler(store *metrics.Store) *stateSampler {
+ return &stateSampler{sampler: metrics.NewSampler(store), done: make(chan int)}
+}
+
+func (s *stateSampler) start(ctx context.Context, t time.Duration) {
+ for {
+ select {
+ case <-s.done:
+ return
+ default:
+ s.sampler.Sample(ctx, t)
+ time.Sleep(20 * time.Millisecond)
Review comment:
Use `t` here, instead of a constant value.
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -51,42 +53,65 @@ func groupByType(minfos []*pipepb.MonitoringInfo) (
}
r := bytes.NewReader(minfo.GetPayload())
-
- switch minfo.GetType() {
- case "beam:metrics:sum_int64:v1":
+ switch minfo.GetUrn() {
+ case "beam:metric:user:sum_int64:v1":
value, err := extractCounterValue(r)
if err != nil {
log.Println(err)
continue
}
counters[key] = value
- case "beam:metrics:distribution_int64:v1":
+ case "beam:metric:user:distribution_int64:v1":
value, err := extractDistributionValue(r)
if err != nil {
log.Println(err)
continue
}
distributions[key] = value
case
- "beam:metrics:latest_int64:v1",
- "beam:metrics:top_n_int64:v1",
- "beam:metrics:bottom_n_int64:v1":
+ "beam:metric:user:latest_int64:v1",
Review comment:
Do the UrnToString thing for *all* the cases please.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -177,12 +180,10 @@ type BundleState struct {
}
// SetPTransformState stores the state of PTransform in its bundle.
-func SetPTransformState(ctx context.Context, state bundleProcState) {
+func SetPTransformState(ctx context.Context, s *PTransformState, state bundleProcState) {
Review comment:
Make this a method on *PTransformState instead of having the PTransformState passed in, and update the code and tests accordingly to have PTransformState values used. Call the method version of this `Set` (so `func (s *PTransformState) Set(ctx context.Context, state bundleProcState)` ) since it doesn't need the repetition.
Also move this code to sampler.go, beside PTransformState.
##########
File path: sdks/go/pkg/beam/core/runtime/exec/pardo.go
##########
@@ -75,6 +77,9 @@ func (n *ParDo) Up(ctx context.Context) error {
n.status = Up
n.inv = newInvoker(n.Fn.ProcessElementFn())
+ // initialize states for current PTransform
Review comment:
Remove the extra comments like this. It's just repeating what the code is doing, which isn't as useful as reading the code. Comments are best used to answer "Why?", or the "How?" when those questions aren't as clear. A comment like this describing "What" the code is doing, is less valuable and more likely to get out of date as the code is changed, which can be hazardous.
Here and below.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -159,6 +159,24 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m
})
},
+ MsecsInt64: func(l string, states *[4]metrics.ExecutionState) {
+ for i, v := range states {
+ label := map[string]string{"PTRANSFORM": l}
Review comment:
Move the label map creation just outside the loop. There's no reason to create the same thing again for every iteration.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731158201
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -189,6 +197,8 @@ func (t kind) String() string {
return "Distribution"
case kindGauge:
return "Gauge"
+ case kindMsec:
+ return "kindMsec"
Review comment:
```suggestion
return "DoFnMsec"
```
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +152,83 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type bundleProcState int
+
+const (
+ StartBundle bundleProcState = 0
+ ProcessBundle bundleProcState = 1
+ FinishBundle bundleProcState = 2
+ TotalBundle bundleProcState = 3
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State bundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTime time.Duration
+}
+
+// executionTracker stores information about a PTransform for execution time metrics.
+type executionTracker struct {
+ numberOfTransitions int64
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ currentState bundleProcState
+ pid string
+}
+
+func SetPTransformState(ctx context.Context, state bundleProcState) {
+ if bctx, ok := ctx.(*beamCtx); ok {
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
+ }
+}
+
+type CurrentStateVal struct {
+ pid string
+ state bundleProcState
+ transitions int64
+}
+
+func loadCurrentState(ctx context.Context) CurrentStateVal {
+ if bctx, ok := ctx.(*beamCtx); ok {
+ bctx.store.mu.Lock()
+ pid := bctx.store.executionStore.pid
+
+ state := bctx.store.executionStore.currentState
+ bctx.store.mu.Unlock()
+ return CurrentStateVal{pid: pid, state: state, transitions: atomic.LoadInt64(&bctx.store.executionStore.numberOfTransitions)}
+ }
+ panic("execution store not yet set.")
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
- store map[Labels]userMetric
+ store map[Labels]userMetric
+ executionStore *executionTracker
+
+ stateRegistry map[string][4]*ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric)}
+ return &Store{store: make(map[Labels]userMetric), executionStore: &executionTracker{}, stateRegistry: make(map[string][4]*ExecutionState)}
+}
+
+// GetRegistry return the state registry.
+func (s *Store) GetRegistry() map[string][4]*ExecutionState {
+ return s.stateRegistry
+}
+
+func (s *Store) GetExecutionStore() *executionTracker {
Review comment:
We never call this, please remove it.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -30,60 +30,24 @@ func NewSampler(ctx context.Context, store *Store) StateSampler {
return StateSampler{store: store}
}
-func initialize() [4]*ExecutionState {
- var v [4]*ExecutionState
- for i := 0; i < 4; i++ {
- v[i] = &ExecutionState{}
- }
- return v
-}
-
func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadPTransformState(ctx)
- pid := PTransformLabels(ps.pid)
-
+ ps := loadCurrentState(ctx)
s.store.mu.Lock()
defer s.store.mu.Unlock()
Review comment:
Note that we're OK with using the lock here, because the StateSampler will acquire the lock relatively infrequenly, once every ~200ms, and not hold it for very long at all. The SetPTransformState call will can be called millions of times in that interval, making it a very contentious lock, and why we must avoid grabbing the lock in that call.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -214,20 +213,21 @@ type Store struct {
css []*ptCounterSet
store map[Labels]userMetric
- executionStore map[Labels]*executionTracker
- stateRegistry map[Labels][4]*ExecutionState
+ executionStore *executionTracker
+
+ stateRegistry map[string][4]*ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric), executionStore: make(map[Labels]*executionTracker), stateRegistry: make(map[Labels][4]*ExecutionState)}
+ return &Store{store: make(map[Labels]userMetric), executionStore: &executionTracker{}, stateRegistry: make(map[string][4]*ExecutionState)}
}
// GetRegistry return the state registry.
-func (s *Store) GetRegistry() map[Labels][4]*ExecutionState {
+func (s *Store) GetRegistry() map[string][4]*ExecutionState {
Review comment:
Since this exported method is only used in the Test, can we get rid of it?
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,63 @@
+// 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 metrics
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestSampler(t *testing.T) {
Review comment:
Time to add a benchmark or three! You can read a little bit about writing micro-benchmarks in Go, in the testing package doc: https://pkg.go.dev/testing#hdr-Benchmarks They're basically special kinds of test, that start with `func Benchmark` instead of `func Test`. A longer description is over here: https://dev.to/mcaci/introduction-to-benchmarks-in-go-3cii
We should have 3 benchmarks for now. One just for the SetPTransformState operation, one for the Sampler's Sample operation, and one for using them both at the same time (yes, you'll finally get to use `go` in a test.).
For those first two:
```
1. Initialize Bundle and PTransform contexts.
2. Repeatedly call the function under test until the benchmark exits.
```
For the combo benchmark:
```
1. Initialize the bundle contexts.
2. Start a sampler goroutine with the bundle context (very similar to how it works in the harness package).
- This goroutine loop and sleep, but every 5ms, instead of 200ms.
3. Initialize 2 different PTransform contexts from the bundle context.
4. Iterate on calling SetPTransformState with each of the PTransform contexts until the benchmark ends.
5. After the end of the benchmark loop, make sure the sampler goroutine is stopped.
```
Note that thees benchmarks are not doing any of their own verification. This is fine, that's the job of the tests, not the benchmarks.
While we are worried about lock contention, we want to most certainly avoid allocations during Sampling and Setting the State as much as possible, so we'll start there. We can use the -benchmem flag when running a benchmark to get this information. It's our goal to get these to 0, if possible.
In the directory with the test, it'll look something like... `go test -benchmem -run=^$ -bench='^BenchmarkMYCODE'`
To end up seeing lock contention and similar, we need to use pprof in conjunction with the benchmarks.
See https://rakyll.org/mutexprofile/ for an example of doing that. For more on the pprof tool see https://golang.org/doc/diagnostics#profiling and here's a good intro to using the tool, in particular the web view: https://jvns.ca/blog/2017/09/24/profiling-go-with-pprof/
Using Go Benchmark with pprof: https://medium.com/@felipedutratine/profile-your-benchmark-with-pprof-fb7070ee1a94
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -30,60 +30,24 @@ func NewSampler(ctx context.Context, store *Store) StateSampler {
return StateSampler{store: store}
}
-func initialize() [4]*ExecutionState {
- var v [4]*ExecutionState
- for i := 0; i < 4; i++ {
- v[i] = &ExecutionState{}
- }
- return v
-}
-
func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadPTransformState(ctx)
- pid := PTransformLabels(ps.pid)
-
+ ps := loadCurrentState(ctx)
s.store.mu.Lock()
defer s.store.mu.Unlock()
- if _, ok := s.store.stateRegistry[pid]; !ok {
- s.store.stateRegistry[pid] = initialize()
- }
- if v, ok := s.store.stateRegistry[pid]; ok {
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
v[ps.state].TotalTime += t
v[TotalBundle].TotalTime += t
- }
- if _, ok := s.store.executionStore[pid]; !ok {
- s.store.executionStore[pid] = &executionTracker{}
- }
- if v, ok := s.store.executionStore[pid]; ok {
+ e := s.store.executionStore
- if v.transitionsAtLastSample != ps.transitions {
- // state change
- v.millisSinceLastTransition = 0
- v.numberOfTransitions = ps.transitions
- v.transitionsAtLastSample = ps.transitions
+ if e.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ e.millisSinceLastTransition = 0
+ e.numberOfTransitions = ps.transitions
Review comment:
Since `numberOfTransitions` is always the same as `transitionsAtLastSample` we should get rid of that field.
To check the expected transitions, get the number of transitions from `loadCurrentState(ctx)` since that will be the accurate count, rather than "the count at last sample" which is what this value is always providing.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -174,36 +174,35 @@ type executionTracker struct {
numberOfTransitions int64
millisSinceLastTransition time.Duration
transitionsAtLastSample int64
-}
-
-// executionState is used to store as atomic.Value in Store.
-type pTransformState struct {
- pid string
- state bundleProcState
-}
-
-func newPTransformState(pid string, state bundleProcState) *pTransformState {
- return &pTransformState{pid: pid, state: state}
+ currentState bundleProcState
+ pid string
}
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
- ps := newPTransformState(bctx.ptransformID, state)
- bctx.pStore.Store(ps)
- atomic.AddInt64(&bctx.transitions, 1)
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
Review comment:
Do this after implementing the benchmarks.
Make the `store.executionStore` field an `atomic.Value` instead and write the states to that instead. We don't want to use `store.mu` to lock these interactions, since that would block the reads from the progress tracking goroutine.
This means the numberOfTransitions will need to move to live on the store as well. That allows the atomic to continue to be used.
Note, this means you should also remove the unused `pStore` field from the `beamCtx` type.
It does mean we'll be allocating a new `type BundleState struct{ pid, state}` for each call to this, but that's fine for now. We'll benchmark this on this iteration, and if the allocations are a problem (and they will be), we'll move to an approach where the 3 versions of the state are pre-allocated, so we can quickly swap between them on each call. This is where some kind of `PTransformState` will come in handy on top of the metrics.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731994073
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -30,60 +30,24 @@ func NewSampler(ctx context.Context, store *Store) StateSampler {
return StateSampler{store: store}
}
-func initialize() [4]*ExecutionState {
- var v [4]*ExecutionState
- for i := 0; i < 4; i++ {
- v[i] = &ExecutionState{}
- }
- return v
-}
-
func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadPTransformState(ctx)
- pid := PTransformLabels(ps.pid)
-
+ ps := loadCurrentState(ctx)
s.store.mu.Lock()
defer s.store.mu.Unlock()
- if _, ok := s.store.stateRegistry[pid]; !ok {
- s.store.stateRegistry[pid] = initialize()
- }
- if v, ok := s.store.stateRegistry[pid]; ok {
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
v[ps.state].TotalTime += t
v[TotalBundle].TotalTime += t
- }
- if _, ok := s.store.executionStore[pid]; !ok {
- s.store.executionStore[pid] = &executionTracker{}
- }
- if v, ok := s.store.executionStore[pid]; ok {
+ e := s.store.executionStore
- if v.transitionsAtLastSample != ps.transitions {
- // state change
- v.millisSinceLastTransition = 0
- v.numberOfTransitions = ps.transitions
- v.transitionsAtLastSample = ps.transitions
+ if e.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ e.millisSinceLastTransition = 0
+ e.numberOfTransitions = ps.transitions
Review comment:
To be clear, sampler should have `transitionsAtLastSample` while `numberOfTransitions` should go away. My original comment wasn't specific enough.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r732242562
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -189,6 +193,8 @@ func (t kind) String() string {
return "Distribution"
case kindGauge:
return "Gauge"
+ case DoFnMsec:
+ return "DoFnMsec"
Review comment:
The goal with my previous comment was to only change the returned String, not to change the variable name, which should be prefixed with `kind` so it's not Exported needlessly.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,55 @@
+// 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 metrics
+
+import (
+ "context"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store) StateSampler {
+ return StateSampler{store: store}
+}
+
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(ctx)
Review comment:
Similarly, since the context is only used because we needed access to the store (which we already have for the sampler), do we need to have the context passed in?
Remember, if we don't need a given parameter or field, we should probably get rid of it. We *do* need the context for the PTransformState calls, since that's *how* the store is passed around for later access by users and such. But this is part of the framework code, and can take the direct approach.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,55 @@
+// 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 metrics
+
+import (
+ "context"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store) StateSampler {
+ return StateSampler{store: store}
+}
+
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(ctx)
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ e := s.store
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ e.transitions = ps.transitions
Review comment:
Why are the transitions getting overwritten here? Following your code, `loadCurrentState` pulls the transitions from the store, and then re-assigns the transitions to the store again, and not doing so atomically.
The only thing the sampler should be doing with the transitions from the store is reading them, so they can be checked at the next call, to see if there are any transitions since the last call. It should not be writing them.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,55 @@
+// 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 metrics
+
+import (
+ "context"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store) StateSampler {
+ return StateSampler{store: store}
+}
+
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(ctx)
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ e := s.store
Review comment:
This doesn't make the code more readable, since it's only used once down below anyway (if at all)
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,202 @@
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func checkStateTime(t *testing.T, s StateSampler, label string, sb, pb, fb, tb time.Duration) {
+ t.Helper()
+ r := s.store.stateRegistry
+ v := r[label]
+ if v[StartBundle].TotalTime != sb || v[ProcessBundle].TotalTime != pb || v[FinishBundle].TotalTime != fb || v[TotalBundle].TotalTime != tb {
+ t.Errorf("got: start: %v, process:%v, finish:%v, total:%v; want start: %v, process:%v, finish:%v, total:%v",
+ v[StartBundle].TotalTime, v[ProcessBundle].TotalTime, v[FinishBundle].TotalTime, v[TotalBundle].TotalTime, sb, pb, fb, tb)
+ }
+}
+
+func checkBundleState(ctx context.Context, t *testing.T, s StateSampler, transitions int64, millisSinceLastTransition time.Duration) {
+ t.Helper()
+ e := atomic.LoadInt64(&s.store.transitions)
+ if e != transitions || s.millisSinceLastTransition != millisSinceLastTransition {
+ t.Errorf("number of transitions: %v, want %v \nmillis since last transition: %vms, want %vms", e, transitions, s.millisSinceLastTransition, millisSinceLastTransition)
+ }
+}
+
+func TestSampler(t *testing.T) {
+ ctx := context.Background()
+ bctx := SetBundleID(ctx, "test")
+
+ st := GetStore(bctx)
+ s := NewSampler(bctx, st)
+
+ pctx := SetPTransformID(bctx, "transform")
+ label := "transform"
+
+ SetPTransformState(pctx, StartBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 0, 0, 200*time.Millisecond)
+ checkBundleState(bctx, t, s, 1, 0)
+
+ SetPTransformState(pctx, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ SetPTransformState(pctx, ProcessBundle)
+ SetPTransformState(pctx, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 400*time.Millisecond, 0, 600*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 0)
+
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 1000*time.Millisecond, 0, 1200*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 600*time.Millisecond)
+
+ SetPTransformState(pctx, FinishBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 1000*time.Millisecond, 200*time.Millisecond, 1400*time.Millisecond)
+ checkBundleState(bctx, t, s, 5, 0)
+}
+
+func TestSampler_TwoPTransforms(t *testing.T) {
+ ctx := context.Background()
+ bctx := SetBundleID(ctx, "bundle")
+
+ st := GetStore(bctx)
+ s := NewSampler(bctx, st)
+
+ ctxA := SetPTransformID(bctx, "transformA")
+ ctxB := SetPTransformID(bctx, "transformB")
+
+ labelA := "transformA"
+ labelB := "transformB"
+
+ SetPTransformState(ctxA, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 0, 200*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 0, 0, 0)
+ checkBundleState(bctx, t, s, 1, 0)
+
+ SetPTransformState(ctxB, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ SetPTransformState(ctxA, ProcessBundle)
+ SetPTransformState(ctxB, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 0, 200*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 400*time.Millisecond, 0, 400*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 0)
+
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 0, 200*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 1000*time.Millisecond, 0, 1000*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 600*time.Millisecond)
+
+ SetPTransformState(ctxA, FinishBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ SetPTransformState(ctxB, FinishBundle)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 200*time.Millisecond, 400*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 1000*time.Millisecond, 0, 1000*time.Millisecond)
+ checkBundleState(bctx, t, s, 6, 0)
+}
+
+// goos: darwin
+// goarch: amd64
+// pkg: github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics
+// cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
+// BenchmarkMsec_SetPTransformState-12 27562231 43.36 ns/op 24 B/op 1 allocs/op
Review comment:
We definitely need to get rid of the allocation if we can. Ideally this call is 0 allocations, and single digit ns/op.
We should be able to do it, if we pre-allocate the structs used for each "state", and use `unsafe.Pointer` to store it with `atomic.StorePointer` (and conversely `atomic.LoadPointer` in the load call the Sampler makes.).
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,55 @@
+// 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 metrics
+
+import (
+ "context"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store) StateSampler {
+ return StateSampler{store: store}
+}
Review comment:
We only needed the context, so we could get the store. But since we already have the store, we don't need the context.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -179,6 +182,7 @@ const (
kindSumCounter
kindDistribution
kindGauge
+ DoFnMsec
Review comment:
kindDoFnMSec
##########
File path: sdks/go/pkg/beam/core/metrics/dumper.go
##########
@@ -74,6 +74,7 @@ func dumperExtractor(store *Store, p func(format string, args ...interface{})) {
GaugeInt64: func(l Labels, v int64, t time.Time) {
m[l] = &gauge{v: v, t: t}
},
+ MsecsInt64: func(labels string, stateRegistry [4]*ExecutionState) {},
Review comment:
Add a comment here saying that MsecInt64 extraction is tested in \<test name here\>.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +152,68 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type bundleProcState int
+
+const (
+ StartBundle bundleProcState = 0
+ ProcessBundle bundleProcState = 1
+ FinishBundle bundleProcState = 2
+ TotalBundle bundleProcState = 3
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State bundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTime time.Duration
+}
+
+// BundleState stores information about a PTransform for execution time metrics.
+type BundleState struct {
+ pid string
+ currentState bundleProcState
+}
+
+// SetPTransformState stores the state of PTransform in its bundle.
+func SetPTransformState(ctx context.Context, state bundleProcState) {
+ if bctx, ok := ctx.(*beamCtx); ok {
+ pid := bctx.ptransformID
+ bctx.store.bundleState.Store(BundleState{pid: pid, currentState: state})
+ atomic.AddInt64(&bctx.store.transitions, 1)
+ }
+}
+
+// CurrentStateVal exports the current state of a bundle wrt PTransform.
+type CurrentStateVal struct {
Review comment:
Note, there's no reason to have this type be exported. Anythnig in the metrics package can access and use it, and Exported types and fields are how we signal the external API to users and ourselves outside of the package.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,202 @@
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func checkStateTime(t *testing.T, s StateSampler, label string, sb, pb, fb, tb time.Duration) {
+ t.Helper()
+ r := s.store.stateRegistry
+ v := r[label]
+ if v[StartBundle].TotalTime != sb || v[ProcessBundle].TotalTime != pb || v[FinishBundle].TotalTime != fb || v[TotalBundle].TotalTime != tb {
+ t.Errorf("got: start: %v, process:%v, finish:%v, total:%v; want start: %v, process:%v, finish:%v, total:%v",
+ v[StartBundle].TotalTime, v[ProcessBundle].TotalTime, v[FinishBundle].TotalTime, v[TotalBundle].TotalTime, sb, pb, fb, tb)
+ }
+}
+
+func checkBundleState(ctx context.Context, t *testing.T, s StateSampler, transitions int64, millisSinceLastTransition time.Duration) {
+ t.Helper()
+ e := atomic.LoadInt64(&s.store.transitions)
+ if e != transitions || s.millisSinceLastTransition != millisSinceLastTransition {
+ t.Errorf("number of transitions: %v, want %v \nmillis since last transition: %vms, want %vms", e, transitions, s.millisSinceLastTransition, millisSinceLastTransition)
+ }
+}
+
+func TestSampler(t *testing.T) {
+ ctx := context.Background()
+ bctx := SetBundleID(ctx, "test")
+
+ st := GetStore(bctx)
+ s := NewSampler(bctx, st)
+
+ pctx := SetPTransformID(bctx, "transform")
+ label := "transform"
+
+ SetPTransformState(pctx, StartBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 0, 0, 200*time.Millisecond)
+ checkBundleState(bctx, t, s, 1, 0)
+
+ SetPTransformState(pctx, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ SetPTransformState(pctx, ProcessBundle)
+ SetPTransformState(pctx, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 400*time.Millisecond, 0, 600*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 0)
+
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 1000*time.Millisecond, 0, 1200*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 600*time.Millisecond)
+
+ SetPTransformState(pctx, FinishBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ // validate states and their time till now
+ checkStateTime(t, s, label, 200*time.Millisecond, 1000*time.Millisecond, 200*time.Millisecond, 1400*time.Millisecond)
+ checkBundleState(bctx, t, s, 5, 0)
+}
+
+func TestSampler_TwoPTransforms(t *testing.T) {
+ ctx := context.Background()
+ bctx := SetBundleID(ctx, "bundle")
+
+ st := GetStore(bctx)
+ s := NewSampler(bctx, st)
+
+ ctxA := SetPTransformID(bctx, "transformA")
+ ctxB := SetPTransformID(bctx, "transformB")
+
+ labelA := "transformA"
+ labelB := "transformB"
+
+ SetPTransformState(ctxA, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 0, 200*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 0, 0, 0)
+ checkBundleState(bctx, t, s, 1, 0)
+
+ SetPTransformState(ctxB, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ SetPTransformState(ctxA, ProcessBundle)
+ SetPTransformState(ctxB, ProcessBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 0, 200*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 400*time.Millisecond, 0, 400*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 0)
+
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(bctx, 200*time.Millisecond)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 0, 200*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 1000*time.Millisecond, 0, 1000*time.Millisecond)
+ checkBundleState(bctx, t, s, 4, 600*time.Millisecond)
+
+ SetPTransformState(ctxA, FinishBundle)
+ s.Sample(bctx, 200*time.Millisecond)
+ SetPTransformState(ctxB, FinishBundle)
+
+ // validate states and their time till now
+ checkStateTime(t, s, labelA, 0, 200*time.Millisecond, 200*time.Millisecond, 400*time.Millisecond)
+ checkStateTime(t, s, labelB, 0, 1000*time.Millisecond, 0, 1000*time.Millisecond)
+ checkBundleState(bctx, t, s, 6, 0)
+}
+
+// goos: darwin
+// goarch: amd64
+// pkg: github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics
+// cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
+// BenchmarkMsec_SetPTransformState-12 27562231 43.36 ns/op 24 B/op 1 allocs/op
+func BenchmarkMsec_SetPTransformState(b *testing.B) {
+ ctx := context.Background()
+ bctx := SetBundleID(ctx, "benchmark")
+ pctx := SetPTransformID(bctx, "transform")
+
+ for i := 0; i < b.N; i++ {
+ SetPTransformState(pctx, StartBundle)
+ }
+}
+
+// goos: darwin
+// goarch: amd64
+// pkg: github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics
+// cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
+// BenchmarkMsec_Sample-12 35614332 33.98 ns/op 0 B/op 0 allocs/op
Review comment:
This is excellent, and what we want to see for the infrequent sampler calls.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r732071146
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -174,36 +174,35 @@ type executionTracker struct {
numberOfTransitions int64
millisSinceLastTransition time.Duration
transitionsAtLastSample int64
-}
-
-// executionState is used to store as atomic.Value in Store.
-type pTransformState struct {
- pid string
- state bundleProcState
-}
-
-func newPTransformState(pid string, state bundleProcState) *pTransformState {
- return &pTransformState{pid: pid, state: state}
+ currentState bundleProcState
+ pid string
}
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
- ps := newPTransformState(bctx.ptransformID, state)
- bctx.pStore.Store(ps)
- atomic.AddInt64(&bctx.transitions, 1)
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
Review comment:
After having a separate lock, we are back at 2 allocs/op
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954911674
PTAL.
--
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] lostluck commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-953351878
Before asking me to PTAL this coming time, please scroll back up and resolve any comments that have been implemented or are no longer relevant. Unless you have a question, or need to discuss something, you don't need to add a response, the "Resolve" is a response.
Easier to ensure things haven't been missed that way.
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954871648
Run Go Flink ValidatesRunner
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954871415
Run Go 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738775769
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ nextLogTime time.Duration
+ logInterval time.Duration
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(store *Store) StateSampler {
+ return StateSampler{store: store, nextLogTime: 5 * time.Minute, logInterval: 5 * time.Minute}
+}
+
+// Sample checks for state transition in processing a DoFn
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ s.transitionsAtLastSample = ps.transitions
+ } else {
+ s.millisSinceLastTransition += t
+ }
+
+ if s.millisSinceLastTransition > s.nextLogTime {
+ log.Info(ctx, "...standard long running operation log ...", ps.pid, ps.state, s.millisSinceLastTransition)
+ s.nextLogTime += s.logInterval
+ }
+ }
+}
+
+func loadCurrentState(s *StateSampler) currentStateVal {
+ if ts := (atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&s.store.bundleState)))); ts == nil {
+ return currentStateVal{}
+ } else {
+ bs := *(*BundleState)(ts)
+ return currentStateVal{pid: bs.pid, state: bs.currentState, transitions: atomic.LoadInt64(s.store.transitions)}
+ }
Review comment:
Minor style nit: If the `if` case returns, don't have an `else`, just have the contents of the else, un-indented. This improves readability.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ nextLogTime time.Duration
+ logInterval time.Duration
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(store *Store) StateSampler {
+ return StateSampler{store: store, nextLogTime: 5 * time.Minute, logInterval: 5 * time.Minute}
+}
+
+// Sample checks for state transition in processing a DoFn
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ s.transitionsAtLastSample = ps.transitions
+ } else {
+ s.millisSinceLastTransition += t
+ }
+
+ if s.millisSinceLastTransition > s.nextLogTime {
+ log.Info(ctx, "...standard long running operation log ...", ps.pid, ps.state, s.millisSinceLastTransition)
Review comment:
The point of me commenting "...standard long running operation log..." meant that you should have looked up the log message that the Python and Java SDKs use and add it here. I can't do all the research and reading for you. If you need help, I'll provide it, but please ask for it.
Part of the job is to do your own thinking about the code you're writing, not simply responding to reviewers without thinking.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ nextLogTime time.Duration
+ logInterval time.Duration
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(store *Store) StateSampler {
+ return StateSampler{store: store, nextLogTime: 5 * time.Minute, logInterval: 5 * time.Minute}
+}
+
+// Sample checks for state transition in processing a DoFn
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ s.transitionsAtLastSample = ps.transitions
Review comment:
We need to reset next log time here: `s.nextLogTime = s.logInterval` since a transition has happened.
Otherwise, we end up with ever increasing nextLogTimes 5 minutes, 10 minutes, 15 minutes etc. even if the state moves, the next "long running state" will take
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -91,6 +91,11 @@ type Extractor struct {
DistributionInt64 func(labels Labels, count, sum, min, max int64)
// GaugeInt64 extracts data from Gauge Int64 counters.
GaugeInt64 func(labels Labels, v int64, t time.Time)
+
+ // Extraction of Msec counters is experimental and subject to change.
+
Review comment:
Remove the blank line. Blank lines disconnect comments and prevent them from being surfaced in the documentation. If you want to have a documented paragraph break, the "blank" line still needs a `//` at it's start.
At this point the comment should be
```
// MsecsInt64 extracts data from StateRegistry of ExecutionState.
//
// Extraction of Msec counters is experimental and subject to change.
```
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r724685672
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
+ CurrentState BundleProcState
+ State ExecutionState
+ NumberOfTransitions int
+ MillisSinceLastTransition int
+ TransitionsAtLastSample int
+ LastSampleTimeMillis int
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
+
+ executionStore ExecutionStateTracker
+ stateRegistry map[Labels][]ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric)}
+ return &Store{store: make(map[Labels]userMetric), stateRegistry: make(map[Labels][]ExecutionState)}
+}
+
+// AddState adds an ExecutionState of a PTransform to the stateRegistry.
+func (s *Store) AddState(e ExecutionState, pid string) {
+ label := PTransformLabels(pid)
+ if s.stateRegistry == nil {
+ s.stateRegistry = make(map[Labels][]ExecutionState)
+ }
+ s.stateRegistry[label] = append(s.stateRegistry[label], e)
+}
+
+// SetState updates the state of a bundle.
+// For the bundle in start state, update the ExecutionState struct as well.
+func (s *Store) SetState(bs BundleProcState) {
+ if bs == StartBundle {
+ s.executionStore.State.State = bs
+ }
+ s.executionStore.CurrentState = bs
+}
+
+// IncTransitions increment the number of transitions by 1.
+func (s *Store) IncTransitions() {
+ s.executionStore.NumberOfTransitions += 1
+}
+
+// GetRegistry return the state registry.
+func (s *Store) GetRegistry() map[Labels][]ExecutionState {
+ return s.stateRegistry
}
Review comment:
I think we can have an abstraction like user metrics for msecs as well and these methods can be added on ExecutionStateTracker. However, I'm not clear on what does "initialize the sampler in each exec.ParDo in the Up method (which is what Up is for), and then pass the context into the methods for it that way instead." is implying.
As of now, when we move the actual sampler code to the metrics package, the harness package has the control to get a new sampler from the metrics package and call methods on it like `start()` and `stop()`. These methods call the methods on the sampler from metrics package.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r726568585
##########
File path: sdks/go/pkg/beam/core/runtime/exec/pardo.go
##########
@@ -121,6 +129,12 @@ func (n *ParDo) ProcessElement(_ context.Context, elm *FullValue, values ...ReSt
return errors.Errorf("invalid status for pardo %v: %v, want Active", n.UID, n.status)
}
+ // set current state for execution time metrics
+ // metrics.GetExecutionStore(n.ctx).SetState(metrics.ProcessBundle)
+ // metrics.GetExecutionStore(n.ctx).IncTransitions()
+ metrics.SetState(n.ctx, metrics.ProcessBundle)
+ metrics.IncTransition(n.ctx)
Review comment:
Why can't we increment transitions in the SetState call? I would think we need to do so on every call SetState?
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -102,6 +103,19 @@ func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
return defaultShortIDCache.shortIdsToInfos(shortids)
}
+func getUrn(i int) metricsx.Urn {
+ switch i {
+ case 0:
+ return metricsx.UrnStartBundle
+ case 1:
+ return metricsx.UrnProcessBundle
+ case 2:
+ return metricsx.UrnFinishBundle
+ default:
Review comment:
We seem to be missing the "metricx.UrnTransformTotalTime" Urn as well.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -102,6 +103,19 @@ func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
return defaultShortIDCache.shortIdsToInfos(shortids)
}
+func getUrn(i int) metricsx.Urn {
Review comment:
Please move this function to the metricsx package. Call it `ExecutionMsecUrn`.
"getUrn" is too general for what it ends up doing.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -112,6 +114,8 @@ func (ctx *beamCtx) Value(key interface{}) interface{} {
}
}
return ctx.store
+ case exStoreKey:
+ return ctx.exStore
Review comment:
This must return a pointer to the atomic.Value (`&ctx.exStore`) otherwise it will end up being copied, which is not allowed.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,77 @@
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *ExecutionTracker
+ store *Store
+ pid string // PTransform ID
Review comment:
You've called this PTransformID but you've passed in the BundleID. The PTransformIDs aren't available until exec.ParDo.
In the Beam Model, a sequence of PTransforms + a set of Data are executed in Bundles. Bundles are the unit of re-trying. But the goal with these metrics is to find where time is being spent in PTransforms, divided up into Start, Process, and Finish segments, and also total time spent with a given PTransform which would be the sum of the Start, Process, Finish portions.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -79,6 +80,7 @@ type beamCtx struct {
bundleID, ptransformID string
store *Store
cs *ptCounterSet
+ exStore atomic.Value
Review comment:
Please add a comment about what type should be stored in this field.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -316,10 +316,17 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
if err != nil {
return fail(ctx, instID, "Failed: %v", err)
}
+ // Start the sampler goroutine here
+ sampler := newSampler(ctx, store, string(instID))
+ sampler.start()
data := NewScopedDataManager(c.data, instID)
state := NewScopedStateReaderWithCache(c.state, instID, c.cache)
err = plan.Execute(ctx, string(instID), exec.DataContext{Data: data, State: state})
+
+ // Plan execution complete, stop sampling for metrics for this bundle.
+ sampler.stop()
Review comment:
Small nit: Move this call to after the data and state `Close` calls, or move the sampler to start between where state is created and before the plan Executed.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,77 @@
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *ExecutionTracker
+ store *Store
+ pid string // PTransform ID
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: getExecutionStore(ctx), store: store, pid: pid}
+}
+
+func (s *StateSampler) Start() {
+ s.startSampler()
+}
+
+func (s *StateSampler) startSampler() {
+ for {
+ select {
+ case <-s.done:
+ return
+ default:
+ if atomic.LoadInt64(&s.e.NumberOfTransitions) != atomic.LoadInt64(&s.e.TransitionsAtLastSample) {
+ s.sample()
+ }
+
+ // TODO: constant for sampling period
+ atomic.AddInt64(&s.e.MillisSinceLastTransition, 200)
+ atomic.AddInt64(&s.e.State.TotalTimeMillis, 200)
+ time.Sleep(200 * time.Millisecond)
+ }
+ }
+}
+
+func (s *StateSampler) Stop() {
+ s.stopSampler()
+}
+
+func (s *StateSampler) stopSampler() {
+ s.done <- 1
+ // collect the remaining metrics (finish bundle metrics)
+ s.sample()
+}
+
+func (s *StateSampler) sample() {
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+ s.e.TransitionsAtLastSample = s.e.NumberOfTransitions
+ s.e.MillisSinceLastTransition = 0
+ s.store.AddState(s.e.State, s.pid)
Review comment:
The AddState call here, should probably be passing in the MillisSinceLastTransition before it gets zeroed, so it can be added to the current State.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,77 @@
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *ExecutionTracker
+ store *Store
+ pid string // PTransform ID
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: getExecutionStore(ctx), store: store, pid: pid}
+}
+
+func (s *StateSampler) Start() {
+ s.startSampler()
+}
+
+func (s *StateSampler) startSampler() {
+ for {
+ select {
+ case <-s.done:
+ return
+ default:
+ if atomic.LoadInt64(&s.e.NumberOfTransitions) != atomic.LoadInt64(&s.e.TransitionsAtLastSample) {
+ s.sample()
+ }
+
+ // TODO: constant for sampling period
+ atomic.AddInt64(&s.e.MillisSinceLastTransition, 200)
+ atomic.AddInt64(&s.e.State.TotalTimeMillis, 200)
+ time.Sleep(200 * time.Millisecond)
Review comment:
Move the contents of this loop to it's own method.
Move the loop and stop management to the harness package sampler.
This separation will make things easier to test, and avoid having a constant, and allow that to be configured.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
+ CurrentState BundleProcState
+ State ExecutionState
+ NumberOfTransitions int
+ MillisSinceLastTransition int
+ TransitionsAtLastSample int
+ LastSampleTimeMillis int
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
+
+ executionStore ExecutionStateTracker
+ stateRegistry map[Labels][]ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric)}
+ return &Store{store: make(map[Labels]userMetric), stateRegistry: make(map[Labels][]ExecutionState)}
+}
+
+// AddState adds an ExecutionState of a PTransform to the stateRegistry.
+func (s *Store) AddState(e ExecutionState, pid string) {
+ label := PTransformLabels(pid)
+ if s.stateRegistry == nil {
+ s.stateRegistry = make(map[Labels][]ExecutionState)
+ }
+ s.stateRegistry[label] = append(s.stateRegistry[label], e)
Review comment:
We still have the memory leak issue.
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-942776133
PTAL.
1. Changed sampler to take in `time`.
2. Added `TransformTotalTime` calculation.
3. Sampler test. verbose but works as expected.
Please ignore the `exec.Pardo` for now. Major changes to metrics package only.
Is there a way we can delete the `Store.stateRegistry` allocated with make and append ? Have a indexed array for it creates problem in assigning the `Struct value to a map field`.
--
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] lostluck merged pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck merged pull request #15657:
URL: https://github.com/apache/beam/pull/15657
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-953120690
PTAL. Will add query filtering for DoFn and PCollection metrics in a separate 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
[GitHub] [beam] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731374488
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -30,60 +30,24 @@ func NewSampler(ctx context.Context, store *Store) StateSampler {
return StateSampler{store: store}
}
-func initialize() [4]*ExecutionState {
- var v [4]*ExecutionState
- for i := 0; i < 4; i++ {
- v[i] = &ExecutionState{}
- }
- return v
-}
-
func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadPTransformState(ctx)
- pid := PTransformLabels(ps.pid)
-
+ ps := loadCurrentState(ctx)
s.store.mu.Lock()
defer s.store.mu.Unlock()
- if _, ok := s.store.stateRegistry[pid]; !ok {
- s.store.stateRegistry[pid] = initialize()
- }
- if v, ok := s.store.stateRegistry[pid]; ok {
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
v[ps.state].TotalTime += t
v[TotalBundle].TotalTime += t
- }
- if _, ok := s.store.executionStore[pid]; !ok {
- s.store.executionStore[pid] = &executionTracker{}
- }
- if v, ok := s.store.executionStore[pid]; ok {
+ e := s.store.executionStore
- if v.transitionsAtLastSample != ps.transitions {
- // state change
- v.millisSinceLastTransition = 0
- v.numberOfTransitions = ps.transitions
- v.transitionsAtLastSample = ps.transitions
+ if e.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ e.millisSinceLastTransition = 0
+ e.numberOfTransitions = ps.transitions
Review comment:
`loadCurrentState(ctx)` reads the `numberOfTransitions ` itself. So we need `transitionsAtLastSample ` somewhere to track the transitions change. After restructuring, we sampler can have these field on its side instead of storing it in the `metrics.Store`
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731375722
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -174,36 +174,35 @@ type executionTracker struct {
numberOfTransitions int64
millisSinceLastTransition time.Duration
transitionsAtLastSample int64
-}
-
-// executionState is used to store as atomic.Value in Store.
-type pTransformState struct {
- pid string
- state bundleProcState
-}
-
-func newPTransformState(pid string, state bundleProcState) *pTransformState {
- return &pTransformState{pid: pid, state: state}
+ currentState bundleProcState
+ pid string
}
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
- ps := newPTransformState(bctx.ptransformID, state)
- bctx.pStore.Store(ps)
- atomic.AddInt64(&bctx.transitions, 1)
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
Review comment:
`allocs/op` changed significantly (from 2 to 12) when I used atomic.Value to store and load BundleState into it. Also, the time per loop changed from 2566 ns/op to 3376 ns/op.
Did this benchmarking without pprof. Now I'll do it with pprof.
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-948108709
PTAL. The pre-allocated state makes the difference. Thanks. For now, I've added them in the `metrics.Store`.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r737870109
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -136,6 +136,10 @@ func SetPTransformID(ctx context.Context, id string) context.Context {
// Checking for *beamCtx is an optimization, so we don't dig deeply
// for ids if not necessary.
if bctx, ok := ctx.(*beamCtx); ok {
+ if _, ok := bctx.store.stateRegistry[id]; !ok {
+ bctx.store.stateRegistry[id] = [4]*ExecutionState{&ExecutionState{}, &ExecutionState{}, &ExecutionState{}, &ExecutionState{}}
Review comment:
Consider whether one can use `[4]ExecutionState{}` instead, which will automatically initialize the values.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -28,28 +29,37 @@ type StateSampler struct {
}
// NewSampler creates a new state sampler.
-func NewSampler(ctx context.Context, store *Store) StateSampler {
+func NewSampler(store *Store) StateSampler {
return StateSampler{store: store}
}
-func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadCurrentState(ctx)
+func (s *StateSampler) Sample(t time.Duration) {
Review comment:
The Sample function should have a context so it can use the beam logger for lull logging, when we hit >=5 minutes without transitions.
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -51,42 +53,65 @@ func groupByType(minfos []*pipepb.MonitoringInfo) (
}
r := bytes.NewReader(minfo.GetPayload())
-
- switch minfo.GetType() {
- case "beam:metrics:sum_int64:v1":
+ switch minfo.GetUrn() {
+ case "beam:metric:user:sum_int64:v1":
value, err := extractCounterValue(r)
if err != nil {
log.Println(err)
continue
}
counters[key] = value
- case "beam:metrics:distribution_int64:v1":
+ case "beam:metric:user:distribution_int64:v1":
value, err := extractDistributionValue(r)
if err != nil {
log.Println(err)
continue
}
distributions[key] = value
case
- "beam:metrics:latest_int64:v1",
- "beam:metrics:top_n_int64:v1",
- "beam:metrics:bottom_n_int64:v1":
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:bottom_n_int64:v1":
value, err := extractGaugeValue(r)
if err != nil {
log.Println(err)
continue
}
gauges[key] = value
+ case
+ "beam:metric:pardo_execution_time:start_bundle_msecs:v1",
Review comment:
Consider using `UrnToString(UrnStartBundle)` and similar to avoid duplicating the string constants where they are prone to typos. Within a package, the compiler is clever enough to inline things.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -45,38 +45,38 @@ func TestSampler(t *testing.T) {
bctx := SetBundleID(ctx, "test")
st := GetStore(bctx)
- s := NewSampler(bctx, st)
+ s := NewSampler(st)
pctx := SetPTransformID(bctx, "transform")
label := "transform"
SetPTransformState(pctx, StartBundle)
- s.Sample(bctx, 200*time.Millisecond)
+ s.Sample(200 * time.Millisecond)
Review comment:
To simplify the test, just make a variable `interval := 200 * time.Millisecond` and re-use it instead of copying the multiplication everywhere. Do it per test, rather than a package level constant. That way readers of the test will know whether what's important about it or not.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -179,36 +180,29 @@ type BundleState struct {
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
pid := bctx.ptransformID
- bctx.store.bundleState.Store(BundleState{pid: pid, currentState: state})
+ bctx.store.states[state].pid = pid
Review comment:
The problem with this specifically is that it's a non-atomic write to something that will be read by some other thread, which means there's a data race. It's possible the Sampler will read the pid as it's being written to, which is bad.
So, we need to have these pre-allocated BundleStates for each PTransform. What if we did those allocations once per PTransform? We can do so by defining a struct
```
type PTransformState struct{
states [3]BundleState
}
func NewPTransformState() *PTransformState {
return &PTransformState{
states: [3]BundleState{
{pid, StartBundle},
{pid, ProcessBundle},
{pid, FinishBundle},
},
}
}
func (p *PTransformState) Set(ctx context.Context, s bundState) {
/// ... existing implementation of SetPTransformState, but using the local s.states values instead ....
}
```
That should avoid the race condition, and mean in `exec.ParDo` we add a field `state *metrics.PTransformState`
which we initialize in the `Up` method, and call as we are currently doing.
Finally, move all this new code for the state to sampler.go, so that everything that's related to sampling is in one place. *This does not mean moving things out of the store struct,just which file it's in. (except for the states field, which has just moved to the the PTransformState, and should not be directly in store.) *
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -179,36 +180,29 @@ type BundleState struct {
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
pid := bctx.ptransformID
- bctx.store.bundleState.Store(BundleState{pid: pid, currentState: state})
+ bctx.store.states[state].pid = pid
+ atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&bctx.store.bundleState)), unsafe.Pointer(&bctx.store.states[state]))
atomic.AddInt64(&bctx.store.transitions, 1)
}
}
-// CurrentStateVal exports the current state of a bundle wrt PTransform.
-type CurrentStateVal struct {
+// currentStateVal exports the current state of a bundle wrt PTransform.
+type currentStateVal struct {
pid string
state bundleProcState
transitions int64
}
-func loadCurrentState(ctx context.Context) CurrentStateVal {
- if bctx, ok := ctx.(*beamCtx); ok {
- bs := bctx.store.bundleState.Load().(BundleState)
- return CurrentStateVal{pid: bs.pid, state: bs.currentState, transitions: atomic.LoadInt64(&bctx.store.transitions)}
- }
- panic("execution store not yet set.")
-}
-
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
- transitions int64
- bundleState atomic.Value
-
+ transitions int64
+ bundleState BundleState
Review comment:
This should be a pointer to a BundleState, so it's only changing an address, not copying the string reference and an integer.
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -51,42 +53,65 @@ func groupByType(minfos []*pipepb.MonitoringInfo) (
}
r := bytes.NewReader(minfo.GetPayload())
-
- switch minfo.GetType() {
- case "beam:metrics:sum_int64:v1":
+ switch minfo.GetUrn() {
+ case "beam:metric:user:sum_int64:v1":
value, err := extractCounterValue(r)
if err != nil {
log.Println(err)
continue
}
counters[key] = value
- case "beam:metrics:distribution_int64:v1":
+ case "beam:metric:user:distribution_int64:v1":
value, err := extractDistributionValue(r)
if err != nil {
log.Println(err)
continue
}
distributions[key] = value
case
- "beam:metrics:latest_int64:v1",
- "beam:metrics:top_n_int64:v1",
- "beam:metrics:bottom_n_int64:v1":
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:bottom_n_int64:v1":
value, err := extractGaugeValue(r)
if err != nil {
log.Println(err)
continue
}
gauges[key] = value
+ case
+ "beam:metric:pardo_execution_time:start_bundle_msecs:v1",
+ "beam:metric:pardo_execution_time:process_bundle_msecs:v1",
+ "beam:metric:pardo_execution_time:finish_bundle_msecs:v1",
+ "beam:metric:ptransform_execution_time:total_msecs:v1":
+ value, err := extractMsecValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ msecs[key] = value
default:
log.Println("unknown metric type")
}
}
- return counters, distributions, gauges
+ return counters, distributions, gauges, msecs
}
func extractKey(mi *pipepb.MonitoringInfo) (metrics.StepKey, error) {
labels := newLabels(mi.GetLabels())
stepName := labels.Transform()
+ urn := mi.GetUrn()
+ switch urn {
+ case "beam:metric:pardo_execution_time:start_bundle_msecs:v1":
Review comment:
Same comment here, WRT using UrnToString and our constants from metricsx/urns.go.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -91,6 +94,8 @@ type Extractor struct {
DistributionInt64 func(labels Labels, count, sum, min, max int64)
// GaugeInt64 extracts data from Gauge Int64 counters.
GaugeInt64 func(labels Labels, v int64, t time.Time)
+ // MsecsInt64 extracts data from StateRegistry of ExecutionState
Review comment:
Please add a comment here saying extracting Msec counters is Experimental and subject to change. That way we can come back and clean things up in a separate PR if needed.
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -28,28 +29,37 @@ type StateSampler struct {
}
// NewSampler creates a new state sampler.
-func NewSampler(ctx context.Context, store *Store) StateSampler {
+func NewSampler(store *Store) StateSampler {
return StateSampler{store: store}
}
-func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
- ps := loadCurrentState(ctx)
+func (s *StateSampler) Sample(t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
s.store.mu.Lock()
defer s.store.mu.Unlock()
if v, ok := s.store.stateRegistry[ps.pid]; ok {
v[ps.state].TotalTime += t
v[TotalBundle].TotalTime += t
- e := s.store
-
if s.transitionsAtLastSample != ps.transitions {
// state change detected
s.millisSinceLastTransition = 0
- e.transitions = ps.transitions
s.transitionsAtLastSample = ps.transitions
} else {
s.millisSinceLastTransition += t
}
Review comment:
Add another `if` clause here to report lulls.
We don't want to log every sample after the log interval, only once every 5 minutes. So we can have two new fields. One variable that has the logging interval (5 minutes) and another that tracks what time "since" we want to log next.
```
if s.millisSinceLastTransition > s.nextLogTime {
log.Info( ...standard long running operation log ... , pid, state, s.millis.SinceLastTransition)
s.nextLogTime += s.logInterval
}
```
This means we need to reset the `s.nextLogTime` every time the samples reset, so we can log the next time a transition takes 5 minutes.
Finally, since we are using time.Duration as the type for `s.millisSinceLastTransition` call it `s.timeSinceLastTransition` instead. It doesn't matter that we're later translating things to milliseconds in this one here.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r739277800
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ nextLogTime time.Duration
+ logInterval time.Duration
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(store *Store) StateSampler {
+ return StateSampler{store: store, nextLogTime: 5 * time.Minute, logInterval: 5 * time.Minute}
+}
+
+// Sample checks for state transition in processing a DoFn
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ s.transitionsAtLastSample = ps.transitions
+ } else {
+ s.millisSinceLastTransition += t
+ }
+
+ if s.millisSinceLastTransition > s.nextLogTime {
+ log.Info(ctx, "...standard long running operation log ...", ps.pid, ps.state, s.millisSinceLastTransition)
Review comment:
Yes. Sorry for rushing to push the changes. I'll make sure things are in order from now on.
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-947075018
PTAL. There is no lock contention in `SetPTransformState` now because we are using atomic.Value for updating the state and pid. However, this has been compensated in the allocs/op. It changed from 0 to 2 allocs/op.
There is a lock contention in case of `s.Sample` . If we can use normal `int` for storing the time instead of `time.Duration` then we can get rid of that lock as well.
--
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] lostluck commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-940436100
Taking a look, but before I start.
> PTAL. Addressed the old comments. However, I didn't understand clearly, the approach of initializing sampler in exec.Pardo's Up method.
I think what I meant could be more clearly said. Instead of "sampler" which is very overloaded, I should have said "state switcher" which tells the underlying metric the current state & PTransform. Basically it could have 3 methods (Start, Process, Finish) which switch the state to the current enum (start process finish), and then do an atomic store with the state. This avoids allocating a new state every time these are called, and avoids any race conditions with the reading sampler goroutine.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731375722
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -174,36 +174,35 @@ type executionTracker struct {
numberOfTransitions int64
millisSinceLastTransition time.Duration
transitionsAtLastSample int64
-}
-
-// executionState is used to store as atomic.Value in Store.
-type pTransformState struct {
- pid string
- state bundleProcState
-}
-
-func newPTransformState(pid string, state bundleProcState) *pTransformState {
- return &pTransformState{pid: pid, state: state}
+ currentState bundleProcState
+ pid string
}
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
- ps := newPTransformState(bctx.ptransformID, state)
- bctx.pStore.Store(ps)
- atomic.AddInt64(&bctx.transitions, 1)
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
Review comment:
`allocs/op` changed significantly (from 2 to 12) when I used atomic.Value to store and load BundleState into it. Also, the time per loop changed from 2566 ns/op to 3376 ns/op.
Did this benchmarking without pprof. Now I'll do it with ppro.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r722462734
##########
File path: sdks/go/pkg/beam/core/runtime/exec/pardo.go
##########
@@ -121,6 +124,10 @@ func (n *ParDo) ProcessElement(_ context.Context, elm *FullValue, values ...ReSt
return errors.Errorf("invalid status for pardo %v: %v, want Active", n.UID, n.status)
}
+ // set current state for execution time metrics
+ metrics.GetStore(n.ctx).SetState(metrics.ProcessBundle)
+ metrics.GetStore(n.ctx).IncTransitions()
Review comment:
GetStore isn't very lightweight (extracting things from contexts never is), so get the Store once, and then call SetState and IncTransitions on it, rather than getting it twice.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *metrics.ExecutionStateTracker
+ store *metrics.Store
+ pid string // PTransform ID
+}
+
+func newSampler(e *metrics.ExecutionStateTracker, store *metrics.Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: e, store: store, pid: pid}
+}
+
+func (s StateSampler) startSampler() {
+ for {
+ select {
+ case <-s.done:
+ return
+ default:
+ if s.e.NumberOfTransitions != s.e.TransitionsAtLastSample {
+ s.sample()
+ }
+ s.e.MillisSinceLastTransition += 200
+ s.e.State.TotalTimeMillis += 200
+ time.Sleep(200 * time.Millisecond)
+ }
+ }
+}
+
+func (s StateSampler) stopSampler() {
+ s.done <- 1
+}
+
+func (s StateSampler) sample() {
+ s.e.TransitionsAtLastSample = s.e.NumberOfTransitions
+ s.e.MillisSinceLastTransition = 0
+ s.store.AddState(s.e.State, s.pid)
+ s.e.State.State = s.e.CurrentState
+ s.e.State.TotalTimeMillis = 0
+}
Review comment:
If we make the default case in the sampler loop thread safe then this method can be not-thread safe (if it's always called in a the lock's critical section (aka when the lock is held).
##########
File path: sdks/go/pkg/beam/core/runtime/exec/pardo.go
##########
@@ -121,6 +124,10 @@ func (n *ParDo) ProcessElement(_ context.Context, elm *FullValue, values ...ReSt
return errors.Errorf("invalid status for pardo %v: %v, want Active", n.UID, n.status)
}
+ // set current state for execution time metrics
+ metrics.GetStore(n.ctx).SetState(metrics.ProcessBundle)
+ metrics.GetStore(n.ctx).IncTransitions()
Review comment:
Note this is sort of obsoleted by another comment in how to change the approach.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
Review comment:
Two comments here:
There's no need to export the BundleProcState type (instead make it bundleProcState). This prevents new states from being defined out side of the package. Users can always use the ones defined here in the constants, because the constants themselves are exported.
2nd. We should not be using strings as a base type. String Comparisons are expensive since they do need to go character by character. Use an int instead which will be much faster to compare. You can always add a `String() string` method to the type that translates 0, 1, 2 to the appropriate Strings.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *metrics.ExecutionStateTracker
+ store *metrics.Store
+ pid string // PTransform ID
+}
+
+func newSampler(e *metrics.ExecutionStateTracker, store *metrics.Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: e, store: store, pid: pid}
+}
+
+func (s StateSampler) startSampler() {
+ for {
+ select {
+ case <-s.done:
+ return
+ default:
+ if s.e.NumberOfTransitions != s.e.TransitionsAtLastSample {
+ s.sample()
+ }
+ s.e.MillisSinceLastTransition += 200
+ s.e.State.TotalTimeMillis += 200
Review comment:
This set of operations is not threadsafe.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *metrics.ExecutionStateTracker
+ store *metrics.Store
+ pid string // PTransform ID
+}
+
+func newSampler(e *metrics.ExecutionStateTracker, store *metrics.Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: e, store: store, pid: pid}
+}
+
+func (s StateSampler) startSampler() {
+ for {
+ select {
+ case <-s.done:
+ return
+ default:
+ if s.e.NumberOfTransitions != s.e.TransitionsAtLastSample {
+ s.sample()
+ }
+ s.e.MillisSinceLastTransition += 200
+ s.e.State.TotalTimeMillis += 200
+ time.Sleep(200 * time.Millisecond)
Review comment:
The harness package should be what handles the times/sleeps however. We can have the tick/increment method on the executions state tracker take in the number of milliseconds that have passed. That will make it easier to test by itself (avoiding real time in tests).
To avoid lock contention or similar, this implies we might need the tick method, and a "sampleAndTick" method to avoid too much lock churn.
##########
File path: sdks/go/pkg/beam/core/runtime/exec/pardo.go
##########
@@ -198,6 +205,10 @@ func (n *ParDo) FinishBundle(_ context.Context) error {
n.status = Up
n.inv.Reset()
+ // set current state for execution time metrics
+ metrics.GetStore(n.ctx).SetState(metrics.FinishBundle)
+ metrics.GetStore(n.ctx).IncTransitions()
Review comment:
Same here.
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
Review comment:
I don't think this interface is doing anything for us? Can we remove it?
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
+ CurrentState BundleProcState
+ State ExecutionState
+ NumberOfTransitions int
+ MillisSinceLastTransition int
+ TransitionsAtLastSample int
+ LastSampleTimeMillis int
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
+
+ executionStore ExecutionStateTracker
+ stateRegistry map[Labels][]ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric)}
+ return &Store{store: make(map[Labels]userMetric), stateRegistry: make(map[Labels][]ExecutionState)}
+}
+
+// AddState adds an ExecutionState of a PTransform to the stateRegistry.
+func (s *Store) AddState(e ExecutionState, pid string) {
+ label := PTransformLabels(pid)
+ if s.stateRegistry == nil {
+ s.stateRegistry = make(map[Labels][]ExecutionState)
+ }
+ s.stateRegistry[label] = append(s.stateRegistry[label], e)
+}
+
+// SetState updates the state of a bundle.
+// For the bundle in start state, update the ExecutionState struct as well.
+func (s *Store) SetState(bs BundleProcState) {
+ if bs == StartBundle {
+ s.executionStore.State.State = bs
+ }
+ s.executionStore.CurrentState = bs
+}
+
+// IncTransitions increment the number of transitions by 1.
+func (s *Store) IncTransitions() {
+ s.executionStore.NumberOfTransitions += 1
+}
+
+// GetRegistry return the state registry.
+func (s *Store) GetRegistry() map[Labels][]ExecutionState {
+ return s.stateRegistry
}
Review comment:
If you look at the other methods for the Store, (I guess that's `storeMetric`) note that it acquires the lock. None of these are concurrency safe operations between reading and writing threads. We need to take that into account.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
Review comment:
Name-wise, we can probably shorten this to `ExecutionTracker` , "State" doesn't add much to it.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -155,6 +155,14 @@ func GetStore(ctx context.Context) *Store {
return nil
}
+// GetExecutionStore extracts the metrics ExecutionStateTracker for the
+// given context of a bundle.
+// Ensure that the store for the bundle exist before calling this function.
+func GetExecutionStore(ctx context.Context) *ExecutionStateTracker {
+ store := GetStore(ctx)
+ return &store.executionStore
+}
Review comment:
Let's try to avoid exposing internal state like this, as it complicates the concurrency handling. This implies we might want to move more of the state sampler handling into the metrics package, with lighter configuration and actuation in the harness. Afterall, the harness is largely signalling "Hey, contribute whatever the current difference is to your current state and start counting again".
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *metrics.ExecutionStateTracker
+ store *metrics.Store
+ pid string // PTransform ID
+}
+
+func newSampler(e *metrics.ExecutionStateTracker, store *metrics.Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: e, store: store, pid: pid}
+}
+
+func (s StateSampler) startSampler() {
Review comment:
Prefer to use pointer methods instead of value methods for the state sampler. Every method call will be copying the state of the various fields which will take time, which the pointer will not.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
+ CurrentState BundleProcState
+ State ExecutionState
+ NumberOfTransitions int
+ MillisSinceLastTransition int
+ TransitionsAtLastSample int
+ LastSampleTimeMillis int
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
+
+ executionStore ExecutionStateTracker
+ stateRegistry map[Labels][]ExecutionState
Review comment:
Changing the "states" to be int's also lets us change these slices to fixed arrays: `[3]ExecutionState`, since we know there's always those 3 states.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
+ CurrentState BundleProcState
+ State ExecutionState
+ NumberOfTransitions int
+ MillisSinceLastTransition int
+ TransitionsAtLastSample int
+ LastSampleTimeMillis int
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
+
+ executionStore ExecutionStateTracker
+ stateRegistry map[Labels][]ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric)}
+ return &Store{store: make(map[Labels]userMetric), stateRegistry: make(map[Labels][]ExecutionState)}
+}
+
+// AddState adds an ExecutionState of a PTransform to the stateRegistry.
+func (s *Store) AddState(e ExecutionState, pid string) {
+ label := PTransformLabels(pid)
+ if s.stateRegistry == nil {
+ s.stateRegistry = make(map[Labels][]ExecutionState)
+ }
+ s.stateRegistry[label] = append(s.stateRegistry[label], e)
+}
+
+// SetState updates the state of a bundle.
+// For the bundle in start state, update the ExecutionState struct as well.
+func (s *Store) SetState(bs BundleProcState) {
+ if bs == StartBundle {
+ s.executionStore.State.State = bs
+ }
+ s.executionStore.CurrentState = bs
+}
+
+// IncTransitions increment the number of transitions by 1.
+func (s *Store) IncTransitions() {
+ s.executionStore.NumberOfTransitions += 1
+}
+
+// GetRegistry return the state registry.
+func (s *Store) GetRegistry() map[Labels][]ExecutionState {
+ return s.stateRegistry
}
Review comment:
Instead of putting these as methods on store, instead lets have them as methods on a "ExecutionStateTracker" object in the metrics package (similar, but not identical) to how the existing user metrics operate. We initialize the sampler in each `exec.ParDo` in the `Up` method (which is what `Up` is for), and then pass the context into the methods for it that way instead.
That way all the concurrency handling is isolated into the metrics package which, as you've observed already has similar mechanisms at work.
It doesn't need to be 100% identical to the existing metrics because we know we're creating one of these per Bundle anyway (which is not always true for user metrics). The main difference is we don't need the separated "counterSet" like we have for the user counters, since that's to deal with runtime declared counters that are constantly created per call.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
Review comment:
While not technically a state, a "TotalBundle" would be a good way to identically track/handle the total metrics.
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
+
+// ExecutionState stores the information about a bundle in a particular state.
+type ExecutionState struct {
+ State BundleProcState
+ IsProcessing bool // set to true when sent as a response to ProcessBundleProgress Request
+ TotalTimeMillis int64
+}
+
+// ExecutionStateTracker stores information about a bundle for execution time metrics.
+type ExecutionStateTracker struct {
+ CurrentState BundleProcState
+ State ExecutionState
+ NumberOfTransitions int
+ MillisSinceLastTransition int
+ TransitionsAtLastSample int
+ LastSampleTimeMillis int
+}
+
// Store retains per transform countersets, intended for per bundle use.
type Store struct {
mu sync.RWMutex
css []*ptCounterSet
store map[Labels]userMetric
+
+ executionStore ExecutionStateTracker
+ stateRegistry map[Labels][]ExecutionState
}
func newStore() *Store {
- return &Store{store: make(map[Labels]userMetric)}
+ return &Store{store: make(map[Labels]userMetric), stateRegistry: make(map[Labels][]ExecutionState)}
+}
+
+// AddState adds an ExecutionState of a PTransform to the stateRegistry.
+func (s *Store) AddState(e ExecutionState, pid string) {
+ label := PTransformLabels(pid)
+ if s.stateRegistry == nil {
+ s.stateRegistry = make(map[Labels][]ExecutionState)
+ }
+ s.stateRegistry[label] = append(s.stateRegistry[label], e)
Review comment:
AddState is being called in the sample method, and nothing clears these away. That means that each individual sample is retained for the life of the bundle.
Probably not noticeable in streaming, but in batch, a bundle can last for hours, and that will add up to an astounding number of allocations.
We need to sum up contributions for the same PTransform + State combination (bundle is implied by construction). This can be facilitated by having the state be an int under the hood, and using a fixed sized array as mentioned elsewhere.
To be concrete a Bundle that has 1 DoFn should have at most, 4 exec metrics reported for it (start, process, finish, and total). 4 DoFns would have 16, etc.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738491007
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -136,6 +136,10 @@ func SetPTransformID(ctx context.Context, id string) context.Context {
// Checking for *beamCtx is an optimization, so we don't dig deeply
// for ids if not necessary.
if bctx, ok := ctx.(*beamCtx); ok {
+ if _, ok := bctx.store.stateRegistry[id]; !ok {
+ bctx.store.stateRegistry[id] = [4]*ExecutionState{&ExecutionState{}, &ExecutionState{}, &ExecutionState{}, &ExecutionState{}}
Review comment:
Other function will make a copy of it and the original registry won't store the written values.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738681699
##########
File path: sdks/go/pkg/beam/core/metrics/sampler_test.go
##########
@@ -0,0 +1,212 @@
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func checkStateTime(t *testing.T, s StateSampler, label string, sb, pb, fb, tb time.Duration) {
+ t.Helper()
+ r := s.store.stateRegistry
+ v := r[label]
Review comment:
I think you can inline the `s.store.stateRegistry` part here, since it's the only use of `r`.
--
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] lostluck commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954209719
Run Go 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738457370
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -136,6 +136,10 @@ func SetPTransformID(ctx context.Context, id string) context.Context {
// Checking for *beamCtx is an optimization, so we don't dig deeply
// for ids if not necessary.
if bctx, ok := ctx.(*beamCtx); ok {
+ if _, ok := bctx.store.stateRegistry[id]; !ok {
+ bctx.store.stateRegistry[id] = [4]*ExecutionState{&ExecutionState{}, &ExecutionState{}, &ExecutionState{}, &ExecutionState{}}
Review comment:
`stateRegistry ` is of pointer value. If we do `[4]*ExecutionState{}` it throws error as `assignment to nil map`.
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-954901448
Run Go 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-940424997
PTAL. Addressed the old comments. However, I didn't understand clearly, the approach of initializing sampler in exec.Pardo's Up method.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r723514613
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -145,16 +150,73 @@ type ptCounterSet struct {
gauges map[nameHash]*gauge
}
+// Bundle processing state (START_BUNDLE, PROCESS_BUNDLE, FINISH_BUNDLE)
+type BundleProcState string
+
+const (
+ StartBundle BundleProcState = "START_BUNDLE"
+ ProcessBundle BundleProcState = "PROCESS_BUNDLE"
+ FinishBundle BundleProcState = "FINISH_BUNDLE"
+)
Review comment:
Got it. Thanks!
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r723505981
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
Review comment:
Yup.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r738778127
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ nextLogTime time.Duration
+ logInterval time.Duration
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(store *Store) StateSampler {
+ return StateSampler{store: store, nextLogTime: 5 * time.Minute, logInterval: 5 * time.Minute}
+}
+
+// Sample checks for state transition in processing a DoFn
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ s.transitionsAtLastSample = ps.transitions
Review comment:
We need to reset next log time here: `s.nextLogTime = s.logInterval` since a transition has happened.
Otherwise, we end up with ever increasing nextLogTimes 5 minutes, 10 minutes, 15 minutes etc. even if the state moves, the next "long running state" will take longer to produce the log.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r732819395
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,55 @@
+// 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 metrics
+
+import (
+ "context"
+ "time"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(ctx context.Context, store *Store) StateSampler {
+ return StateSampler{store: store}
+}
+
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(ctx)
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ e := s.store
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ e.transitions = ps.transitions
Review comment:
You're right.
--
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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r723505782
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -273,6 +273,54 @@ func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) {
return plan, nil
}
+// Sampler provides methods for implementing state sampling
+// to track execution time metrics.
+type Sampler interface {
+ startSampler()
+ stopSampler()
+ sample()
+}
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ done chan (int) // signal to stop sampling
+ e *metrics.ExecutionStateTracker
+ store *metrics.Store
+ pid string // PTransform ID
+}
+
+func newSampler(e *metrics.ExecutionStateTracker, store *metrics.Store, pid string) StateSampler {
+ return StateSampler{done: make(chan int), e: e, store: store, pid: pid}
+}
+
+func (s StateSampler) startSampler() {
Review comment:
Got 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] riteshghorse commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r730078534
##########
File path: sdks/go/pkg/beam/core/runtime/harness/harness.go
##########
@@ -316,13 +316,20 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe
if err != nil {
return fail(ctx, instID, "Failed: %v", err)
}
+ // Start the sampler goroutine here
+ sampler := newSampler(ctx, store)
+ sampler.start(ctx, time.Millisecond*200)
Review comment:
Got it. Missed this. Thanks!
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -17,58 +17,73 @@ package metrics
import (
"context"
- "sync/atomic"
"time"
)
// StateSampler tracks the state of a bundle.
type StateSampler struct {
- done chan (int) // signal to stop sampling
- e *ExecutionTracker
- store *Store
- total int64
+ store *Store // used to store states into state registry
}
// NewSampler creates a new state sampler.
func NewSampler(ctx context.Context, store *Store) StateSampler {
- return StateSampler{done: make(chan int), e: getExecutionStore(ctx), store: store}
+ return StateSampler{store: store}
}
-// Start is called from the harness package repeatedly whenever required
-func (s *StateSampler) Start(ctx context.Context, t time.Duration) {
- s.startSampler(ctx, t)
+func initialize() [4]*ExecutionState {
+ var v [4]*ExecutionState
+ for i := 0; i < 4; i++ {
+ v[i] = &ExecutionState{}
+ }
+ return v
}
-func (s *StateSampler) startSampler(ctx context.Context, t time.Duration) {
- if loadTransitions(ctx) != atomic.LoadInt64(&(s.e.TransitionsAtLastSample)) {
- s.sample(ctx)
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadPTransformState(ctx)
+ pid := PTransformLabels(ps.pid)
+
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if _, ok := s.store.stateRegistry[pid]; !ok {
+ s.store.stateRegistry[pid] = initialize()
+ }
+ if v, ok := s.store.stateRegistry[pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
}
- atomic.AddInt64(&s.e.MillisSinceLastTransition, int64(t))
- atomic.AddInt64(&s.e.State.TotalTimeMillis, int64(t))
-}
+ if _, ok := s.store.executionStore[pid]; !ok {
+ s.store.executionStore[pid] = &executionTracker{}
+ }
+ if v, ok := s.store.executionStore[pid]; ok {
-func (s *StateSampler) Stop(ctx context.Context, t time.Duration) {
- s.stopSampler(ctx, t)
+ if v.transitionsAtLastSample != ps.transitions {
+ // state change
+ v.millisSinceLastTransition = 0
+ v.numberOfTransitions = ps.transitions
+ v.transitionsAtLastSample = ps.transitions
+ } else {
+ v.millisSinceLastTransition += t
+ }
+ }
}
-func (s *StateSampler) stopSampler(ctx context.Context, t time.Duration) {
- // collect the remaining metrics (finish bundle metrics)
- s.sample(ctx)
- // add final state
-
- ex := ExecutionState{State: TotalBundle, TotalTimeMillis: s.total}
- s.store.AddState(ex, loadPTransformState(ctx).pid)
+// Start is called from the harness package repeatedly whenever required
+func (s *StateSampler) Start(ctx context.Context, t time.Duration) {
+ s.Sample(ctx, t)
}
-func (s *StateSampler) sample(ctx context.Context) {
+func (s *StateSampler) Stop(ctx context.Context) {
Review comment:
Okay. This would mean that in some cases when the PTransform transitions to FinishBundle and immediately stops, we won't have record for that. But as we are aiming for approximate time, this won't make much of a difference.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r739375972
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -55,24 +56,31 @@ func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
// state change detected
s.millisSinceLastTransition = 0
s.transitionsAtLastSample = ps.transitions
+ s.nextLogTime = s.logInterval
} else {
s.millisSinceLastTransition += t
}
if s.millisSinceLastTransition > s.nextLogTime {
- log.Info(ctx, "...standard long running operation log ...", ps.pid, ps.state, s.millisSinceLastTransition)
+ log.Info(ctx, "Operation ongoing in transform "+ps.pid+
+ " for at least "+fmt.Sprint(s.millisSinceLastTransition)+
+ " without outputting or completing in state "+getState(ps.state))
Review comment:
In Go, don't concat messages, use formatting directives. In this case use Infof.
```suggestion
log.Infof(ctx, "Operation ongoing in transform %v for at least %v without outputting or completing in state %v",
ps.pid, s.millisSinceLastTransition, getState(ps.state))
```
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -96,3 +104,16 @@ func (s *PTransformState) Set(ctx context.Context, state bundleProcState) {
atomic.AddInt64(bctx.store.transitions, 1)
}
}
+
+func getState(s bundleProcState) string {
+ switch s {
+ case 0:
+ return "START_BUNDLE"
+ case 1:
+ return "PROCESS_BUNDLE"
+ case 2:
+ return "FINISH BUNDLE"
Review comment:
```suggestion
return "FINISH_BUNDLE"
```
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r739391168
##########
File path: sdks/go/pkg/beam/core/metrics/sampler.go
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// 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 metrics
+
+import (
+ "context"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+)
+
+// StateSampler tracks the state of a bundle.
+type StateSampler struct {
+ store *Store // used to store states into state registry
+ millisSinceLastTransition time.Duration
+ transitionsAtLastSample int64
+ nextLogTime time.Duration
+ logInterval time.Duration
+}
+
+// NewSampler creates a new state sampler.
+func NewSampler(store *Store) StateSampler {
+ return StateSampler{store: store, nextLogTime: 5 * time.Minute, logInterval: 5 * time.Minute}
+}
+
+// Sample checks for state transition in processing a DoFn
+func (s *StateSampler) Sample(ctx context.Context, t time.Duration) {
+ ps := loadCurrentState(s)
+ if ps.pid == "" {
+ return
+ }
+ s.store.mu.Lock()
+ defer s.store.mu.Unlock()
+
+ if v, ok := s.store.stateRegistry[ps.pid]; ok {
+ v[ps.state].TotalTime += t
+ v[TotalBundle].TotalTime += t
+
+ if s.transitionsAtLastSample != ps.transitions {
+ // state change detected
+ s.millisSinceLastTransition = 0
+ s.transitionsAtLastSample = ps.transitions
+ } else {
+ s.millisSinceLastTransition += t
+ }
+
+ if s.millisSinceLastTransition > s.nextLogTime {
+ log.Info(ctx, "...standard long running operation log ...", ps.pid, ps.state, s.millisSinceLastTransition)
Review comment:
No worries. It's easy to get caught up in apparent urgency.
There's a saying, Slow is Fast. Going fast often leads to technical debt and mistakes that will slow things down later, when compared to taking a bit more time initially, to get it right. Usually discussed in context of larger product roadmaps and team scale development (eg https://www.infoq.com/articles/slow-down-go-faster/). It does apply on an individual level too though.
In this instance, a trick is to review your own code, and try to catch issues yourself before sending them out. Yes, you won't always catch your own mistakes (being blind to mistakes in code you just wrote is common), but over time you will get better at it. The way I've approached this in the past has been to make the changes, go work on something else for a bit (go workout, eat, another change etc.) then take a look at the new changes, and see if there's anything else to polish up. Then again at the whole current PR. Make those fixes and then take another look, and then send it out for review by someone else.
It's important to not get too caught up in increasing the scope when doing this (eg. Oh I can just do this part now...), but that's another issue.
--
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] lostluck commented on a change in pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #15657:
URL: https://github.com/apache/beam/pull/15657#discussion_r731992100
##########
File path: sdks/go/pkg/beam/core/metrics/store.go
##########
@@ -174,36 +174,35 @@ type executionTracker struct {
numberOfTransitions int64
millisSinceLastTransition time.Duration
transitionsAtLastSample int64
-}
-
-// executionState is used to store as atomic.Value in Store.
-type pTransformState struct {
- pid string
- state bundleProcState
-}
-
-func newPTransformState(pid string, state bundleProcState) *pTransformState {
- return &pTransformState{pid: pid, state: state}
+ currentState bundleProcState
+ pid string
}
func SetPTransformState(ctx context.Context, state bundleProcState) {
if bctx, ok := ctx.(*beamCtx); ok {
- ps := newPTransformState(bctx.ptransformID, state)
- bctx.pStore.Store(ps)
- atomic.AddInt64(&bctx.transitions, 1)
+ pid := bctx.ptransformID
+ bctx.store.mu.Lock()
+ bctx.store.executionStore.pid = pid
+ bctx.store.executionStore.currentState = state
+ bctx.store.mu.Unlock()
+ atomic.AddInt64(&bctx.store.executionStore.numberOfTransitions, 1)
Review comment:
Interesting! That's a very good point. This is why we measure.
The reason atomic.Value is allocating so much is because it's wrapping things into `interface{}`, which requires values to escape to the heap most of the time. Since that's the case, we may want to use a 2nd lock for this sort of management (rather than overburden the main store lock with multiple jobs). This also means we can remove the atomic from transitons, and just use the lock in question.
--
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] riteshghorse commented on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse commented on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-945959699
PTAL. The second test works as expected.
The precommit here is failing in the `metrics_test`. When we do the metrics dump we expected only the counter values earlier. But now the addition of [MsecsInt64](https://github.com/apache/beam/blob/57c81d7aedb1c334fa17dc399bd75af808ed8901/sdks/go/pkg/beam/core/metrics/store.go#L97) to the [Extractor](https://github.com/apache/beam/blob/57c81d7aedb1c334fa17dc399bd75af808ed8901/sdks/go/pkg/beam/core/metrics/store.go#L89) will dump execution state metric value as well and the test fails.
--
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] riteshghorse edited a comment on pull request #15657: [BEAM-13001] DoFn metrics for Go SDK
Posted by GitBox <gi...@apache.org>.
riteshghorse edited a comment on pull request #15657:
URL: https://github.com/apache/beam/pull/15657#issuecomment-948108709
PTAL. The pre-allocated states makes the difference. Thanks. For now, I've added them in the `metrics.Store`.
--
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