refactor(handler): move the metrics lifecycle to the handler

AGENTS.md

@@ -65,7 +65,7 @@ The top-level split is by **responsibility**, not by domain: `controller/` handl
 
 The controller is the YARPC service implementation. It owns transport-adjacent concerns: request validation, metrics, cancellation linkage, response streaming, output filtering, comparison fan-out, and compared-target caching. It does **not** own workspace creation, git operations, or graph computation.
 
-Every RPC records operation metrics, classifies invalid input as a user error, preserves error chains, attempts applicable cache reads before expensive work, delegates graph production to the orchestrator, and converts the final error at the wire boundary.
+The handler records the operation metrics of every RPC, under the `handler` scope. Every RPC classifies invalid input as a user error, preserves error chains, attempts applicable cache reads before expensive work, delegates graph production to the orchestrator, and converts the final error at the wire boundary.
 
 ### Orchestrator
 

controller/BUILD.bazel

@@ -44,7 +44,6 @@ go_test(
         "getchangedtargets_test.go",
         "getchangedtargets_tgb_test.go",
         "gettargetgraph_test.go",
-        "metrics_test.go",
         "output_filter_test.go",
         "testhelper_test.go",
     ],

controller/README.md

@@ -1,7 +1,7 @@
 # Controller
 
-The controller is the request-handling layer of Tango. It implements the
-generated YARPC service surface and turns each streaming RPC into a
+The controller is the request-handling layer of Tango. The handler calls
+it, and it turns each streaming RPC into a
 deterministic pipeline of cache lookup, graph computation, comparison, and
 response streaming.
 
@@ -24,9 +24,9 @@ sit between the wire protocol and the rest of the system:
 - **Streaming and chunking.** Responses are emitted as multiple stream
   messages sized to stay below the gRPC per-message limit. Targets,
   metadata, and topology deltas are chunked independently.
-- **Observability.** Every RPC emits per-call counters, per-phase timers,
-  and a classified failure metric that distinguishes user from
-  infrastructure errors.
+- **Observability.** Every RPC emits per-phase timers. The handler emits the
+  per-call counters and the classified failure metric that distinguishes
+  user from infrastructure errors.
 
 ## Collaborators
 
@@ -46,5 +46,4 @@ behavior:
 The controller is built once at startup with its logger, storage,
 orchestrator, optional metrics scope, and optional max-message-bytes
 configuration. The constructor returns the
-generated server interface, so the controller can be registered with a
-YARPC dispatcher without additional adaptation.
+`Controller` interface, which the handler calls.

controller/controller.go

@@ -30,6 +30,8 @@ import (
 	"go.uber.org/zap"
 )
 
+// Duplicated in the handler. Temporary: the controller drops this copy when it
+// takes the resolved repository from the handler.
 const unknownRepositoryMetricLabel = "unknown"
 
 // Params are the parameters for the controller.
