feat(controller): emit result-tagged cache-lookup counters

controller/getchangedtargets.go

@@ -137,7 +137,7 @@ func (c *controller) GetChangedTargets(request *pb.GetChangedTargetsRequest, str
 // rather than silent degradation.
 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())
+	treehash1, treehash2, err := readTreehashParallel(ctx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), e, opGetChangedTargets)
 	if err != nil {
 		return false, fmt.Errorf("read revision treehash: %w", err)
 	}
@@ -152,6 +152,7 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metric
 		return false, nil
 	}
 	if cachedReader == nil {
+		recordCacheLookup(e, opGetChangedTargets, _metricComparedTargetsCacheLookup, cacheErr)
 		return false, nil
 	}
 
@@ -180,7 +181,7 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metric
 	_ = cachedReader.Close()
 
 	if readErr != nil {
-		// Blob is corrupt (likely an incomplete write). Log and fall through to recompute.
+		// Blob is corrupt (likely an incomplete write); recompute.
 		logger.Warn("GetChangedTargets: Cached result is incomplete, recomputing", zap.Error(readErr))
 		return false, nil
 	}
@@ -189,7 +190,7 @@ func (c *controller) serveChangedTargetsFromCache(ctx context.Context, e *metric
 	logger.Info("GetChangedTargets: Cache hit, streaming from storage",
 		zap.Duration("cache_read_duration", cacheReadDuration),
 	)
-	e.Counter(opGetChangedTargets, "cache_hit").Inc(1)
+	recordCacheLookup(e, opGetChangedTargets, _metricComparedTargetsCacheLookup, nil)
 	e.DurationHistogram(opGetChangedTargets, "cache_read_duration", metrics.FastDurationBuckets).RecordDuration(cacheReadDuration)
 	if sendErr := sendTrimmedChangedTargets(stream, cached, maxDist, request.GetOutputConfig()); sendErr != nil {
 		return false, fmt.Errorf("send cached response: %w", sendErr)
@@ -343,7 +344,10 @@ func (c *controller) cacheComparedTargets(logger *zap.Logger, request *pb.GetCha
 		// is cancelled on shutdown. Per-operation deadlines are the storage
 		// backend's responsibility — the controller is backend-agnostic and
 		// must not encode any one implementation's I/O budget.
-		treehash1, treehash2, err := readTreehashParallel(c.appCtx, c.storage, request.GetFirstRevision(), request.GetSecondRevision())
+		// The treehash reads here are for building the write key, not a cache
+		// serve attempt, so they pass a no-op emitter to avoid skewing the
+		// treehash cache hit rate.
+		treehash1, treehash2, err := readTreehashParallel(c.appCtx, c.storage, request.GetFirstRevision(), request.GetSecondRevision(), metrics.Nop(), opGetChangedTargets)
 		if err != nil {
 			// Goroutine outlives the handler so we can't return; log loudly and
 			// abandon the cache write. Surfacing infra failures matters more than
@@ -731,7 +735,7 @@ func validateGetChangedTargetsRequest(request *pb.GetChangedTargetsRequest) erro
 // wasting work on a result that will be discarded anyway. The cancelled sibling's error
 // is dropped — only the original failure is returned, so a self-inflicted
 // context.Canceled never masks the real reason the lookup failed.
-func readTreehashParallel(ctx context.Context, st storage.Storage, first, second *pb.BuildDescription) (string, string, error) {
+func readTreehashParallel(ctx context.Context, st storage.Storage, first, second *pb.BuildDescription, e *metrics.Emitter, op string) (string, string, error) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -744,7 +748,7 @@ func readTreehashParallel(ctx context.Context, st storage.Storage, first, second
 	results := make(chan result, len(descs))
 	for i, desc := range descs {
 		go func(idx int, d *pb.BuildDescription) {
-			hash, err := readTreehash(ctx, st, d)
+			hash, err := readTreehash(ctx, st, d, e, op)
 			results <- result{idx: idx, hash: hash, err: err}
 		}(i, desc)
 	}
@@ -771,13 +775,14 @@ func readTreehashParallel(ctx context.Context, st storage.Storage, first, second
 // Returns ("", nil) on a cache miss (not-found is the normal "not yet computed" state).
 // Returns ("", err) on any other storage or read failure so callers can decide whether to
 // surface the error or fall back. Returns (treehash, nil) on a successful read.
-func readTreehash(ctx context.Context, st storage.Storage, buildDescription *pb.BuildDescription) (string, error) {
+func readTreehash(ctx context.Context, st storage.Storage, buildDescription *pb.BuildDescription, e *metrics.Emitter, op string) (string, error) {
 	entityBuild, err := mapper.ProtoToBuildDescription(buildDescription)
 	if err != nil {
 		return "", err
 	}
 	key := cachekey.GetTreehashCachePath(entityBuild)
 	resp, err := st.Get(ctx, storage.DownloadRequest{Key: key})
+	recordCacheLookup(e, op, _metricTreehashCacheLookup, err)
 	if err != nil {
 		if storage.IsNotFound(err) {
 			return "", nil

controller/getchangedtargets_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/uber/tango/core/storage"
 	storagemock "github.com/uber/tango/core/storage/storagemock"
 	"github.com/uber/tango/entity"
+	"github.com/uber/tango/observability/metrics"
 	orchestratormock "github.com/uber/tango/orchestrator/orchestratormock"
 	pb "github.com/uber/tango/tangopb"
 	tangomock "github.com/uber/tango/tangopb/tangopbmock"
@@ -236,7 +237,7 @@ func TestReadTreehash(t *testing.T) {
 		st.EXPECT().Get(gomock.Any(), gomock.Any()).
 			Return(storage.DownloadResponse{}, storage.NewNotFoundError("missing"))
 
-		val, err := readTreehash(t.Context(), st, bd)
+		val, err := readTreehash(t.Context(), st, bd, metrics.Nop(), opGetChangedTargets)
 		require.NoError(t, err)
 		assert.Empty(t, val)
 	})
@@ -248,7 +249,7 @@ func TestReadTreehash(t *testing.T) {
 		st.EXPECT().Get(gomock.Any(), gomock.Any()).
 			Return(storage.DownloadResponse{}, injected)
 
-		val, err := readTreehash(t.Context(), st, bd)
+		val, err := readTreehash(t.Context(), st, bd, metrics.Nop(), opGetChangedTargets)
 		require.Error(t, err)
 		assert.ErrorIs(t, err, injected)
 		assert.Empty(t, val)
@@ -260,7 +261,7 @@ func TestReadTreehash(t *testing.T) {
 		st.EXPECT().Get(gomock.Any(), gomock.Any()).
 			Return(storage.DownloadResponse{ReadCloser: io.NopCloser(strings.NewReader("deadbeef"))}, nil)
 
-		val, err := readTreehash(t.Context(), st, bd)
+		val, err := readTreehash(t.Context(), st, bd, metrics.Nop(), opGetChangedTargets)
 		require.NoError(t, err)
 		assert.Equal(t, "deadbeef", val)
 	})

controller/gettargetgraph.go

@@ -103,6 +103,7 @@ func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entit
 		// Look up the the git treehash based on cache path
 		treehashCachePath := cachekey.GetTreehashCachePath(req.Build)
 		treehashResponse, err := c.storage.Get(ctx, storage.DownloadRequest{Key: treehashCachePath})
+		recordCacheLookup(e, opGetTargetGraph, _metricTreehashCacheLookup, err)
 		if err != nil {
 			if storage.IsNotFound(err) {
 				// Cache miss - blob doesn't exist, need to compute and store target graph
@@ -122,10 +123,11 @@ func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entit
 			// Download the target graph based on treehash.
 			storageStart := time.Now()
 			graphReader, err := storage.NewGraphReader(ctx, c.storage, treehashPath)
+			if ctx.Err() != nil {
+				err = ctx.Err()
+			}
+			recordCacheLookup(e, opGetTargetGraph, _metricGraphCacheLookup, err)
 			if err != nil {
-				if ctx.Err() != nil {
-					err = ctx.Err()
-				}
 				if !storage.IsNotFound(err) {
 					return nil, fmt.Errorf("graph reader: %w", err)
 				}
@@ -135,7 +137,6 @@ func (c *controller) getGraph(ctx context.Context, e *metrics.Emitter, req entit
 					zap.Duration("storage_duration", time.Since(storageStart)),
 					zap.Duration("total_duration", time.Since(start)),
 				)
-				e.Counter(opGetTargetGraph, "cache_hit").Inc(1)
 				e.DurationHistogram(opGetTargetGraph, "download_graph", metrics.SlowDurationBuckets).RecordDuration(time.Since(storageStart))
 				return graphReader, nil
 			}

controller/metrics.go

@@ -14,10 +14,41 @@
 
 package controller
 
+import (
+	"github.com/uber/tango/core/storage"
+	"github.com/uber/tango/observability/metrics"
+)
+
 // 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"
 )
+
+// Cache-lookup metric names. Each is emitted under its parent RPC op with a
+// result=hit|miss tag, so a hit rate is derivable per cache layer.
+const (
+	_metricTreehashCacheLookup        = "treehash_cache_lookup"
+	_metricGraphCacheLookup           = "graph_cache_lookup"
+	_metricComparedTargetsCacheLookup = "compared_targets_cache_lookup"
+)
+
+// recordCacheLookup emits a result-tagged counter for a cache lookup under the
+// given parent op: a nil error is a hit and a not-found error is a miss. Any
+// other error is an infra failure (already tracked by the failure metric), so
+// nothing is emitted — an infra error is not a cache miss and must not skew the
+// hit rate.
+func recordCacheLookup(e *metrics.Emitter, parentOp, name string, err error) {
+	var result string
+	switch {
+	case err == nil:
+		result = metrics.ResultHit
+	case storage.IsNotFound(err):
+		result = metrics.ResultMiss
+	default:
+		return
+	}
+	e.Tagged(map[string]string{metrics.TagResult: result}).Counter(parentOp, name).Inc(1)
+}