feat(stovepipe): Add project-status metrics

doc/rfc/stovepipe/steps/record.md

@@ -147,7 +147,11 @@ The equal case matters as much as the newer one. A redelivery that finds its own
 
 ## Observability
 
-Two things are measured here, because this is where greenness becomes known. Both need a `SourceControl` lookup of a commit's creation time.
+Project-status work has two operation lifecycles under the record controller scope: `resolve_project_results` covers resolver selection, execution, and result validation; `persist_project_facts` covers the project fact writes, degree indexes, and completion-log materialization. Both use `LongLatencyBuckets` because resolution can call external analysis services and persistence spans an entire batch of storage round trips. Each emits a `.start` counter and `.finish` duration histogram tagged with `result=success|error|cancel`, inheriting metric tags from the consumer context, including the queue. The finish histogram also supplies completion counts; no separate completion counter is emitted.
+
+Successful resolution records the validated result count in `resolve_project_results.projects` using `CountBuckets`, including zero for valid empty results. Empty results do not distinguish disabled resolution, skipped analysis, and a completed analysis with no projects; that distinction belongs to the resolver implementation. These are per-attempt measurements: redelivery repeats resolution and persistence and emits another sample, even when existing immutable facts are adopted. Dependency-specific resolution metrics belong to the resolver implementation rather than this controller.
+
+Two additional signals are measured here, because this is where greenness becomes known. Both need a `SourceControl` lookup of a commit's creation time.
 
 - **Last-green age.** A gauge carrying the creation timestamp of the commit the bookmark now points at, emitted once the bookmark write is durable. It is a timestamp rather than an elapsed time so that subtracting it at scrape time gives the current age, and a queue that stops going green keeps reporting a staler value without the stage re-emitting anything.
 - **Failure-detection latency.** A histogram of how long the break a build failed on went undetected, measured from the creation time of the base URI it validated against. A distribution rather than a gauge, because what matters is how long breaks typically survive, not how long the last one did. There is no later moment to sample it from, since an elapsed time only means something against the failure that just became known, so this lookup cannot move off the delivery path onto a clock.

stovepipe/controller/record/record.go

@@ -69,7 +69,6 @@ type Controller struct {
 // Verify Controller implements consumer.Controller interface at compile time.
 var _ consumer.Controller = (*Controller)(nil)
 
-// _opName is the metric operation name shared by every emit in this file.
 const _opName = "record"
 
 // wholeRepositoryProject is the project component of a fact covering the whole
@@ -184,23 +183,32 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	}
 }
 
