[
https://issues.apache.org/jira/browse/BEAM-3545?focusedWorklogId=85045&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-85045
]
ASF GitHub Bot logged work on BEAM-3545:
----------------------------------------
Author: ASF GitHub Bot
Created on: 27/Mar/18 22:18
Start Date: 27/Mar/18 22:18
Worklog Time Spent: 10m
Work Description: lostluck commented on a change in pull request #4899:
[BEAM-3545] Go SDK UserCounters
URL: https://github.com/apache/beam/pull/4899#discussion_r177589501
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -0,0 +1,479 @@
+// 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 implement the Beam metrics API, described at
+// http://s.apache.org/beam-metrics-api
+//
+// Metrics in the Beam model are uniquely identified by a namespace, a name,
+// and the PTransform context in which they are used. Further, they are
+// reported as a delta against the bundle being processed, so that overcounting
+// doesn't occur if a bundle needs to be retried. Each metric is scoped to
+// their bundle, and ptransform.
+//
+// Cells (or metric cells) are defined for each Beam model metric
+// type, and the serve as concurrency safe storage of a given metric's values.
+// Proxys are exported values representing the metric, for use in user
+// ptransform code. They don't retain their cells, since they don't have
+// the context to be able to store them for export back to the pipeline runner.
+//
+// Metric cells aren't initialized until their first mutation, which
+// follows from the Beam model design, where metrics are only sent for a bundle
+// if they have changed. This is particularly convenient for distributions
which
+// means their min and max fields can be set to the first value on creation
+// rather than have some marker of uninitialized state, which would otherwise
+// need to be checked for on every update.
+//
+// Metric values are implemented as lightweight proxies of the user provided
+// namespace and name. This allows them to be declared globally, and used in
+// any ParDo. Further, as per the design, they can be declared dynamically
+// at runtime.
+//
+// To handle reporting deltas on the metrics by bundle, metrics
+// are keyed by bundleID,PTransformID,namespace, and name, so metrics that
+// are identical except for bundles are treated as distinct, effectively
+// providing per bundle deltas, since a new value cell is used per bundle.
+package metrics
+
+import (
+ "context"
+ "fmt"
+ "sort"
+ "sync"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/log"
+ "github.com/apache/beam/sdks/go/pkg/beam/model/fnexecution_v1"
+ "github.com/golang/protobuf/ptypes"
+)
+
+// Metric cells are named and scoped by ptransform, and bundle,
+// the latter two of which are only known at runtime. We propagate
+// the PTransformID and BundleID via a context.Context. Consequently
+// using metrics requires the PTransform have a context.Context
+// argument.
+
+type ctxKey string
+
+const bundleKey ctxKey = "beam:bundle"
+const ptransformKey ctxKey = "beam:ptransform"
+
+// SetBundleID sets the id of the current Bundle.
+func SetBundleID(ctx context.Context, id string) context.Context {
+ return context.WithValue(ctx, bundleKey, id)
+}
+
+// SetPTransformID sets the id of the current PTransform.
+func SetPTransformID(ctx context.Context, id string) context.Context {
+ return context.WithValue(ctx, ptransformKey, id)
+}
+
+func getContextKey(ctx context.Context, n name) key {
+ key := key{name: n, bundle: "(bundle id unset)", ptransform:
"(ptransform id unset)"}
+ if id := ctx.Value(bundleKey); id != nil {
+ key.bundle = id.(string)
+ }
+ if id := ctx.Value(ptransformKey); id != nil {
+ key.ptransform = id.(string)
+ }
+ return key
+}
+
+// userMetric knows how to convert it's value to a Metrics_User proto.
+type userMetric interface {
+ toProto() *fnexecution_v1.Metrics_User
+}
+
+// name is a pair of strings identifying a specific metric.
+type name struct {
+ namespace, name string
+}
+
+func (n name) String() string {
+ return fmt.Sprintf("%s.%s", n.namespace, n.name)
+}
+
+func newName(ns, n string) name {
+ if len(n) == 0 || len(ns) == 0 {
+ panic(fmt.Sprintf("namespace and name are required to be
non-empty, got %q and %q", ns, n))
+ }
+ return name{namespace: ns, name: n}
+}
+
+type key struct {
+ name name
+ bundle, ptransform string
+}
+
+var (
+ // metricMu protects access to metricStorage
+ metricMu sync.RWMutex
+ // metricStorage is a map of BundleIDs to PtransformIDs to userMetrics.
+ // it permits us to extract metric protos for runners per data Bundle,
and
+ // per PTransform.
+ metricStorage = make(map[string]map[string]map[name]userMetric)
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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: 85045)
Time Spent: 8h 50m (was: 8h 40m)
> Fn API metrics in Go SDK harness
> --------------------------------
>
> Key: BEAM-3545
> URL: https://issues.apache.org/jira/browse/BEAM-3545
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-go
> Reporter: Kenneth Knowles
> Assignee: Robert Burke
> Priority: Major
> Labels: portability
> Time Spent: 8h 50m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
