diff --git a/controller/BUILD.bazel b/controller/BUILD.bazel index d74b30f..9f25e63 100644 --- a/controller/BUILD.bazel +++ b/controller/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "getchangedtargetgraph.go", "getchangedtargets.go", "gettargetgraph.go", + "metrics.go", "output_filter.go", ], importpath = "github.com/uber/tango/controller", @@ -24,6 +25,7 @@ go_library( "//internal/mapper/idmapper", "//internal/targetdiff", "//internal/url", + "//observability/metrics", "//orchestrator", "//tangopb", "@com_github_uber_go_tally//:tally", @@ -39,6 +41,7 @@ go_test( "distance_filter_test.go", "getchangedtargets_test.go", "gettargetgraph_test.go", + "metrics_test.go", "output_filter_test.go", "testhelper_test.go", ], @@ -48,6 +51,7 @@ go_test( "//core/storage", "//core/storage/storagemock", "//entity", + "//observability/metrics", "//orchestrator/orchestratormock", "//tangopb", "//tangopb/tangopbmock", diff --git a/controller/controller.go b/controller/controller.go index 34ca33c..ab482ad 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -16,12 +16,12 @@ package controller import ( "context" - "time" "github.com/uber-go/tally" "github.com/uber/tango/config" "github.com/uber/tango/core/common" "github.com/uber/tango/core/storage" + "github.com/uber/tango/observability/metrics" "github.com/uber/tango/orchestrator" pb "github.com/uber/tango/tangopb" "go.uber.org/fx" @@ -38,18 +38,14 @@ type Params struct { ChunkConfig config.ChunkConfig `optional:"true"` } -// _totalDurationBuckets covers 0–15 minutes in 10-second linear intervals. -var _totalDurationBuckets = tally.MustMakeLinearDurationBuckets(10*time.Second, 10*time.Second, 90) - type controller struct { logger *zap.Logger storage storage.Storage orchestrator orchestrator.Orchestrator - scope tally.Scope + emitter *metrics.Emitter targetChunkSize int changedTargetChunkSize int metadataMapChunkSize int - totalDurationBuckets tally.Buckets // appCtx is the application lifetime; cancel it on process shutdown. // Used by linkRequestCtx and any fire-and-forget goroutines so they @@ -64,6 +60,7 @@ func NewController(appCtx context.Context, p Params) pb.TangoYARPCServer { if scope == nil { scope = tally.NoopScope } + emitter, _ := metrics.New(scope.SubScope("controller")) targetChunkSize := p.ChunkConfig.TargetChunkSize if targetChunkSize <= 0 { targetChunkSize = common.DefaultTargetChunkSize @@ -80,11 +77,10 @@ func NewController(appCtx context.Context, p Params) pb.TangoYARPCServer { logger: p.Logger, storage: p.Storage, orchestrator: p.Orchestrator, - scope: scope.SubScope("controller"), + emitter: emitter, targetChunkSize: targetChunkSize, changedTargetChunkSize: changedTargetChunkSize, metadataMapChunkSize: metadataMapChunkSize, - totalDurationBuckets: _totalDurationBuckets, appCtx: appCtx, } } diff --git a/controller/errors.go b/controller/errors.go index 7a0d069..af49a48 100644 --- a/controller/errors.go +++ b/controller/errors.go @@ -18,8 +18,8 @@ import ( "context" "errors" - "github.com/uber-go/tally" "github.com/uber/tango/core/common" + "github.com/uber/tango/observability/metrics" ) // failure_reason tag values for errors that originate in the controller itself. @@ -38,8 +38,9 @@ const ( // emitFailureMetric tags the failure counter with the reason and type from the // error's ClassifiedError. Context errors are recognised explicitly; everything -// else falls back to unknown/infra. -func emitFailureMetric(scope tally.Scope, err error) { +// else falls back to unknown/infra. e should already carry the repo tag; op is +// the operation subscope the counter lands under. +func emitFailureMetric(e *metrics.Emitter, op string, err error) { var ce common.ClassifiedError switch { case errors.As(err, &ce): @@ -51,8 +52,8 @@ func emitFailureMetric(scope tally.Scope, err error) { default: ce = common.WithReason(common.FailureReasonUnknown, common.ErrorTypeInfra, err) } - scope.Tagged(map[string]string{ + e.Tagged(map[string]string{ "failure_type": ce.Type(), "failure_reason": ce.Reason(), - }).Counter("failure_type").Inc(1) + }).Counter(op, "failures").Inc(1) } diff --git a/controller/getchangedtargetgraph.go b/controller/getchangedtargetgraph.go index c8b65c0..6e823e7 100644 --- a/controller/getchangedtargetgraph.go +++ b/controller/getchangedtargetgraph.go @@ -15,20 +15,15 @@ package controller import ( + "github.com/uber/tango/observability/metrics" pb "github.com/uber/tango/tangopb" ) // GetChangedTargetGraph is the streaming RPC that will return the subgraph // induced by the changed targets between two revisions. It is currently a -// stub: it records success/failure metrics and returns no data. +// stub: it records the start/finish lifecycle and returns no data. func (c *controller) GetChangedTargetGraph(request *pb.GetChangedTargetGraphRequest, stream pb.TangoServiceGetChangedTargetGraphYARPCServer) (retErr error) { - scope := c.scope.SubScope("get_changed_target_graph") - defer func() { - if retErr != nil { - scope.Counter("failure").Inc(1) - } else { - scope.Counter("success").Inc(1) - } - }() + op := metrics.Begin(c.emitter, opGetChangedTargetGraph, slowDurationBuckets) + defer func() { op.Complete(retErr) }() return nil } diff --git a/controller/getchangedtargets.go b/controller/getchangedtargets.go index 72ec2bf..c5bd2dc 100644 --- a/controller/getchangedtargets.go +++ b/controller/getchangedtargets.go @@ -21,7 +21,6 @@ import ( "io" "time" - "github.com/uber-go/tally" "github.com/uber/tango/core/cachekey" "github.com/uber/tango/core/common" tangoerrors "github.com/uber/tango/core/errors" @@ -31,6 +30,7 @@ import ( "github.com/uber/tango/internal/mapper/idmapper" "github.com/uber/tango/internal/targetdiff" "github.com/uber/tango/internal/url" + "github.com/uber/tango/observability/metrics" pb "github.com/uber/tango/tangopb" "go.uber.org/zap" ) @@ -49,26 +49,23 @@ type job struct { // client disconnects, the stream's context is cancelled and the function // returns with context.Canceled. func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer) (retErr error) { - scope := c.scope.SubScope("get_changed_targets") - scope.Counter("calls").Inc(1) + repo := url.ToShortRemote(request.GetFirstRevision().GetRemote()) + e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repo}) + op := metrics.Begin(e, opGetChangedTargets, slowDurationBuckets) logger := c.logger.WithLazy( zap.Any("first_revision", request.GetFirstRevision()), zap.Any("second_revision", request.GetSecondRevision()), ) defer func() { + op.Complete(retErr) if retErr != nil { logger.Error("GetChangedTargets failed", tangoerrors.Fields(retErr)...) - scope.Counter("failure").Inc(1) - emitFailureMetric(scope, retErr) - retErr = mapper.ToProtoError(retErr) - } else { - scope.Counter("success").Inc(1) + emitFailureMetric(e, opGetChangedTargets, retErr) } }() if err := validateGetChangedTargetsRequest(request); err != nil { return tangoerrors.NewUser(err) } - scope = scope.Tagged(map[string]string{"repo": url.ToShortRemote(request.GetFirstRevision().GetRemote())}) ctx, cancelLink := c.linkRequestCtx(stream.Context()) defer cancelLink() start := time.Now() @@ -85,7 +82,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // Fast path: stream a previously computed result straight from cache. if !request.GetBypassCache() { - served, err := c.serveChangedTargetsFromCache(ctx, scope, logger, request, stream, maxDist, start) + served, err := c.serveChangedTargetsFromCache(ctx, e, logger, request, stream, maxDist, start) if err != nil { return fmt.Errorf("serve from cache: %w", err) } @@ -95,12 +92,12 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str } // Fetch both revisions' target graphs concurrently. - firstGraph, secondGraph, err := c.fetchTargetGraphs(ctx, scope, logger, request) + firstGraph, secondGraph, err := c.fetchTargetGraphs(ctx, e, logger, request) if err != nil { return fmt.Errorf("fetch target graphs: %w", err) } - changedTargetsResponses, err := c.compareTargetGraphs(ctx, scope, logger, firstGraph, secondGraph, maxDist) + changedTargetsResponses, err := c.compareTargetGraphs(ctx, e, logger, firstGraph, secondGraph, maxDist) // Allow GC of raw graph data while the caching goroutine runs. firstGraph = nil secondGraph = nil @@ -119,15 +116,12 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str return fmt.Errorf("send response: %w", err) } sendDuration := time.Since(sendStart) - scope.Timer("send_duration").Record(sendDuration) + e.DurationHistogram(opGetChangedTargets, "send_duration", fastDurationBuckets).RecordDuration(sendDuration) - totalDuration := time.Since(start) logger.Info("GetChangedTargets: Successfully processed request", zap.Duration("send_duration", sendDuration), - zap.Duration("total_duration", totalDuration), + zap.Duration("total_duration", time.Since(start)), ) - scope.Timer("total_duration").Record(totalDuration) - scope.Histogram("total_duration.histogram", c.totalDurationBuckets).RecordDuration(totalDuration) return nil } @@ -141,7 +135,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str // real storage error surfaces here so an infra failure that disables the cache // (e.g. a missing-deadline "missing TTL" reject) becomes a visible request failure // rather than silent degradation. -func (c *controller) serveChangedTargetsFromCache(ctx context.Context, scope tally.Scope, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time) (bool, error) { +func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer, maxDist int32, start time.Time) (bool, error) { cacheStart := time.Now() treehash1, treehash2, err := readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision()) if err != nil { @@ -195,17 +189,14 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, scope tal logger.Info("GetChangedTargets: Cache hit, streaming from storage", zap.Duration("cache_read_duration", cacheReadDuration), ) - scope.Counter("changed_targets_cache_hit").Inc(1) - scope.Timer("cache_read_duration").Record(cacheReadDuration) + e.Counter(opGetChangedTargets, "cache_hit").Inc(1) + e.DurationHistogram(opGetChangedTargets, "cache_read_duration", fastDurationBuckets).RecordDuration(cacheReadDuration) if sendErr := sendTrimmedChangedTargets(stream, cached, maxDist, request.GetOutputConfig()); sendErr != nil { return false, fmt.Errorf("send cached response: %w", sendErr) } - totalDuration := time.Since(start) logger.Info("GetChangedTargets: Successfully streamed from cache", - zap.Duration("total_duration", totalDuration), + zap.Duration("total_duration", time.Since(start)), ) - scope.Timer("total_duration").Record(totalDuration) - scope.Histogram("total_duration.histogram", c.totalDurationBuckets).RecordDuration(totalDuration) return true, nil } @@ -215,7 +206,7 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, scope tal // Errors caused solely by that induced cancellation are dropped; only the // original failure is returned. A client disconnect surfaces as a user-cancelled // error. -func (c *controller) fetchTargetGraphs(ctx context.Context, scope tally.Scope, logger *zap.Logger, request *pb.GetChangedTargetsRequest) ([]*pb.GetTargetGraphResponse, []*pb.GetTargetGraphResponse, error) { +func (c *controller) fetchTargetGraphs(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, request *pb.GetChangedTargetsRequest) ([]*pb.GetTargetGraphResponse, []*pb.GetTargetGraphResponse, error) { jobs := make([]*job, 2) for i := 0; i < 2; i++ { // create independent contexts for each job; if one of the jobs fails, the other one should be cancelled to save resources and improve latency @@ -258,7 +249,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, scope tally.Scope, l ExcludeFilesRegex: request.GetRequestOptions().GetExtraExcludeFilesRegex(), BypassCache: request.GetBypassCache(), } - graphReader, err := c.getGraph(jobs[idx].ctx, entityReq) + graphReader, err := c.getGraph(jobs[idx].ctx, e, entityReq) if err != nil || graphReader == nil { results <- graphResult{order: idx, err: err} return @@ -310,7 +301,7 @@ func (c *controller) fetchTargetGraphs(ctx context.Context, scope tally.Scope, l logger.Info("GetChangedTargets: Both graphs fetched", zap.Duration("graph_fetch_duration", graphFetchDuration), ) - scope.Timer("graph_fetch_duration").Record(graphFetchDuration) + e.DurationHistogram(opGetChangedTargets, "graph_fetch_duration", slowDurationBuckets).RecordDuration(graphFetchDuration) if ctx.Err() != nil { // If the context was cancelled by the upstream, just return the original error without additional augmentation @@ -381,13 +372,13 @@ func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetCha // are re-mapped into a canonical per-call ID namespace so the response metadata // only carries the names actually referenced. See internal/targetdiff for the // classification and distance rules. -func (c *controller) compareTargetGraphs(ctx context.Context, scope tally.Scope, logger *zap.Logger, firstGraph, secondGraph []*pb.GetTargetGraphResponse, maxDist int32) ([]*pb.GetChangedTargetsResponse, error) { - start := time.Now() - compareScope := scope.SubScope("compare_target_graphs") +func (c *controller) compareTargetGraphs(ctx context.Context, e *metrics.Emitter, logger *zap.Logger, firstGraph, secondGraph []*pb.GetTargetGraphResponse, maxDist int32) (_ []*pb.GetChangedTargetsResponse, retErr error) { + op := metrics.Begin(e, opCompareTargetGraphs, slowDurationBuckets) + defer func() { op.Complete(retErr) }() logger.Info("compareTargetGraphs: Computing differences between target graphs") // 1) Decode each stream into a semantic graph keyed by canonical target name. - indexStart := time.Now() + decodeStart := time.Now() firstTargetsByID, firstMetadata, err := getTargetsAndMetadata(ctx, firstGraph) if err != nil { return nil, err @@ -412,11 +403,10 @@ func (c *controller) compareTargetGraphs(ctx context.Context, scope tally.Scope, } secondTargetsByID = nil secondMetadata = nil - indexDuration := time.Since(indexStart) - compareScope.Timer("index_duration").Record(indexDuration) + e.DurationHistogram(opCompareTargetGraphs, "decode_duration", fastDurationBuckets).RecordDuration(time.Since(decodeStart)) // 2) Compare the two semantic graphs. - computeStart := time.Now() + diffStart := time.Now() result, err := targetdiff.Compare(ctx, targetdiff.Request{ Before: before, After: after, @@ -428,7 +418,8 @@ func (c *controller) compareTargetGraphs(ctx context.Context, scope tally.Scope, // Release the input graphs; only result is needed from here on. before = nil after = nil - compareScope.Timer("compute_duration").Record(time.Since(computeStart)) + e.DurationHistogram(opCompareTargetGraphs, "diff_duration", fastDurationBuckets).RecordDuration(time.Since(diffStart)) + e.ValueHistogram(opGetChangedTargets, "target_count", changedTargetCountBuckets).RecordValue(float64(len(result.ChangedTargets))) if ctx.Err() != nil { return nil, ctx.Err() @@ -484,14 +475,7 @@ func (c *controller) compareTargetGraphs(ctx context.Context, scope tally.Scope, }, }) } - totalDuration := time.Since(start) - compareScope.Timer("total_duration").Record(totalDuration) - // This helper owns its own timing/log on the request scope (mirroring - // fetchTargetGraphs) rather than leaving it to the caller. - logger.Info("GetChangedTargets: Target graphs compared", - zap.Duration("compare_duration", totalDuration), - ) - scope.Timer("compare_duration").Record(totalDuration) + logger.Info("GetChangedTargets: Target graphs compared") return results, nil } diff --git a/controller/getchangedtargets_test.go b/controller/getchangedtargets_test.go index b2e303b..0c203c8 100644 --- a/controller/getchangedtargets_test.go +++ b/controller/getchangedtargets_test.go @@ -146,7 +146,7 @@ func TestCompareTargetGraphs(t *testing.T) { }, } - response, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), []*pb.GetTargetGraphResponse{firstGraph}, []*pb.GetTargetGraphResponse{secondGraph}, -1) + response, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), []*pb.GetTargetGraphResponse{firstGraph}, []*pb.GetTargetGraphResponse{secondGraph}, -1) require.NoError(t, err) require.NotNil(t, response) } @@ -609,7 +609,7 @@ func TestCompareTargetGraphs_NewTarget_CanonicalIDs(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) require.Len(t, res, 2) cs := res[0].GetChangedTargets() @@ -681,7 +681,7 @@ func TestCompareTargetGraphs_SourceFileDirectAndPropagation(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -754,7 +754,7 @@ func TestCompareTargetGraphs_ChangedRuleUnreachableFromAnySeed(t *testing.T) { // Hash-only change on a rule with no own-config change and no reachable // seed: under "trust the hasher" semantics, an orphan CHANGED rule with // no upstream explanation becomes a distance-0 seed itself. - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -821,7 +821,7 @@ func TestCompareTargetGraphs_ChangedWhenDependenciesChanged(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -903,7 +903,7 @@ func TestCompareTargetGraphs_ChangedWhenAttributesChanged(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -985,7 +985,7 @@ func TestCompareTargetGraphs_ChangedWhenNewAttributeAdded(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -1186,7 +1186,7 @@ func TestCompareTargetGraphs_HashOnlyChangePropagatesViaBFS(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -1268,7 +1268,7 @@ func TestCompareTargetGraphs_SiblingRuleNotPromotedToSeed(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -1322,7 +1322,7 @@ func TestCompareTargetGraphs_DeletedTargetEmitted(t *testing.T) { }, }, } - res, err := c.compareTargetGraphs(t.Context(), c.scope, zap.NewNop(), first, second, -1) + res, err := c.compareTargetGraphs(t.Context(), c.emitter, zap.NewNop(), first, second, -1) require.NoError(t, err) cs := res[0].GetChangedTargets() require.NotNil(t, cs) @@ -1410,7 +1410,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { c.storage = st stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.scope, c.logger, changedTargetsRequest(), stream, -1, time.Now()) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now()) require.NoError(t, err) assert.False(t, served, "a cache miss must not be served") }) @@ -1447,7 +1447,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) // No Send expectation: a corrupt blob must not send anything to the client. - served, err := c.serveChangedTargetsFromCache(t.Context(), c.scope, c.logger, changedTargetsRequest(), stream, -1, time.Now()) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now()) require.NoError(t, err) assert.False(t, served, "a corrupt blob must trigger recompute, not a partial send") }) @@ -1485,7 +1485,7 @@ func TestServeChangedTargetsFromCache(t *testing.T) { stream := tangomock.NewMockTangoServiceGetChangedTargetsYARPCServer(ctrl) stream.EXPECT().Send(gomock.Any()).Return(nil).Times(2) - served, err := c.serveChangedTargetsFromCache(t.Context(), c.scope, c.logger, changedTargetsRequest(), stream, -1, time.Now()) + served, err := c.serveChangedTargetsFromCache(t.Context(), c.emitter, c.logger, changedTargetsRequest(), stream, -1, time.Now()) require.NoError(t, err) assert.True(t, served, "a clean cache hit must be served") }) @@ -1514,7 +1514,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - first, second, err := c.fetchTargetGraphs(t.Context(), c.scope, c.logger, bypassRequest()) + first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) require.NoError(t, err) require.Len(t, first, 1) require.Len(t, second, 1) @@ -1535,7 +1535,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - first, second, err := c.fetchTargetGraphs(t.Context(), c.scope, c.logger, bypassRequest()) + first, second, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) require.Error(t, err) assert.ErrorIs(t, err, injected) assert.Nil(t, first) @@ -1553,7 +1553,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - _, _, err := c.fetchTargetGraphs(t.Context(), c.scope, c.logger, bypassRequest()) + _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) require.Error(t, err) }) @@ -1568,7 +1568,7 @@ func TestFetchTargetGraphs(t *testing.T) { c := newTestController(zaptest.NewLogger(t)) c.orchestrator = orch - _, _, err := c.fetchTargetGraphs(t.Context(), c.scope, c.logger, bypassRequest()) + _, _, err := c.fetchTargetGraphs(t.Context(), c.emitter, c.logger, bypassRequest()) require.Error(t, err) }) } diff --git a/controller/gettargetgraph.go b/controller/gettargetgraph.go index f085126..95e7362 100644 --- a/controller/gettargetgraph.go +++ b/controller/gettargetgraph.go @@ -25,6 +25,7 @@ import ( "github.com/uber/tango/entity" "github.com/uber/tango/internal/mapper" "github.com/uber/tango/internal/url" + "github.com/uber/tango/observability/metrics" "github.com/uber/tango/core/storage" pb "github.com/uber/tango/tangopb" @@ -33,19 +34,17 @@ import ( // GetTargetGraph returns the target graph for a given request. func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb.TangoServiceGetTargetGraphYARPCServer) (retErr error) { - scope := c.scope.SubScope("get_target_graph") - scope.Counter("calls").Inc(1) + repo := url.ToShortRemote(request.GetBuildDescription().GetRemote()) + e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repo}) + op := metrics.Begin(e, opGetTargetGraph, slowDurationBuckets) logger := c.logger.WithLazy( zap.Any("build_description", request.GetBuildDescription()), ) defer func() { + op.Complete(retErr) if retErr != nil { logger.Error("GetTargetGraph failed", tangoerrors.Fields(retErr)...) - scope.Counter("failure").Inc(1) - emitFailureMetric(scope, retErr) - retErr = mapper.ToProtoError(retErr) - } else { - scope.Counter("success").Inc(1) + emitFailureMetric(e, opGetTargetGraph, retErr) } }() start := time.Now() @@ -55,8 +54,7 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb if err != nil { return tangoerrors.NewUser(fmt.Errorf("convert get target graph request: %w", err)) } - scope = scope.Tagged(map[string]string{"repo": url.ToShortRemote(entityReq.Build.Remote)}) - graphReader, err := c.getGraph(ctx, entityReq) + graphReader, err := c.getGraph(ctx, e, entityReq) if err != nil { return fmt.Errorf("get graph: %w", err) } @@ -71,13 +69,11 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb graphStreamChunk, err := graphReader.Read() if err == io.EOF { sendDuration := time.Since(sendStart) - totalDuration := time.Since(start) logger.Info("GetTargetGraph: Done streaming", zap.Duration("send_duration", sendDuration), - zap.Duration("total_duration", totalDuration), + zap.Duration("total_duration", time.Since(start)), ) - scope.Timer("send_duration").Record(sendDuration) - scope.Timer("total_duration").Record(totalDuration) + e.DurationHistogram(opGetTargetGraph, "send_duration", fastDurationBuckets).RecordDuration(sendDuration) return nil } if err != nil { @@ -97,7 +93,7 @@ func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb // entries store the full payload and stripping happens at send time, so // letting an orchestrator see it could poison the shared cache with // stripped graphs. -func (c *controller) getGraph(ctx context.Context, req entity.GetTargetGraphRequest) (storage.GraphReader, error) { +func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entity.GetTargetGraphRequest) (storage.GraphReader, error) { start := time.Now() logger := c.logger.With( zap.Any("build_description", req.Build), @@ -138,10 +134,8 @@ func (c *controller) getGraph(ctx context.Context, req entity.GetTargetGraphRequ zap.Duration("storage_duration", time.Since(storageStart)), zap.Duration("total_duration", time.Since(start)), ) - scope := c.scope.SubScope("get_graph") - scope.Counter("graph_cache_hit").Inc(1) - scope.Timer("storage_duration").Record(time.Since(storageStart)) - scope.Timer("total_duration").Record(time.Since(start)) + e.Counter(opGetTargetGraph, "cache_hit").Inc(1) + e.DurationHistogram(opGetTargetGraph, "download_graph", slowDurationBuckets).RecordDuration(time.Since(storageStart)) return graphReader, nil } } @@ -160,8 +154,5 @@ func (c *controller) getGraph(ctx context.Context, req entity.GetTargetGraphRequ zap.Duration("compute_duration", time.Since(computeStart)), zap.Duration("total_duration", time.Since(start)), ) - scope := c.scope.SubScope("get_graph") - scope.Timer("compute_duration").Record(time.Since(computeStart)) - scope.Timer("total_duration").Record(time.Since(start)) return graphReader, nil } diff --git a/controller/metrics.go b/controller/metrics.go new file mode 100644 index 0000000..5e67806 --- /dev/null +++ b/controller/metrics.go @@ -0,0 +1,49 @@ +// 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 controller + +import ( + "time" + + "github.com/uber-go/tally" +) + +// Operation names, snake_cased after the RPC interface methods they measure. +const ( + opGetTargetGraph = "get_target_graph" + opGetChangedTargets = "get_changed_targets" + opGetChangedTargetGraph = "get_changed_target_graph" + opCompareTargetGraphs = "compare_target_graphs" +) + +// Metric buckets for the controller's operations. Per the observability/metrics +// design, buckets are declared as package-level values next to the handlers and +// passed to Begin (finish) or the custom-metric callsites. +var ( + // fastDurationBuckets covers cheap sub-operations that are normally + // milliseconds to seconds (sends, cache reads, decode, diff). Fine-grained + // so sub-second latencies are distinguishable: exponential 1ms..~35m. + fastDurationBuckets = tally.MustMakeExponentialDurationBuckets(time.Millisecond, 2, 22) + + // slowDurationBuckets covers whole-operation finishes and hours-scale work + // (graph fetch, download). The edges match the orchestrator and graphrunner + // slow buckets so the same compute is comparable across nesting layers: + // exponential 1ms..~12h. + slowDurationBuckets = tally.MustMakeExponentialDurationBuckets(time.Millisecond, 3, 17) + + // changedTargetCountBuckets covers changed-target counts. Graphs already + // reach ~4M targets, so a diff can be large: exponential 1..~100M. + changedTargetCountBuckets = tally.MustMakeExponentialValueBuckets(1, 10, 9) +) diff --git a/controller/metrics_test.go b/controller/metrics_test.go new file mode 100644 index 0000000..80e506f --- /dev/null +++ b/controller/metrics_test.go @@ -0,0 +1,42 @@ +// 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 controller + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/tango/observability/metrics" + "go.uber.org/zap" +) + +// TestControllerMetricsPathShape drives the stub GetChangedTargetGraph handler +// through a TestScope-backed emitter and asserts the start/finish lifecycle +// lands under .. with the outcome tag. +func TestControllerMetricsPathShape(t *testing.T) { + ts := tally.NewTestScope("controller", nil) + e, err := metrics.New(ts) + require.NoError(t, err) + + c := &controller{logger: zap.NewNop(), emitter: e, appCtx: context.Background()} + require.NoError(t, c.GetChangedTargetGraph(nil, nil)) + + snap := ts.Snapshot() + assert.Contains(t, snap.Counters(), "controller.get_changed_target_graph.start+") + assert.Contains(t, snap.Histograms(), "controller.get_changed_target_graph.finish+result=success") +} diff --git a/controller/testhelper_test.go b/controller/testhelper_test.go index 56ff439..18c04c0 100644 --- a/controller/testhelper_test.go +++ b/controller/testhelper_test.go @@ -17,19 +17,18 @@ package controller import ( "context" - "github.com/uber-go/tally" "github.com/uber/tango/core/common" + "github.com/uber/tango/observability/metrics" "go.uber.org/zap" ) func newTestController(logger *zap.Logger) *controller { return &controller{ logger: logger, - scope: tally.NoopScope, + emitter: metrics.Nop(), targetChunkSize: common.DefaultTargetChunkSize, changedTargetChunkSize: common.DefaultChangedTargetChunkSize, metadataMapChunkSize: common.DefaultMetadataMapChunkSize, - totalDurationBuckets: _totalDurationBuckets, appCtx: context.Background(), } }