feat(stovepipe): report the last-green change timestamp from record

platform/metrics/README.md

@@ -1,6 +1,6 @@
 # Metrics Utilities (`platform/metrics`)
 
-The `metrics` package provides reusable helpers for emitting counters and histograms on a `tally.Scope`.
+The `metrics` package provides reusable helpers for emitting counters, gauges, and histograms on a `tally.Scope`.
 
 ## Design
 
@@ -48,6 +48,7 @@ For ad-hoc metrics that do not fit the operation lifecycle:
 | Function | Emits | Example |
 |----------|-------|---------|
 | `NamedCounter(scope, name, counter, value, ...tags)` | `{name}.{counter}` counter | `publish.attempts` |
+| `NamedGauge(scope, name, gauge, value, ...tags)` | `{name}.{gauge}` gauge | `record.last_green_timestamp_seconds` |
 | `NamedHistogram(scope, name, histogram, buckets, ...tags)` | `{name}.{histogram}` histogram | `process.duration` |
 
 ```go
@@ -57,7 +58,9 @@ 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.
+Use gauges only for state whose latest value is the whole answer, such as a bookmark timestamp. A gauge is reported once per update rather than continuously, so a gauge set on a discrete event produces a sparse series: it carries no value between updates or after a restart, and each replica reports only the updates it made, so queries must aggregate across replicas with `max` or last-value. State that must be readable at any moment needs a periodic re-emit rather than an event-driven one.
+
+Represent operation latency and completion count with lifecycle histograms; do not emit timers.
 
 ### Why histograms, not timers
 

platform/metrics/metrics.go

@@ -171,6 +171,11 @@ func NamedHistogram(scope tally.Scope, name string, histogram string, buckets ta
 	return tagged(scope, tags).SubScope(name).Histogram(histogram, buckets)
 }
 
+// NamedGauge sets the {name}.{gauge} gauge to value.
+func NamedGauge(scope tally.Scope, name string, gauge string, value float64, tags ...Tag) {
+	tagged(scope, tags).SubScope(name).Gauge(gauge).Update(value)
+}
+
 // tagsToMap converts a slice of Tag to a map for tally.
 func tagsToMap(tags []Tag) map[string]string {
 	m := make(map[string]string, len(tags))

platform/metrics/metrics_test.go

@@ -149,6 +149,15 @@ func TestNamedHistogram(t *testing.T) {
 	assert.True(t, ok, "expected process.duration histogram")
 }
 
+func TestNamedGauge(t *testing.T) {
+	scope := tally.NewTestScope("", nil)
+	NamedGauge(scope, "process", "in_flight", 42, NewTag("queue", "monorepo/main"))
+
+	g, ok := scope.Snapshot().Gauges()["process.in_flight+queue=monorepo/main"]
+	assert.True(t, ok, "expected tagged process.in_flight gauge")
+	assert.Equal(t, float64(42), g.Value())
+}
+
 func TestLatencyBuckets_Sorted(t *testing.T) {
 	sets := map[string]tally.DurationBuckets{
 		"FastLatencyBuckets":    FastLatencyBuckets,

service/stovepipe/server/main.go

@@ -426,7 +426,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, scf, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
 	}

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/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",
@@ -26,6 +27,8 @@ go_test(
         "//platform/consumer/mock:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//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",

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/sourcecontrol"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
@@ -41,11 +42,12 @@ import (
 // advances the queue's last-green bookmark when that fact is green. Implements
 // consumer.Controller.
 type Controller struct {
-	logger        *zap.SugaredLogger
-	metricsScope  tally.Scope
-	stores        storage.Factory
-	topicKey      consumer.TopicKey
-	consumerGroup string
+	logger         *zap.SugaredLogger
+	metricsScope   tally.Scope
+	stores         storage.Factory
+	sourceControls sourcecontrol.Factory
+	topicKey       consumer.TopicKey
+	consumerGroup  string
 }
 
 // Verify Controller implements consumer.Controller interface at compile time.
@@ -64,15 +66,17 @@ func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	stores storage.Factory,
+	sourceControls sourcecontrol.Factory,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
 	return &Controller{
-		logger:        logger.Named("record_controller"),
-		metricsScope:  scope.SubScope("record_controller"),
-		stores:        stores,
-		topicKey:      topicKey,
-		consumerGroup: consumerGroup,
+		logger:         logger.Named("record_controller"),
+		metricsScope:   scope.SubScope("record_controller"),
+		stores:         stores,
+		sourceControls: sourceControls,
+		topicKey:       topicKey,
+		consumerGroup:  consumerGroup,
 	}
 }
 
@@ -256,10 +260,63 @@ func (c *Controller) advanceLastGreen(ctx context.Context, store storage.Storage
 			"request_id", request.ID,
 			"last_green_uri", request.URI,
 		)
+		c.emitLastGreenTimestamp(ctx, request)
 		return nil
 	}
 }
 
+// emitLastGreenTimestamp emits the creation time of the change the bookmark now
+// points at, once that bookmark is durable. Reporting is best-effort so an
+// observability failure cannot turn a successful record operation into a retry,
+// which is why each cause is counted and logged separately instead of returned.
+func (c *Controller) emitLastGreenTimestamp(ctx context.Context, request entity.Request) {
+	queueTag := metrics.NewTag("queue", request.Queue)
+
+	sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: request.Queue})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_resolve_errors", 1, queueTag)
+		c.logger.Warnw("failed to resolve source control to report the last green timestamp",
+			"queue", request.Queue,
+			"error", err,
+		)
+		return
+	}
+
+	info, err := sourceControl.ChangeInfo(ctx, request.URI)
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_errors", 1, queueTag)
+		c.logger.Warnw("failed to look up the last green change timestamp",
+			"queue", request.Queue,
+			"uri", request.URI,
+			"error", err,
+		)
+		return
+	}
+
+	// SourceControl must report a positive creation timestamp, so a missing one
+	// is a broken extension contract rather than a lookup failure. Emitting it
+	// anyway would publish a 1970 timestamp and read as an infinitely stale queue.
+	if info.CreatedAt <= 0 {
+		metrics.NamedCounter(c.metricsScope, _opName, "last_green_timestamp_invalid", 1, queueTag)
+		c.logger.Warnw("source control reported no creation timestamp for the last green change",
+			"queue", request.Queue,
+			"uri", request.URI,
+			"created_at", info.CreatedAt,
+		)
+		return
+	}
+
+	// The gauge carries the creation time as Unix seconds, so subtracting it
+	// from the current time yields the age of the last-green change in seconds.
+	metrics.NamedGauge(
+		c.metricsScope,
+		_opName,
+		"last_green_timestamp_seconds",
+		float64(time.UnixMilli(info.CreatedAt).Unix()),
+		queueTag,
+	)
+}
+
 // isNewerRequest reports whether candidate was ingested after current. An empty
 // current means the bookmark has never been set, so any candidate is newer.
 func isNewerRequest(queue, candidate, current string) (bool, error) {

stovepipe/controller/record/record_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -26,6 +27,8 @@ import (
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
+	"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"
@@ -38,12 +41,16 @@ const (
 	testURI   = "git://remote/monorepo/main/head-sha"
 )
 
+var testChangeTime = time.Unix(1_700_000_000, 0).UTC()
+
 // recordMocks bundles the mocks a record controller test case wires
 // expectations on.
 type recordMocks struct {
-	reqStore   *storagemock.MockRequestStore
-	queueStore *storagemock.MockQueueStore
-	factStore  *storagemock.MockValidationFactStore
+	reqStore      *storagemock.MockRequestStore
+	queueStore    *storagemock.MockQueueStore
+	factStore     *storagemock.MockValidationFactStore
+	sourceControl *sourcecontrolmock.MockSourceControl
+	metricsScope  tally.TestScope
 }
 
 // expectFactCreated wires a successful fact write and captures it, so a case can
@@ -62,13 +69,31 @@ type staticStorageFactory struct{ store storage.Storage }
 // For returns the fixed store aggregate for any queue.
 func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { return f.store, nil }
 
+type staticSourceControlFactory struct {
+	sourceControl sourcecontrol.SourceControl
+}
+
+func (f staticSourceControlFactory) For(sourcecontrol.Config) (sourcecontrol.SourceControl, error) {
+	return f.sourceControl, nil
+}
+
+// failingSourceControlFactory resolves no queue.
+type failingSourceControlFactory struct{}
+
+func (failingSourceControlFactory) For(sourcecontrol.Config) (sourcecontrol.SourceControl, error) {
+	return nil, errors.New("no source control for queue")
+}
+
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMocks) {
 	t.Helper()
 
+	scope := tally.NewTestScope("", nil)
 	m := recordMocks{
-		reqStore:   storagemock.NewMockRequestStore(ctrl),
-		queueStore: storagemock.NewMockQueueStore(ctrl),
-		factStore:  storagemock.NewMockValidationFactStore(ctrl),
+		reqStore:      storagemock.NewMockRequestStore(ctrl),
+		queueStore:    storagemock.NewMockQueueStore(ctrl),
+		factStore:     storagemock.NewMockValidationFactStore(ctrl),
+		sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl),
+		metricsScope:  scope,
 	}
 
 	store := storagemock.NewMockStorage(ctrl)
@@ -78,8 +103,9 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, recordMo
 
 	c := NewController(
 		zap.NewNop().Sugar(),
-		tally.NewTestScope("test", nil),
+		scope,
 		staticStorageFactory{store: store},
+		staticSourceControlFactory{sourceControl: m.sourceControl},
 		stovepipemq.TopicKeyRecord,
 		"stovepipe-record",
 	)
@@ -158,11 +184,17 @@ func TestProcess_AdvancesBookmarkOnSuccess(t *testing.T) {
 					written = q
 					return nil
 				})
+			m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).
+				Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil)
 
 			require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
 			assert.Equal(t, tt.wantURI, written.LastGreenURI)
 			assert.Equal(t, testID, written.LastGreenRequestID)
 
+			gauge, ok := m.metricsScope.Snapshot().Gauges()["record_controller.record.last_green_timestamp_seconds+queue=monorepo/main"]
+			require.True(t, ok)
+			assert.Equal(t, float64(testChangeTime.Unix()), gauge.Value())
+
 			// The green fact is what authorises the advance.
 			assert.Equal(t, entity.DegreeGreen, fact.Degree)
 			assert.Equal(t, testURI, fact.URI)
@@ -173,6 +205,68 @@ func TestProcess_AdvancesBookmarkOnSuccess(t *testing.T) {
 	}
 }
 
+func TestProcess_TimestampReportingFailureDoesNotFailRecord(t *testing.T) {
+	tests := []struct {
+		name        string
+		info        sourcecontrol.ChangeInfo
+		err         error
+		wantCounter string
+	}{
+		{
+			name:        "lookup fails",
+			err:         errors.New("boom"),
+			wantCounter: "record_controller.record.last_green_timestamp_errors+queue=monorepo/main",
+		},
+		{
+			// A zero timestamp breaks the extension contract, so it is counted
+			// apart from a lookup failure rather than emitted as a 1970 gauge.
+			name:        "timestamp missing",
+			info:        sourcecontrol.ChangeInfo{CreatedAt: 0},
+			wantCounter: "record_controller.record.last_green_timestamp_invalid+queue=monorepo/main",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			c, m := newController(t, ctrl)
+
+			m.reqStore.EXPECT().Get(gomock.Any(), testID).
+				Return(requestWithState(entity.RequestStateSucceeded), nil)
+			var fact entity.ValidationFact
+			m.expectFactCreated(&fact)
+			m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil)
+			m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
+			m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).Return(tt.info, tt.err)
+
+			require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+			assert.Empty(t, m.metricsScope.Snapshot().Gauges())
+			counter, ok := m.metricsScope.Snapshot().Counters()[tt.wantCounter]
+			require.True(t, ok)
+			assert.Equal(t, int64(1), counter.Value())
+		})
+	}
+}
+
+func TestProcess_UnresolvableSourceControlDoesNotFailRecord(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+	c.sourceControls = failingSourceControlFactory{}
+
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).
+		Return(requestWithState(entity.RequestStateSucceeded), nil)
+	var fact entity.ValidationFact
+	m.expectFactCreated(&fact)
+	m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil)
+	m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
+
+	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+	assert.Empty(t, m.metricsScope.Snapshot().Gauges())
+	counter, ok := m.metricsScope.Snapshot().Counters()["record_controller.record.last_green_timestamp_resolve_errors+queue=monorepo/main"]
+	require.True(t, ok)
+	assert.Equal(t, int64(1), counter.Value())
+}
+
 func TestProcess_RecordsBrokenFactWithoutAdvancing(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	c, m := newController(t, ctrl)
@@ -221,6 +315,8 @@ func TestProcess_AdoptsExistingFactFromSameRequest(t *testing.T) {
 			if tt.wantUpdate {
 				m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow("", "", 1), nil)
 				m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(nil)
+				m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).
+					Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil)
 			}
 
 			require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
@@ -270,6 +366,11 @@ func TestProcess_SkipsBookmarkWhenNotNewer(t *testing.T) {
 			// No Update: the bookmark only moves forward.
 
 			require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
+			assert.NotContains(
+				t,
+				m.metricsScope.Snapshot().Gauges(),
+				"record_controller.record.last_green_timestamp_seconds+queue=monorepo/main",
+			)
 		})
 	}
 }
@@ -338,6 +439,8 @@ func TestProcess_RetriesBookmarkOnVersionMismatch(t *testing.T) {
 		m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), fresh.Version, fresh.Version+1).
 			Return(nil),
 	)
+	m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).
+		Return(sourcecontrol.ChangeInfo{CreatedAt: testChangeTime.UnixMilli()}, nil)
 
 	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, recordPayload(t, testID))))
 }