refactor(controller): route metrics through observability/metrics Emitter

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",
@@ -22,6 +23,7 @@ go_library(
         "//internal/mapper",
         "//internal/mapper/idmapper",
         "//internal/targetdiff",
+        "//observability/metrics",
         "//orchestrator",
         "//tangopb",
         "@com_github_uber_go_tally//:tally",
@@ -37,6 +39,7 @@ go_test(
         "distance_filter_test.go",
         "getchangedtargets_test.go",
         "gettargetgraph_test.go",
+        "metrics_test.go",
         "output_filter_test.go",
         "testhelper_test.go",
     ],
@@ -46,6 +49,7 @@ go_test(
         "//core/storage",
         "//core/storage/storagemock",
         "//entity",
+        "//observability/metrics",
         "//orchestrator/orchestratormock",
         "//tangopb",
         "//tangopb/tangopbmock",

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,
 	}
 }

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)
 }

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, requestFinishBuckets)
+	defer func() { op.Complete(retErr) }()
 	return nil
 }

controller/getchangedtargets.go

@@ -21,14 +21,14 @@ import (
 	"io"
 	"time"
 
-	"github.com/uber-go/tally"
 	"github.com/uber/tango/core/common"
 	"github.com/uber/tango/core/storage"
 	"github.com/uber/tango/entity"
 	"github.com/uber/tango/internal/cachekey"
 	"github.com/uber/tango/internal/mapper"
 	"github.com/uber/tango/internal/mapper/idmapper"
 	"github.com/uber/tango/internal/targetdiff"
+	"github.com/uber/tango/observability/metrics"
 	pb "github.com/uber/tango/tangopb"
 	"go.uber.org/zap"
 )
@@ -47,25 +47,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 := common.ToShortRemote(request.GetFirstRevision().GetRemote())
+	e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repo})
+	op := metrics.Begin(e, opGetChangedTargets, requestFinishBuckets)
 	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", zap.Error(retErr))
-			scope.Counter("failure").Inc(1)
-			emitFailureMetric(scope, retErr)
-		} else {
-			scope.Counter("success").Inc(1)
+			emitFailureMetric(e, opGetChangedTargets, retErr)
 		}
 	}()
 	if err := validateGetChangedTargetsRequest(request); err != nil {
 		return common.WithReason(common.FailureReasonValidation, common.ErrorTypeUser, err)
 	}
-	scope = scope.Tagged(map[string]string{"repo": common.ToShortRemote(request.GetFirstRevision().GetRemote())})
 	ctx, cancelLink := c.linkRequestCtx(stream.Context())
 	defer cancelLink()
 	start := time.Now()
@@ -82,7 +80,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 err
 		}
@@ -92,12 +90,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 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
@@ -116,15 +114,12 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str
 		return common.WithReason(failureReasonSend, common.ErrorTypeInfra, fmt.Errorf("failed to send response: %w", err))
 	}
 	sendDuration := time.Since(sendStart)
-	scope.Timer("send_duration").Record(sendDuration)
+	e.DurationHistogram(opGetChangedTargets, "send_duration", stepDurationBuckets).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
 }
 
@@ -138,7 +133,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 {
@@ -192,17 +187,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", stepDurationBuckets).RecordDuration(cacheReadDuration)
 	if sendErr := sendTrimmedChangedTargets(stream, cached, maxDist, request.GetOutputConfig()); sendErr != nil {
 		return false, common.WithReason(failureReasonSend, common.ErrorTypeInfra, fmt.Errorf("failed to 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
 }
 
@@ -212,7 +204,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
@@ -255,7 +247,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
@@ -307,7 +299,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", stepDurationBuckets).RecordDuration(graphFetchDuration)
 
 	if ctx.Err() != nil {
 		// If the context was cancelled by the upstream, just return the original error without additional augmentation
@@ -378,13 +370,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, requestFinishBuckets)
+	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
@@ -409,11 +401,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", stepDurationBuckets).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,
@@ -425,7 +416,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", stepDurationBuckets).RecordDuration(time.Since(diffStart))
+	e.ValueHistogram(opGetChangedTargets, "target_count", changedTargetCountBuckets).RecordValue(float64(len(result.ChangedTargets)))
 
 	if ctx.Err() != nil {
 		return nil, ctx.Err()
@@ -481,14 +473,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
 }