@@ -48,6 +50,9 @@ type Params struct {
 // for a validated request. Invalid requests retain the common unknown label and
 // their existing error; valid requests must exactly match the configured
 // repository allowlist before controller cache I/O.
+//
+// Duplicated in the handler. Temporary: the controller drops this copy when it
+// takes the resolved repository from the handler.
 func (c *controller) resolveRequestRepository(remote string, requestErr error) (config.RepositoryConfig, string, error) {
 	if requestErr != nil {
 		return config.RepositoryConfig{}, unknownRepositoryMetricLabel, requestErr

controller/getchangedtargetgraph.go

@@ -16,7 +16,6 @@ package controller
 
 import (
 	tangoerrors "github.com/uber/tango/core/errors"
-	"github.com/uber/tango/observability/metrics"
 	pb "github.com/uber/tango/tangopb"
 	"go.uber.org/yarpc/yarpcerrors"
 )
@@ -25,13 +24,8 @@ import (
 // induced by the changed targets between two revisions. NOT YET IMPLEMENTED:
 // returns a YARPC Unimplemented error so callers get an explicit failure
 // instead of a silent empty stream.
-func (c *controller) GetChangedTargetGraph(request *pb.GetChangedTargetGraphRequest, stream pb.TangoServiceGetChangedTargetGraphYARPCServer) (retErr error) {
-	op := metrics.Begin(c.emitter, opGetChangedTargetGraph, metrics.SlowDurationBuckets)
-	retErr = tangoerrors.NewInfra(
+func (c *controller) GetChangedTargetGraph(request *pb.GetChangedTargetGraphRequest, stream pb.TangoServiceGetChangedTargetGraphYARPCServer) error {
+	return tangoerrors.NewInfra(
 		yarpcerrors.Newf(yarpcerrors.CodeUnimplemented, "GetChangedTargetGraph is not yet implemented"),
 	)
-	defer func() {
-		op.Complete(retErr)
-	}()
-	return retErr
 }

controller/getchangedtargets.go

@@ -82,20 +82,16 @@ type job struct {
 // GetChangedTargets returns the changed targets between two revisions. If the
 // 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) {
+func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer) error {
 	entityReq, mappingErr := proto.ProtoToGetChangedTargetsRequest(request)
 	if mappingErr != nil {
 		mappingErr = tangoerrors.NewUser(fmt.Errorf("convert get changed targets request: %w", mappingErr))
 	}
 	repoCfg, repo, repositoryErr := c.resolveRequestRepository(entityReq.First.Remote, mappingErr)
 	e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repo})
-	op := metrics.Begin(e, opGetChangedTargets, metrics.SlowDurationBuckets)
 	logger := c.logger.WithLazy(
 		zap.String("repository", repo),
 	)
-	defer func() {
-		op.Complete(retErr)
-	}()
 	if repositoryErr != nil {
 		return repositoryErr
 	}

controller/gettargetgraph.go

@@ -34,20 +34,16 @@ import (
 )
 
 // GetTargetGraph returns the target graph for a given request.
-func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb.TangoServiceGetTargetGraphYARPCServer) (retErr error) {
+func (c *controller) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb.TangoServiceGetTargetGraphYARPCServer) error {
 	entityReq, mappingErr := mapper.ProtoToGetTargetGraphRequest(request)
 	if mappingErr != nil {
 		mappingErr = tangoerrors.NewUser(fmt.Errorf("convert get target graph request: %w", mappingErr))
 	}
 	repoCfg, repo, repositoryErr := c.resolveRequestRepository(entityReq.Build.Remote, mappingErr)
 	e := c.emitter.Tagged(map[string]string{metrics.TagRepo: repo})
-	op := metrics.Begin(e, opGetTargetGraph, metrics.SlowDurationBuckets)
 	logger := c.logger.WithLazy(
 		zap.String("repository", repo),
 	)
-	defer func() {
-		op.Complete(retErr)
-	}()
 	start := time.Now()
 	ctx, cancelLink := c.linkRequestCtx(stream.Context())
 	defer cancelLink()

controller/metrics.go

@@ -16,7 +16,6 @@ package controller
 
 // 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"
+	opGetTargetGraph    = "get_target_graph"
+	opGetChangedTargets = "get_changed_targets"
 )

docs/observability/metrics.md

@@ -208,16 +208,16 @@ if err == nil {
 
 ```
 # operation rate
-fetch service:tango name:controller.get_changed_targets.start
+fetch service:tango name:handler.get_changed_targets.start
 
 # success and classified error counts
-fetch service:tango name:controller.get_changed_targets.finish | sum by (result)
+fetch service:tango name:handler.get_changed_targets.finish | sum by (result)
 
 # P95 latency of successful requests
-fetch service:tango name:controller.get_changed_targets.finish result:success | histogram_percentile(95)
+fetch service:tango name:handler.get_changed_targets.finish result:success | histogram_percentile(95)
 
 # scoped to a repo
-fetch service:tango name:controller.get_changed_targets.finish result:success repo:my-monorepo | histogram_percentile(95)
+fetch service:tango name:handler.get_changed_targets.finish result:success repo:my-monorepo | histogram_percentile(95)
 
 # custom value metric — changed-target count distribution
 fetch service:tango name:controller.get_changed_targets.target_count | histogram_percentile(95)

example/main.go

@@ -109,7 +109,7 @@ func run() error {
 		GraphConfig:     cfg,
 	})
 
-	h := handler.New(handler.Params{Logger: logger, Controller: ctrl})
+	h := handler.New(handler.Params{Logger: logger, Controller: ctrl, RepoConfig: cfg})
 
 	// YARPC transports and dispatcher
 	grpcTransport := yarpcgrpc.NewTransport()

handler/BUILD.bazel

@@ -8,14 +8,20 @@ go_library(
         "getchangedtargets.go",
         "gettargetgraph.go",
         "handler.go",
+        "metrics.go",
     ],
     importpath = "github.com/uber/tango/handler",
     visibility = ["//visibility:public"],
     deps = [
+        "//config",
         "//controller",
         "//core/errors",
+        "//internal/mapper",
         "//mapper/proto",
+        "//observability/metrics",
         "//tangopb",
+        "@com_github_uber_go_tally//:tally",
+        "@org_uber_go_fx//:fx",
         "@org_uber_go_zap//:zap",
     ],
 )
@@ -26,6 +32,8 @@ go_test(
         "getchangedtargetgraph_test.go",
         "getchangedtargets_test.go",
         "gettargetgraph_test.go",
+        "handler_test.go",
+        "metrics_test.go",
         "testhelper_test.go",
         "wireerror_test.go",
     ],
@@ -41,9 +49,11 @@ go_test(
         "//tangopb/tangopbmock",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally//:tally",
         "@org_uber_go_mock//gomock",
         "@org_uber_go_yarpc//encoding/protobuf",
         "@org_uber_go_yarpc//yarpcerrors",
+        "@org_uber_go_zap//:zap",
         "@org_uber_go_zap//zaptest",
     ],
 )

handler/getchangedtargetgraph.go

@@ -15,9 +15,14 @@
 package handler
 
 import (
+	"github.com/uber/tango/observability/metrics"
 	pb "github.com/uber/tango/tangopb"
 )
 
-func (h *handler) GetChangedTargetGraph(request *pb.GetChangedTargetGraphRequest, stream pb.TangoServiceGetChangedTargetGraphYARPCServer) error {
+func (h *handler) GetChangedTargetGraph(request *pb.GetChangedTargetGraphRequest, stream pb.TangoServiceGetChangedTargetGraphYARPCServer) (retErr error) {
+	op := metrics.Begin(h.emitter, opGetChangedTargetGraph, metrics.SlowDurationBuckets)
+	defer func() {
+		op.Complete(retErr)
+	}()
 	return h.controller.GetChangedTargetGraph(request, stream)
 }

handler/getchangedtargets.go

@@ -15,16 +15,35 @@
 package handler
 
 import (
+	"fmt"
+
 	tangoerrors "github.com/uber/tango/core/errors"
+	"github.com/uber/tango/mapper/proto"
+	"github.com/uber/tango/observability/metrics"
 	pb "github.com/uber/tango/tangopb"
+	"go.uber.org/zap"
 )
 
 func (h *handler) GetChangedTargets(request *pb.GetChangedTargetsRequest, stream pb.TangoServiceGetChangedTargetsYARPCServer) (retErr error) {
+	// Temporary: the handler maps the request only to tag the metrics. The
+	// controller maps it again until it takes the entity request.
+	entityReq, mappingErr := proto.ProtoToGetChangedTargetsRequest(request)
+	if mappingErr != nil {
+		mappingErr = tangoerrors.NewUser(fmt.Errorf("convert get changed targets request: %w", mappingErr))
+	}
+	_, repo, repositoryErr := h.resolveRequestRepository(entityReq.First.Remote, mappingErr)
+	e := h.emitter.Tagged(map[string]string{metrics.TagRepo: repo})
+	op := metrics.Begin(e, opGetChangedTargets, metrics.SlowDurationBuckets)
+	logger := h.logger.WithLazy(zap.String("repository", repo))
 	defer func() {
+		op.Complete(retErr)
 		if retErr != nil {
-			h.logger.Error("GetChangedTargets failed", tangoerrors.Fields(retErr)...)
+			logger.Error("GetChangedTargets failed", tangoerrors.Fields(retErr)...)
 			retErr = toWireError(retErr)
 		}
 	}()
+	if repositoryErr != nil {
+		return repositoryErr
+	}
 	return h.controller.GetChangedTargets(request, stream)
 }

handler/getchangedtargets_test.go

@@ -31,9 +31,12 @@ type fakeGetChangedTargetsStream struct {
 
 func TestGetChangedTargetsForwardsToController(t *testing.T) {
 	ctrl := mock_controller.NewMockController(gomock.NewController(t))
-	h := New(Params{Logger: zaptest.NewLogger(t), Controller: ctrl})
+	h := New(Params{Logger: zaptest.NewLogger(t), RepoConfig: allowAnyRepositoryConfigProvider{}, Controller: ctrl})
 
-	req := &pb.GetChangedTargetsRequest{}
+	req := &pb.GetChangedTargetsRequest{
+		FirstRevision:  &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha1"},
+		SecondRevision: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha2"},
+	}
 	stream := &fakeGetChangedTargetsStream{}
 	ctrl.EXPECT().GetChangedTargets(gomock.Any(), gomock.Any()).DoAndReturn(
 		func(gotReq *pb.GetChangedTargetsRequest, gotStream pb.TangoServiceGetChangedTargetsYARPCServer) error {

handler/gettargetgraph.go

@@ -15,16 +15,35 @@
 package handler
 
 import (
+	"fmt"
+
 	tangoerrors "github.com/uber/tango/core/errors"
+	"github.com/uber/tango/internal/mapper"
+	"github.com/uber/tango/observability/metrics"
 	pb "github.com/uber/tango/tangopb"
+	"go.uber.org/zap"
 )
 
 func (h *handler) GetTargetGraph(request *pb.GetTargetGraphRequest, stream pb.TangoServiceGetTargetGraphYARPCServer) (retErr error) {
+	// Temporary: the handler maps the request only to tag the metrics. The
+	// controller maps it again until it takes the entity request.
+	entityReq, mappingErr := mapper.ProtoToGetTargetGraphRequest(request)
+	if mappingErr != nil {
+		mappingErr = tangoerrors.NewUser(fmt.Errorf("convert get target graph request: %w", mappingErr))
+	}
+	_, repo, repositoryErr := h.resolveRequestRepository(entityReq.Build.Remote, mappingErr)
+	e := h.emitter.Tagged(map[string]string{metrics.TagRepo: repo})
+	op := metrics.Begin(e, opGetTargetGraph, metrics.SlowDurationBuckets)
+	logger := h.logger.WithLazy(zap.String("repository", repo))
 	defer func() {
+		op.Complete(retErr)
 		if retErr != nil {
-			h.logger.Error("GetTargetGraph failed", tangoerrors.Fields(retErr)...)
+			logger.Error("GetTargetGraph failed", tangoerrors.Fields(retErr)...)
 			retErr = toWireError(retErr)
 		}
 	}()
+	if repositoryErr != nil {
+		return repositoryErr
+	}
 	return h.controller.GetTargetGraph(request, stream)
 }

handler/gettargetgraph_test.go

@@ -31,9 +31,9 @@ type fakeGetTargetGraphStream struct {
 
 func TestGetTargetGraphForwardsToController(t *testing.T) {
 	ctrl := mock_controller.NewMockController(gomock.NewController(t))
-	h := New(Params{Logger: zaptest.NewLogger(t), Controller: ctrl})
+	h := New(Params{Logger: zaptest.NewLogger(t), RepoConfig: allowAnyRepositoryConfigProvider{}, Controller: ctrl})
 
-	req := &pb.GetTargetGraphRequest{}
+	req := &pb.GetTargetGraphRequest{BuildDescription: &pb.BuildDescription{Strategy: pb.COMPUTATION_STRATEGY_UNSET, Remote: "repo:go-code", BaseSha: "sha"}}
 	stream := &fakeGetTargetGraphStream{}
 	ctrl.EXPECT().GetTargetGraph(gomock.Any(), gomock.Any()).DoAndReturn(
 		func(gotReq *pb.GetTargetGraphRequest, gotStream pb.TangoServiceGetTargetGraphYARPCServer) error {