From 7e5f0739ef4351846683aacc179b1fcd67f1578a Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Wed, 28 Oct 2026 14:23:38 +0000 Subject: [PATCH] feat(stovepipe): Add project-status metrics --- doc/rfc/stovepipe/steps/record.md | 6 +- stovepipe/controller/record/record.go | 16 +- stovepipe/controller/record/record_test.go | 181 ++++++++++++++++++++- 3 files changed, 195 insertions(+), 8 deletions(-) diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index 00df9f18e..e420f1ae0 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/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. diff --git a/stovepipe/controller/record/record.go b/stovepipe/controller/record/record.go index 45624122d..33d9b24d5 100644 --- a/stovepipe/controller/record/record.go +++ b/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,12 +183,15 @@ 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) } @@ -197,10 +199,16 @@ func (c *Controller) resolveProjectResults(ctx context.Context, request entity.R 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, diff --git a/stovepipe/controller/record/record_test.go b/stovepipe/controller/record/record_test.go index 95dc05b93..074946d7b 100644 --- a/stovepipe/controller/record/record_test.go +++ b/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())) + } }) } }