-func (c *Controller) resolveProjectResults(ctx context.Context, request entity.Request) ([]projectstatus.Result, error) {
+func (c *Controller) resolveProjectResults(ctx context.Context, request entity.Request) (results []projectstatus.Result, retErr error) {
+	op := metrics.Begin(c.metricsScope, "resolve_project_results", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
 	resolver, err := c.projectStatusResolverFactory.For(projectstatus.ResolverConfig{QueueName: request.Queue})
 	if err != nil {
 		return nil, fmt.Errorf("resolve project status resolver for queue %q: %w", request.Queue, err)
 	}
-	results, err := resolver.Resolve(ctx, request, request.TerminalBuildID)
+	results, err = resolver.Resolve(ctx, request, request.TerminalBuildID)
 	if err != nil {
 		return nil, fmt.Errorf("resolve project status results for request %q: %w", request.ID, err)
 	}
 
 	if err := validateProjectStatusResults(request, results); err != nil {
 		return nil, err
 	}
+	metrics.NamedHistogram(c.metricsScope, "resolve_project_results", "projects", metrics.CountBuckets,
+		metrics.TagsFromContext(ctx)...,
+	).RecordValue(float64(len(results)))
 	return results, nil
 }
 
-func (c *Controller) persistProjectFacts(ctx context.Context, store storage.Storage, request entity.Request, results []projectstatus.Result) error {
+func (c *Controller) persistProjectFacts(ctx context.Context, store storage.Storage, request entity.Request, results []projectstatus.Result) (retErr error) {
+	op := metrics.Begin(c.metricsScope, "persist_project_facts", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
 	for _, result := range results {
 		fact, _, err := recordValidationFact(ctx, c.logger, c.metricsScope, store.GetValidationFactStore(), entity.ValidationFact{
 			URI:       request.URI,

stovepipe/controller/record/record_test.go

@@ -304,6 +304,8 @@ func TestProcess_WaitsForProjectResolutionBeforeRecordingOrPromotion(t *testing.
 	resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return(nil, assert.AnError)
 	require.ErrorIs(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))), assert.AnError)
 	assert.Empty(t, m.hooks.events)
+	assertRecordOperationMetrics(t, m.metricsScope, "resolve_project_results", "error")
+	assert.NotContains(t, m.metricsScope.Snapshot().Counters(), "record_controller.persist_project_facts.start+queue="+testQueue)
 }
 
 func TestProcess_RejectsLegacyBookmark(t *testing.T) {
@@ -386,6 +388,8 @@ func TestProcess_RecordsNamedProjectStatusResults(t *testing.T) {
 	assert.Equal(t, entity.RequestEventValidationFactRecorded, logs[0].Event)
 	assert.Equal(t, entity.RequestEventProjectFactsRecorded, logs[1].Event)
 	assert.Equal(t, "2", logs[1].Metadata[requestlog.MetadataKeyProjectFactCount])
+	assertRecordOperationMetrics(t, m.metricsScope, "resolve_project_results", "success")
+	assertRecordOperationMetrics(t, m.metricsScope, "persist_project_facts", "success")
 }
 
 func TestRecordProjectFacts_PassesEmptyTerminalBuild(t *testing.T) {
@@ -405,6 +409,167 @@ func TestRecordProjectFacts_PassesEmptyTerminalBuild(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestResolveProjectResultsMetrics(t *testing.T) {
+	validResults := []projectstatus.Result{
+		{Project: "project-a", Degree: entity.DegreeGreen},
+		{Project: "project-b", Degree: entity.DegreeBroken},
+	}
+	for _, tt := range []struct {
+		name       string
+		results    []projectstatus.Result
+		factoryErr error
+		resolveErr error
+		invalid    bool
+		wantResult string
+	}{
+		{name: "projects", results: validResults, wantResult: "success"},
+		{name: "nil results", wantResult: "success"},
+		{name: "empty results", results: []projectstatus.Result{}, wantResult: "success"},
+		{name: "factory error", factoryErr: assert.AnError, wantResult: "error"},
+		{name: "factory cancellation", factoryErr: context.Canceled, wantResult: "cancel"},
+		{name: "resolver error with partial results", results: validResults, resolveErr: assert.AnError, wantResult: "error"},
+		{name: "resolver cancellation", resolveErr: context.Canceled, wantResult: "cancel"},
+		{name: "resolver deadline", resolveErr: context.DeadlineExceeded, wantResult: "error"},
+		{name: "invalid project", results: []projectstatus.Result{{Degree: entity.DegreeGreen}}, invalid: true, wantResult: "error"},
+		{name: "invalid degree", results: []projectstatus.Result{{Project: "project-a", Degree: math.NaN()}}, invalid: true, wantResult: "error"},
+		{name: "duplicate projects", results: []projectstatus.Result{validResults[0], validResults[0]}, invalid: true, wantResult: "error"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			c, m := newController(t, ctrl)
+			request := failedRequest()
+			factory := projectstatusmock.NewMockResolverFactory(ctrl)
+			resolver := projectstatusmock.NewMockResolver(ctrl)
+			c.projectStatusResolverFactory = factory
+			factory.EXPECT().For(projectstatus.ResolverConfig{QueueName: testQueue}).Return(resolver, tt.factoryErr)
+			if tt.factoryErr == nil {
+				resolver.EXPECT().Resolve(gomock.Any(), request, request.TerminalBuildID).Return(tt.results, tt.resolveErr)
+			}
+
+			got, err := c.resolveProjectResults(queueContext(), request)
+
+			switch {
+			case tt.factoryErr != nil:
+				require.ErrorIs(t, err, tt.factoryErr)
+			case tt.resolveErr != nil:
+				require.ErrorIs(t, err, tt.resolveErr)
+			case tt.invalid:
+				require.Error(t, err)
+			default:
+				require.NoError(t, err)
+				assert.Equal(t, tt.results, got)
+			}
+			assertRecordOperationMetrics(t, m.metricsScope, "resolve_project_results", tt.wantResult)
+			histograms := m.metricsScope.Snapshot().Histograms()
+			const projectsKey = "record_controller.resolve_project_results.projects+queue=" + testQueue
+			if err != nil {
+				assert.NotContains(t, histograms, projectsKey)
+				return
+			}
+			projects, ok := histograms[projectsKey]
+			require.True(t, ok)
+			assert.EqualValues(t, 1, projects.Values()[float64(len(tt.results))])
+			var samples int64
+			for _, count := range projects.Values() {
+				samples += count
+			}
+			assert.EqualValues(t, 1, samples)
+		})
+	}
+}
+
+func TestPersistProjectFactsMetrics(t *testing.T) {
+	for _, tt := range []struct {
+		name          string
+		empty         bool
+		existing      bool
+		factErr       error
+		indexErr      error
+		completionErr error
+		wantResult    string
+	}{
+		{name: "projects", wantResult: "success"},
+		{name: "empty results", empty: true, wantResult: "success"},
+		{name: "existing facts and indexes", existing: true, wantResult: "success"},
+		{name: "fact write error", factErr: assert.AnError, wantResult: "error"},
+		{name: "fact write cancellation", factErr: context.Canceled, wantResult: "cancel"},
+		{name: "index write error", indexErr: assert.AnError, wantResult: "error"},
+		{name: "index write cancellation", indexErr: context.Canceled, wantResult: "cancel"},
+		{name: "completion error", completionErr: assert.AnError, wantResult: "error"},
+		{name: "completion cancellation", completionErr: context.Canceled, wantResult: "cancel"},
+		{name: "empty completion error", empty: true, completionErr: assert.AnError, wantResult: "error"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			c, m := newController(t, ctrl)
+			request := failedRequest()
+			var results []projectstatus.Result
+			if !tt.empty {
+				results = []projectstatus.Result{
+					{Project: "project-a", Degree: entity.DegreeGreen},
+					{Project: "project-b", Degree: entity.DegreeBroken},
+				}
+			}
+			materializer := requestlogmock.NewMockMaterializer(ctrl)
+			c.materializer = materializer
+			for _, result := range results {
+				fact := entity.ValidationFact{URI: request.URI, Project: result.Project, Degree: result.Degree, RequestID: request.ID, CreatedAt: testChangeTime.UnixMilli()}
+				createErr := tt.factErr
+				if tt.existing {
+					createErr = storage.ErrAlreadyExists
+				}
+				m.factStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(createErr)
+				if tt.factErr != nil {
+					break
+				}
+				indexErr := tt.indexErr
+				if tt.existing {
+					m.factStore.EXPECT().Get(gomock.Any(), request.URI, result.Project).Return(fact, nil)
+					indexErr = storage.ErrAlreadyExists
+				}
+				entry, err := entity.NewProjectFactByDegree(fact)
+				require.NoError(t, err)
+				m.indexStore.EXPECT().Create(gomock.Any(), entry).Return(indexErr)
+				if tt.indexErr != nil {
+					break
+				}
+			}
+			if tt.factErr == nil && tt.indexErr == nil {
+				materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).DoAndReturn(
+					func(_ context.Context, _ storage.Storage, log entity.RequestLog) error {
+						assert.Equal(t, entity.RequestEventProjectFactsRecorded, log.Event)
+						return tt.completionErr
+					})
+			}
+
+			err := c.persistProjectFacts(queueContext(), m.store, request, results)
+
+			switch {
+			case tt.factErr != nil:
+				require.ErrorIs(t, err, tt.factErr)
+			case tt.indexErr != nil:
+				require.ErrorIs(t, err, tt.indexErr)
+			case tt.completionErr != nil:
+				require.ErrorIs(t, err, tt.completionErr)
+			default:
+				require.NoError(t, err)
+			}
+			assertRecordOperationMetrics(t, m.metricsScope, "persist_project_facts", tt.wantResult)
+		})
+	}
+}
+
+func assertRecordOperationMetrics(t *testing.T, scope tally.TestScope, operation, result string) {
+	t.Helper()
+	snapshot := scope.Snapshot()
+	start, ok := snapshot.Counters()["record_controller."+operation+".start+queue="+testQueue]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, start.Value())
+	finish, ok := snapshot.Histograms()["record_controller."+operation+".finish+queue="+testQueue+",result="+result]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, totalSamples(finish.Durations()))
+}
+
 func TestProcessRecordsEmptyProjectCompletion(t *testing.T) {
 	resolveErr := errors.New("project resolution unavailable")
 	writeErr := errors.New("completion write unavailable")
@@ -652,7 +817,7 @@ func TestProcess_FullBuildFailureHasNoBaseline(t *testing.T) {
 	require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))))
 
 	snapshot := m.metricsScope.Snapshot()
-	assert.Empty(t, snapshot.Histograms(), "a build with no baseline has no latency to report")
+	assert.NotContains(t, snapshot.Histograms(), "record_controller.record.failure_detection_latency+queue="+testQueue+",strategy=full")
 	counter, ok := snapshot.Counters()[failureDetectionMissing]
 	require.True(t, ok)
 	assert.EqualValues(t, 1, counter.Value())
@@ -671,7 +836,7 @@ func TestProcess_RedeliveredFailureIsNotResampled(t *testing.T) {
 		Return(entity.ValidationFact{URI: testURI, Degree: entity.DegreeBroken, RequestID: testID, CreatedAt: testChangeTime.UnixMilli()}, nil)
 
 	require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))))
