This is an automated email from the ASF dual-hosted git repository. lostluck pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push: new b426d34f622 [BEAM-14557] Read and Seek Runner Capabilities in Go SDK (#17821) b426d34f622 is described below commit b426d34f62271f8ccb4d24e1f82b28cd26ec8af3 Author: Ritesh Ghorse <riteshgho...@gmail.com> AuthorDate: Mon Jun 13 22:04:28 2022 -0400 [BEAM-14557] Read and Seek Runner Capabilities in Go SDK (#17821) --- sdks/go/container/boot.go | 4 + sdks/go/pkg/beam/core/runtime/graphx/translate.go | 10 +- sdks/go/pkg/beam/core/runtime/harness/harness.go | 25 +++- .../go/pkg/beam/core/runtime/harness/monitoring.go | 133 +++++++++++---------- 4 files changed, 102 insertions(+), 70 deletions(-) diff --git a/sdks/go/container/boot.go b/sdks/go/container/boot.go index 88b654230c1..b180f54bf18 100644 --- a/sdks/go/container/boot.go +++ b/sdks/go/container/boot.go @@ -111,6 +111,10 @@ func main() { args = append(args, "--status_endpoint="+info.GetStatusEndpoint().GetUrl()) } + if len(info.GetRunnerCapabilities()) > 0 { + os.Setenv("RUNNER_CAPABILITIES", strings.Join(info.GetRunnerCapabilities(), " ")) + } + log.Fatalf("User program exited: %v", execx.Execute(prog, args...)) } diff --git a/sdks/go/pkg/beam/core/runtime/graphx/translate.go b/sdks/go/pkg/beam/core/runtime/graphx/translate.go index bde9f9b676a..cb42da25397 100644 --- a/sdks/go/pkg/beam/core/runtime/graphx/translate.go +++ b/sdks/go/pkg/beam/core/runtime/graphx/translate.go @@ -63,9 +63,10 @@ const ( URNWindowMappingFixed = "beam:go:windowmapping:fixed:v1" URNWindowMappingSliding = "beam:go:windowmapping:sliding:v1" - URNLegacyProgressReporting = "beam:protocol:progress_reporting:v0" - URNMultiCore = "beam:protocol:multi_core_bundle_processing:v1" - URNWorkerStatus = "beam:protocol:worker_status:v1" + URNProgressReporting = "beam:protocol:progress_reporting:v1" + URNMultiCore = "beam:protocol:multi_core_bundle_processing:v1" + URNWorkerStatus = "beam:protocol:worker_status:v1" + URNMonitoringInfoShortID = "beam:protocol:monitoring_info_short_ids:v1" URNRequiresSplittableDoFn = "beam:requirement:pardo:splittable_dofn:v1" URNRequiresBundleFinalization = "beam:requirement:pardo:finalization:v1" @@ -86,10 +87,11 @@ const ( func goCapabilities() []string { capabilities := []string{ - URNLegacyProgressReporting, + URNProgressReporting, URNMultiCore, URNTruncate, URNWorkerStatus, + URNMonitoringInfoShortID, // TOOD(BEAM-9614): Make this versioned. "beam:version:sdk_base:go", } diff --git a/sdks/go/pkg/beam/core/runtime/harness/harness.go b/sdks/go/pkg/beam/core/runtime/harness/harness.go index 04a2baf41b4..4ed4cf6d4b2 100644 --- a/sdks/go/pkg/beam/core/runtime/harness/harness.go +++ b/sdks/go/pkg/beam/core/runtime/harness/harness.go @@ -20,6 +20,8 @@ import ( "context" "fmt" "io" + "os" + "strings" "sync" "sync/atomic" "time" @@ -40,6 +42,9 @@ import ( // StatusAddress is a type of status endpoint address as an optional argument to harness.Main(). type StatusAddress string +// URNMonitoringInfoShortID is a URN indicating support for short monitoring info IDs. +const URNMonitoringInfoShortID = "beam:protocol:monitoring_info_short_ids:v1" + // TODO(herohde) 2/8/2017: for now, assume we stage a full binary (not a plugin). // Main is the main entrypoint for the Go harness. It runs at "runtime" -- not @@ -58,6 +63,17 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string, options } } + // Extract environment variables. These are optional runner supported capabilities. + // Expected env variables: + // RUNNER_CAPABILITIES : list of runner supported capability urn. + runnerCapabilities := strings.Split(os.Getenv("RUNNER_CAPABILITIES"), " ") + rcMap := make(map[string]bool) + if len(runnerCapabilities) > 0 { + for _, capability := range runnerCapabilities { + rcMap[capability] = true + } + } + // Pass in the logging endpoint for use w/the default remote logging hook. ctx = context.WithValue(ctx, loggingEndpointCtxKey, loggingEndpoint) ctx, err := hooks.RunInitHooks(ctx) @@ -139,6 +155,7 @@ func Main(ctx context.Context, loggingEndpoint, controlEndpoint string, options data: &DataChannelManager{}, state: &StateChannelManager{}, cache: &sideCache, + runnerCapabilities: rcMap, } // gRPC requires all readers of a stream be the same goroutine, so this goroutine // is responsible for managing the network data. All it does is pull data from @@ -275,7 +292,8 @@ type control struct { data *DataChannelManager state *StateChannelManager // TODO(BEAM-11097): Cache is currently unused. - cache *statecache.SideInputCache + cache *statecache.SideInputCache + runnerCapabilities map[string]bool } func (c *control) getOrCreatePlan(bdID bundleDescriptorID) (*exec.Plan, error) { @@ -370,7 +388,8 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe c.cache.CompleteBundle(tokens...) - mons, pylds := monitoring(plan, store) + mons, pylds := monitoring(plan, store, c.runnerCapabilities[URNMonitoringInfoShortID]) + requiresFinalization := false // Move the plan back to the candidate state c.mu.Lock() @@ -493,7 +512,7 @@ func (c *control) handleInstruction(ctx context.Context, req *fnpb.InstructionRe } } - mons, pylds := monitoring(plan, store) + mons, pylds := monitoring(plan, store, c.runnerCapabilities[URNMonitoringInfoShortID]) return &fnpb.InstructionResponse{ InstructionId: string(instID), diff --git a/sdks/go/pkg/beam/core/runtime/harness/monitoring.go b/sdks/go/pkg/beam/core/runtime/harness/monitoring.go index 76557c32f4f..ef6fbc1d1fc 100644 --- a/sdks/go/pkg/beam/core/runtime/harness/monitoring.go +++ b/sdks/go/pkg/beam/core/runtime/harness/monitoring.go @@ -102,11 +102,10 @@ func shortIdsToInfos(shortids []string) map[string]*pipepb.MonitoringInfo { return defaultShortIDCache.shortIdsToInfos(shortids) } -func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, map[string][]byte) { +func monitoring(p *exec.Plan, store *metrics.Store, supportShortID bool) ([]*pipepb.MonitoringInfo, map[string][]byte) { if store == nil { return nil, nil } - defaultShortIDCache.mu.Lock() defer defaultShortIDCache.mu.Unlock() @@ -119,14 +118,15 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m panic(err) } payloads[getShortID(l, metricsx.UrnUserSumInt64)] = payload - - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(metricsx.UrnUserSumInt64), - Type: metricsx.UrnToType(metricsx.UrnUserSumInt64), - Labels: l.Map(), - Payload: payload, - }) + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(metricsx.UrnUserSumInt64), + Type: metricsx.UrnToType(metricsx.UrnUserSumInt64), + Labels: l.Map(), + Payload: payload, + }) + } }, DistributionInt64: func(l metrics.Labels, count, sum, min, max int64) { payload, err := metricsx.Int64Distribution(count, sum, min, max) @@ -134,14 +134,15 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m panic(err) } payloads[getShortID(l, metricsx.UrnUserDistInt64)] = payload - - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(metricsx.UrnUserDistInt64), - Type: metricsx.UrnToType(metricsx.UrnUserDistInt64), - Labels: l.Map(), - Payload: payload, - }) + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(metricsx.UrnUserDistInt64), + Type: metricsx.UrnToType(metricsx.UrnUserDistInt64), + Labels: l.Map(), + Payload: payload, + }) + } }, GaugeInt64: func(l metrics.Labels, v int64, t time.Time) { payload, err := metricsx.Int64Latest(t, v) @@ -149,15 +150,15 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m panic(err) } payloads[getShortID(l, metricsx.UrnUserLatestMsInt64)] = payload - - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(metricsx.UrnUserLatestMsInt64), - Type: metricsx.UrnToType(metricsx.UrnUserLatestMsInt64), - Labels: l.Map(), - Payload: payload, - }) - + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(metricsx.UrnUserLatestMsInt64), + Type: metricsx.UrnToType(metricsx.UrnUserLatestMsInt64), + Labels: l.Map(), + Payload: payload, + }) + } }, MsecsInt64: func(l string, states *[4]metrics.ExecutionState) { label := map[string]string{"PTRANSFORM": l} @@ -168,13 +169,15 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m } ul := metricsx.ExecutionMsecUrn(i) payloads[getShortID(metrics.PTransformLabels(l), ul)] = payload - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(ul), - Type: metricsx.UrnToType(ul), - Labels: label, - Payload: payload, - }) + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(ul), + Type: metricsx.UrnToType(ul), + Labels: label, + Payload: payload, + }) + } } }, }.ExtractFrom(store) @@ -194,16 +197,17 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m // TODO(BEAM-9934): This metric should account for elements in multiple windows. payloads[getShortID(metrics.PCollectionLabels(pcol.ID), metricsx.UrnElementCount)] = payload - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(metricsx.UrnElementCount), - Type: metricsx.UrnToType(metricsx.UrnElementCount), - Labels: map[string]string{ - "PCOLLECTION": pcol.ID, - }, - Payload: payload, - }) - + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(metricsx.UrnElementCount), + Type: metricsx.UrnToType(metricsx.UrnElementCount), + Labels: map[string]string{ + "PCOLLECTION": pcol.ID, + }, + Payload: payload, + }) + } // Skip pcollections without size if pcol.SizeCount != 0 { payload, err := metricsx.Int64Distribution(pcol.SizeCount, pcol.SizeSum, pcol.SizeMin, pcol.SizeMax) @@ -212,15 +216,17 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m } payloads[getShortID(metrics.PCollectionLabels(pcol.ID), metricsx.UrnSampledByteSize)] = payload - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(metricsx.UrnSampledByteSize), - Type: metricsx.UrnToType(metricsx.UrnSampledByteSize), - Labels: map[string]string{ - "PCOLLECTION": pcol.ID, - }, - Payload: payload, - }) + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(metricsx.UrnSampledByteSize), + Type: metricsx.UrnToType(metricsx.UrnSampledByteSize), + Labels: map[string]string{ + "PCOLLECTION": pcol.ID, + }, + Payload: payload, + }) + } } } @@ -230,15 +236,16 @@ func monitoring(p *exec.Plan, store *metrics.Store) ([]*pipepb.MonitoringInfo, m } payloads[getShortID(metrics.PTransformLabels(snapshot.Source.ID), metricsx.UrnDataChannelReadIndex)] = payload - monitoringInfo = append(monitoringInfo, - &pipepb.MonitoringInfo{ - Urn: metricsx.UrnToString(metricsx.UrnDataChannelReadIndex), - Type: metricsx.UrnToType(metricsx.UrnDataChannelReadIndex), - Labels: map[string]string{ - "PTRANSFORM": snapshot.Source.ID, - }, - Payload: payload, - }) - + if !supportShortID { + monitoringInfo = append(monitoringInfo, + &pipepb.MonitoringInfo{ + Urn: metricsx.UrnToString(metricsx.UrnDataChannelReadIndex), + Type: metricsx.UrnToType(metricsx.UrnDataChannelReadIndex), + Labels: map[string]string{ + "PTRANSFORM": snapshot.Source.ID, + }, + Payload: payload, + }) + } return monitoringInfo, payloads }