|
| 1 | +// Copyright (c) 2025 Uber Technologies, Inc. |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | +// |
| 15 | +// Package periodicmetrics holds the periodic-metrics queue controller. It consumes |
| 16 | +// PeriodicMetrics messages (a queue name) and emits metrics describing that queue's |
| 17 | +// current health, sampled from state the pipeline already persists. |
| 18 | +// |
| 19 | +// It is the one stage no other stage feeds: the deployment publishes to this topic |
| 20 | +// on a schedule. That is deliberate rather than incidental. The health reported here |
| 21 | +// degrades while nothing happens — a queue whose last-known-green commit stops |
| 22 | +// advancing ages silently — so an observation triggered by pipeline activity would go |
| 23 | +// quiet in exactly the outage worth alerting on. Being driven by a clock instead of by |
| 24 | +// work gives the observation a cadence that holds while the pipeline is idle. |
| 25 | +// |
| 26 | +// The stage advances no entity and publishes nothing onward. It reads through the |
| 27 | +// storage and source-control extensions and writes only metrics. |
| 28 | +package periodicmetrics |
| 29 | + |
| 30 | +import ( |
| 31 | + "context" |
| 32 | + "fmt" |
| 33 | + "time" |
| 34 | + |
| 35 | + "github.com/uber-go/tally" |
| 36 | + "github.com/uber/submitqueue/platform/consumer" |
| 37 | + "github.com/uber/submitqueue/platform/metrics" |
| 38 | + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" |
| 39 | + "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" |
| 40 | + "github.com/uber/submitqueue/stovepipe/extension/storage" |
| 41 | + "go.uber.org/zap" |
| 42 | +) |
| 43 | + |
| 44 | +// Controller consumes PeriodicMetrics messages and observes the named queue. |
| 45 | +// Implements consumer.Controller. |
| 46 | +type Controller struct { |
| 47 | + logger *zap.SugaredLogger |
| 48 | + metricsScope tally.Scope |
| 49 | + stores storage.Factory |
| 50 | + sourceControls sourcecontrol.Factory |
| 51 | + topicKey consumer.TopicKey |
| 52 | + consumerGroup string |
| 53 | +} |
| 54 | + |
| 55 | +// Verify Controller implements consumer.Controller interface at compile time. |
| 56 | +var _ consumer.Controller = (*Controller)(nil) |
| 57 | + |
| 58 | +const ( |
| 59 | + // _opName is the metric operation name for this stage's own handling counters. |
| 60 | + _opName = "periodicmetrics" |
| 61 | + |
| 62 | + // _opLastGreen is the metric operation name for the last-known-green |
| 63 | + // observation. It is named for what is measured rather than for this stage, so |
| 64 | + // the series an operator alerts on does not move if the stage does. |
| 65 | + _opLastGreen = "last_green" |
| 66 | +) |
| 67 | + |
| 68 | +// NewController creates a new periodic metrics controller. |
| 69 | +func NewController( |
| 70 | + logger *zap.SugaredLogger, |
| 71 | + scope tally.Scope, |
| 72 | + stores storage.Factory, |
| 73 | + sourceControls sourcecontrol.Factory, |
| 74 | + topicKey consumer.TopicKey, |
| 75 | + consumerGroup string, |
| 76 | +) *Controller { |
| 77 | + return &Controller{ |
| 78 | + logger: logger.Named("periodicmetrics_controller"), |
| 79 | + metricsScope: scope.SubScope("periodicmetrics_controller"), |
| 80 | + stores: stores, |
| 81 | + sourceControls: sourceControls, |
| 82 | + topicKey: topicKey, |
| 83 | + consumerGroup: consumerGroup, |
| 84 | + } |
| 85 | +} |
| 86 | + |
| 87 | +// Process observes the queue named in the delivery. Returns nil to ack (success) or |
| 88 | +// an error to nack (retry) / reject (DLQ). |
| 89 | +// |
| 90 | +// Only a message that violates the payload contract is rejected; a failed observation |
| 91 | +// acks. Nothing downstream depends on this stage, so an error would buy nothing but |
| 92 | +// retries of a sample whose moment has passed — and since the schedule keeps producing |
| 93 | +// messages, a persistently failing observation would fill the dead-letter queue at the |
| 94 | +// publishing rate. Every reason an observation cannot be made is counted with the step |
| 95 | +// that failed instead, which is where a reporting fault belongs. |
| 96 | +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { |
| 97 | + msg := delivery.Message() |
| 98 | + |
| 99 | + req := &stovepipemq.PeriodicMetrics{} |
| 100 | + if err := stovepipemq.Unmarshal(msg.Payload, req); err != nil { |
| 101 | + metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1) |
| 102 | + // Non-retryable: a malformed message will never succeed regardless of retries. |
| 103 | + return fmt.Errorf("failed to deserialize periodic metrics request: %w", err) |
| 104 | + } |
| 105 | + |
| 106 | + queue := req.GetQueueName() |
| 107 | + if queue == "" { |
| 108 | + metrics.NamedCounter(c.metricsScope, _opName, "missing_queue", 1) |
| 109 | + // Non-retryable: the queue to observe is the whole payload. |
| 110 | + return fmt.Errorf("periodic metrics request has no queue name") |
| 111 | + } |
| 112 | + |
| 113 | + c.reportLastGreenAge(ctx, queue) |
| 114 | + |
| 115 | + metrics.NamedCounter(c.metricsScope, _opName, "observed", 1, metrics.NewTag("queue", queue)) |
| 116 | + return nil |
| 117 | +} |
| 118 | + |
| 119 | +// reportLastGreenAge updates the gauge holding the current age of the queue's |
| 120 | +// last-known-green commit. A gauge rather than a histogram because the answer is the |
| 121 | +// latest observation, not a distribution: how stale the bookmark is *now*. |
| 122 | +// |
| 123 | +// Callers gate deployments on that commit, so its age is the staleness of the newest |
| 124 | +// thing they are allowed to ship: a queue whose green bookmark stopped advancing looks |
| 125 | +// healthy from the pipeline's perspective — nothing is failing — while the answer it |
| 126 | +// serves silently ages. |
| 127 | +func (c *Controller) reportLastGreenAge(ctx context.Context, queue string) { |
| 128 | + queueTag := metrics.NewTag("queue", queue) |
| 129 | + |
| 130 | + store, err := c.stores.For(storage.Config{QueueName: queue}) |
| 131 | + if err != nil { |
| 132 | + c.ageError(queueTag, "resolve_storage", queue, err) |
| 133 | + return |
| 134 | + } |
| 135 | + |
| 136 | + queueRow, err := store.GetQueueStore().Get(ctx, queue) |
| 137 | + if err != nil { |
| 138 | + c.ageError(queueTag, "get_queue", queue, err) |
| 139 | + return |
| 140 | + } |
| 141 | + |
| 142 | + // A queue that has never gone green has no age to report. Emitting zero |
| 143 | + // would read as "green as of right now", the opposite of the truth. |
| 144 | + if queueRow.LastGreenURI == "" { |
| 145 | + metrics.NamedCounter(c.metricsScope, _opLastGreen, "age_missing", 1, queueTag) |
| 146 | + return |
| 147 | + } |
| 148 | + |
| 149 | + sourceControl, err := c.sourceControls.For(sourcecontrol.Config{QueueName: queue}) |
| 150 | + if err != nil { |
| 151 | + c.ageError(queueTag, "resolve_source_control", queue, err) |
| 152 | + return |
| 153 | + } |
| 154 | + |
| 155 | + info, err := sourceControl.ChangeInfo(ctx, queueRow.LastGreenURI) |
| 156 | + if err != nil || info.CreatedAt.IsZero() { |
| 157 | + c.ageError(queueTag, "get_change_info", queue, err) |
| 158 | + return |
| 159 | + } |
| 160 | + |
| 161 | + // A commit dated in the future means the provider's clock disagrees with |
| 162 | + // ours; a negative age would corrupt the series rather than describe it. |
| 163 | + age := time.Since(info.CreatedAt) |
| 164 | + if age < 0 { |
| 165 | + c.ageError(queueTag, "future_change", queue, nil) |
| 166 | + return |
| 167 | + } |
| 168 | + |
| 169 | + metrics.NamedGauge(c.metricsScope, _opLastGreen, "age_seconds", age.Seconds(), queueTag) |
| 170 | +} |
| 171 | + |
| 172 | +// ageError counts an observation that could not be made, tagged with the step that |
| 173 | +// failed so a silent gauge can be told apart from a broken dependency. |
| 174 | +func (c *Controller) ageError(queueTag metrics.Tag, step, queue string, err error) { |
| 175 | + metrics.NamedCounter(c.metricsScope, _opLastGreen, "age_errors", 1, queueTag, metrics.NewTag("step", step)) |
| 176 | + c.logger.Errorw("failed to observe last green age", "queue", queue, "step", step, "error", err) |
| 177 | +} |
| 178 | + |
| 179 | +// Name returns the controller name for logging and metrics. |
| 180 | +func (c *Controller) Name() string { |
| 181 | + return "periodicmetrics" |
| 182 | +} |
| 183 | + |
| 184 | +// TopicKey returns the topic key this controller subscribes to. |
| 185 | +func (c *Controller) TopicKey() consumer.TopicKey { |
| 186 | + return c.topicKey |
| 187 | +} |
| 188 | + |
| 189 | +// ConsumerGroup returns the consumer group for offset tracking. |
| 190 | +func (c *Controller) ConsumerGroup() string { |
| 191 | + return c.consumerGroup |
| 192 | +} |
0 commit comments