test(stovepipe): cover freshness metrics

Verify failed-build detection age emission and last-known-green reporting
against the source-control and storage boundaries.

Author: mnoah1

platform/metrics/README.md

@@ -57,7 +57,7 @@ h := metrics.NamedHistogram(c.scope, "process", "duration", metrics.FastLatencyB
 h.RecordDuration(elapsed)
 ```
 
-Do not emit gauges or timers. Represent operation latency and completion count with lifecycle histograms, and represent instantaneous quantities as sampled histogram values when needed.
+Do not emit timers. Represent operation latency and completion count with lifecycle histograms. Use a gauge only for a periodically refreshed, current-state value whose latest observation is the query result; use a histogram for distributions of observations over time.
 
 ### Why histograms, not timers
 

platform/metrics/metrics.go

@@ -106,6 +106,25 @@ var (
 		2 * time.Hour,
 		4 * time.Hour,
 	}
+
+	// ChangeAgeBuckets suits age-based signals for source-control changes,
+	// including time to failure detection and last-known-green freshness.
+	ChangeAgeBuckets = tally.DurationBuckets{
+		1 * time.Minute,
+		5 * time.Minute,
+		15 * time.Minute,
+		30 * time.Minute,
+		1 * time.Hour,
+		2 * time.Hour,
+		4 * time.Hour,
+		8 * time.Hour,
+		12 * time.Hour,
+		24 * time.Hour,
+		48 * time.Hour,
+		7 * 24 * time.Hour,
+		14 * 24 * time.Hour,
+		30 * 24 * time.Hour,
+	}
 )
 
 // Op tracks the lifecycle of a named operation. It captures the start time on

stovepipe/controller/buildsignal/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
+        "//stovepipe/extension/sourcecontrol:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
@@ -33,6 +34,8 @@ go_test(
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/buildrunner:go_default_library",
         "//stovepipe/extension/buildrunner/mock:go_default_library",
+        "//stovepipe/extension/sourcecontrol:go_default_library",
+        "//stovepipe/extension/sourcecontrol/mock:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "//stovepipe/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",

stovepipe/controller/buildsignal/buildsignal.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/uber-go/tally"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -33,6 +34,7 @@ import (
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
+	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
@@ -61,6 +63,7 @@ type Controller struct {
 	metricsScope  tally.Scope
 	stores        storage.Factory
 	buildRunners  buildrunner.Factory
+	sourceControl sourcecontrol.Factory
 	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
@@ -72,6 +75,16 @@ var _ consumer.Controller = (*Controller)(nil)
 // _opName is the metric operation name shared by every emit in this file.
 const _opName = "buildsignal"
 
+// Option configures a Controller.
+type Option func(*Controller)
+
+// WithSourceControl enables base-change-age metrics for failed builds.
+func WithSourceControl(factory sourcecontrol.Factory) Option {
+	return func(c *Controller) {
+		c.sourceControl = factory
+	}
+}
+
 // NewController creates a new buildsignal controller.
 func NewController(
 	logger *zap.SugaredLogger,
@@ -81,8 +94,9 @@ func NewController(
 	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
+	options ...Option,
 ) *Controller {
-	return &Controller{
+	controller := &Controller{
 		logger:        logger.Named("buildsignal_controller"),
 		metricsScope:  scope.SubScope("buildsignal_controller"),
 		stores:        stores,
@@ -91,6 +105,10 @@ func NewController(
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
+	for _, option := range options {
+		option(controller)
+	}
+	return controller
 }
 
 // Process reloads the build referenced by the delivery, polls its runner for
@@ -270,10 +288,52 @@ func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, req
 		metrics.NamedCounter(c.metricsScope, _opName, "outcomes", 1,
 			metrics.NewTag("state", string(state)),
 		)
+		if state == entity.RequestStateFailed {
+			c.emitBaseChangeAge(ctx, request)
+		}
 		return nil
 	}
 }
 
+func (c *Controller) emitBaseChangeAge(ctx context.Context, request *entity.Request) {
+	if c.sourceControl == nil || request.BaseURI == "" {
+		metrics.NamedCounter(c.metricsScope, "build_failure", "base_change_unavailable", 1,
+			metrics.NewTag("queue", request.Queue),
+			metrics.NewTag("strategy", string(request.BuildStrategy)),
+		)
+		return
+	}
+
+	control, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, "build_failure", "change_info_errors", 1,
+			metrics.NewTag("queue", request.Queue),
+			metrics.NewTag("stage", "resolve_source_control"),
+		)
+		return
+	}
+	info, err := control.ChangeInfo(ctx, request.BaseURI)
+	if err != nil || info.CreatedAt.IsZero() {
+		metrics.NamedCounter(c.metricsScope, "build_failure", "change_info_errors", 1,
+			metrics.NewTag("queue", request.Queue),
+			metrics.NewTag("stage", "get_change_info"),
+		)
+		return
+	}
+	age := time.Since(info.CreatedAt)
+	if age < 0 {
+		metrics.NamedCounter(c.metricsScope, "build_failure", "change_info_errors", 1,
+			metrics.NewTag("queue", request.Queue),
+			metrics.NewTag("stage", "future_change"),
+		)
+		return
+	}
+	metrics.NamedHistogram(c.metricsScope, "build_failure", "time_to_detection", metrics.ChangeAgeBuckets,
+		metrics.NewTag("queue", request.Queue),
+		metrics.NewTag("strategy", string(request.BuildStrategy)),
+	).RecordDuration(age)
+}
+
 // releaseBuildSlot CAS-decrements the queue's in_flight_count, reopening the process
 // concurrency gate now that this request's build is over. It decrements relatively
 // (preserving concurrent updates), clamps at zero, and retries on version conflicts.

stovepipe/controller/buildsignal/buildsignal_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -31,6 +32,8 @@ import (
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
 	buildrunnermock "github.com/uber/submitqueue/stovepipe/extension/buildrunner/mock"
+	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
+	sourcecontrolmock "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/mock"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
 	"go.uber.org/mock/gomock"
@@ -143,6 +146,38 @@ func expectFinish(m buildsignalMocks, state entity.RequestState) {
 	m.reqStore.EXPECT().Update(gomock.Any(), requestWithState(state), int32(1), int32(2)).Return(nil)
 }
 
+func TestEmitBaseChangeAge(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	scope := tally.NewTestScope("test", nil)
+	sourceControls := sourcecontrolmock.NewMockFactory(ctrl)
+	source := sourcecontrolmock.NewMockSourceControl(ctrl)
+	baseURI := "git://github.com/uber-code/repo/refs%2Fheads%2Fmain/abc"
+
+	sourceControls.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(source, nil)
+	source.EXPECT().ChangeInfo(gomock.Any(), baseURI).Return(sourcecontrol.ChangeInfo{
+		CreatedAt: time.Now().Add(-time.Hour),
+	}, nil)
+
+	controller := NewController(
+		zap.NewNop().Sugar(),
+		scope,
+		nil,
+		nil,
+		consumer.TopicRegistry{},
+		stovepipemq.TopicKeyBuildSignal,
+		"stovepipe-buildsignal",
+		WithSourceControl(sourceControls),
+	)
+	controller.emitBaseChangeAge(context.Background(), &entity.Request{
+		Queue:         testQueue,
+		BaseURI:       baseURI,
+		BuildStrategy: entity.BuildStrategyIncrementalSinceGreen,
+	})
+
+	_, ok := scope.Snapshot().Histograms()["test.buildsignal_controller.build_failure.time_to_detection+queue=monorepo/main,strategy=incremental_since_green"]
+	assert.True(t, ok)
+}
+
 func TestProcess(t *testing.T) {
 	tests := []struct {
 		name    string

stovepipe/controller/record/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/observability:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",

stovepipe/controller/record/record.go

@@ -33,6 +33,7 @@ import (
 	"github.com/uber/submitqueue/stovepipe/core/loader"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/observability"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
@@ -44,6 +45,7 @@ type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	stores        storage.Factory
+	reporter      observability.Reporter
 	topicKey      consumer.TopicKey
 	consumerGroup string
 }
@@ -59,21 +61,36 @@ const _opName = "record"
 // attribution that this stage does not do, so every fact it writes is whole-repository.
 const wholeRepositoryProject = ""
 
+// Option configures a Controller.
+type Option func(*Controller)
+
+// WithReporter configures best-effort queue observability reporting.
+func WithReporter(reporter observability.Reporter) Option {
+	return func(c *Controller) {
+		c.reporter = reporter
+	}
+}
+
 // NewController creates a new record controller.
 func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	stores storage.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
+	options ...Option,
 ) *Controller {
-	return &Controller{
+	controller := &Controller{
 		logger:        logger.Named("record_controller"),
 		metricsScope:  scope.SubScope("record_controller"),
 		stores:        stores,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 	}
+	for _, option := range options {
+		option(controller)
+	}
+	return controller
 }
 
 // Process loads the request referenced by the delivery and, when its build
@@ -112,6 +129,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
 		return fmt.Errorf("payload queue %q does not match queue %q of request %s", rec.GetQueueName(), request.Queue, request.ID)
 	}
+	if c.reporter != nil {
+		defer c.reporter.Report(ctx, request.Queue)
+	}
 
 	switch request.State {
 	case entity.RequestStateSucceeded, entity.RequestStateFailed:

stovepipe/extension/observability/BUILD.bazel

@@ -0,0 +1,8 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["observability.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/extension/observability",
+    visibility = ["//visibility:public"],
+)

stovepipe/extension/observability/lastgreen/BUILD.bazel

@@ -0,0 +1,31 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["lastgreen.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/extension/observability/lastgreen",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//stovepipe/extension/observability:go_default_library",
+        "//stovepipe/extension/sourcecontrol:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["lastgreen_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/sourcecontrol:go_default_library",
+        "//stovepipe/extension/sourcecontrol/mock:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)