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/11/05 16:07:34 UTC
[GitHub] [beam] kamilwu opened a new pull request #13272: Metric Extraction via proto RPC API
kamilwu opened a new pull request #13272:
URL: https://github.com/apache/beam/pull/13272
This PR is an implementation of the following proposal: https://s.apache.org/get-metrics-api. The ultimate goal is to offer a way to query user metrics after a job. The API follows the design of existing APIs in Python and Java SDKs.
How it works from user's perspective:
```go
result, _ := beamx.RunPipelineWithMetrics(ctx, p)
queryResult := result.Metrics().Query()
fmt.Println(queryResult.GetCounters())
fmt.Println(queryResult.GetDistributions())
fmt.Println(queryResult.GetGauges())
```
An example of output:
```
[{1 1 {fn/read/n2/SplitAndSize0:0 }} {20000 20000 {main.counterOperationFn counter-0 counterOperationFn}} {1 1 {fn/read/n1:0 }}]
[]
[]
```
------------------------
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 | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.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.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726219556
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
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r521706785
##########
File path: sdks/go/pkg/beam/core/metrics/monitoring_infos.go
##########
@@ -0,0 +1,176 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
Review comment:
Consider moving the contents of this file into the new metricx package, which is where I thought it was going to move last time. Any specific reason this code can't go there instead of this metrics package?
Part of the separation we have is to avoid having the protos dependended on by the same user construction packages, which is why there are x and non-x versions of packages for the most part.
Eg. Graph which is a pure go representation of handling the pipeline graph, and graphx which translates that graph into the proto representation. Similarly, the coder and coderx packages which are the representation of the coders, and the actual execution of the coders.
This notionally allows much of the Go SDK front end to be re-used for a non-beam backend without depending on the beam proto structure, and keeps the protos away from pipeline authors, for whome the protos should be implementation details.
Specifically, we try to hide the proto details from pipeline authors, not runner authors. Runner authors have to deal with the protos by definition, but pipeline authors should need a limited subset of those, and certainly not the protos. We try to have most user facing things be part of the beam package, but that's not appropriate for everything (it makes the package too big, too central, and harder to maintain).
##########
File path: sdks/go/pkg/beam/core/metrics/monitoring_infos_test.go
##########
@@ -0,0 +1,165 @@
+// 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 (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/runtime/metricsx"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
Review comment:
Similarly, these tests should be in the metricsx package.
##########
File path: sdks/go/pkg/beam/runners/universal/runnerlib/execute.go
##########
@@ -79,15 +81,50 @@ func Execute(ctx context.Context, p *pipepb.Pipeline, endpoint string, opt *JobO
jobID, err := Submit(ctx, client, prepID, token)
if err != nil {
- return "", err
+ return presult, err
}
log.Infof(ctx, "Submitted job: %v", jobID)
// (4) Wait for completion.
if async {
- return jobID, nil
+ return presult, nil
}
- return jobID, WaitForCompletion(ctx, client, jobID)
+ err = WaitForCompletion(ctx, client, jobID)
+
+ res, err := newUniversalPipelineResult(ctx, jobID, client)
+ if err != nil {
+ return presult, err
+ }
+ presult = res
+
+ return presult, err
+}
+
+type universalPipelineResult struct {
+ JobID string
+ metrics *metrics.Results
+}
+
+func newUniversalPipelineResult(ctx context.Context, jobID string, client jobpb.JobServiceClient) (*universalPipelineResult, error) {
+ metrics, err := getMetrics(ctx, jobID, client)
+ if err != nil {
+ return &universalPipelineResult{jobID, nil}, err
+ }
+ return &universalPipelineResult{jobID, metrics}, err
+}
+
+func (pr universalPipelineResult) Metrics() metrics.Results {
+ return *pr.metrics
+}
+
+func getMetrics(ctx context.Context, jobID string, client jobpb.JobServiceClient) (*metrics.Results, error) {
Review comment:
Is it necessary to isolate this part in it's own function? It's only used in the "newUniversalPipelineResult" which takes all the same arguments. If this is tested separately sure, but I don't see a reason to do that separately from the newUniversalPipelineResult for the previous reason.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// Results represents all metrics gathered during the job's execution.
+// It allows for querying metrics using a provided filter.
+type Results struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// AllMetrics returns all metrics from a Results instance.
+func (mr Results) AllMetrics() QueryResults {
+ return QueryResults{mr.counters, mr.distributions, mr.gauges}
+}
+
+// TODO(BEAM-11217): Implement Query(Filter) and metrics filtering
+
+// QueryResults is the result of a query. Allows accessing all of the
+// metrics that matched the filter.
+type QueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// GetCounters returns an array of counter metrics.
+func (qr QueryResults) GetCounters() []CounterResult {
Review comment:
That would still be idiomatic go. The question is whether we need to "hide" this behind an interface or not. I'm leaning no to that.
The real "trick" is probably copying out the results so users can query/mutate their own copies freely, without worrying about aliasing issues. Slices are not immutable, and are a pointer to a backing array which can be shared by multiple headers. The current method sets have just as many aliasing issues as having raw fields, so the fields aren't helping much.
Idiomatically BTW if we were keeping these methods, we wouldn't have the Get* prefix which is very much a javaism.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520036345
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -13,10 +13,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package harness
+package monitoring
Review comment:
Please move the new package to under beam/core/runtime/monitoring instead of just in raw core. runtime is where code that depends on the protocol buffers tends to live.
I'm not against a new package for this handling, but I think I'm against moving the code that is harness only being moved out of the harness. That code isn't intended to be part of any external API. There's also nothing relating the code beyond "they use the MonitoringInfos". That's not reason enough to put the code into the same package. To match the other convention around protos, perhaps we can call the package "metricsx" to go along with runtime/graphx which also handles protocol buffers.
Further, WRT Testing, you only seem to be using the encoding functions. Please simply copy those helpers into the test file instead of trying to deduplicate. A little bit of copying is better than a little bit of dependency.
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -199,11 +201,11 @@ func getShortID(l metrics.Labels, urn mUrn) string {
return defaultShortIDCache.getShortID(l, urn)
}
-func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
+func ShortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
Review comment:
Every exported identifier should have a doc comment.
https://golang.org/doc/effective_go.html#commentary
// ShortIDsToInfos translates a slice shortids from a monitoring request to their fully qualified MonitoringInfos
// based on the contents of the defaultShortIDCache.
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
Review comment:
Why not return a metrics.Results instead? Is the intent to force a breaking change if a new class of results is added?
Conversely, by using a metrics.Results, no such breaking change needs to happen, and things can change more gracefully (such as when generics land in Go).
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -199,11 +201,11 @@ func getShortID(l metrics.Labels, urn mUrn) string {
return defaultShortIDCache.getShortID(l, urn)
}
-func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
+func ShortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
return defaultShortIDCache.shortIdsToInfos(shortids)
}
-func monitoring(p *exec.Plan) ([]*pipepb.MonitoringInfo, map[string][]byte) {
+func Monitoring(p *exec.Plan) ([]*pipepb.MonitoringInfo, map[string][]byte) {
Review comment:
Every exported identifier should have a doc comment.
https://golang.org/doc/effective_go.html#commentary
// Monitoring extracts and translates metrics from an execution plan into the associated MonitoringInfos and short id mappings.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// MetricResults queries for all metric values that match a given filter.
+type MetricResults struct {
Review comment:
Please drop the "Metric" prefix when putting things into the metrics package.
Otherwise users of this type will always be typing metrics.MetricResults.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// MetricResults queries for all metric values that match a given filter.
+type MetricResults struct {
+ Counters []CounterResult
+ Distributions []DistributionResult
+ Gauges []GaugeResult
+}
+
+// AllMetrics returns all metrics from a MetricResults instance.
+func (mr MetricResults) AllMetrics() MetricQueryResults {
+ return MetricQueryResults{mr.Counters, mr.Distributions, mr.Gauges}
+}
+
+// TODO: Implement Query(MetricsFilter) and metrics filtering
+
+// MetricQueryResults is the results of a query. Allows accessing all of the
+// metrics that matched the filter.
+type MetricQueryResults struct {
Review comment:
Please drop the "Metric" prefix when putting things into the metrics package.
Otherwise users of this type will always be typing metrics.MetricQueryResult.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// MetricResults queries for all metric values that match a given filter.
+type MetricResults struct {
+ Counters []CounterResult
+ Distributions []DistributionResult
+ Gauges []GaugeResult
+}
+
+// AllMetrics returns all metrics from a MetricResults instance.
+func (mr MetricResults) AllMetrics() MetricQueryResults {
+ return MetricQueryResults{mr.Counters, mr.Distributions, mr.Gauges}
+}
+
+// TODO: Implement Query(MetricsFilter) and metrics filtering
+
+// MetricQueryResults is the results of a query. Allows accessing all of the
+// metrics that matched the filter.
+type MetricQueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// GetCounters returns an array of counter metrics.
+func (qr MetricQueryResults) GetCounters() []CounterResult {
+ return qr.counters
+}
+
+// GetDistributions returns an array of distribution metrics.
+func (qr MetricQueryResults) GetDistributions() []DistributionResult {
+ return qr.distributions
+}
+
+// GetGauges returns an array of gauges metrics.
+func (qr MetricQueryResults) GetGauges() []GaugeResult {
+ return qr.gauges
+}
+
+// CounterResult is an attempted and a commited value of a Counter metric plus
+// key.
+type CounterResult struct {
+ Attempted, Committed int64
+ Key MetricKey
+}
+
+// DistributionResult is an attempted and a commited value of a Distribution
+// metric plus key.
+type DistributionResult struct {
+ Attempted, Committed DistributionValue
+ Key MetricKey
+}
+
+// GaugeResult is an attempted and a commited value of a Gauge metric plus
+// key.
+type GaugeResult struct {
+ Attempted, Committed GaugeValue
+ Key MetricKey
+}
+
+// MetricKey includes the namespace and the name of the metric, as well as
+// the step that reported the metric.
+type MetricKey struct {
Review comment:
Please drop the "Metric" prefix when putting things into the metrics package.
Otherwise users of this type will always be typing metrics.MetricKey
There are also metrics that aren't associated with DoFns (eg PCollection specific ones), that may warrant different key types, rather than a "global" key. Maybe DoFnKey or StepKey?
Consider documenting what this is used for, rather than what it is. "StepKey uniquely identifies a metric"
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
Review comment:
Same comment that I had before about consider changing this if-else ladder into a switch/case statement.
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.MetricKey]int64,
+ committed map[metrics.MetricKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.MetricKey]metrics.DistributionValue,
+ committed map[metrics.MetricKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.MetricKey]metrics.GaugeValue,
+ committed map[metrics.MetricKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.MetricKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := getStepName(labels)
+ if stepName == "" {
+ return metrics.MetricKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.MetricKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
+ }
+ return value, nil
+}
+
+func extractDistributionValue(reader *bytes.Reader) (metrics.DistributionValue, error) {
+ values, err := decodeMany(reader, 4)
+ if err != nil {
+ return metrics.DistributionValue{}, err
+ }
+ return metrics.DistributionValue{Count: values[0], Sum: values[1], Min: values[2], Max: values[3]}, nil
+}
+
+func extractGaugeValue(reader *bytes.Reader) (metrics.GaugeValue, error) {
+ values, err := decodeMany(reader, 2)
+ if err != nil {
+ return metrics.GaugeValue{}, err
+ }
+ return metrics.GaugeValue{Timestamp: time.Unix(0, values[0]*int64(time.Millisecond)), Value: values[1]}, nil
+}
+
+func newLabels(miLabels map[string]string) *metrics.Labels {
+ labels := metrics.UserLabels(
+ miLabels["PTRANSFORM"],
+ miLabels[pipepb.MonitoringInfo_NAMESPACE.String()],
Review comment:
Consider just using the constants"NAMESPACE" and "NAME". Or go the other way, and pull PTRANSFORM from the enum instead. The inconsistency is odd.
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.MetricKey]int64,
+ committed map[metrics.MetricKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.MetricKey]metrics.DistributionValue,
+ committed map[metrics.MetricKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.MetricKey]metrics.GaugeValue,
+ committed map[metrics.MetricKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.MetricKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := getStepName(labels)
+ if stepName == "" {
+ return metrics.MetricKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.MetricKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
+ }
+ return value, nil
+}
+
+func extractDistributionValue(reader *bytes.Reader) (metrics.DistributionValue, error) {
+ values, err := decodeMany(reader, 4)
+ if err != nil {
+ return metrics.DistributionValue{}, err
+ }
+ return metrics.DistributionValue{Count: values[0], Sum: values[1], Min: values[2], Max: values[3]}, nil
+}
+
+func extractGaugeValue(reader *bytes.Reader) (metrics.GaugeValue, error) {
+ values, err := decodeMany(reader, 2)
+ if err != nil {
+ return metrics.GaugeValue{}, err
+ }
+ return metrics.GaugeValue{Timestamp: time.Unix(0, values[0]*int64(time.Millisecond)), Value: values[1]}, nil
+}
+
+func newLabels(miLabels map[string]string) *metrics.Labels {
+ labels := metrics.UserLabels(
+ miLabels["PTRANSFORM"],
+ miLabels[pipepb.MonitoringInfo_NAMESPACE.String()],
+ miLabels[pipepb.MonitoringInfo_NAME.String()])
+ return &labels
+}
+
+func getStepName(labels *metrics.Labels) string {
+ return labels.Transform()
+}
Review comment:
This function doesn't do anything other than serve as a comment. Consider simply calling labels.Transform() directly at the only call site.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-723951348
Run Seed Job
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520608852
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -199,11 +201,11 @@ func getShortID(l metrics.Labels, urn mUrn) string {
return defaultShortIDCache.getShortID(l, urn)
}
-func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
+func ShortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
Review comment:
No longer relevant.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-728817282
You're welcome! And thank you, too.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-722660388
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
[GitHub] [beam] lostluck commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-724185807
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
[GitHub] [beam] kamilwu commented on pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-722478309
R: @lostluck Note this PR is not a final one, work is in progress and things might change. However, I'd be grateful for an initial review. In particular, I wonder if those changes conform to Go SDK's vision.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-724176833
Thanks @lostluck, I've already created a ticket: https://issues.apache.org/jira/browse/BEAM-11207. For any follow-up work not covered by this PR, we can create a subtask.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520623472
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.MetricKey]int64,
+ committed map[metrics.MetricKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.MetricKey]metrics.DistributionValue,
+ committed map[metrics.MetricKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.MetricKey]metrics.GaugeValue,
+ committed map[metrics.MetricKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.MetricKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := getStepName(labels)
+ if stepName == "" {
+ return metrics.MetricKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.MetricKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
+ }
+ return value, nil
+}
+
+func extractDistributionValue(reader *bytes.Reader) (metrics.DistributionValue, error) {
+ values, err := decodeMany(reader, 4)
+ if err != nil {
+ return metrics.DistributionValue{}, err
+ }
+ return metrics.DistributionValue{Count: values[0], Sum: values[1], Min: values[2], Max: values[3]}, nil
+}
+
+func extractGaugeValue(reader *bytes.Reader) (metrics.GaugeValue, error) {
+ values, err := decodeMany(reader, 2)
+ if err != nil {
+ return metrics.GaugeValue{}, err
+ }
+ return metrics.GaugeValue{Timestamp: time.Unix(0, values[0]*int64(time.Millisecond)), Value: values[1]}, nil
+}
+
+func newLabels(miLabels map[string]string) *metrics.Labels {
+ labels := metrics.UserLabels(
+ miLabels["PTRANSFORM"],
+ miLabels[pipepb.MonitoringInfo_NAMESPACE.String()],
Review comment:
`pipepb.MonitoringInfo_PTRANSFORM.String()` keeps returning TRANSFORM, without the "P" prefix. I'm not sure why is that, TRANSFORM is invalid in this context and we should use PTRANSFORM. I changed it so that only the constants are used.
Just out of my curiosity, why the generated .pb.go files are committed into a git repository? for example: https://github.com/apache/beam/tree/master/sdks/go/pkg/beam/model/pipeline_v1. Is there some kind of problem with generating those files locally?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520613198
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
Review comment:
Changed it into a switch/case statement. Do you think that using raw strings is ok? Or should I define a new enum type for representing different types of 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-724186454
I put my changes into newly created `monitoring` package and merged `monitoring_info.go` with `monitoring.go`. The monitoring.go file already contains similar code. It also contains many functions which are helpful in testing my changes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726274397
At present, we haven't been adding to CHANGES.md as it's experimental. While the rate of changes has not been high, the messaging would be weird at this point? When we do announce experimental exit (SOON), we'll have to do a fairly complete rundown of all the features of the SDK anyway, and that's easier for me to do a mental dump than trying to extract from the CHANGES file.
Something this size would certainly be appropriate to message in releases after that point though.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520070443
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -13,10 +13,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package harness
+package monitoring
Review comment:
That said, doing the inverse: Having the new metricx package have Exported payload encoding helpers, and have the harness specific monitoring code depend on those, would be a good move, if you don't like the copying.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522276853
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// Results represents all metrics gathered during the job's execution.
+// It allows for querying metrics using a provided filter.
+type Results struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// AllMetrics returns all metrics from a Results instance.
+func (mr Results) AllMetrics() QueryResults {
+ return QueryResults{mr.counters, mr.distributions, mr.gauges}
+}
+
+// TODO(BEAM-11217): Implement Query(Filter) and metrics filtering
+
+// QueryResults is the result of a query. Allows accessing all of the
+// metrics that matched the filter.
+type QueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// GetCounters returns an array of counter metrics.
+func (qr QueryResults) GetCounters() []CounterResult {
Review comment:
Let's keep these methods, but remove the Get* prefix.
Regarding the results, a safer option is probably to return their copies instead. The structs don't have any underlying pointers, so it's not a problem to use built-in `copy` function.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-727926020
@lostluck That should be 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-725707007
> Thanks @lostluck. I'm still new to Go, so my knowledge of Go idioms and best practices can be limited. This is a valuable feedback for me.
Good to know! Overall pretty good though. Package organization is a tricky topic.
Taking another look now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r521694702
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.MetricKey]int64,
+ committed map[metrics.MetricKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.MetricKey]metrics.DistributionValue,
+ committed map[metrics.MetricKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.MetricKey]metrics.GaugeValue,
+ committed map[metrics.MetricKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.MetricKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := getStepName(labels)
+ if stepName == "" {
+ return metrics.MetricKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.MetricKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
+ }
+ return value, nil
+}
+
+func extractDistributionValue(reader *bytes.Reader) (metrics.DistributionValue, error) {
+ values, err := decodeMany(reader, 4)
+ if err != nil {
+ return metrics.DistributionValue{}, err
+ }
+ return metrics.DistributionValue{Count: values[0], Sum: values[1], Min: values[2], Max: values[3]}, nil
+}
+
+func extractGaugeValue(reader *bytes.Reader) (metrics.GaugeValue, error) {
+ values, err := decodeMany(reader, 2)
+ if err != nil {
+ return metrics.GaugeValue{}, err
+ }
+ return metrics.GaugeValue{Timestamp: time.Unix(0, values[0]*int64(time.Millisecond)), Value: values[1]}, nil
+}
+
+func newLabels(miLabels map[string]string) *metrics.Labels {
+ labels := metrics.UserLabels(
+ miLabels["PTRANSFORM"],
+ miLabels[pipepb.MonitoringInfo_NAMESPACE.String()],
Review comment:
I think it's a legacy issue that hasn't migrated yet. I'm surprised, I thought lcwik handled that already. I'd add a comment about the discrepancy then. It's likely a legacy issue then. I see the existing monitoring code uses "PTRANSFORM" as a result too.
-------
Regarding the protos: Great question! Please bear with the length of this topic.
The short answer is: that's the idiom for generated Go files: submit them.
Java relies heavily on external build/package management to create jars, which contain the code generated from the PBs.
Python uses a one time setup script users use to do so locally, and an abundance of external management (pip, pyenv etc)
They also deliver everything via archives/wheels/jars.
Go on the other hand deliberately avoids those approaches, and pointedly delivers everything via open source code. This works reasonably well in this day and age, of micro services. If you really want to deliver something a client runs on their machine, without delivering the source, a binary happens, or you simply spin up a service on the internet. No source needed.
One of Go's better touted features is that binaries are generally statically compiled without needing specific DLLs or other libaries resident on the OS.
While this can lead to larger binaries, that's certainly less of a problem than before with how cheap storage and RAM is these days.
Further, Go has made it's way by having excellent tooling around testing and packages and execution. If you want to install a binary that's available as go code? You `go install` it and the tool takes care of the rest. However, there are limits to that. Go's build cycle deliberately doesn't invoke other processes, including additional code generators like the proto compiler.
There is support for pre-build time code generation, using the `go generate` command
Specifically, idiomatically, package users should never need to run `go generate` to get one of their received libraries to work. `go generate` is for package authors, not users.
As a result, anything generated by that command ends up needing to be committed to allow the users to access it.
So to answer the question again: that approach isn't desirable from a user point of view, as it assumes and forces our implementation details on the user, instead of centralizing them on the project.
----
This can then devolve into Go Package versioning, and the short answer there is, the Go Beam SDK doesn't currently respect versions, and won't until we migrate it to use Go Modules. Hopefully soon. After I finish Schemas and this Beam Release work...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522330612
##########
File path: sdks/go/pkg/beam/core/metrics/monitoring_infos.go
##########
@@ -0,0 +1,176 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
Review comment:
That's a good move. Go packages can be spread over multiple files as needed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r523128139
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -0,0 +1,181 @@
+// 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 metricsx
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from monitored states and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) *metrics.Results {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ return metrics.NewResults(mergeCounters(ac, cc), mergeDistributions(ad, cd), mergeGauges(ag, cg))
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.StepKey]int64,
+ map[metrics.StepKey]metrics.DistributionValue,
+ map[metrics.StepKey]metrics.GaugeValue) {
+ counters := make(map[metrics.StepKey]int64)
+ distributions := make(map[metrics.StepKey]metrics.DistributionValue)
+ gauges := make(map[metrics.StepKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ switch minfo.GetType() {
+ case "beam:metrics:sum_int64:v1":
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ case "beam:metrics: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":
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ default:
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.StepKey]int64,
+ committed map[metrics.StepKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
Review comment:
If the user created a counter, and no increments occurred, the counter will never appear in the final results. It would be missing entirely. Beam counters operate on a "diff per bundle" set up, and the counters aren't known to the runner a-priory, so a runner can't known about them in advanced.
0 would be correct for committed values that have an attempted value, as no count has been committed.
I can't think of a useful case of distinguishing between "does not exist" for committed only, and simply saying 0.
Another reason: users can also use negative values for counters/increments if they so choose. It's not incorrect. So -1 isn't a useful way to signal non-existance.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520633641
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// Results represents all metrics gathered during the job's execution.
+// It allows for querying metrics using a provided filter.
+type Results struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// AllMetrics returns all metrics from a Results instance.
+func (mr Results) AllMetrics() QueryResults {
+ return QueryResults{mr.counters, mr.distributions, mr.gauges}
+}
+
+// TODO(BEAM-11217): Implement Query(Filter) and metrics filtering
+
+// QueryResults is the result of a query. Allows accessing all of the
+// metrics that matched the filter.
+type QueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// GetCounters returns an array of counter metrics.
+func (qr QueryResults) GetCounters() []CounterResult {
Review comment:
Perhaps we could skip these getters and expose QueryResults's fields directly. We don't need to follow Java API that much. @lostluck What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522347395
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,116 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// Results represents all metrics gathered during the job's execution.
+// It allows for querying metrics using a provided filter.
+type Results struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// NewResults creates a new Results.
+func NewResults(
+ counters []CounterResult,
+ distributions []DistributionResult,
+ gauges []GaugeResult) *Results {
+ return &Results{counters, distributions, gauges}
+}
+
+// AllMetrics returns all metrics from a Results instance.
+func (mr Results) AllMetrics() QueryResults {
+ return QueryResults{mr.counters, mr.distributions, mr.gauges}
+}
+
+// TODO(BEAM-11217): Implement Query(Filter) and metrics filtering
+
+// QueryResults is the result of a query. Allows accessing all of the
+// metrics that matched the filter.
+type QueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// Counters returns an array of counter metrics.
Review comment:
In go comments, it's reasonable to call this "a slice". In particular, arrays in Go are fixed size and have different implications compared to variable sized slices. Here and below.
```
var slice []int // a slice of ints
var array [4]int // an array of ints of length 4
slice = []int{1,2,3,4}
array = [4]int{1,2,3,4}
array = [...]int{1,2,3,4} // Have the compiler determine the array size based on the construction time parameters
// array = [2]int{1, 2} // a compiler error
slice = array[1:len(array)] // a slice of array, excluding the first element.
// This is backed by array, which will lead to aliasing issues if care isn't taken...
```
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -0,0 +1,181 @@
+// 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 metricsx
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from monitored states and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) *metrics.Results {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ return metrics.NewResults(mergeCounters(ac, cc), mergeDistributions(ad, cd), mergeGauges(ag, cg))
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.StepKey]int64,
+ map[metrics.StepKey]metrics.DistributionValue,
+ map[metrics.StepKey]metrics.GaugeValue) {
+ counters := make(map[metrics.StepKey]int64)
+ distributions := make(map[metrics.StepKey]metrics.DistributionValue)
+ gauges := make(map[metrics.StepKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ switch minfo.GetType() {
+ case "beam:metrics:sum_int64:v1":
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ case "beam:metrics: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":
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ default:
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.StepKey]int64,
+ committed map[metrics.StepKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.StepKey]metrics.DistributionValue,
+ committed map[metrics.StepKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
Review comment:
Since this is a map to values (rather than pointers), v will already be the zero value for the type. In short, this code is identical to simply `v := commited[k]`
https://golang.org/doc/effective_go.html#maps
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -0,0 +1,181 @@
+// 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 metricsx
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from monitored states and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) *metrics.Results {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ return metrics.NewResults(mergeCounters(ac, cc), mergeDistributions(ad, cd), mergeGauges(ag, cg))
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.StepKey]int64,
+ map[metrics.StepKey]metrics.DistributionValue,
+ map[metrics.StepKey]metrics.GaugeValue) {
+ counters := make(map[metrics.StepKey]int64)
+ distributions := make(map[metrics.StepKey]metrics.DistributionValue)
+ gauges := make(map[metrics.StepKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ switch minfo.GetType() {
+ case "beam:metrics:sum_int64:v1":
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ case "beam:metrics: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":
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ default:
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.StepKey]int64,
+ committed map[metrics.StepKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
Review comment:
Is the -1 part of the spec here? Wouldn't a 0 notionally be correct if there are no commited values for a key?
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ panic(err)
Review comment:
Same comment here, prefer a t.Fatalf here instead of a panic.
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ panic(err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, expected: %v", size, 1)
+ }
+ if got[0] != want {
+ t.Fatalf("Invalid counter: got: %v, want: %v",
+ got[0], want)
+ }
+}
+
+func TestDistributionExtraction(t *testing.T) {
+ var count, sum, min, max int64 = 100, 5, -12, 30
+
+ want := metrics.DistributionResult{
+ Attempted: metrics.DistributionValue{
+ Count: 100,
+ Sum: 5,
+ Min: -12,
+ Max: 30,
+ },
+ Committed: metrics.DistributionValue{},
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customDist",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Distribution(count, sum, min, max)
+ if err != nil {
+ panic(err)
Review comment:
Prefer a t.Fatal here instead, and be clear that the encoding the Int64Distribution returned an error.
As a rule, never panic if you don't have to. In tests, it's better to clearly explain why you're failing the test run anyway. https://golang.org/doc/effective_go.html#panic
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ panic(err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, expected: %v", size, 1)
+ }
+ if got[0] != want {
+ t.Fatalf("Invalid counter: got: %v, want: %v",
+ got[0], want)
+ }
+}
+
+func TestDistributionExtraction(t *testing.T) {
+ var count, sum, min, max int64 = 100, 5, -12, 30
+
+ want := metrics.DistributionResult{
+ Attempted: metrics.DistributionValue{
+ Count: 100,
+ Sum: 5,
+ Min: -12,
+ Max: 30,
+ },
+ Committed: metrics.DistributionValue{},
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customDist",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Distribution(count, sum, min, max)
+ if err != nil {
+ panic(err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customDist",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserDistInt64),
+ Type: UrnToType(UrnUserDistInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Distributions()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, expected: %v", size, 1)
+ }
+ if got[0] != want {
+ t.Fatalf("Invalid distribution: got: %v, want: %v",
+ got[0], want)
+ }
+}
+
+func TestGaugeExtraction(t *testing.T) {
+ var value int64 = 100
+ loc, _ := time.LoadLocation("Local")
+ tm := time.Date(2020, 11, 9, 17, 52, 28, 462*int(time.Millisecond), loc)
+
+ want := metrics.GaugeResult{
+ Attempted: metrics.GaugeValue{
+ Value: 100,
+ Timestamp: tm,
+ },
+ Committed: metrics.GaugeValue{},
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customGauge",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Latest(tm, value)
+ if err != nil {
+ panic(err)
Review comment:
Same comment here, prefer a t.Fatalf here instead of a panic.
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,116 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// Results represents all metrics gathered during the job's execution.
+// It allows for querying metrics using a provided filter.
+type Results struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// NewResults creates a new Results.
+func NewResults(
+ counters []CounterResult,
+ distributions []DistributionResult,
+ gauges []GaugeResult) *Results {
+ return &Results{counters, distributions, gauges}
+}
+
+// AllMetrics returns all metrics from a Results instance.
+func (mr Results) AllMetrics() QueryResults {
+ return QueryResults{mr.counters, mr.distributions, mr.gauges}
+}
+
+// TODO(BEAM-11217): Implement Query(Filter) and metrics filtering
+
+// QueryResults is the result of a query. Allows accessing all of the
+// metrics that matched the filter.
+type QueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// Counters returns an array of counter metrics.
+func (qr QueryResults) Counters() []CounterResult {
+ out := make([]CounterResult, len(qr.counters))
+ copy(out, qr.counters)
+ return out
+}
+
+// Distributions returns an array of distribution metrics.
+func (qr QueryResults) Distributions() []DistributionResult {
+ out := make([]DistributionResult, len(qr.distributions))
+ copy(out, qr.distributions)
+ return out
+}
+
+// Gauges returns an array of gauge metrics.
+func (qr QueryResults) Gauges() []GaugeResult {
+ out := make([]GaugeResult, len(qr.gauges))
+ copy(out, qr.gauges)
+ return out
+}
+
+// CounterResult is an attempted and a commited value of a counter metric plus
+// key.
+type CounterResult struct {
+ Attempted, Committed int64
+ Key StepKey
+}
+
+// Result returns committed metrics. Falls back to attempted metrics if committed
+// are not populated (e.g. due to not being supported on a given runner).
+func (r CounterResult) Result() int64 {
+ if r.Committed != -1 {
+ return r.Committed
+ }
+ return r.Attempted
+}
+
+// DistributionResult is an attempted and a commited value of a distribution
+// metric plus key.
+type DistributionResult struct {
+ Attempted, Committed DistributionValue
+ Key StepKey
+}
+
+// Result returns committed metrics. Falls back to attempted metrics if committed
+// are not populated (e.g. due to not being supported on a given runner).
+func (r DistributionResult) Result() DistributionValue {
+ empty := DistributionValue{}
+ if r.Committed != empty {
+ return r.Committed
+ }
+ return r.Attempted
+}
+
+// GaugeResult is an attempted and a commited value of a gauge metric plus
+// key.
+type GaugeResult struct {
+ Attempted, Committed GaugeValue
+ Key StepKey
+}
+
+// Result returns committed metrics. Falls back to attempted metrics if committed
+// are not populated (e.g. due to not being supported on a given runner).
+func (r GaugeResult) Result() GaugeValue {
+ empty := GaugeValue{}
+ if r.Committed != empty {
+ return r.Committed
+ }
+ return r.Attempted
+}
+
+// StepKey uniquely identifies a metric.
Review comment:
```suggestion
// StepKey uniquely identifies a metric within a pipeline graph.
```
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -0,0 +1,181 @@
+// 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 metricsx
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from monitored states and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) *metrics.Results {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ return metrics.NewResults(mergeCounters(ac, cc), mergeDistributions(ad, cd), mergeGauges(ag, cg))
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.StepKey]int64,
+ map[metrics.StepKey]metrics.DistributionValue,
+ map[metrics.StepKey]metrics.GaugeValue) {
+ counters := make(map[metrics.StepKey]int64)
+ distributions := make(map[metrics.StepKey]metrics.DistributionValue)
+ gauges := make(map[metrics.StepKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ switch minfo.GetType() {
+ case "beam:metrics:sum_int64:v1":
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ case "beam:metrics: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":
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ default:
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.StepKey]int64,
+ committed map[metrics.StepKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.StepKey]metrics.DistributionValue,
+ committed map[metrics.StepKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.StepKey]metrics.GaugeValue,
+ committed map[metrics.StepKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.StepKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := labels.Transform()
+ if stepName == "" {
+ return metrics.StepKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.StepKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
Review comment:
In Go, if a non-nil error is returned, conventionally the non-error returns should be ignored. There's no need to have a special marker value for an error, since that's what the error is for. Specifically if the value returned is still valid in some way when an error is returned, that is the exception, and should be documented.
(As it literally doesn't matter, no change required here, so just pointing it out for teaching purposes.)
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ panic(err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, expected: %v", size, 1)
+ }
+ if got[0] != want {
Review comment:
Do this comparison using the cmp package instead.
"github.com/google/go-cmp/cmp"
```
if d := cmp.Diff(want, got[0]); d != "" {
t.Fatalf("Invalid counter: %v", d)
}
```
Here and below.
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ panic(err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, expected: %v", size, 1)
Review comment:
In go tests, never say "expected" when you can simply say "want". Otherwise this test output is excellent. Here and below.
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
Review comment:
Test names are related to the code as much as possible. In this case, the thing we're properly testing is FromMonitoringInfos, and specifically Counters so `func TestFromMonitoringInfos_Counters` is an idiomatic test name.
There's also an opportunity for subtests... but that's difficult here without generics right now., and that's not until 2022. You can see more here https://gobyexample.com/testing and https://blog.golang.org/subtests
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-727912681
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
[GitHub] [beam] lostluck merged pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck merged pull request #13272:
URL: https://github.com/apache/beam/pull/13272
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r518966479
##########
File path: sdks/go/pkg/beam/runners/universal/runnerlib/execute.go
##########
@@ -79,15 +81,79 @@ func Execute(ctx context.Context, p *pipepb.Pipeline, endpoint string, opt *JobO
jobID, err := Submit(ctx, client, prepID, token)
if err != nil {
- return "", err
+ return presult, err
}
log.Infof(ctx, "Submitted job: %v", jobID)
// (4) Wait for completion.
if async {
- return jobID, nil
+ return presult, nil
}
- return jobID, WaitForCompletion(ctx, client, jobID)
+ err = WaitForCompletion(ctx, client, jobID)
+
+ res, err := newUniversalPipelineResult(ctx, jobID, client)
+ if err != nil {
+ return presult, err
+ }
+ presult = res
+
+ return presult, err
+}
+
+type UniversalPipelineResult struct {
+ JobID string
+ metrics *UniversalMetrics
+}
+
+func newUniversalPipelineResult(ctx context.Context, jobID string, client jobpb.JobServiceClient) (*UniversalPipelineResult, error) {
+ metrics, err := newUniversalMetrics(ctx, jobID, client)
+ if err != nil {
+ return &UniversalPipelineResult{jobID, nil}, err
+ }
+ return &UniversalPipelineResult{jobID, metrics}, err
+}
+
+func (pr UniversalPipelineResult) Metrics() metrics.MetricResults {
+ return pr.metrics
+}
+
+func newUniversalMetrics(ctx context.Context, jobID string, client jobpb.JobServiceClient) (*UniversalMetrics, error) {
+ request := &jobpb.GetJobMetricsRequest{JobId: jobID}
+ response, err := client.GetJobMetrics(ctx, request)
+ if err != nil {
+ return nil, errors.Wrap(err, "failed to get metrics")
+ }
+ c, d, g := metrics.FromMonitoringInfos(response.GetMetrics())
+ return &UniversalMetrics{c, d, g}, err
+}
+
+type UniversalMetrics struct {
+ counters []metrics.CounterResult
+ distributions []metrics.DistributionResult
+ gauges []metrics.GaugeResult
+}
+
+func (um UniversalMetrics) Query() metrics.MetricQueryResults {
+ // TODO: Implement metrics filtering
Review comment:
For consistency and ease of finding related work, also put the same JIRA id here.
##########
File path: sdks/go/pkg/beam/core/metrics/monitoring_infos.go
##########
@@ -0,0 +1,211 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ jobpb "github.com/apache/beam/sdks/go/pkg/beam/model/jobmanagement_v1"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(resultsProto *jobpb.MetricResults) (
+ []CounterResult,
+ []DistributionResult,
+ []GaugeResult) {
+ ac, ad, ag := groupByType(resultsProto.Attempted)
+ cc, cd, cg := groupByType(resultsProto.Committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[MetricKey]int64,
+ map[MetricKey]DistributionValue,
+ map[MetricKey]GaugeValue) {
+ counters := make(map[MetricKey]int64)
+ distributions := make(map[MetricKey]DistributionValue)
+ gauges := make(map[MetricKey]GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterResult(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(attempted map[MetricKey]int64, committed map[MetricKey]int64) []CounterResult {
+ res := make([]CounterResult, 0)
+
+ for k := range attempted {
+ if v, ok := committed[k]; ok {
+ res = append(res, CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ } else {
+ res = append(res, CounterResult{Attempted: attempted[k], Committed: -1, Key: k})
+ }
Review comment:
Style nit: Makes it clearer that the only difference between the two clauses is the value of v.
```suggestion
v, ok := committed[k]
if !ok {
v = -1
}
res = append(res, CounterResult{Attempted: attempted[k], Committed: v, Key: k})
```
If you make this change, please make the change for distributions and gauges below for consistency,
##########
File path: sdks/go/pkg/beam/testing/ptest/ptest.go
##########
@@ -73,7 +73,8 @@ func Run(p *beam.Pipeline) error {
if *Runner == "" {
*Runner = defaultRunner
}
- return beam.Run(context.Background(), *Runner, p)
+ _, err := beam.Run(context.Background(), *Runner, p)
+ return err
Review comment:
We should consider changing ptest or at least having a new ptest.RunWithMetrics endpoint that also returns the pipeline results, as metrics are very valuable and convenient to check in testing situations. That can be handled as a TODO (BEAM--####) however, rather than as a part of this PR, since ptest is pervasive.
##########
File path: sdks/go/pkg/beam/x/beamx/run.go
##########
@@ -40,5 +40,13 @@ var runner = flag.String("runner", "direct", "Pipeline runner.")
// defaults to the direct runner, but all beam-distributed runners and textio
// filesystems are implicitly registered.
func Run(ctx context.Context, p *beam.Pipeline) error {
+ _, err := beam.Run(ctx, *runner, p)
+ return err
+}
+
+// RunPipelineWithMetrics invokes beam.Run with the runner supplied by the
+// flag "runner". Returns a beam.PipelineResult objects, which can be
+// accessed to query the pipeline's metrics.
+func RunPipelineWithMetrics(ctx context.Context, p *beam.Pipeline) (beam.PipelineResult, error) {
Review comment:
I'd omit the "Pipeline" as it's already clear from the parameter type, and the other Run function that pipelines are going to be executed.
##########
File path: sdks/go/pkg/beam/core/metrics/monitoring_infos.go
##########
@@ -0,0 +1,211 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ jobpb "github.com/apache/beam/sdks/go/pkg/beam/model/jobmanagement_v1"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(resultsProto *jobpb.MetricResults) (
+ []CounterResult,
+ []DistributionResult,
+ []GaugeResult) {
+ ac, ad, ag := groupByType(resultsProto.Attempted)
+ cc, cd, cg := groupByType(resultsProto.Committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[MetricKey]int64,
+ map[MetricKey]DistributionValue,
+ map[MetricKey]GaugeValue) {
+ counters := make(map[MetricKey]int64)
+ distributions := make(map[MetricKey]DistributionValue)
+ gauges := make(map[MetricKey]GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
Review comment:
Style nit: Consider changing this to a switch case statement.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726312084
Run Go Precommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r523124215
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,166 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+ "github.com/google/go-cmp/cmp"
+)
+
+func TestFromMonitoringInfos_Counters(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ t.Fatalf("Failed to encode Int64Counter: %v", err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, want: %v", size, 1)
+ }
+ if d := cmp.Diff(got[0], want); d != "" {
Review comment:
You'll want to have want before got when using cmp.Diff. Otherwise your diff guidance is wrong.
cmp.Diff(want, got[0]) means the diff will be (-want,+got) that is saying what is missing from want, and extra in got, which is a useful way to read them. cmp.Diff(want, got[0]) means the diff will be (+want,-got), which will be confusing.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu removed a comment on pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu removed a comment on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-723951348
Run Seed Job
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r524143231
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -0,0 +1,181 @@
+// 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 metricsx
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from monitored states and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) *metrics.Results {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ return metrics.NewResults(mergeCounters(ac, cc), mergeDistributions(ad, cd), mergeGauges(ag, cg))
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.StepKey]int64,
+ map[metrics.StepKey]metrics.DistributionValue,
+ map[metrics.StepKey]metrics.GaugeValue) {
+ counters := make(map[metrics.StepKey]int64)
+ distributions := make(map[metrics.StepKey]metrics.DistributionValue)
+ gauges := make(map[metrics.StepKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ switch minfo.GetType() {
+ case "beam:metrics:sum_int64:v1":
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ case "beam:metrics: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":
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ default:
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.StepKey]int64,
+ committed map[metrics.StepKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
Review comment:
TIL. In that case I have on further objections. Let's use 0.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726243986
What's a policy regarding Go SDK and the `CHANGES.md` file? Are major changes to Go SDK highlighted there? The sdk is experimental, so I wonder if such changes is something that users should know about.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-724755477
Thanks @lostluck. I'm still new to Go, so my knowledge of Go idioms and best practices can be limited. This is a valuable feedback for me.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r518385153
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,50 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// MetricResults queries for all metric values that match a given filter.
+type MetricResults interface {
+ // TODO: Implement metrics filtering
Review comment:
Please file a JIRA (https://issues.apache.org/jira/projects/BEAM/issues) under the sdk-go component for this PRs and followup work. Then add the JIRA [BEAM-#####] as a prefix to the PR title.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522344928
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.MetricKey]int64,
+ committed map[metrics.MetricKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.MetricKey]metrics.DistributionValue,
+ committed map[metrics.MetricKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.MetricKey]metrics.GaugeValue,
+ committed map[metrics.MetricKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.MetricKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := getStepName(labels)
+ if stepName == "" {
+ return metrics.MetricKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.MetricKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
+ }
+ return value, nil
+}
+
+func extractDistributionValue(reader *bytes.Reader) (metrics.DistributionValue, error) {
+ values, err := decodeMany(reader, 4)
+ if err != nil {
+ return metrics.DistributionValue{}, err
+ }
+ return metrics.DistributionValue{Count: values[0], Sum: values[1], Min: values[2], Max: values[3]}, nil
+}
+
+func extractGaugeValue(reader *bytes.Reader) (metrics.GaugeValue, error) {
+ values, err := decodeMany(reader, 2)
+ if err != nil {
+ return metrics.GaugeValue{}, err
+ }
+ return metrics.GaugeValue{Timestamp: time.Unix(0, values[0]*int64(time.Millisecond)), Value: values[1]}, nil
+}
+
+func newLabels(miLabels map[string]string) *metrics.Labels {
+ labels := metrics.UserLabels(
+ miLabels["PTRANSFORM"],
+ miLabels[pipepb.MonitoringInfo_NAMESPACE.String()],
Review comment:
This will certainly become easier once we're on Go Modules. Right now it's tricky due to ensuring we have the right versions of the proto code generators. Heck, at a certain point, one could script installing protoc and installing the go proto plugin so go generate for the protos is fairly self contained for contributors.
Nothing you've been doing in this PR requires re-generating the protos, or code generated type assertion "shims", so hopefully you haven't run into places where it was implied that was necessary.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520610373
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
Review comment:
Good spot! Of course it should return `metrics.Results`. 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
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522889229
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go
##########
@@ -0,0 +1,181 @@
+// 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 metricsx
+
+import (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+// FromMonitoringInfos extracts metrics from monitored states and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) *metrics.Results {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ return metrics.NewResults(mergeCounters(ac, cc), mergeDistributions(ad, cd), mergeGauges(ag, cg))
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.StepKey]int64,
+ map[metrics.StepKey]metrics.DistributionValue,
+ map[metrics.StepKey]metrics.GaugeValue) {
+ counters := make(map[metrics.StepKey]int64)
+ distributions := make(map[metrics.StepKey]metrics.DistributionValue)
+ gauges := make(map[metrics.StepKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ switch minfo.GetType() {
+ case "beam:metrics:sum_int64:v1":
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ case "beam:metrics: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":
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ default:
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.StepKey]int64,
+ committed map[metrics.StepKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
Review comment:
What if a user created a counter in the pipeline, but the counter hasn't been incremented? If the runner supports committed metrics, the correct value is 0. I'm leaning towards we can't use the same value in "does not exist" context. -1 is a good candidate, because it will never be a correct value for a counter.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520607900
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -13,10 +13,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-package harness
+package monitoring
Review comment:
I created the new metricsx package which exports payload helpers and urn constants (which are required both by harness and my changes). I don't like copying over 100 lines of code, so the package seems like a good idea.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726714008
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522272434
##########
File path: sdks/go/pkg/beam/core/metrics/monitoring_infos.go
##########
@@ -0,0 +1,176 @@
+// 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 (
+ "bytes"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
Review comment:
Thanks, now I understand that distinction between x and non-x versions of packages. I moved this code to the metricsx package.
When it comes to urns constants and encoding helpers, I moved them to `urns.go` file within the same package (metricsx)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r518385153
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,50 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// MetricResults queries for all metric values that match a given filter.
+type MetricResults interface {
+ // TODO: Implement metrics filtering
Review comment:
Please file a JIRA (https://issues.apache.org/jira/projects/BEAM/issues) under the sdk-go component for this PRs and followup work. Then add the JIRA [BEAM-###] as a prefix to the PR title.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726312144
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
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522423683
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,165 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+)
+
+func TestCounterExtraction(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ panic(err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, expected: %v", size, 1)
+ }
+ if got[0] != want {
Review comment:
Here's an example in repo: https://github.com/apache/beam/blob/e7a3392595e14cdb377fe95fb679305ea9a32249/sdks/go/pkg/beam/core/graph/coder/bytes_test.go#L46
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520611568
##########
File path: sdks/go/pkg/beam/core/metrics/metrics.go
##########
@@ -448,3 +453,73 @@ func (m *gauge) get() (int64, time.Time) {
defer m.mu.Unlock()
return m.v, m.t
}
+
+// GaugeValue is the value of a Gauge metric.
+type GaugeValue struct {
+ Value int64
+ Timestamp time.Time
+}
+
+// MetricResults queries for all metric values that match a given filter.
+type MetricResults struct {
+ Counters []CounterResult
+ Distributions []DistributionResult
+ Gauges []GaugeResult
+}
+
+// AllMetrics returns all metrics from a MetricResults instance.
+func (mr MetricResults) AllMetrics() MetricQueryResults {
+ return MetricQueryResults{mr.Counters, mr.Distributions, mr.Gauges}
+}
+
+// TODO: Implement Query(MetricsFilter) and metrics filtering
+
+// MetricQueryResults is the results of a query. Allows accessing all of the
+// metrics that matched the filter.
+type MetricQueryResults struct {
+ counters []CounterResult
+ distributions []DistributionResult
+ gauges []GaugeResult
+}
+
+// GetCounters returns an array of counter metrics.
+func (qr MetricQueryResults) GetCounters() []CounterResult {
+ return qr.counters
+}
+
+// GetDistributions returns an array of distribution metrics.
+func (qr MetricQueryResults) GetDistributions() []DistributionResult {
+ return qr.distributions
+}
+
+// GetGauges returns an array of gauges metrics.
+func (qr MetricQueryResults) GetGauges() []GaugeResult {
+ return qr.gauges
+}
+
+// CounterResult is an attempted and a commited value of a Counter metric plus
+// key.
+type CounterResult struct {
+ Attempted, Committed int64
+ Key MetricKey
+}
+
+// DistributionResult is an attempted and a commited value of a Distribution
+// metric plus key.
+type DistributionResult struct {
+ Attempted, Committed DistributionValue
+ Key MetricKey
+}
+
+// GaugeResult is an attempted and a commited value of a Gauge metric plus
+// key.
+type GaugeResult struct {
+ Attempted, Committed GaugeValue
+ Key MetricKey
+}
+
+// MetricKey includes the namespace and the name of the metric, as well as
+// the step that reported the metric.
+type MetricKey struct {
Review comment:
`StepKey` sounds good. Thanks for suggestion.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r522288697
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -341,3 +343,189 @@ func int64Distribution(count, sum, min, max int64) ([]byte, error) {
}
return buf.Bytes(), nil
}
+
+// FromMonitoringInfos extracts metrics from GetJobMetrics's response and
+// groups them into counters, distributions and gauges.
+func FromMonitoringInfos(attempted []*pipepb.MonitoringInfo, committed []*pipepb.MonitoringInfo) (
+ []metrics.CounterResult,
+ []metrics.DistributionResult,
+ []metrics.GaugeResult) {
+ ac, ad, ag := groupByType(attempted)
+ cc, cd, cg := groupByType(committed)
+
+ c := mergeCounters(ac, cc)
+ d := mergeDistributions(ad, cd)
+ g := mergeGauges(ag, cg)
+
+ return c, d, g
+}
+
+// IsCounter returns true if the monitoring info is a counter metric.
+func IsCounter(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:sum_int64:v1"
+}
+
+// IsDistribution returns true if the monitoring info is a distribution metric.
+func IsDistribution(mi *pipepb.MonitoringInfo) bool {
+ return mi.GetType() == "beam:metrics:distribution_int64:v1"
+}
+
+// IsGauge returns true if the monitoring info is a gauge metric.
+func IsGauge(mi *pipepb.MonitoringInfo) bool {
+ switch mi.GetType() {
+ case
+ "beam:metrics:latest_int64:v1",
+ "beam:metrics:top_n_int64:v1",
+ "beam:metrics:bottom_n_int64:v1":
+ return true
+ }
+ return false
+}
+
+func groupByType(minfos []*pipepb.MonitoringInfo) (
+ map[metrics.MetricKey]int64,
+ map[metrics.MetricKey]metrics.DistributionValue,
+ map[metrics.MetricKey]metrics.GaugeValue) {
+ counters := make(map[metrics.MetricKey]int64)
+ distributions := make(map[metrics.MetricKey]metrics.DistributionValue)
+ gauges := make(map[metrics.MetricKey]metrics.GaugeValue)
+
+ for _, minfo := range minfos {
+ key, err := extractKey(minfo)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+
+ r := bytes.NewReader(minfo.GetPayload())
+
+ if IsCounter(minfo) {
+ value, err := extractCounterValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ counters[key] = value
+ } else if IsDistribution(minfo) {
+ value, err := extractDistributionValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ distributions[key] = value
+ } else if IsGauge(minfo) {
+ value, err := extractGaugeValue(r)
+ if err != nil {
+ log.Println(err)
+ continue
+ }
+ gauges[key] = value
+ } else {
+ log.Println("unknown metric type")
+ }
+ }
+ return counters, distributions, gauges
+}
+
+func mergeCounters(
+ attempted map[metrics.MetricKey]int64,
+ committed map[metrics.MetricKey]int64) []metrics.CounterResult {
+ res := make([]metrics.CounterResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = -1
+ }
+ res = append(res, metrics.CounterResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeDistributions(
+ attempted map[metrics.MetricKey]metrics.DistributionValue,
+ committed map[metrics.MetricKey]metrics.DistributionValue) []metrics.DistributionResult {
+ res := make([]metrics.DistributionResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.DistributionValue{}
+ }
+ res = append(res, metrics.DistributionResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func mergeGauges(
+ attempted map[metrics.MetricKey]metrics.GaugeValue,
+ committed map[metrics.MetricKey]metrics.GaugeValue) []metrics.GaugeResult {
+ res := make([]metrics.GaugeResult, 0)
+
+ for k := range attempted {
+ v, ok := committed[k]
+ if !ok {
+ v = metrics.GaugeValue{}
+ }
+ res = append(res, metrics.GaugeResult{Attempted: attempted[k], Committed: v, Key: k})
+ }
+ return res
+}
+
+func extractKey(mi *pipepb.MonitoringInfo) (metrics.MetricKey, error) {
+ labels := newLabels(mi.GetLabels())
+ stepName := getStepName(labels)
+ if stepName == "" {
+ return metrics.MetricKey{}, fmt.Errorf("Failed to deduce Step from MonitoringInfo: %v", mi)
+ }
+ return metrics.MetricKey{Step: stepName, Name: labels.Name(), Namespace: labels.Namespace()}, nil
+}
+
+func extractCounterValue(reader *bytes.Reader) (int64, error) {
+ value, err := coder.DecodeVarInt(reader)
+ if err != nil {
+ return -1, err
+ }
+ return value, nil
+}
+
+func extractDistributionValue(reader *bytes.Reader) (metrics.DistributionValue, error) {
+ values, err := decodeMany(reader, 4)
+ if err != nil {
+ return metrics.DistributionValue{}, err
+ }
+ return metrics.DistributionValue{Count: values[0], Sum: values[1], Min: values[2], Max: values[3]}, nil
+}
+
+func extractGaugeValue(reader *bytes.Reader) (metrics.GaugeValue, error) {
+ values, err := decodeMany(reader, 2)
+ if err != nil {
+ return metrics.GaugeValue{}, err
+ }
+ return metrics.GaugeValue{Timestamp: time.Unix(0, values[0]*int64(time.Millisecond)), Value: values[1]}, nil
+}
+
+func newLabels(miLabels map[string]string) *metrics.Labels {
+ labels := metrics.UserLabels(
+ miLabels["PTRANSFORM"],
+ miLabels[pipepb.MonitoringInfo_NAMESPACE.String()],
Review comment:
That's a wonderful summary! From my point of view, as a contributor, I can add that it was tricky to get the `go generate` command to work. Missing system-wide dependencies (like protobuf headers), path problems in the `gen.go` file, etc. In Python SDK it's much easier, as all you need to do is too install build dependencies into a virtual environment.
It makes sense now to commit those files to the repo.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-726228539
@lostluck Thanks, it's good to hear that we are closer. In addition to your proposed fixes, I also added `Result` methods. The `Result` method will return committed metrics, but if they do don't exist, attempted metrics will be returned instead. This method exists in Python SDK, and AFAIK in Java SDK too.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on pull request #13272:
URL: https://github.com/apache/beam/pull/13272#issuecomment-724755649
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
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520608620
##########
File path: sdks/go/pkg/beam/core/monitoring/monitoring.go
##########
@@ -199,11 +201,11 @@ func getShortID(l metrics.Labels, urn mUrn) string {
return defaultShortIDCache.getShortID(l, urn)
}
-func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
+func ShortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo {
return defaultShortIDCache.shortIdsToInfos(shortids)
}
-func monitoring(p *exec.Plan) ([]*pipepb.MonitoringInfo, map[string][]byte) {
+func Monitoring(p *exec.Plan) ([]*pipepb.MonitoringInfo, map[string][]byte) {
Review comment:
No longer relevant.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kamilwu commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
kamilwu commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r520629281
##########
File path: sdks/go/pkg/beam/testing/ptest/ptest.go
##########
@@ -73,7 +73,8 @@ func Run(p *beam.Pipeline) error {
if *Runner == "" {
*Runner = defaultRunner
}
- return beam.Run(context.Background(), *Runner, p)
+ _, err := beam.Run(context.Background(), *Runner, p)
+ return err
Review comment:
Filled https://issues.apache.org/jira/browse/BEAM-11218
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a change in pull request #13272: [BEAM-11207] Metric Extraction via proto RPC API
Posted by GitBox <gi...@apache.org>.
lostluck commented on a change in pull request #13272:
URL: https://github.com/apache/beam/pull/13272#discussion_r523124215
##########
File path: sdks/go/pkg/beam/core/runtime/metricsx/metricsx_test.go
##########
@@ -0,0 +1,166 @@
+// 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 metricsx
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/metrics"
+ pipepb "github.com/apache/beam/sdks/go/pkg/beam/model/pipeline_v1"
+ "github.com/google/go-cmp/cmp"
+)
+
+func TestFromMonitoringInfos_Counters(t *testing.T) {
+ var value int64 = 15
+ want := metrics.CounterResult{
+ Attempted: 15,
+ Committed: -1,
+ Key: metrics.StepKey{
+ Step: "main.customDoFn",
+ Name: "customCounter",
+ Namespace: "customDoFn",
+ }}
+
+ payload, err := Int64Counter(value)
+ if err != nil {
+ t.Fatalf("Failed to encode Int64Counter: %v", err)
+ }
+
+ labels := map[string]string{
+ "PTRANSFORM": "main.customDoFn",
+ "NAMESPACE": "customDoFn",
+ "NAME": "customCounter",
+ }
+
+ mInfo := &pipepb.MonitoringInfo{
+ Urn: UrnToString(UrnUserSumInt64),
+ Type: UrnToType(UrnUserSumInt64),
+ Labels: labels,
+ Payload: payload,
+ }
+
+ attempted := []*pipepb.MonitoringInfo{mInfo}
+ committed := []*pipepb.MonitoringInfo{}
+
+ got := FromMonitoringInfos(attempted, committed).AllMetrics().Counters()
+ size := len(got)
+ if size < 1 {
+ t.Fatalf("Invalid array's size: got: %v, want: %v", size, 1)
+ }
+ if d := cmp.Diff(got[0], want); d != "" {
Review comment:
You'll want to have want before got when using cmp.Diff. Otherwise your diff guidance is wrong.
cmp.Diff(want, got[0]) means the diff will be (-want,+got) that is saying what is missing from want, and extra in got, which is a useful way to read them. cmp.Diff(got[0],want) means the diff will be (+want,-got), which will be confusing.
----------------------------------------------------------------
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