diff --git a/platform/metrics/README.md b/platform/metrics/README.md index f3161ac6..f98f9e77 100644 --- a/platform/metrics/README.md +++ b/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 diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go index e2bf0db9..37d5669a 100644 --- a/platform/metrics/metrics.go +++ b/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)) diff --git a/platform/metrics/metrics_test.go b/platform/metrics/metrics_test.go index 4e00bf3c..69364104 100644 --- a/platform/metrics/metrics_test.go +++ b/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, diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index a162adb0..1c1d3554 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/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", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index a3a1f05d..78babda8 100644 --- a/service/stovepipe/server/main.go +++ b/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", diff --git a/stovepipe/controller/periodicmetrics/BUILD.bazel b/stovepipe/controller/periodicmetrics/BUILD.bazel new file mode 100644 index 00000000..d3f8821e --- /dev/null +++ b/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", + ], +) diff --git a/stovepipe/controller/periodicmetrics/periodicmetrics.go b/stovepipe/controller/periodicmetrics/periodicmetrics.go new file mode 100644 index 00000000..39679554 --- /dev/null +++ b/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 +} diff --git a/stovepipe/controller/periodicmetrics/periodicmetrics_test.go b/stovepipe/controller/periodicmetrics/periodicmetrics_test.go new file mode 100644 index 00000000..61465b12 --- /dev/null +++ b/stovepipe/controller/periodicmetrics/periodicmetrics_test.go @@ -0,0 +1,242 @@ +// 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 + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + 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" + "go.uber.org/zap" +) + +const ( + testQueue = "monorepo/main" + testURI = "git://github.com/uber-code/repo/refs%2Fheads%2Fmain/abc" + + // Metric names as they appear in a snapshot, so a case asserts on the series an + // operator queries rather than on how the emit is composed. + ageSeconds = "stovepipe.periodicmetrics_controller.last_green.age_seconds+queue=monorepo/main" + ageMissing = "stovepipe.periodicmetrics_controller.last_green.age_missing+queue=monorepo/main" + ageErrors = "stovepipe.periodicmetrics_controller.last_green.age_errors+queue=monorepo/main,step=" +) + +// periodicMetricsMocks bundles the mocks a test case wires expectations on. +type periodicMetricsMocks struct { + stores *storagemock.MockFactory + store *storagemock.MockStorage + queueStore *storagemock.MockQueueStore + sourceControls *sourcecontrolmock.MockFactory + sourceControl *sourcecontrolmock.MockSourceControl +} + +// expectQueue wires resolution down to a queue row holding the given bookmark. +func (m periodicMetricsMocks) expectQueue(lastGreenURI string) { + m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil) + m.store.EXPECT().GetQueueStore().Return(m.queueStore) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue). + Return(entity.Queue{Name: testQueue, LastGreenURI: lastGreenURI}, nil) +} + +// expectChangeInfo wires the dating of the bookmarked commit. +func (m periodicMetricsMocks) expectChangeInfo(info sourcecontrol.ChangeInfo, err error) { + m.sourceControls.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) + m.sourceControl.EXPECT().ChangeInfo(gomock.Any(), testURI).Return(info, err) +} + +func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, periodicMetricsMocks, tally.TestScope) { + t.Helper() + + m := periodicMetricsMocks{ + stores: storagemock.NewMockFactory(ctrl), + store: storagemock.NewMockStorage(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + sourceControls: sourcecontrolmock.NewMockFactory(ctrl), + sourceControl: sourcecontrolmock.NewMockSourceControl(ctrl), + } + + scope := tally.NewTestScope("stovepipe", nil) + c := NewController( + zap.NewNop().Sugar(), + scope, + m.stores, + m.sourceControls, + stovepipemq.TopicKeyPeriodicMetrics, + "stovepipe-periodicmetrics", + ) + return c, m, scope +} + +func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery { + t.Helper() + d := consumermock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(entityqueue.NewMessage(testQueue, payload, testQueue, nil)).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + return d +} + +func payload(t *testing.T, queue string) []byte { + t.Helper() + b, err := stovepipemq.Marshal(&stovepipemq.PeriodicMetrics{QueueName: queue}) + require.NoError(t, err) + return b +} + +func TestProcess_EmitsLastGreenAge(t *testing.T) { + ctrl := gomock.NewController(t) + c, m, scope := newController(t, ctrl) + createdAt := time.Now().Add(-time.Hour) + + m.expectQueue(testURI) + m.expectChangeInfo(sourcecontrol.ChangeInfo{CreatedAt: createdAt}, nil) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, payload(t, testQueue)))) + + gauge, ok := scope.Snapshot().Gauges()[ageSeconds] + require.True(t, ok) + assert.InDelta(t, time.Since(createdAt).Seconds(), gauge.Value(), 1) +} + +// TestProcess_RecordsMissingLastGreen covers a queue that has never gone green: it +// has no age, and emitting zero would read as "green as of right now". +func TestProcess_RecordsMissingLastGreen(t *testing.T) { + ctrl := gomock.NewController(t) + c, m, scope := newController(t, ctrl) + + m.expectQueue("") + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, payload(t, testQueue)))) + + snapshot := scope.Snapshot() + assert.Empty(t, snapshot.Gauges(), "a queue that has never gone green has no age to report") + counter, ok := snapshot.Counters()[ageMissing] + require.True(t, ok) + assert.EqualValues(t, 1, counter.Value()) +} + +// TestProcess_AcksWhenAgeCannotBeObserved covers the stage's error posture: every way +// the observation can fail is counted with the step that failed and acked, because the +// schedule supersedes the missed sample and nothing downstream reads this stage. +func TestProcess_AcksWhenAgeCannotBeObserved(t *testing.T) { + tests := []struct { + name string + step string + setup func(m periodicMetricsMocks) + }{ + { + name: "storage cannot be resolved", + step: "resolve_storage", + setup: func(m periodicMetricsMocks) { + m.stores.EXPECT().For(gomock.Any()).Return(nil, errors.New("boom")) + }, + }, + { + name: "the queue cannot be read", + step: "get_queue", + setup: func(m periodicMetricsMocks) { + m.stores.EXPECT().For(gomock.Any()).Return(m.store, nil) + m.store.EXPECT().GetQueueStore().Return(m.queueStore) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue). + Return(entity.Queue{}, errors.New("boom")) + }, + }, + { + name: "source control cannot be resolved", + step: "resolve_source_control", + setup: func(m periodicMetricsMocks) { + m.expectQueue(testURI) + m.sourceControls.EXPECT().For(gomock.Any()).Return(nil, errors.New("boom")) + }, + }, + { + name: "change info fails", + step: "get_change_info", + setup: func(m periodicMetricsMocks) { + m.expectQueue(testURI) + m.expectChangeInfo(sourcecontrol.ChangeInfo{}, errors.New("boom")) + }, + }, + { + name: "the change is undated", + step: "get_change_info", + setup: func(m periodicMetricsMocks) { + m.expectQueue(testURI) + m.expectChangeInfo(sourcecontrol.ChangeInfo{}, nil) + }, + }, + { + name: "the change is dated in the future", + step: "future_change", + setup: func(m periodicMetricsMocks) { + m.expectQueue(testURI) + m.expectChangeInfo(sourcecontrol.ChangeInfo{CreatedAt: time.Now().Add(time.Hour)}, nil) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m, scope := newController(t, ctrl) + tt.setup(m) + + require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, payload(t, testQueue)))) + + snapshot := scope.Snapshot() + assert.Empty(t, snapshot.Gauges(), "no age may be reported when it cannot be observed") + counter, ok := snapshot.Counters()[ageErrors+tt.step] + require.True(t, ok) + assert.EqualValues(t, 1, counter.Value()) + }) + } +} + +func TestProcess_RejectsMalformedMessages(t *testing.T) { + tests := []struct { + name string + payload func(t *testing.T) []byte + }{ + { + name: "not protojson", + payload: func(*testing.T) []byte { return []byte("not-protojson") }, + }, + { + name: "no queue to observe", + payload: func(t *testing.T) []byte { return payload(t, "") }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, _, _ := newController(t, ctrl) + + require.Error(t, c.Process(context.Background(), delivery(t, ctrl, tt.payload(t)))) + }) + } +} diff --git a/stovepipe/core/messagequeue/README.md b/stovepipe/core/messagequeue/README.md index 15889fe1..37dddf43 100644 --- a/stovepipe/core/messagequeue/README.md +++ b/stovepipe/core/messagequeue/README.md @@ -9,5 +9,7 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali - **process** (`TopicKeyProcess`, `ProcessRequest`) — ingest publishes the minted request id here once it accepts a new head; the process controller reloads the `Request` from storage and decides the build strategy. Only the id travels: producer and consumer share the store, so messages stay small and redelivery is idempotent. - **build** (`TopicKeyBuild`, `BuildRequest`) — process/analyze publishes the request id here once its build scope (`BuildStrategy`/`BaseURI`) is decided; the build controller reloads the `Request`, triggers the build, and persists the resulting `Build`. Partitioned by request id. - **buildsignal** (`TopicKeyBuildSignal`, `BuildSignal`) — build publishes the build id here after triggering; buildsignal re-publishes to itself between polls until the build reaches a terminal status. Partitioned by build id, so each build's poll loop is an independent partition. See [doc/rfc/stovepipe/steps/build.md](../../../doc/rfc/stovepipe/steps/build.md) and [buildsignal.md](../../../doc/rfc/stovepipe/steps/buildsignal.md). +- **record** (`TopicKeyRecord`, `Record`) — buildsignal publishes the request id here once, and only once, a build reaches a terminal status; the record controller writes the validation fact and advances the queue's last-green bookmark. Partitioned by request id. +- **periodicmetrics** (`TopicKeyPeriodicMetrics`, `PeriodicMetrics`) — the one topic no stage publishes to: the deployment publishes a queue name here on whatever schedule it wants queue-health observations, and the periodicmetrics controller samples the queue and emits the result as metrics. Partitioned by queue name. Driving it by clock rather than by pipeline activity is the point — the health it reports degrades while the pipeline is idle. See [doc/rfc/messagequeue-contract.md](../../../doc/rfc/messagequeue-contract.md) for the contract conventions and `api/runway/messagequeue` for the external reference example. diff --git a/stovepipe/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index 867eb3aa..1f69a17b 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -53,6 +53,10 @@ type ( // Record is the payload buildsignal publishes to the record stage once a // build reaches a terminal status: the build id to record. Record = protopb.Record + + // PeriodicMetrics is the payload the deployment publishes on a schedule to + // the periodicmetrics stage: the name of the queue to observe. + PeriodicMetrics = protopb.PeriodicMetrics ) // marshalOpts keeps the JSON field names identical to the proto field names diff --git a/stovepipe/core/messagequeue/messagequeue_test.go b/stovepipe/core/messagequeue/messagequeue_test.go index 141dddda..b3ea2aba 100644 --- a/stovepipe/core/messagequeue/messagequeue_test.go +++ b/stovepipe/core/messagequeue/messagequeue_test.go @@ -66,6 +66,17 @@ func TestRecordRoundTrip(t *testing.T) { assert.True(t, proto.Equal(rec, got), "round-tripped Record should equal the original") } +func TestPeriodicMetricsRoundTrip(t *testing.T) { + msg := &PeriodicMetrics{QueueName: "monorepo/main"} + + data, err := Marshal(msg) + require.NoError(t, err) + + got := &PeriodicMetrics{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(msg, got), "round-tripped PeriodicMetrics should equal the original") +} + // TestWireFormat locks the protojson encoding decision the contract relies on: // snake_case field names (UseProtoNames). func TestWireFormat(t *testing.T) { @@ -80,7 +91,7 @@ func TestWireFormat(t *testing.T) { // no topic_keys option names an unknown key. func TestTopicKeysBindEveryTopicKey(t *testing.T) { bound := map[string]int{} - for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &Record{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &Record{}, &PeriodicMetrics{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -93,6 +104,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { TopicKeyBuild, TopicKeyBuildSignal, TopicKeyRecord, + TopicKeyPeriodicMetrics, } valid := map[string]bool{} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index 5f2390a8..572116d6 100644 --- a/stovepipe/core/messagequeue/proto/BUILD.bazel +++ b/stovepipe/core/messagequeue/proto/BUILD.bazel @@ -2,6 +2,7 @@ exports_files( [ "build.proto", "buildsignal.proto", + "periodicmetrics.proto", "process.proto", "record.proto", ], diff --git a/stovepipe/core/messagequeue/proto/periodicmetrics.proto b/stovepipe/core/messagequeue/proto/periodicmetrics.proto new file mode 100644 index 00000000..98e4ac13 --- /dev/null +++ b/stovepipe/core/messagequeue/proto/periodicmetrics.proto @@ -0,0 +1,41 @@ +// 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. + +syntax = "proto3"; + +package uber.stovepipe.messagequeue; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb"; +option java_multiple_files = true; +option java_outer_classname = "PeriodicMetricsProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// PeriodicMetrics asks for one observation of a queue's current health, emitted as +// metrics. Unlike every other payload on Stovepipe's pipeline queues, no stage +// produces it: the deployment publishes it on a schedule. That is the point — the +// health it reports (how stale the queue's last-known-green commit is) degrades +// while the pipeline is idle, so an observation triggered by pipeline activity goes +// quiet in exactly the outage worth alerting on. +// +// The payload carries no request or build id because it observes the queue rather +// than any one unit of work. Redelivery is harmless: a repeated observation +// overwrites the same current-state gauge. +message PeriodicMetrics { + option (uber.base.messagequeue.topic_keys) = "periodicmetrics"; + + // queue_name is the name of the queue to observe (e.g. "monorepo/main"). + string queue_name = 1; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index 7f61fb89..12331b3f 100644 --- a/stovepipe/core/messagequeue/protopb/BUILD.bazel +++ b/stovepipe/core/messagequeue/protopb/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build.pb.go", "buildsignal.pb.go", + "periodicmetrics.pb.go", "process.pb.go", "record.pb.go", ], diff --git a/stovepipe/core/messagequeue/protopb/periodicmetrics.pb.go b/stovepipe/core/messagequeue/protopb/periodicmetrics.pb.go new file mode 100644 index 00000000..46d0837f --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/periodicmetrics.pb.go @@ -0,0 +1,151 @@ +// 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.10 +// protoc v5.29.3 +// source: periodicmetrics.proto + +package protopb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + _ "github.com/uber/submitqueue/api/base/messagequeue/protopb" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// PeriodicMetrics asks for one observation of a queue's current health, emitted as +// metrics. Unlike every other payload on Stovepipe's pipeline queues, no stage +// produces it: the deployment publishes it on a schedule. That is the point — the +// health it reports (how stale the queue's last-known-green commit is) degrades +// while the pipeline is idle, so an observation triggered by pipeline activity goes +// quiet in exactly the outage worth alerting on. +// +// The payload carries no request or build id because it observes the queue rather +// than any one unit of work. Redelivery is harmless: a repeated observation +// overwrites the same current-state gauge. +type PeriodicMetrics struct { + state protoimpl.MessageState `protogen:"open.v1"` + // queue_name is the name of the queue to observe (e.g. "monorepo/main"). + QueueName string `protobuf:"bytes,1,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PeriodicMetrics) Reset() { + *x = PeriodicMetrics{} + mi := &file_periodicmetrics_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PeriodicMetrics) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PeriodicMetrics) ProtoMessage() {} + +func (x *PeriodicMetrics) ProtoReflect() protoreflect.Message { + mi := &file_periodicmetrics_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PeriodicMetrics.ProtoReflect.Descriptor instead. +func (*PeriodicMetrics) Descriptor() ([]byte, []int) { + return file_periodicmetrics_proto_rawDescGZIP(), []int{0} +} + +func (x *PeriodicMetrics) GetQueueName() string { + if x != nil { + return x.QueueName + } + return "" +} + +var File_periodicmetrics_proto protoreflect.FileDescriptor + +const file_periodicmetrics_proto_rawDesc = "" + + "\n" + + "\x15periodicmetrics.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"E\n" + + "\x0fPeriodicMetrics\x12\x1d\n" + + "\n" + + "queue_name\x18\x01 \x01(\tR\tqueueName:\x13\x8a\xb5\x18\x0fperiodicmetricsB\x86\x01\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\x14PeriodicMetricsProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_periodicmetrics_proto_rawDescOnce sync.Once + file_periodicmetrics_proto_rawDescData []byte +) + +func file_periodicmetrics_proto_rawDescGZIP() []byte { + file_periodicmetrics_proto_rawDescOnce.Do(func() { + file_periodicmetrics_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_periodicmetrics_proto_rawDesc), len(file_periodicmetrics_proto_rawDesc))) + }) + return file_periodicmetrics_proto_rawDescData +} + +var file_periodicmetrics_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_periodicmetrics_proto_goTypes = []any{ + (*PeriodicMetrics)(nil), // 0: uber.stovepipe.messagequeue.PeriodicMetrics +} +var file_periodicmetrics_proto_depIdxs = []int32{ + 0, // [0:0] is the sub-list for method output_type + 0, // [0:0] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_periodicmetrics_proto_init() } +func file_periodicmetrics_proto_init() { + if File_periodicmetrics_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_periodicmetrics_proto_rawDesc), len(file_periodicmetrics_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_periodicmetrics_proto_goTypes, + DependencyIndexes: file_periodicmetrics_proto_depIdxs, + MessageInfos: file_periodicmetrics_proto_msgTypes, + }.Build() + File_periodicmetrics_proto = out.File + file_periodicmetrics_proto_goTypes = nil + file_periodicmetrics_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index bac3f71d..0acba8b7 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -48,4 +48,13 @@ const ( // terminal status; non-terminal polls never publish here. Partitioned by // request id. TopicKeyRecord TopicKey = "record" + + // TopicKeyPeriodicMetrics carries requests to observe a queue's current + // health to the periodicmetrics stage. No stage publishes here: the + // deployment publishes a PeriodicMetrics (the queue name) on whatever + // schedule it wants observations, which is what makes the observation + // independent of whether the pipeline is doing anything. Partitioned by + // queue name, so one queue's observations stay serialized and a slow + // observation cannot delay another queue's. + TopicKeyPeriodicMetrics TopicKey = "periodicmetrics" ) diff --git a/stovepipe/extension/sourcecontrol/README.md b/stovepipe/extension/sourcecontrol/README.md index da7a8212..9ad9e65d 100644 --- a/stovepipe/extension/sourcecontrol/README.md +++ b/stovepipe/extension/sourcecontrol/README.md @@ -9,6 +9,7 @@ A `SourceControl` is **bound to a single queue** (a repo+ref) when its `Factory` - **Latest** resolves the queue's ref to the URI of its latest commit — the commit a new validation `Request` is minted against during `ingest`. - **IsAncestor** answers whether one URI is an ancestor of another. The `process` stage uses it to choose a build strategy: if the queue's last-green URI is no longer an ancestor of the latest commit, history was rewritten and a full build is required rather than an incremental one. - **History** returns a bounded, newest-first page of commit URIs on the ref, using the shared generic `page.Page[string]` (`platform/base/page`). It is paginated with an opaque cursor: callers pass an empty cursor for the newest page and the page's `NextCursor` to walk further back, stopping when it is empty. Pagination keeps a remote backend cheap; callers join the URIs against the request store to render the greenness of each commit. +- **ChangeInfo** returns immutable metadata about the commit a URI names — today, when the VCS recorded it. It dates the *commit*, not the ref that points at it or the moment Stovepipe noticed the ref move, which is what makes it usable as an observation baseline: the age of the queue's last-known-green commit is measured from it. ## Errors diff --git a/stovepipe/extension/sourcecontrol/fake/fake.go b/stovepipe/extension/sourcecontrol/fake/fake.go index ae57a8c1..aadb9f1e 100644 --- a/stovepipe/extension/sourcecontrol/fake/fake.go +++ b/stovepipe/extension/sourcecontrol/fake/fake.go @@ -21,27 +21,36 @@ package fake import ( "context" + "time" "github.com/uber/submitqueue/platform/base/page" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" ) +// commitInterval is how much older each successive commit in the seeded history +// is dated, giving the fake a plausible commit cadence. +const commitInterval = time.Minute + // sourceControlFake serves a single queue's linear history. history[0] is the // latest commit; higher indices are progressively older ancestors. type sourceControlFake struct { // cfg is the per-queue identity this source control was built for. cfg sourcecontrol.Config history []string + // latestCreatedAt dates history[0]; older commits are dated back from it. + latestCreatedAt time.Time } // New returns a sourcecontrol.SourceControl bound to the queue named in cfg, // backed by the given ref history, ordered newest-first (history[0] is the // latest commit). The slice is copied so later mutation by the caller does not -// affect the fake. +// affect the fake. The history is dated ending at construction time, so a caller +// measuring how old a commit is sees a plausible age rather than one anchored to +// an arbitrary epoch. func New(cfg sourcecontrol.Config, history []string) sourcecontrol.SourceControl { cp := make([]string, len(history)) copy(cp, history) - return sourceControlFake{cfg: cfg, history: cp} + return sourceControlFake{cfg: cfg, history: cp, latestCreatedAt: time.Now().UTC()} } // Latest returns the newest commit URI, or ErrNotFound when the history is empty. @@ -94,6 +103,20 @@ func (s sourceControlFake) History(_ context.Context, cursor string, limit int) return page.Page[string]{Items: uris, NextCursor: next}, nil } +// ChangeInfo dates a URI by its position in the history: the latest commit is +// dated at construction time and each older one commitInterval further back, so +// the timestamps agree with the ancestry the fake reports. Returns ErrNotFound +// when the URI is not on the ref. +func (s sourceControlFake) ChangeInfo(_ context.Context, uri string) (sourcecontrol.ChangeInfo, error) { + index := s.indexOf(uri) + if index < 0 { + return sourcecontrol.ChangeInfo{}, sourcecontrol.ErrNotFound + } + return sourcecontrol.ChangeInfo{ + CreatedAt: s.latestCreatedAt.Add(-time.Duration(index) * commitInterval), + }, nil +} + // indexOf returns the index of uri in the history, or -1 if absent. func (s sourceControlFake) indexOf(uri string) int { for i, u := range s.history { diff --git a/stovepipe/extension/sourcecontrol/fake/fake_test.go b/stovepipe/extension/sourcecontrol/fake/fake_test.go index 77b3bab3..01d830a9 100644 --- a/stovepipe/extension/sourcecontrol/fake/fake_test.go +++ b/stovepipe/extension/sourcecontrol/fake/fake_test.go @@ -17,6 +17,7 @@ package fake import ( "context" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -140,3 +141,20 @@ func TestHistory(t *testing.T) { }) } } + +func TestChangeInfo(t *testing.T) { + source := New(testCfg, history) + + latest, err := source.ChangeInfo(context.Background(), "git://repo/ref/c") + require.NoError(t, err) + assert.WithinDuration(t, time.Now(), latest.CreatedAt, time.Minute) + + // Timestamps must agree with the ancestry the fake reports: an ancestor is + // dated before its descendant. + oldest, err := source.ChangeInfo(context.Background(), "git://repo/ref/a") + require.NoError(t, err) + assert.Equal(t, 2*commitInterval, latest.CreatedAt.Sub(oldest.CreatedAt)) + + _, err = source.ChangeInfo(context.Background(), "git://repo/ref/x") + require.ErrorIs(t, err, sourcecontrol.ErrNotFound) +} diff --git a/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go b/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go index 92d52526..a7a3c05f 100644 --- a/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go +++ b/stovepipe/extension/sourcecontrol/mock/sourcecontrol_mock.go @@ -42,6 +42,21 @@ func (m *MockSourceControl) EXPECT() *MockSourceControlMockRecorder { return m.recorder } +// ChangeInfo mocks base method. +func (m *MockSourceControl) ChangeInfo(ctx context.Context, uri string) (sourcecontrol.ChangeInfo, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ChangeInfo", ctx, uri) + ret0, _ := ret[0].(sourcecontrol.ChangeInfo) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ChangeInfo indicates an expected call of ChangeInfo. +func (mr *MockSourceControlMockRecorder) ChangeInfo(ctx, uri any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChangeInfo", reflect.TypeOf((*MockSourceControl)(nil).ChangeInfo), ctx, uri) +} + // History mocks base method. func (m *MockSourceControl) History(ctx context.Context, cursor string, limit int) (page.Page[string], error) { m.ctrl.T.Helper() diff --git a/stovepipe/extension/sourcecontrol/sourcecontrol.go b/stovepipe/extension/sourcecontrol/sourcecontrol.go index d61a1013..b7fd0331 100644 --- a/stovepipe/extension/sourcecontrol/sourcecontrol.go +++ b/stovepipe/extension/sourcecontrol/sourcecontrol.go @@ -28,10 +28,19 @@ import ( "context" "errors" "fmt" + "time" "github.com/uber/submitqueue/platform/base/page" ) +// ChangeInfo describes immutable metadata about the source-control change +// represented by a URI. +type ChangeInfo struct { + // CreatedAt is when the source-control provider recorded the immutable + // change. It does not represent ref-update or fetch time. + CreatedAt time.Time +} + // ErrNotFound is returned when a queue, ref, or URI cannot be resolved by the // implementation (for example an unknown queue, or an ancestry query referencing // a URI that is not on the ref). @@ -71,6 +80,10 @@ type SourceControl interface { // the greenness/status of each commit. Returns ErrNotFound if the cursor does // not refer to a position on the ref. History(ctx context.Context, cursor string, limit int) (page.Page[string], error) + + // ChangeInfo returns immutable metadata for uri. The returned CreatedAt must + // be non-zero. Returns ErrNotFound when uri cannot be resolved. + ChangeInfo(ctx context.Context, uri string) (ChangeInfo, error) } // Config carries the per-queue identity handed to a Factory. The system knows diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 5b9ebca1..f5be3266 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -69,6 +69,7 @@ go_proto_generated_files( srcs = [ "//stovepipe/core/messagequeue/proto:build.proto", "//stovepipe/core/messagequeue/proto:buildsignal.proto", + "//stovepipe/core/messagequeue/proto:periodicmetrics.proto", "//stovepipe/core/messagequeue/proto:process.proto", "//stovepipe/core/messagequeue/proto:record.proto", ],