feat(stovepipe): report last-green freshness

platform/metrics/README.md

@@ -49,6 +49,7 @@ For ad-hoc metrics that do not fit the operation lifecycle:
 |----------|-------|---------|
 | `NamedCounter(scope, name, counter, value, ...tags)` | `{name}.{counter}` counter | `publish.attempts` |
 | `NamedHistogram(scope, name, histogram, buckets, ...tags)` | `{name}.{histogram}` histogram | `process.duration` |
+| `NamedGauge(scope, name, gauge, value, ...tags)` | `{name}.{gauge}` gauge | `last_green.age_seconds` |
 
 ```go
 metrics.NamedCounter(c.scope, "publish", "attempts", 1)
@@ -57,7 +58,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

@@ -171,6 +171,13 @@ 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. Reserved for current-state
+// values whose latest observation is the answer; use a histogram for anything
+// whose distribution over time is the point.
+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,17 @@ func TestNamedHistogram(t *testing.T) {
 	assert.True(t, ok, "expected process.duration histogram")
 }
 
+func TestNamedGauge(t *testing.T) {
+	scope := tally.NewTestScope("", nil)
+	NamedGauge(scope, "last_green", "age_seconds", 42, NewTag("queue", "monorepo/main"))
+
+	snapshot := scope.Snapshot()
+	gauges := snapshot.Gauges()
+	g, ok := gauges["last_green.age_seconds+queue=monorepo/main"]
+	assert.True(t, ok, "expected tagged last_green.age_seconds gauge")
+	assert.Equal(t, float64(42), g.Value())
+}
+
 func TestLatencyBuckets_Sorted(t *testing.T) {
 	sets := map[string]tally.DurationBuckets{
 		"FastLatencyBuckets":    FastLatencyBuckets,

service/stovepipe/server/BUILD.bazel

@@ -21,6 +21,7 @@ go_library(
         "//stovepipe/controller/build:go_default_library",
         "//stovepipe/controller/buildsignal:go_default_library",
         "//stovepipe/controller/dlq:go_default_library",
+        "//stovepipe/controller/periodicmetrics:go_default_library",
         "//stovepipe/controller/process:go_default_library",
         "//stovepipe/controller/record:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",

service/stovepipe/server/main.go

@@ -43,6 +43,7 @@ import (
 	"github.com/uber/submitqueue/stovepipe/controller/build"
 	"github.com/uber/submitqueue/stovepipe/controller/buildsignal"
 	"github.com/uber/submitqueue/stovepipe/controller/dlq"
+	"github.com/uber/submitqueue/stovepipe/controller/periodicmetrics"
 	"github.com/uber/submitqueue/stovepipe/controller/process"
 	"github.com/uber/submitqueue/stovepipe/controller/record"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
@@ -432,6 +433,19 @@ func registerPrimaryControllers(
 	}
 	count++
 
+	periodicMetricsController := periodicmetrics.NewController(
+		logger,
+		scope,
+		store,
+		scf,
+		stovepipemq.TopicKeyPeriodicMetrics,
+		"stovepipe-periodicmetrics",
+	)
+	if err := c.Register(periodicMetricsController); err != nil {
+		return count, fmt.Errorf("failed to register periodic metrics controller: %w", err)
+	}
+	count++
+
 	return count, nil
 }
 
@@ -467,6 +481,9 @@ func registerDLQControllers(
 // topic and the buildsignal consumer subscribes to it, and also republishes to itself while
 // polling. buildsignal publishes to the record topic once a build reaches a terminal status,
 // and the record consumer subscribes to it.
+//
+// The periodicmetrics topic is the exception: no stage publishes to it. The deployment does,
+// on whatever schedule it wants queue-health observations.
 func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
 	return consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{
@@ -501,6 +518,14 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 				subscriberName, "stovepipe-record",
 			),
 		},
+		{
+			Key:   stovepipemq.TopicKeyPeriodicMetrics,
+			Name:  "periodicmetrics",
+			Queue: q,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "stovepipe-periodicmetrics",
+			),
+		},
 		{
 			Key:          dlq.TopicKey(stovepipemq.TopicKeyProcess),
 			Name:         "process_dlq",

stovepipe/controller/periodicmetrics/BUILD.bazel

@@ -0,0 +1,38 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["periodicmetrics.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/controller/periodicmetrics",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
+        "//stovepipe/core/messagequeue: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",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["periodicmetrics_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//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",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+        "@org_uber_go_zap//:go_default_library",
+    ],
+)

stovepipe/controller/periodicmetrics/periodicmetrics.go

@@ -0,0 +1,192 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Package periodicmetrics holds the periodic-metrics queue controller. It consumes
+// PeriodicMetrics messages (a queue name) and emits metrics describing that queue's
+// current health, sampled from state the pipeline already persists.
+//
+// It is the one stage no other stage feeds: the deployment publishes to this topic
+// on a schedule. That is deliberate rather than incidental. The health reported here
+// degrades while nothing happens — a queue whose last-known-green commit stops
+// advancing ages silently — so an observation triggered by pipeline activity would go
+// quiet in exactly the outage worth alerting on. Being driven by a clock instead of by
+// work gives the observation a cadence that holds while the pipeline is idle.
+//
+// The stage advances no entity and publishes nothing onward. It reads through the
+// storage and source-control extensions and writes only metrics.
+package periodicmetrics
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	"go.uber.org/zap"
+)
+
+// Controller consumes PeriodicMetrics messages and observes the named queue.
+// Implements consumer.Controller.
+type Controller struct {
+	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.
+var _ consumer.Controller = (*Controller)(nil)
+
+const (
+	// _opName is the metric operation name for this stage's own handling counters.
+	_opName = "periodicmetrics"
+
+	// _opLastGreen is the metric operation name for the last-known-green
+	// observation. It is named for what is measured rather than for this stage, so
+	// the series an operator alerts on does not move if the stage does.
+	_opLastGreen = "last_green"
+)
+
+// NewController creates a new periodic metrics controller.
+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("periodicmetrics_controller"),
+		metricsScope:   scope.SubScope("periodicmetrics_controller"),
+		stores:         stores,
+		sourceControls: sourceControls,
+		topicKey:       topicKey,
+		consumerGroup:  consumerGroup,
+	}
+}
+
+// Process observes the queue named in the delivery. Returns nil to ack (success) or
+// an error to nack (retry) / reject (DLQ).
+//
+// Only a message that violates the payload contract is rejected; a failed observation
+// acks. Nothing downstream depends on this stage, so an error would buy nothing but
+// retries of a sample whose moment has passed — and since the schedule keeps producing
+// messages, a persistently failing observation would fill the dead-letter queue at the
+// publishing rate. Every reason an observation cannot be made is counted with the step
+// that failed instead, which is where a reporting fault belongs.
+func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
+	msg := delivery.Message()
+
+	req := &stovepipemq.PeriodicMetrics{}
+	if err := stovepipemq.Unmarshal(msg.Payload, req); err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1)
+		// Non-retryable: a malformed message will never succeed regardless of retries.
+		return fmt.Errorf("failed to deserialize periodic metrics request: %w", err)
+	}
+
+	queue := req.GetQueueName()
+	if queue == "" {
+		metrics.NamedCounter(c.metricsScope, _opName, "missing_queue", 1)
+		// Non-retryable: the queue to observe is the whole payload.
+		return fmt.Errorf("periodic metrics request has no queue name")
+	}
+
+	c.reportLastGreenAge(ctx, queue)
+
+	metrics.NamedCounter(c.metricsScope, _opName, "observed", 1, metrics.NewTag("queue", queue))
+	return nil
+}
+
+// reportLastGreenAge updates the gauge holding the current age of the queue's
+// last-known-green commit. A gauge rather than a histogram because the answer is the
+// latest observation, not a distribution: how stale the bookmark is *now*.
+//
+// Callers gate deployments on that commit, so its age is the staleness of the newest
+// thing they are allowed to ship: a queue whose green bookmark stopped advancing looks
+// healthy from the pipeline's perspective — nothing is failing — while the answer it
+// serves silently ages.
+func (c *Controller) reportLastGreenAge(ctx context.Context, queue string) {
+	queueTag := metrics.NewTag("queue", queue)
+
+	store, err := c.stores.For(storage.Config{QueueName: queue})
+	if err != nil {
+		c.ageError(queueTag, "resolve_storage", queue, err)
+		return
+	}
+
+	queueRow, err := store.GetQueueStore().Get(ctx, queue)
+	if err != nil {
+		c.ageError(queueTag, "get_queue", queue, err)
+		return
+	}
+
+	// A queue that has never gone green has no age to report. Emitting zero
+	// would read as "green as of right now", the opposite of the truth.
+	if queueRow.LastGreenURI == "" {
+		metrics.NamedCounter(c.metricsScope, _opLastGreen, "age_missing", 1, queueTag)
+		return
+	}
+
+	sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: queue})
+	if err != nil {
+		c.ageError(queueTag, "resolve_source_control", queue, err)
+		return
+	}
+
+	info, err := sourceControl.ChangeInfo(ctx, queueRow.LastGreenURI)
+	if err != nil || info.CreatedAt.IsZero() {
+		c.ageError(queueTag, "get_change_info", queue, err)
+		return
+	}
+
+	// A commit dated in the future means the provider's clock disagrees with
+	// ours; a negative age would corrupt the series rather than describe it.
+	age := time.Since(info.CreatedAt)
+	if age < 0 {
+		c.ageError(queueTag, "future_change", queue, nil)
+		return
+	}
+
+	metrics.NamedGauge(c.metricsScope, _opLastGreen, "age_seconds", age.Seconds(), queueTag)
+}
+
+// ageError counts an observation that could not be made, tagged with the step that
+// failed so a silent gauge can be told apart from a broken dependency.
+func (c *Controller) ageError(queueTag metrics.Tag, step, queue string, err error) {
+	metrics.NamedCounter(c.metricsScope, _opLastGreen, "age_errors", 1, queueTag, metrics.NewTag("step", step))
+	c.logger.Errorw("failed to observe last green age", "queue", queue, "step", step, "error", err)
+}
+
+// Name returns the controller name for logging and metrics.
+func (c *Controller) Name() string {
+	return "periodicmetrics"
+}
+
+// TopicKey returns the topic key this controller subscribes to.
+func (c *Controller) TopicKey() consumer.TopicKey {
+	return c.topicKey
+}
+
+// ConsumerGroup returns the consumer group for offset tracking.
+func (c *Controller) ConsumerGroup() string {
+	return c.consumerGroup
+}