-	assert.Empty(t, m.metricsScope.Snapshot().Histograms())
+	assert.NotContains(t, m.metricsScope.Snapshot().Histograms(), failureDetectionLatency)
 }
 
 // TestProcess_UnobservableDetectionLatencyDoesNotFailRecord covers the observation's
@@ -729,7 +894,7 @@ func TestProcess_UnobservableDetectionLatencyDoesNotFailRecord(t *testing.T) {
 			require.NoError(t, c.Process(queueContext(), delivery(t, ctrl, recordPayload(t, testID))))
 
 			snapshot := m.metricsScope.Snapshot()
-			assert.Empty(t, snapshot.Histograms(), "no latency may be reported when it cannot be observed")
+			assert.NotContains(t, snapshot.Histograms(), failureDetectionLatency)
 			counter, ok := snapshot.Counters()[failureDetectionErrors+tt.step]
 			require.True(t, ok)
 			assert.EqualValues(t, 1, counter.Value())
@@ -932,6 +1097,9 @@ func TestProcess_TerminalWithoutFactDoesNotTouchStores(t *testing.T) {
 				got = append(got, event.GetType())
 			}
 			assert.Equal(t, tt.wantEventTypes, got)
+			for _, operation := range []string{"resolve_project_results", "persist_project_facts"} {
+				assert.NotContains(t, m.metricsScope.Snapshot().Counters(), "record_controller."+operation+".start+queue="+testQueue)
+			}
 		})
 	}
 }
@@ -1218,6 +1386,13 @@ func TestPersistProjectFactsRepairsIndexOnRetry(t *testing.T) {
 			)
 			require.ErrorIs(t, c.persistProjectFacts(queueContext(), m.store, request, []projectstatus.Result{{Project: "p", Degree: 0.25}}), writeErr)
 			require.NoError(t, c.persistProjectFacts(queueContext(), m.store, request, []projectstatus.Result{{Project: "p", Degree: 0.75}}))
+			snapshot := m.metricsScope.Snapshot()
+			assert.EqualValues(t, 2, snapshot.Counters()["record_controller.persist_project_facts.start+queue="+testQueue].Value())
+			for _, result := range []string{"success", "error"} {
+				finish, ok := snapshot.Histograms()["record_controller.persist_project_facts.finish+queue="+testQueue+",result="+result]
+				require.True(t, ok)
+				assert.EqualValues(t, 1, totalSamples(finish.Durations()))
+			}
 		})
 	}
 }