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 2020/03/26 00:29:55 UTC
[GitHub] [beam] lostluck opened a new pull request #11231: [BEAM-4374]
Shortids for the Go SDK
lostluck opened a new pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231
A quick unblocker for current proto changes. Adds Short ID handling for the Go SDK.
There are likely some performance improvements that can go in around the single lock, either moving them to a RW lock or a sync.Map, but this should be fine for Batch cases at present, until we measure differently.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
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.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] TheNeuralBit commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r402425925
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -28,10 +31,180 @@ import (
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+
+// TODO: Pull these from the protos.
+var sUrns = [...]string{
+ "beam:metric:user:sum_int64:v1",
+ "beam:metric:user:sum_double:v1",
+ "beam:metric:user:distribution_int64:v1",
+ "beam:metric:user:distribution_double:v1",
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:latest_double:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:top_n_double:v1",
+ "beam:metric:user:bottom_n_int64:v1",
+ "beam:metric:user:bottom_n_double:v1",
+
+ "beam:metric:element_count:v1",
+ "beam:metric:sampled_byte_size:v1",
+
+ "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",
+
+ "beam:metric:ptransform_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUserSumInt64 mUrn = iota
+ urnUserSumFloat64
+ urnUserDistInt64
+ urnUserDistFloat64
+ urnUserLatestMsInt64
+ urnUserLatestMsFloat64
+ urnUserTopNInt64
+ urnUserTopNFloat64
+ urnUserBottomNInt64
+ urnUserBottomNFloat64
+
+ urnElementCount
+ urnSampledByteSize
+
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnTransformTotalTime
+
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+// urnToType maps the urn to it's encoding type.
+// This function is written to be inlinable by the compiler.
+func urnToType(u mUrn) string {
+ switch u {
+ case urnUserSumInt64, urnElementCount, urnStartBundle, urnProcessBundle, urnFinishBundle, urnTransformTotalTime:
+ return "beam:metrics:sum_int64:v1"
+ case urnUserSumFloat64:
+ return "beam:metrics:sum_double:v1"
+ case urnUserDistInt64, urnSampledByteSize:
+ return "beam:metrics:distribution_int64:v1"
+ case urnUserDistFloat64:
+ return "beam:metrics:distribution_double:v1"
+ case urnUserLatestMsInt64:
+ return "beam:metrics:latest_int64:v1"
+ case urnUserLatestMsFloat64:
+ return "beam:metrics:latest_double:v1"
+ case urnUserTopNInt64:
+ return "beam:metrics:top_n_int64:v1"
+ case urnUserTopNFloat64:
+ return "beam:metrics:top_n_double:v1"
+ case urnUserBottomNInt64:
+ return "beam:metrics:bottom_n_int64:v1"
+ case urnUserBottomNFloat64:
+ return "beam:metrics:bottom_n_double:v1"
+
+ case urnProgressRemaining, urnProgressCompleted:
+ return "beam:metrics:progress:v1"
+
+ // Monitoring Table isn't currently in the protos.
+ // case ???:
+ // return "beam:metrics:monitoring_table:v1"
+
+ case urnTestSentinel:
+ return "TestingSentinelType"
+
+ default:
+ panic("metric urn without specified type" + sUrns[u])
+ }
+}
+
+type shortKey struct {
+ metrics.Labels
+ Urn mUrn // Urns fully specify their type.
+}
+
+// shortIDCache retains lookup caches for short ids to the full monitoring
+// info metadata.
+//
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+type shortIDCache struct {
+ mu sync.Mutex
+ labels2ShortIds map[shortKey]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+}
+
+func newShortIDCache() *shortIDCache {
+ return &shortIDCache{
+ labels2ShortIds: make(map[shortKey]string),
+ shortIds2Infos: make(map[string]*ppb.MonitoringInfo),
+ }
+}
+
+func (c *shortIDCache) getNextShortID() string {
+ id := atomic.AddInt64(&c.lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
+ return strconv.FormatInt(id, 36)
+}
+
+// getShortID returns the short id for the given metric, and if
+// it doesn't exist yet, stores the metadata.
+// Assumes c.mu lock is held.
+func (c *shortIDCache) getShortID(l metrics.Labels, urn mUrn) string {
+ k := shortKey{l, urn}
+ s, ok := c.labels2ShortIds[k]
+ if ok {
+ return s
+ }
+ s = c.getNextShortID()
+ c.labels2ShortIds[k] = s
+ c.shortIds2Infos[s] = &ppb.MonitoringInfo{
+ Urn: sUrns[urn],
+ Type: urnToType(urn),
+ Labels: userLabels(l),
+ }
+ return s
+}
+
+func (c *shortIDCache) shortIdsToInfos(shortids []string) map[string]*ppb.MonitoringInfo {
+ c.mu.Lock()
+ defer c.mu.Unlock()
Review comment:
Oops sorry for the noise @lostluck, thanks for the explanation
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-604162515
This code drifts slightly from https://github.com/apache/beam/pull/11184, which removes the OneOf in favour of always using the payload format. This version still uses the oneof, but only populates payload.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] TheNeuralBit commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r401969044
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -28,10 +31,180 @@ import (
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+
+// TODO: Pull these from the protos.
+var sUrns = [...]string{
+ "beam:metric:user:sum_int64:v1",
+ "beam:metric:user:sum_double:v1",
+ "beam:metric:user:distribution_int64:v1",
+ "beam:metric:user:distribution_double:v1",
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:latest_double:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:top_n_double:v1",
+ "beam:metric:user:bottom_n_int64:v1",
+ "beam:metric:user:bottom_n_double:v1",
+
+ "beam:metric:element_count:v1",
+ "beam:metric:sampled_byte_size:v1",
+
+ "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",
+
+ "beam:metric:ptransform_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUserSumInt64 mUrn = iota
+ urnUserSumFloat64
+ urnUserDistInt64
+ urnUserDistFloat64
+ urnUserLatestMsInt64
+ urnUserLatestMsFloat64
+ urnUserTopNInt64
+ urnUserTopNFloat64
+ urnUserBottomNInt64
+ urnUserBottomNFloat64
+
+ urnElementCount
+ urnSampledByteSize
+
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnTransformTotalTime
+
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+// urnToType maps the urn to it's encoding type.
+// This function is written to be inlinable by the compiler.
+func urnToType(u mUrn) string {
+ switch u {
+ case urnUserSumInt64, urnElementCount, urnStartBundle, urnProcessBundle, urnFinishBundle, urnTransformTotalTime:
+ return "beam:metrics:sum_int64:v1"
+ case urnUserSumFloat64:
+ return "beam:metrics:sum_double:v1"
+ case urnUserDistInt64, urnSampledByteSize:
+ return "beam:metrics:distribution_int64:v1"
+ case urnUserDistFloat64:
+ return "beam:metrics:distribution_double:v1"
+ case urnUserLatestMsInt64:
+ return "beam:metrics:latest_int64:v1"
+ case urnUserLatestMsFloat64:
+ return "beam:metrics:latest_double:v1"
+ case urnUserTopNInt64:
+ return "beam:metrics:top_n_int64:v1"
+ case urnUserTopNFloat64:
+ return "beam:metrics:top_n_double:v1"
+ case urnUserBottomNInt64:
+ return "beam:metrics:bottom_n_int64:v1"
+ case urnUserBottomNFloat64:
+ return "beam:metrics:bottom_n_double:v1"
+
+ case urnProgressRemaining, urnProgressCompleted:
+ return "beam:metrics:progress:v1"
+
+ // Monitoring Table isn't currently in the protos.
+ // case ???:
+ // return "beam:metrics:monitoring_table:v1"
+
+ case urnTestSentinel:
+ return "TestingSentinelType"
+
+ default:
+ panic("metric urn without specified type" + sUrns[u])
+ }
+}
+
+type shortKey struct {
+ metrics.Labels
+ Urn mUrn // Urns fully specify their type.
+}
+
+// shortIDCache retains lookup caches for short ids to the full monitoring
+// info metadata.
+//
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+type shortIDCache struct {
+ mu sync.Mutex
+ labels2ShortIds map[shortKey]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+}
+
+func newShortIDCache() *shortIDCache {
+ return &shortIDCache{
+ labels2ShortIds: make(map[shortKey]string),
+ shortIds2Infos: make(map[string]*ppb.MonitoringInfo),
+ }
+}
+
+func (c *shortIDCache) getNextShortID() string {
+ id := atomic.AddInt64(&c.lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
+ return strconv.FormatInt(id, 36)
+}
+
+// getShortID returns the short id for the given metric, and if
+// it doesn't exist yet, stores the metadata.
+// Assumes c.mu lock is held.
+func (c *shortIDCache) getShortID(l metrics.Labels, urn mUrn) string {
+ k := shortKey{l, urn}
+ s, ok := c.labels2ShortIds[k]
+ if ok {
+ return s
+ }
+ s = c.getNextShortID()
+ c.labels2ShortIds[k] = s
+ c.shortIds2Infos[s] = &ppb.MonitoringInfo{
+ Urn: sUrns[urn],
+ Type: urnToType(urn),
+ Labels: userLabels(l),
+ }
+ return s
+}
+
+func (c *shortIDCache) shortIdsToInfos(shortids []string) map[string]*ppb.MonitoringInfo {
+ c.mu.Lock()
+ defer c.mu.Unlock()
Review comment:
Is the locking correct here? I'd think you'd want to acquire the lock in `getShortId` since that's where the race condition could occur. This function is only reading `shortIds2Infos` which should only gain more entries over time and entries are never modified, so I'd think it would be thread-safe even without acquiring the lock. Am I missing something?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-605395052
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398760431
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,71 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+var (
+ shortMu sync.Mutex
+ labels2ShortIds map[metrics.Labels]string
Review comment:
This won't be enough as the key. There are things that have the same labels but different URNs, e.g start/process/finish msecs
I would suggest using the MonitoringInfo with the payload field "blanked" as the key of the 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-604713667
Just as a note, I'll wait until https://github.com/apache/beam/pull/11184 is in, and resolve the merge conflicts on my end before we merge this one.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-604159962
R: @lukecwik
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398818391
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,71 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+var (
+ shortMu sync.Mutex
+ labels2ShortIds map[metrics.Labels]string
Review comment:
Ah good to know. 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398949690
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring_test.go
##########
@@ -0,0 +1,122 @@
+// 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 (
+ "testing"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+)
+
+func TestGetShortID(t *testing.T) {
+ tests := []struct {
+ id string
+ urn mUrn
+ typ mType
+ expectedUrn string
+ expectedType string
+ }{
+ {
+ id: "1",
+ urn: urnUser,
Review comment:
Can you add the case where the same urn but unique labels are used gets a different short id?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398886664
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,71 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+var (
+ shortMu sync.Mutex
+ labels2ShortIds map[metrics.Labels]string
Review comment:
Ah good point.
Can't use protos as Go Map keys, because of all the magic fields they have, but I can use other things.
I've put in aligned constants, types, and string arrays for the proto specified strings, so these lookups don't end up hashing the strings every time (and instead use a uint32, which is very fast for go maps to deal with.) There's still the hashing of the fields in metrics.Labels, but we can do the same hashing in the metrics code at a later time, to allow for faster lookups for those instead.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r399602340
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,165 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+type mType uint32
+
+// TODO: Pull these from the protos.
+var sUrns = []string{
+ "beam:metric:user:v1",
+ "beam:metric:element_count:v1",
+ "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_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUser mUrn = iota
+ urnElementCount
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+var sTypes = []string{
+ "beam:metrics:sum_int64:v1",
+ "beam:metrics:sum_double:v1",
+ "beam:metrics:distribution_int64:v1",
+ "beam:metrics:distribution_double:v1",
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:latest_double:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:top_n_double:v1",
+ "beam:metrics:bottom_n_int64:v1",
+ "beam:metrics:bottom_n_double:v1",
+ "beam:metrics:monitoring_table:v1",
+ "beam:metrics:progress:v1",
+
+ "TestingSentinelType", // Must remain last.
+}
+
+const (
Review comment:
Ack. I've kept the extra marker type around for the moment, but I might collapse things into the function to simplify some thing.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r402026279
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -28,10 +31,180 @@ import (
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+
+// TODO: Pull these from the protos.
+var sUrns = [...]string{
+ "beam:metric:user:sum_int64:v1",
+ "beam:metric:user:sum_double:v1",
+ "beam:metric:user:distribution_int64:v1",
+ "beam:metric:user:distribution_double:v1",
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:latest_double:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:top_n_double:v1",
+ "beam:metric:user:bottom_n_int64:v1",
+ "beam:metric:user:bottom_n_double:v1",
+
+ "beam:metric:element_count:v1",
+ "beam:metric:sampled_byte_size:v1",
+
+ "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",
+
+ "beam:metric:ptransform_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUserSumInt64 mUrn = iota
+ urnUserSumFloat64
+ urnUserDistInt64
+ urnUserDistFloat64
+ urnUserLatestMsInt64
+ urnUserLatestMsFloat64
+ urnUserTopNInt64
+ urnUserTopNFloat64
+ urnUserBottomNInt64
+ urnUserBottomNFloat64
+
+ urnElementCount
+ urnSampledByteSize
+
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnTransformTotalTime
+
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+// urnToType maps the urn to it's encoding type.
+// This function is written to be inlinable by the compiler.
+func urnToType(u mUrn) string {
+ switch u {
+ case urnUserSumInt64, urnElementCount, urnStartBundle, urnProcessBundle, urnFinishBundle, urnTransformTotalTime:
+ return "beam:metrics:sum_int64:v1"
+ case urnUserSumFloat64:
+ return "beam:metrics:sum_double:v1"
+ case urnUserDistInt64, urnSampledByteSize:
+ return "beam:metrics:distribution_int64:v1"
+ case urnUserDistFloat64:
+ return "beam:metrics:distribution_double:v1"
+ case urnUserLatestMsInt64:
+ return "beam:metrics:latest_int64:v1"
+ case urnUserLatestMsFloat64:
+ return "beam:metrics:latest_double:v1"
+ case urnUserTopNInt64:
+ return "beam:metrics:top_n_int64:v1"
+ case urnUserTopNFloat64:
+ return "beam:metrics:top_n_double:v1"
+ case urnUserBottomNInt64:
+ return "beam:metrics:bottom_n_int64:v1"
+ case urnUserBottomNFloat64:
+ return "beam:metrics:bottom_n_double:v1"
+
+ case urnProgressRemaining, urnProgressCompleted:
+ return "beam:metrics:progress:v1"
+
+ // Monitoring Table isn't currently in the protos.
+ // case ???:
+ // return "beam:metrics:monitoring_table:v1"
+
+ case urnTestSentinel:
+ return "TestingSentinelType"
+
+ default:
+ panic("metric urn without specified type" + sUrns[u])
+ }
+}
+
+type shortKey struct {
+ metrics.Labels
+ Urn mUrn // Urns fully specify their type.
+}
+
+// shortIDCache retains lookup caches for short ids to the full monitoring
+// info metadata.
+//
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+type shortIDCache struct {
+ mu sync.Mutex
+ labels2ShortIds map[shortKey]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+}
+
+func newShortIDCache() *shortIDCache {
+ return &shortIDCache{
+ labels2ShortIds: make(map[shortKey]string),
+ shortIds2Infos: make(map[string]*ppb.MonitoringInfo),
+ }
+}
+
+func (c *shortIDCache) getNextShortID() string {
+ id := atomic.AddInt64(&c.lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
+ return strconv.FormatInt(id, 36)
+}
+
+// getShortID returns the short id for the given metric, and if
+// it doesn't exist yet, stores the metadata.
+// Assumes c.mu lock is held.
+func (c *shortIDCache) getShortID(l metrics.Labels, urn mUrn) string {
+ k := shortKey{l, urn}
+ s, ok := c.labels2ShortIds[k]
+ if ok {
+ return s
+ }
+ s = c.getNextShortID()
+ c.labels2ShortIds[k] = s
+ c.shortIds2Infos[s] = &ppb.MonitoringInfo{
+ Urn: sUrns[urn],
+ Type: urnToType(urn),
+ Labels: userLabels(l),
+ }
+ return s
+}
+
+func (c *shortIDCache) shortIdsToInfos(shortids []string) map[string]*ppb.MonitoringInfo {
+ c.mu.Lock()
+ defer c.mu.Unlock()
Review comment:
Thanks for the explanation again, I knew this before but blanked when @TheNeuralBit brought it up.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398818636
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,71 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+var (
+ shortMu sync.Mutex
+ labels2ShortIds map[metrics.Labels]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+)
+
+func getNextShortID() string {
+ id := atomic.AddInt64(&lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
Review comment:
Base36: It's what efficiency craves.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-605395102
R: @youngoli @lukecwik
I can't wait to get rid of both of the legacy stuff and the older monitoring info listing. That function is becoming crufty.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-604159898
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r401974455
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -28,10 +31,180 @@ import (
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+
+// TODO: Pull these from the protos.
+var sUrns = [...]string{
+ "beam:metric:user:sum_int64:v1",
+ "beam:metric:user:sum_double:v1",
+ "beam:metric:user:distribution_int64:v1",
+ "beam:metric:user:distribution_double:v1",
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:latest_double:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:top_n_double:v1",
+ "beam:metric:user:bottom_n_int64:v1",
+ "beam:metric:user:bottom_n_double:v1",
+
+ "beam:metric:element_count:v1",
+ "beam:metric:sampled_byte_size:v1",
+
+ "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",
+
+ "beam:metric:ptransform_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUserSumInt64 mUrn = iota
+ urnUserSumFloat64
+ urnUserDistInt64
+ urnUserDistFloat64
+ urnUserLatestMsInt64
+ urnUserLatestMsFloat64
+ urnUserTopNInt64
+ urnUserTopNFloat64
+ urnUserBottomNInt64
+ urnUserBottomNFloat64
+
+ urnElementCount
+ urnSampledByteSize
+
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnTransformTotalTime
+
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+// urnToType maps the urn to it's encoding type.
+// This function is written to be inlinable by the compiler.
+func urnToType(u mUrn) string {
+ switch u {
+ case urnUserSumInt64, urnElementCount, urnStartBundle, urnProcessBundle, urnFinishBundle, urnTransformTotalTime:
+ return "beam:metrics:sum_int64:v1"
+ case urnUserSumFloat64:
+ return "beam:metrics:sum_double:v1"
+ case urnUserDistInt64, urnSampledByteSize:
+ return "beam:metrics:distribution_int64:v1"
+ case urnUserDistFloat64:
+ return "beam:metrics:distribution_double:v1"
+ case urnUserLatestMsInt64:
+ return "beam:metrics:latest_int64:v1"
+ case urnUserLatestMsFloat64:
+ return "beam:metrics:latest_double:v1"
+ case urnUserTopNInt64:
+ return "beam:metrics:top_n_int64:v1"
+ case urnUserTopNFloat64:
+ return "beam:metrics:top_n_double:v1"
+ case urnUserBottomNInt64:
+ return "beam:metrics:bottom_n_int64:v1"
+ case urnUserBottomNFloat64:
+ return "beam:metrics:bottom_n_double:v1"
+
+ case urnProgressRemaining, urnProgressCompleted:
+ return "beam:metrics:progress:v1"
+
+ // Monitoring Table isn't currently in the protos.
+ // case ???:
+ // return "beam:metrics:monitoring_table:v1"
+
+ case urnTestSentinel:
+ return "TestingSentinelType"
+
+ default:
+ panic("metric urn without specified type" + sUrns[u])
+ }
+}
+
+type shortKey struct {
+ metrics.Labels
+ Urn mUrn // Urns fully specify their type.
+}
+
+// shortIDCache retains lookup caches for short ids to the full monitoring
+// info metadata.
+//
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+type shortIDCache struct {
+ mu sync.Mutex
+ labels2ShortIds map[shortKey]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+}
+
+func newShortIDCache() *shortIDCache {
+ return &shortIDCache{
+ labels2ShortIds: make(map[shortKey]string),
+ shortIds2Infos: make(map[string]*ppb.MonitoringInfo),
+ }
+}
+
+func (c *shortIDCache) getNextShortID() string {
+ id := atomic.AddInt64(&c.lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
+ return strconv.FormatInt(id, 36)
+}
+
+// getShortID returns the short id for the given metric, and if
+// it doesn't exist yet, stores the metadata.
+// Assumes c.mu lock is held.
+func (c *shortIDCache) getShortID(l metrics.Labels, urn mUrn) string {
+ k := shortKey{l, urn}
+ s, ok := c.labels2ShortIds[k]
+ if ok {
+ return s
+ }
+ s = c.getNextShortID()
+ c.labels2ShortIds[k] = s
+ c.shortIds2Infos[s] = &ppb.MonitoringInfo{
+ Urn: sUrns[urn],
+ Type: urnToType(urn),
+ Labels: userLabels(l),
+ }
+ return s
+}
+
+func (c *shortIDCache) shortIdsToInfos(shortids []string) map[string]*ppb.MonitoringInfo {
+ c.mu.Lock()
+ defer c.mu.Unlock()
Review comment:
CC: @lostluck
Thats a good point, it looks like we should also acquire the lock in getShortID after we generate the short id.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik merged pull request #11231: [BEAM-4374] Shortids
for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik merged pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r401974636
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -28,10 +31,180 @@ import (
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+
+// TODO: Pull these from the protos.
+var sUrns = [...]string{
+ "beam:metric:user:sum_int64:v1",
+ "beam:metric:user:sum_double:v1",
+ "beam:metric:user:distribution_int64:v1",
+ "beam:metric:user:distribution_double:v1",
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:latest_double:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:top_n_double:v1",
+ "beam:metric:user:bottom_n_int64:v1",
+ "beam:metric:user:bottom_n_double:v1",
+
+ "beam:metric:element_count:v1",
+ "beam:metric:sampled_byte_size:v1",
+
+ "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",
+
+ "beam:metric:ptransform_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUserSumInt64 mUrn = iota
+ urnUserSumFloat64
+ urnUserDistInt64
+ urnUserDistFloat64
+ urnUserLatestMsInt64
+ urnUserLatestMsFloat64
+ urnUserTopNInt64
+ urnUserTopNFloat64
+ urnUserBottomNInt64
+ urnUserBottomNFloat64
+
+ urnElementCount
+ urnSampledByteSize
+
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnTransformTotalTime
+
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+// urnToType maps the urn to it's encoding type.
+// This function is written to be inlinable by the compiler.
+func urnToType(u mUrn) string {
+ switch u {
+ case urnUserSumInt64, urnElementCount, urnStartBundle, urnProcessBundle, urnFinishBundle, urnTransformTotalTime:
+ return "beam:metrics:sum_int64:v1"
+ case urnUserSumFloat64:
+ return "beam:metrics:sum_double:v1"
+ case urnUserDistInt64, urnSampledByteSize:
+ return "beam:metrics:distribution_int64:v1"
+ case urnUserDistFloat64:
+ return "beam:metrics:distribution_double:v1"
+ case urnUserLatestMsInt64:
+ return "beam:metrics:latest_int64:v1"
+ case urnUserLatestMsFloat64:
+ return "beam:metrics:latest_double:v1"
+ case urnUserTopNInt64:
+ return "beam:metrics:top_n_int64:v1"
+ case urnUserTopNFloat64:
+ return "beam:metrics:top_n_double:v1"
+ case urnUserBottomNInt64:
+ return "beam:metrics:bottom_n_int64:v1"
+ case urnUserBottomNFloat64:
+ return "beam:metrics:bottom_n_double:v1"
+
+ case urnProgressRemaining, urnProgressCompleted:
+ return "beam:metrics:progress:v1"
+
+ // Monitoring Table isn't currently in the protos.
+ // case ???:
+ // return "beam:metrics:monitoring_table:v1"
+
+ case urnTestSentinel:
+ return "TestingSentinelType"
+
+ default:
+ panic("metric urn without specified type" + sUrns[u])
+ }
+}
+
+type shortKey struct {
+ metrics.Labels
+ Urn mUrn // Urns fully specify their type.
+}
+
+// shortIDCache retains lookup caches for short ids to the full monitoring
+// info metadata.
+//
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+type shortIDCache struct {
+ mu sync.Mutex
+ labels2ShortIds map[shortKey]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+}
+
+func newShortIDCache() *shortIDCache {
+ return &shortIDCache{
+ labels2ShortIds: make(map[shortKey]string),
+ shortIds2Infos: make(map[string]*ppb.MonitoringInfo),
+ }
+}
+
+func (c *shortIDCache) getNextShortID() string {
+ id := atomic.AddInt64(&c.lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
+ return strconv.FormatInt(id, 36)
+}
+
+// getShortID returns the short id for the given metric, and if
+// it doesn't exist yet, stores the metadata.
+// Assumes c.mu lock is held.
+func (c *shortIDCache) getShortID(l metrics.Labels, urn mUrn) string {
+ k := shortKey{l, urn}
+ s, ok := c.labels2ShortIds[k]
+ if ok {
+ return s
+ }
+ s = c.getNextShortID()
+ c.labels2ShortIds[k] = s
+ c.shortIds2Infos[s] = &ppb.MonitoringInfo{
+ Urn: sUrns[urn],
+ Type: urnToType(urn),
+ Labels: userLabels(l),
+ }
+ return s
+}
+
+func (c *shortIDCache) shortIdsToInfos(shortids []string) map[string]*ppb.MonitoringInfo {
+ c.mu.Lock()
+ defer c.mu.Unlock()
Review comment:
Locking by itself is a reasonably expensive operation so I made the choice to lock per metrics request. getShortId is called multiple times in that critical section.
shortIdsToInfo though is called outside that critical section, by potentially a different thread, so the lock here is to protect the reads from hitting the mutations in getShortId.
Though come to think of it, the Go SDK harness handles all runner requests other than the ProcessBundle instructions on the grpc goroutine, so we might not need any locking at all, which would definitely simplify things and make it faster. In order to unblock the proto changes I didn't optimize this section entirely so there's still work and clean up that can happen 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398948604
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,165 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+type mType uint32
+
+// TODO: Pull these from the protos.
+var sUrns = []string{
+ "beam:metric:user:v1",
+ "beam:metric:element_count:v1",
+ "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_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUser mUrn = iota
+ urnElementCount
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+var sTypes = []string{
+ "beam:metrics:sum_int64:v1",
+ "beam:metrics:sum_double:v1",
+ "beam:metrics:distribution_int64:v1",
+ "beam:metrics:distribution_double:v1",
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:latest_double:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:top_n_double:v1",
+ "beam:metrics:bottom_n_int64:v1",
+ "beam:metrics:bottom_n_double:v1",
+ "beam:metrics:monitoring_table:v1",
+ "beam:metrics:progress:v1",
+
+ "TestingSentinelType", // Must remain last.
+}
+
+const (
Review comment:
Since the urns uniquely identify the type now, you don't need this anymore and a monitoring info is uniquely described by urn + labels.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398948198
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,165 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+type mType uint32
+
+// TODO: Pull these from the protos.
+var sUrns = []string{
+ "beam:metric:user:v1",
Review comment:
heads up that this has now been exploded so that each MonitoringInfoSpec has a unique urn meaning that you'll see:
beam:metric:user:sum_int64:v1, beam:metric:user:sum_double:v1, ...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r402020087
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -28,10 +31,180 @@ import (
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+type mUrn uint32
+
+// TODO: Pull these from the protos.
+var sUrns = [...]string{
+ "beam:metric:user:sum_int64:v1",
+ "beam:metric:user:sum_double:v1",
+ "beam:metric:user:distribution_int64:v1",
+ "beam:metric:user:distribution_double:v1",
+ "beam:metric:user:latest_int64:v1",
+ "beam:metric:user:latest_double:v1",
+ "beam:metric:user:top_n_int64:v1",
+ "beam:metric:user:top_n_double:v1",
+ "beam:metric:user:bottom_n_int64:v1",
+ "beam:metric:user:bottom_n_double:v1",
+
+ "beam:metric:element_count:v1",
+ "beam:metric:sampled_byte_size:v1",
+
+ "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",
+
+ "beam:metric:ptransform_progress:remaining:v1",
+ "beam:metric:ptransform_progress:completed:v1",
+
+ "TestingSentinelUrn", // Must remain last.
+}
+
+const (
+ urnUserSumInt64 mUrn = iota
+ urnUserSumFloat64
+ urnUserDistInt64
+ urnUserDistFloat64
+ urnUserLatestMsInt64
+ urnUserLatestMsFloat64
+ urnUserTopNInt64
+ urnUserTopNFloat64
+ urnUserBottomNInt64
+ urnUserBottomNFloat64
+
+ urnElementCount
+ urnSampledByteSize
+
+ urnStartBundle
+ urnProcessBundle
+ urnFinishBundle
+ urnTransformTotalTime
+
+ urnProgressRemaining
+ urnProgressCompleted
+
+ urnTestSentinel // Must remain last.
+)
+
+// urnToType maps the urn to it's encoding type.
+// This function is written to be inlinable by the compiler.
+func urnToType(u mUrn) string {
+ switch u {
+ case urnUserSumInt64, urnElementCount, urnStartBundle, urnProcessBundle, urnFinishBundle, urnTransformTotalTime:
+ return "beam:metrics:sum_int64:v1"
+ case urnUserSumFloat64:
+ return "beam:metrics:sum_double:v1"
+ case urnUserDistInt64, urnSampledByteSize:
+ return "beam:metrics:distribution_int64:v1"
+ case urnUserDistFloat64:
+ return "beam:metrics:distribution_double:v1"
+ case urnUserLatestMsInt64:
+ return "beam:metrics:latest_int64:v1"
+ case urnUserLatestMsFloat64:
+ return "beam:metrics:latest_double:v1"
+ case urnUserTopNInt64:
+ return "beam:metrics:top_n_int64:v1"
+ case urnUserTopNFloat64:
+ return "beam:metrics:top_n_double:v1"
+ case urnUserBottomNInt64:
+ return "beam:metrics:bottom_n_int64:v1"
+ case urnUserBottomNFloat64:
+ return "beam:metrics:bottom_n_double:v1"
+
+ case urnProgressRemaining, urnProgressCompleted:
+ return "beam:metrics:progress:v1"
+
+ // Monitoring Table isn't currently in the protos.
+ // case ???:
+ // return "beam:metrics:monitoring_table:v1"
+
+ case urnTestSentinel:
+ return "TestingSentinelType"
+
+ default:
+ panic("metric urn without specified type" + sUrns[u])
+ }
+}
+
+type shortKey struct {
+ metrics.Labels
+ Urn mUrn // Urns fully specify their type.
+}
+
+// shortIDCache retains lookup caches for short ids to the full monitoring
+// info metadata.
+//
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+type shortIDCache struct {
+ mu sync.Mutex
+ labels2ShortIds map[shortKey]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+}
+
+func newShortIDCache() *shortIDCache {
+ return &shortIDCache{
+ labels2ShortIds: make(map[shortKey]string),
+ shortIds2Infos: make(map[string]*ppb.MonitoringInfo),
+ }
+}
+
+func (c *shortIDCache) getNextShortID() string {
+ id := atomic.AddInt64(&c.lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
+ return strconv.FormatInt(id, 36)
+}
+
+// getShortID returns the short id for the given metric, and if
+// it doesn't exist yet, stores the metadata.
+// Assumes c.mu lock is held.
+func (c *shortIDCache) getShortID(l metrics.Labels, urn mUrn) string {
+ k := shortKey{l, urn}
+ s, ok := c.labels2ShortIds[k]
+ if ok {
+ return s
+ }
+ s = c.getNextShortID()
+ c.labels2ShortIds[k] = s
+ c.shortIds2Infos[s] = &ppb.MonitoringInfo{
+ Urn: sUrns[urn],
+ Type: urnToType(urn),
+ Labels: userLabels(l),
+ }
+ return s
+}
+
+func (c *shortIDCache) shortIdsToInfos(shortids []string) map[string]*ppb.MonitoringInfo {
+ c.mu.Lock()
+ defer c.mu.Unlock()
Review comment:
Also, I did comment that getShortId assumes the lock is held.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11231:
[BEAM-4374] Shortids for the Go SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#discussion_r398761123
##########
File path: sdks/go/pkg/beam/core/runtime/harness/monitoring.go
##########
@@ -16,20 +16,71 @@
package harness
import (
+ "bytes"
+ "strconv"
+ "sync"
+ "sync/atomic"
"time"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/mtime"
"github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
fnpb "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
ppb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
"github.com/golang/protobuf/ptypes"
)
-func monitoring(p *exec.Plan) (*fnpb.Metrics, []*ppb.MonitoringInfo) {
+// TODO: 2020/03/26 - measure mutex overhead vs sync.Map for this case.
+// sync.Map might have lower contention for this read heavy load.
+var (
+ shortMu sync.Mutex
+ labels2ShortIds map[metrics.Labels]string
+ shortIds2Infos map[string]*ppb.MonitoringInfo
+
+ lastShortID int64
+)
+
+func getNextShortID() string {
+ id := atomic.AddInt64(&lastShortID, 1)
+ // No reason not to use the smallest string short ids possible.
Review comment:
+1
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-604680300
A rather large refactoring occurred. For one, I wasn't initializing the maps properly which correctly caused postcommit failures.
Now I've moved that verification to tests, and made the cache easier to test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-604681904
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lostluck commented on issue #11231: [BEAM-4374] Shortids for
the Go SDK
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #11231: [BEAM-4374] Shortids for the Go SDK
URL: https://github.com/apache/beam/pull/11231#issuecomment-605375733
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services