[
https://issues.apache.org/jira/browse/BEAM-9167?focusedWorklogId=380226&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-380226
]
ASF GitHub Bot logged work on BEAM-9167:
----------------------------------------
Author: ASF GitHub Bot
Created on: 31/Jan/20 22:24
Start Date: 31/Jan/20 22:24
Worklog Time Spent: 10m
Work Description: lostluck commented on pull request #10716: [BEAM-9167]
Metrics extraction refactoring.
URL: https://github.com/apache/beam/pull/10716#discussion_r373708761
##########
File path: sdks/go/pkg/beam/core/metrics/dumper.go
##########
@@ -0,0 +1,130 @@
+// 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"
+ "fmt"
+ "sort"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+)
+
+// DumpToLog is a debugging function that outputs all metrics available locally
+// to beam.Log.
+func DumpToLog(ctx context.Context) {
+ store := GetStore(ctx)
+ if store == nil {
+ log.Errorf(ctx, "Unable to dump metrics: provided context
doesn't contain metrics Store.")
+ return
+ }
+ DumpToLogFromStore(ctx, store)
+}
+
+// DumpToLogFromStore dumps the metrics in the provided Store to beam.Log.
+func DumpToLogFromStore(ctx context.Context, store *Store) {
+ dumperExtractor(store, func(format string, args ...interface{}) {
+ log.Errorf(ctx, format, args...)
+ })
+}
+
+// DumpToOutFromContext is a debugging function that outputs all metrics
+// available locally to std out,
+// extracting the metric store from the context.
+func DumpToOutFromContext(ctx context.Context) {
+ store := GetStore(ctx)
+ if store == nil {
+ fmt.Printf("Unable to dump metrics: provided context doesn't
contain metrics Store.")
+ return
+ }
+ DumpToOutFromStore(store)
+}
+
+// DumpToOutFromStore is a debugging function that outputs all metrics
+// available locally to std out directly from the store.
+func DumpToOutFromStore(store *Store) {
+ dumperExtractor(store, func(format string, args ...interface{}) {
+ fmt.Printf(format+"\n", args...)
+ })
+}
+
+func dumperExtractor(store *Store, p func(format string, args ...interface{}))
{
+ m := make(map[Labels]interface{})
+ e := &Extractor{
+ SumInt64: func(l Labels, v int64) {
+ m[l] = &counter{value: v}
+ },
+ DistributionInt64: func(l Labels, count, sum, min, max int64) {
+ m[l] = &distribution{count: count, sum: sum, min: min,
max: max}
+ },
+ GaugeInt64: func(l Labels, v int64, t time.Time) {
+ m[l] = &gauge{v: v, t: t}
+ },
+ }
+ e.ExtractFrom(store)
+ dumpTo(m, p)
+}
+
+type metricDumper struct {
Review comment:
Good catch. Oversight from a previous prototype.
----------------------------------------------------------------
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:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 380226)
Time Spent: 2h 50m (was: 2h 40m)
> Reduce overhead of Go SDK side metrics
> --------------------------------------
>
> Key: BEAM-9167
> URL: https://issues.apache.org/jira/browse/BEAM-9167
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-go
> Reporter: Robert Burke
> Assignee: Robert Burke
> Priority: Major
> Time Spent: 2h 50m
> Remaining Estimate: 0h
>
> Locking overhead due to the global store and local caches of SDK counter data
> can dominate certain workloads, which means we can do better.
> Instead of having a global store of metrics data to extract counters, we
> should use per ptransform (or per bundle) counter sets, which would avoid
> requiring locking per counter operation. The main detriment compared to the
> current implementation is that a user would need to add their own locking if
> they were to spawn multiple goroutines to process a Bundle's work in a DoFn.
> Given that self multithreaded DoFns aren't recommended/safe in Java, largely
> impossible in Python, and the other beam Go SDK provided constructs (like
> Iterators and Emitters) are not thread safe, this is a small concern,
> provided the documentation is clear on this.
> Removing the locking and switching to atomic ops reduces the overhead
> significantly in example jobs and in the benchmarks.
> A second part of this change should be to move the exec package to manage
> it's own per bundle state, rather than relying on a global datastore to
> extract the per bundle,per ptransform values.
> Related: https://issues.apache.org/jira/browse/BEAM-6541
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
