From 9a824291f7fc4c2579fa9f46b99752e2e640ca40 Mon Sep 17 00:00:00 2001 From: Xiaoyang Tan Date: Fri, 17 Jul 2026 14:03:59 -0700 Subject: [PATCH] refactor(orchestrator): route metrics through observability/metrics Emitter Port native_orchestrator's GetTargetGraph metrics off raw tally.Scope to the Begin/Complete lifecycle helper. - Params keeps Scope tally.Scope; the orchestrator self-subscopes to "orchestrator" and builds its own emitter (still forwards the scope to the graph runner). - start + result-tagged finish (replaces calls/success/failure); recordFailure keeps the failure_type/failure_reason axis as a "failures" counter. - Buckets are package-level in orchestrator/metrics.go. --- orchestrator/BUILD.bazel | 3 ++ orchestrator/metrics.go | 29 ++++++++++++++++ orchestrator/native_orchestrator.go | 52 ++++++++++++++++++----------- 3 files changed, 65 insertions(+), 19 deletions(-) create mode 100644 orchestrator/metrics.go diff --git a/orchestrator/BUILD.bazel b/orchestrator/BUILD.bazel index effa0b95..9baa3b5d 100644 --- a/orchestrator/BUILD.bazel +++ b/orchestrator/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "orchestrator", srcs = [ "errors.go", + "metrics.go", "native_orchestrator.go", "orchestrator.go", ], @@ -21,6 +22,8 @@ go_library( "//core/workspace", "//entity", "//graphrunner", + "//internal/url", + "//observability/metrics", "@com_github_uber_go_tally//:tally", "@org_uber_go_zap//:zap", ], diff --git a/orchestrator/metrics.go b/orchestrator/metrics.go new file mode 100644 index 00000000..83de9616 --- /dev/null +++ b/orchestrator/metrics.go @@ -0,0 +1,29 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed 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 orchestrator + +import ( + "time" + + "github.com/uber-go/tally" +) + +// opGetTargetGraph is snake_cased after the Orchestrator.GetTargetGraph method. +const _opGetTargetGraph = "get_target_graph" + +// _stepDurationBuckets covers whole-operation and individual pipeline steps +// (lease, checkout, apply, cache read/write, compute). A compute on a large +// repo can run for hours, so the range extends to ~4h: exponential 1ms..~4h. +var _stepDurationBuckets = tally.MustMakeExponentialDurationBuckets(time.Millisecond, 3, 16) diff --git a/orchestrator/native_orchestrator.go b/orchestrator/native_orchestrator.go index 32725dd9..47ac4225 100644 --- a/orchestrator/native_orchestrator.go +++ b/orchestrator/native_orchestrator.go @@ -34,6 +34,8 @@ import ( "github.com/uber/tango/core/workspace" "github.com/uber/tango/entity" "github.com/uber/tango/graphrunner" + "github.com/uber/tango/internal/url" + "github.com/uber/tango/observability/metrics" "go.uber.org/zap" ) @@ -42,7 +44,10 @@ type nativeOrchestrator struct { storage storage.Storage repoManager repomanager.RepoManager logger *zap.SugaredLogger - scope tally.Scope + // scope is subscoped to the orchestrator component and forwarded to the + // graph runner so its metrics nest under orchestrator.*. + scope tally.Scope + emitter *metrics.Emitter // gitFactory allows injecting a git.Interface constructor for testing gitFactory func(directory string) git.Interface graphRunner graphrunner.GraphRunner @@ -83,12 +88,18 @@ func NewNativeOrchestrator(appCtx context.Context, p Params) (Orchestrator, erro if scope == nil { scope = tally.NoopScope } + scope = scope.SubScope("orchestrator") + emitter, err := metrics.New(scope) + if err != nil { + emitter = metrics.Nop() + } return &nativeOrchestrator{ storage: p.Storage, repoManager: p.RepoManager, logger: p.Logger, - scope: scope.SubScope("orchestrator"), + scope: scope, + emitter: emitter, gitFactory: p.GitFactory, graphRunner: p.GraphRunner, appCtx: appCtx, @@ -99,23 +110,9 @@ func NewNativeOrchestrator(appCtx context.Context, p Params) (Orchestrator, erro // GetTargetGraph is used to compute the target graph locally. // It leases a workspace, checks out the base revision, applies the change requests, and computes the target graph. func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetTargetGraphRequest) (_ storage.GraphReader, retErr error) { - scope := b.scope.SubScope("get_target_graph") - scope.Counter("calls").Inc(1) - defer func() { - if retErr != nil { - scope.Counter("failure").Inc(1) - var ce common.ClassifiedError - if !errors.As(retErr, &ce) { - ce = common.WithReason(common.FailureReasonUnknown, common.ErrorTypeInfra, retErr) - } - scope.Tagged(map[string]string{ - "failure_type": ce.Type(), - "failure_reason": ce.Reason(), - }).Counter("failure_type").Inc(1) - } else { - scope.Counter("success").Inc(1) - } - }() + e := b.emitter.Tagged(map[string]string{metrics.TagRepo: url.ToShortRemote(req.Build.Remote)}) + op := metrics.Begin(e, _opGetTargetGraph, _stepDurationBuckets) + defer func() { op.Complete(retErr) }() build := req.Build logger := b.logger.With(zap.Any("build_description", build)) logger.Infow("GetTargetGraph: Processing request") @@ -125,7 +122,9 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT if !ok { return nil, fmt.Errorf("no repository configuration found for remote %q", remote) } + leaseStart := time.Now() ws, err := b.repoManager.Lease(ctx, build) + recordStep(e, "lease_duration", leaseStart) if err != nil { return nil, classifyLeaseError(err) } @@ -138,7 +137,9 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT } } }() + checkoutStart := time.Now() err = ws.Checkout(ctx, build.Remote, build.BaseSha) + recordStep(e, "checkout_duration", checkoutStart) if err != nil { return nil, fmt.Errorf("checkout %s@%s: %w", build.Remote, build.BaseSha, err) } @@ -158,7 +159,9 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT } requests = append(requests, request) } + applyStart := time.Now() err = ws.ApplyRequests(ctx, requests) + recordStep(e, "apply_requests_duration", applyStart) if err != nil { return nil, fmt.Errorf("apply requests: %w", err) } @@ -171,7 +174,9 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT } treehashPath := cachekey.GetGraphByTreeHash(build.Remote, treehash, build.Strategy, req.ExcludeFilesRegex) if !req.BypassCache { + cacheReadStart := time.Now() graphReader, err := storage.NewGraphReader(ctx, b.storage, treehashPath) + recordStep(e, "cache_read_duration", cacheReadStart) if err == nil { logger.Infow("GetTargetGraph: Cache hit on treehash", zap.String("treehash", treehash)) return graphReader, nil @@ -205,7 +210,9 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT Scope: b.scope, }) } + computeStart := time.Now() result, err := runner.Compute(ctx, ws) + recordStep(e, "compute_duration", computeStart) if err != nil { return nil, fmt.Errorf("compute target graph: %w", err) } @@ -216,6 +223,7 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT if err != nil { return nil, fmt.Errorf("convert target graph to response: %w", err) } + cacheWriteStart := time.Now() err = storage.WriteGraphStream(ctx, b.storage, treehashPath, responses) if err != nil { return nil, fmt.Errorf("write graph to storage at %s: %w", treehashPath, err) @@ -226,6 +234,7 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT if err != nil { return nil, fmt.Errorf("store treehash mapping at %s: %w", treehashCachePath, err) } + recordStep(e, "cache_write_duration", cacheWriteStart) graphReader, err := storage.NewGraphReader(ctx, b.storage, treehashPath) if err != nil { return nil, fmt.Errorf("create graph reader at %s: %w", treehashPath, err) @@ -233,3 +242,8 @@ func (b *nativeOrchestrator) GetTargetGraph(ctx context.Context, req entity.GetT logger.Infow("GetTargetGraph: Done computing and storing target graph", zap.String("treehash", treehash)) return graphReader, nil } + +// recordStep records a pipeline step's duration under the get_target_graph op. +func recordStep(e *metrics.Emitter, name string, start time.Time) { + e.DurationHistogram(_opGetTargetGraph, name, _stepDurationBuckets).RecordDuration(time.Since(start)) +}