From 5f16a7ef612c1c68a70e9b50d3a793b7d43cd144 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 04:08:33 +0900 Subject: [PATCH 01/10] feat(sqs/monitoring): per-queue depth gauges + Grafana dashboard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds elastickv_sqs_queue_messages{queue,state} (visible / not_visible / delayed) on top of the existing partition-counter metric (PR 7a), plus the Grafana dashboard the operator drops into the existing provisioning chain. Pattern mirrors the Raft / Redis observer split: - adapter/sqs_depth_source.go: *SQSServer.SnapshotQueueDepths(ctx) returns one SQSQueueDepth per known queue, or an empty slice on followers. Leader-only emission keeps gauge state consistent with what AdminListQueues would return at the same instant — followers scanning concurrently would race the leader's writes and emit conflicting numbers for the same series. - monitoring/sqs.go (extended): SQSMetrics gains a queueDepth GaugeVec + ObserveQueueDepth / ForgetQueue helpers. New SQSObserver type owns the tick loop and the current-vs-previous queue diff; queues that disappeared since the last tick get ForgetQueue'd so the dashboard doesn't show a frozen backlog after DeleteQueue. Same nil-tolerant contract as RaftObserver. - monitoring/registry.go: r.SQSObserver() accessor mirrors r.RaftObserver(). - main.go: startSQSDepthObserver wires the SQS adapter into the observer after startSQSServer; thin sqsDepthSourceAdapter type bridges adapter.SQSQueueDepth to monitoring.SQSQueueDepth so the adapter package doesn't have to import monitoring. Dashboard (monitoring/grafana/dashboards/elastickv-sqs.json): - Top stat row: queue count, total backlog, HT-FIFO ops rate, active leader. - Per-queue depth timeseries (visible / in-flight / delayed) with $queue templating. - Top 20 queues by backlog table. - HT-FIFO partition activity: ops rate by action and by (queue, partition) — the partition split is what the original PR 7a counter was added for; the dashboard surfaces it. Tests: - TestSQSMetrics_ObserveQueueDepth_EmitsThreeStates pins the gauge label triple (visible / not_visible / delayed). - TestSQSMetrics_ObserveQueueDepth_ClampsNegativeToZero pins the defensive clamp against -1 sentinels from a future failed scan. - TestSQSMetrics_ObserveQueueDepth_DropsEmptyQueue pins the empty-name drop (mirrors the partition-counter rule). - TestSQSMetrics_ForgetQueue_DropsThreeSeries pins that ForgetQueue clears every state series for the queue while leaving the cumulative counter untouched. - TestSQSMetrics_DepthNilReceiverIsSafe + DepthRegistryWiring pin the nil-tolerant contract and registry plumbing. - TestSQSObserver_ObserveOnce_EmitsAndForgets pins the diff state machine: queues present in the current tick get gauges, queues that disappeared get ForgetQueue'd. - TestSQSObserver_ObserveOnce_LeaderStepDownClearsAll pins the step-down branch (source returns empty slice -> all gauges cleared). - TestSQSObserver_NilTolerant pins nil source / nil observer no-op behaviour. Self-review (5 lenses): 1. Data loss — N/A; metrics-only, no storage / Raft / FSM touch. 2. Concurrency — observer.mu protects lastSeen; SQSMetrics.mu protects trackedQueues. Source returns a fresh slice per call (no shared mutable state). Ticker goroutine stops on ctx.Done. 3. Performance — one catalog scan + one approx-counter scan per queue per 30s on the leader. scanApproxCounters already self-caps the per-queue scan so a 1M-message queue doesn't block. Per-queue cardinality cap (sqsMaxTrackedQueues = 512) bounds Prometheus series budget. 4. Data consistency — leader-only emission keeps gauges in lock- step with AdminListQueues / AdminDescribeQueue. Per-queue scan error -> ForgetQueue (not zero) so a transient blip surfaces as missing data, not as a false "drained" reading. 5. Test coverage — 8 new tests cover gauge math, observer state machine, leader step-down, nil-tolerance, and registry wiring. --- adapter/sqs_depth_source.go | 81 +++ go.mod | 28 +- go.sum | 56 +- main.go | 57 +++ .../grafana/dashboards/elastickv-sqs.json | 477 ++++++++++++++++++ monitoring/registry.go | 13 + monitoring/sqs.go | 177 ++++++- monitoring/sqs_test.go | 214 ++++++++ 8 files changed, 1058 insertions(+), 45 deletions(-) create mode 100644 adapter/sqs_depth_source.go create mode 100644 monitoring/grafana/dashboards/elastickv-sqs.json diff --git a/adapter/sqs_depth_source.go b/adapter/sqs_depth_source.go new file mode 100644 index 000000000..2d8580135 --- /dev/null +++ b/adapter/sqs_depth_source.go @@ -0,0 +1,81 @@ +package adapter + +import ( + "context" + "log/slog" +) + +// SQSQueueDepth is one queue's depth-attribute snapshot, the unit +// the SQSServer hands to monitoring.SQSObserver on each tick. The +// fields mirror sqsApproxCounters byte-for-byte and the public +// AdminQueueCounters JSON shape — operators see consistent numbers +// in dashboards and the admin SPA. +type SQSQueueDepth struct { + Queue string + Visible int64 + NotVisible int64 + Delayed int64 +} + +// SnapshotQueueDepths satisfies monitoring.SQSDepthSource. The +// observer Start loop calls this on every tick; the SQSServer +// returns one entry per known queue when this node is the verified +// Raft leader, or an empty slice on followers. Leader-only +// emission keeps the dashboard's queue-depth gauges consistent +// with what AdminListQueues / AdminDescribeQueue would return at +// the same instant — followers that scanned the catalog at the +// same time would race the leader's writes and emit conflicting +// values for the same series. +// +// Per-queue scan errors are logged and the offending queue is +// dropped from this tick's snapshot. The observer detects the +// disappearance and ForgetQueue's the gauges so the dashboard +// surfaces "scrape failed" as a missing series rather than as a +// pinned stale backlog. +func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth { + if s == nil || s.coordinator == nil || s.store == nil || !s.coordinator.IsLeader() { + return nil + } + names, err := s.scanQueueNames(ctx) + if err != nil { + slog.Warn("sqs depth snapshot: scanQueueNames failed", "err", err) + return nil + } + out := make([]SQSQueueDepth, 0, len(names)) + for _, name := range names { + if err := ctx.Err(); err != nil { + return out + } + if snap, ok := s.snapshotOneQueueDepth(ctx, name); ok { + out = append(out, snap) + } + } + return out +} + +// snapshotOneQueueDepth runs the per-queue catalog read pair +// (loadQueueMetaAt + scanApproxCounters) and returns the resulting +// snapshot. Pulled out of the loop body so SnapshotQueueDepths +// stays under the cyclop budget; ok=false means "skip this queue +// from this tick" (queue gone, transient catalog read failure). +// Per-queue scan errors are logged and the offending queue is +// dropped from this tick's snapshot rather than aborting the +// entire pass. +func (s *SQSServer) snapshotOneQueueDepth(ctx context.Context, name string) (SQSQueueDepth, bool) { + readTS := s.nextTxnReadTS(ctx) + meta, exists, err := s.loadQueueMetaAt(ctx, name, readTS) + if err != nil || !exists { + return SQSQueueDepth{}, false + } + counters, err := s.scanApproxCounters(ctx, name, meta.Generation, readTS) + if err != nil { + slog.Warn("sqs depth snapshot: counters failed", "queue", name, "err", err) + return SQSQueueDepth{}, false + } + return SQSQueueDepth{ + Queue: name, + Visible: counters.Visible, + NotVisible: counters.NotVisible, + Delayed: counters.Delayed, + }, true +} diff --git a/go.mod b/go.mod index 4948c8ee2..68a4d1bfe 100644 --- a/go.mod +++ b/go.mod @@ -6,11 +6,11 @@ toolchain go1.26.2 require ( github.com/Jille/grpc-multi-resolver v1.3.0 - github.com/aws/aws-sdk-go-v2 v1.41.6 - github.com/aws/aws-sdk-go-v2/config v1.32.16 - github.com/aws/aws-sdk-go-v2/credentials v1.19.15 + github.com/aws/aws-sdk-go-v2 v1.41.7 + github.com/aws/aws-sdk-go-v2/config v1.32.17 + github.com/aws/aws-sdk-go-v2/credentials v1.19.16 github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.2 - github.com/aws/smithy-go v1.25.0 + github.com/aws/smithy-go v1.25.1 github.com/cockroachdb/errors v1.12.0 github.com/cockroachdb/pebble/v2 v2.1.4 github.com/emirpasic/gods v1.18.1 @@ -40,17 +40,17 @@ require ( github.com/DataDog/zstd v1.5.7 // indirect github.com/RaduBerinde/axisds v0.1.0 // indirect github.com/RaduBerinde/btreemap v0.0.0-20250419174037-3d62b7205d54 // indirect - github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.22 // indirect - github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.22 // indirect - github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.22 // indirect - github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.23 // indirect - github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.8 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.23 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 // indirect github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.22 // indirect - github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.22 // indirect - github.com/aws/aws-sdk-go-v2/service/signin v1.0.10 // indirect - github.com/aws/aws-sdk-go-v2/service/sso v1.30.16 // indirect - github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.20 // indirect - github.com/aws/aws-sdk-go-v2/service/sts v1.42.0 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.0.11 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.30.17 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.21 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.42.1 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cockroachdb/crlib v0.0.0-20241112164430-1264a2edc35b // indirect diff --git a/go.sum b/go.sum index 3726c7ed1..48a4e3b5a 100644 --- a/go.sum +++ b/go.sum @@ -8,38 +8,38 @@ github.com/RaduBerinde/btreemap v0.0.0-20250419174037-3d62b7205d54 h1:bsU8Tzxr/P github.com/RaduBerinde/btreemap v0.0.0-20250419174037-3d62b7205d54/go.mod h1:0tr7FllbE9gJkHq7CVeeDDFAFKQVy5RnCSSNBOvdqbc= github.com/aclements/go-perfevent v0.0.0-20240301234650-f7843625020f h1:JjxwchlOepwsUWcQwD2mLUAGE9aCp0/ehy6yCHFBOvo= github.com/aclements/go-perfevent v0.0.0-20240301234650-f7843625020f/go.mod h1:tMDTce/yLLN/SK8gMOxQfnyeMeCg8KGzp0D1cbECEeo= -github.com/aws/aws-sdk-go-v2 v1.41.6 h1:1AX0AthnBQzMx1vbmir3Y4WsnJgiydmnJjiLu+LvXOg= -github.com/aws/aws-sdk-go-v2 v1.41.6/go.mod h1:dy0UzBIfwSeot4grGvY1AqFWN5zgziMmWGzysDnHFcQ= -github.com/aws/aws-sdk-go-v2/config v1.32.16 h1:Q0iQ7quUgJP0F/SCRTieScnaMdXr9h/2+wze1u3cNeM= -github.com/aws/aws-sdk-go-v2/config v1.32.16/go.mod h1:duCCnJEFqpt2RC6no1iK6q+8HpwOAkiUua0pY507dQc= -github.com/aws/aws-sdk-go-v2/credentials v1.19.15 h1:fyvgWTszojq8hEnMi8PPBTvZdTtEVmAVyo+NFLHBhH4= -github.com/aws/aws-sdk-go-v2/credentials v1.19.15/go.mod h1:gJiYyMOjNg8OEdRWOf3CrFQxM2a98qmrtjx1zuiQfB8= -github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.22 h1:IOGsJ1xVWhsi+ZO7/NW8OuZZBtMJLZbk4P5HDjJO0jQ= -github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.22/go.mod h1:b+hYdbU+jGKfXE8kKM6g1+h+L/Go3vMvzlxBsiuGsxg= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.22 h1:GmLa5Kw1ESqtFpXsx5MmC84QWa/ZrLZvlJGa2y+4kcQ= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.22/go.mod h1:6sW9iWm9DK9YRpRGga/qzrzNLgKpT2cIxb7Vo2eNOp0= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.22 h1:dY4kWZiSaXIzxnKlj17nHnBcXXBfac6UlsAx2qL6XrU= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.22/go.mod h1:KIpEUx0JuRZLO7U6cbV204cWAEco2iC3l061IxlwLtI= -github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.23 h1:FPXsW9+gMuIeKmz7j6ENWcWtBGTe1kH8r9thNt5Uxx4= -github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.23/go.mod h1:7J8iGMdRKk6lw2C+cMIphgAnT8uTwBwNOsGkyOCm80U= +github.com/aws/aws-sdk-go-v2 v1.41.7 h1:DWpAJt66FmnnaRIOT/8ASTucrvuDPZASqhhLey6tLY8= +github.com/aws/aws-sdk-go-v2 v1.41.7/go.mod h1:4LAfZOPHNVNQEckOACQx60Y8pSRjIkNZQz1w92xpMJc= +github.com/aws/aws-sdk-go-v2/config v1.32.17 h1:FpL4/758/diKwqbytU0prpuiu60fgXKUWCpDJtApclU= +github.com/aws/aws-sdk-go-v2/config v1.32.17/go.mod h1:OXqUMzgXytfoF9JaKkhrOYsyh72t9G+MJH8mMRaexOE= +github.com/aws/aws-sdk-go-v2/credentials v1.19.16 h1:r3RJBuU7X9ibt8RHbMjWE6y60QbKBiII6wSrXnapxSU= +github.com/aws/aws-sdk-go-v2/credentials v1.19.16/go.mod h1:6cx7zqDENJDbBIIWX6P8s0h6hqHC8Avbjh9Dseo27ug= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.23 h1:UuSfcORqNSz/ey3VPRS8TcVH2Ikf0/sC+Hdj400QI6U= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.23/go.mod h1:+G/OSGiOFnSOkYloKj/9M35s74LgVAdJBSD5lsFfqKg= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 h1:GpT/TrnBYuE5gan2cZbTtvP+JlHsutdmlV2YfEyNde0= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23/go.mod h1:xYWD6BS9ywC5bS3sz9Xh04whO/hzK2plt2Zkyrp4JuA= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 h1:bpd8vxhlQi2r1hiueOw02f/duEPTMK59Q4QMAoTTtTo= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23/go.mod h1:15DfR2nw+CRHIk0tqNyifu3G1YdAOy68RftkhMDDwYk= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 h1:OQqn11BtaYv1WLUowvcA30MpzIu8Ti4pcLPIIyoKZrA= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24/go.mod h1:X5ZJyfwVrWA96GzPmUCWFQaEARPR7gCrpq2E92PJwAE= github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.2 h1:J2ibOhlMLx1o6QwDFsHHfbQjaZ6t5LXodiLNuK6jbZA= github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.2/go.mod h1:Tj8VcffnduuewrM8HN8xQ9wzzez0CJ0FGSGEovq7Sgs= -github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.8 h1:HtOTYcbVcGABLOVuPYaIihj6IlkqubBwFj10K5fxRek= -github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.8/go.mod h1:VsK9abqQeGlzPgUr+isNWzPlK2vKe9INMLWnY65f5Xs= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 h1:FLudkZLt5ci0ozzgkVo8BJGwvqNaZbTWb3UcucAateA= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9/go.mod h1:w7wZ/s9qK7c8g4al+UyoF1Sp/Z45UwMGcqIzLWVQHWk= github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.22 h1:8IXbJCgOn8ztzvRUOm27iCeTSxmPW45JsSDW3EGi16M= github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.22/go.mod h1:l53RbOWvncp4DEmlEz6dSXJS913AIxtFqkJZ+Xz7pHs= -github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.22 h1:PUmZeJU6Y1Lbvt9WFuJ0ugUK2xn6hIWUBBbKuOWF30s= -github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.22/go.mod h1:nO6egFBoAaoXze24a2C0NjQCvdpk8OueRoYimvEB9jo= -github.com/aws/aws-sdk-go-v2/service/signin v1.0.10 h1:a1Fq/KXn75wSzoJaPQTgZO0wHGqE9mjFnylnqEPTchA= -github.com/aws/aws-sdk-go-v2/service/signin v1.0.10/go.mod h1:p6+MXNxW7IA6dMgHfTAzljuwSKD0NCm/4lbS4t6+7vI= -github.com/aws/aws-sdk-go-v2/service/sso v1.30.16 h1:x6bKbmDhsgSZwv6q19wY/u3rLk/3FGjJWyqKcIRufpE= -github.com/aws/aws-sdk-go-v2/service/sso v1.30.16/go.mod h1:CudnEVKRtLn0+3uMV0yEXZ+YZOKnAtUJ5DmDhilVnIw= -github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.20 h1:oK/njaL8GtyEihkWMD4k3VgHCT64RQKkZwh0DG5j8ak= -github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.20/go.mod h1:JHs8/y1f3zY7U5WcuzoJ/yAYGYtNIVPKLIbp61euvmg= -github.com/aws/aws-sdk-go-v2/service/sts v1.42.0 h1:ks8KBcZPh3PYISr5dAiXCM5/Thcuxk8l+PG4+A0exds= -github.com/aws/aws-sdk-go-v2/service/sts v1.42.0/go.mod h1:pFw33T0WLvXU3rw1WBkpMlkgIn54eCB5FYLhjDc9Foo= -github.com/aws/smithy-go v1.25.0 h1:Sz/XJ64rwuiKtB6j98nDIPyYrV1nVNJ4YU74gttcl5U= -github.com/aws/smithy-go v1.25.0/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 h1:pbrxO/kuIwgEsOPLkaHu0O+m4fNgLU8B3vxQ+72jTPw= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23/go.mod h1:/CMNUqoj46HpS3MNRDEDIwcgEnrtZlKRaHNaHxIFpNA= +github.com/aws/aws-sdk-go-v2/service/signin v1.0.11 h1:TdJ+HdzOBhU8+iVAOGUTU63VXopcumCOF1paFulHWZc= +github.com/aws/aws-sdk-go-v2/service/signin v1.0.11/go.mod h1:R82ZRExE/nheo0N+T8zHPcLRTcH8MGsnR3BiVGX0TwI= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.17 h1:7byT8HUWrgoRp6sXjxtZwgOKfhss5fW6SkLBtqzgRoE= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.17/go.mod h1:xNWknVi4Ezm1vg1QsB/5EWpAJURq22uqd38U8qKvOJc= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.21 h1:+1Kl1zx6bWi4X7cKi3VYh29h8BvsCoHQEQ6ST9X8w7w= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.21/go.mod h1:4vIRDq+CJB2xFAXZ+YgGUTiEft7oAQlhIs71xcSeuVg= +github.com/aws/aws-sdk-go-v2/service/sts v1.42.1 h1:F/M5Y9I3nwr2IEpshZgh1GeHpOItExNM9L1euNuh/fk= +github.com/aws/aws-sdk-go-v2/service/sts v1.42.1/go.mod h1:mTNxImtovCOEEuD65mKW7DCsL+2gjEH+RPEAexAzAio= +github.com/aws/smithy-go v1.25.1 h1:J8ERsGSU7d+aCmdQur5Txg6bVoYelvQJgtZehD12GkI= +github.com/aws/smithy-go v1.25.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= diff --git a/main.go b/main.go index f94f60896..201ba1755 100644 --- a/main.go +++ b/main.go @@ -1126,6 +1126,56 @@ func startMonitoringCollectors(ctx context.Context, reg *monitoring.Registry, ru } } +// startSQSDepthObserver wires the SQS adapter (when enabled on this +// node) into the monitoring registry's SQSObserver so the +// elastickv_sqs_queue_messages gauges start updating. Mirrors the +// Raft / Redis pattern: the source is plugged in once after startup, +// then the observer owns the ticker. nil sqsServer (e.g. +// --sqsAddress empty on this node) is a no-op. +// +// The thin adapter exists because monitoring.SQSQueueDepth and +// adapter.SQSQueueDepth are intentionally distinct types — having +// the adapter import monitoring would invert the dependency +// direction (every adapter would then know about Prometheus). +// Conversion is a fixed 4-field copy and a shape mismatch surfaces +// at compile time here, not at runtime on the metrics path. +func startSQSDepthObserver(ctx context.Context, reg *monitoring.Registry, sqsServer *adapter.SQSServer) { + if reg == nil || sqsServer == nil { + return + } + if observer := reg.SQSObserver(); observer != nil { + observer.Start(ctx, sqsDepthSourceAdapter{inner: sqsServer}, 0) + } +} + +// sqsDepthSourceAdapter bridges *adapter.SQSServer (which returns +// []adapter.SQSQueueDepth) to monitoring.SQSDepthSource (which +// expects []monitoring.SQSQueueDepth). Same shape both sides; the +// loop is a fixed-size copy. +type sqsDepthSourceAdapter struct { + inner *adapter.SQSServer +} + +func (a sqsDepthSourceAdapter) SnapshotQueueDepths(ctx context.Context) []monitoring.SQSQueueDepth { + if a.inner == nil { + return nil + } + snaps := a.inner.SnapshotQueueDepths(ctx) + if len(snaps) == 0 { + return nil + } + out := make([]monitoring.SQSQueueDepth, len(snaps)) + for i, s := range snaps { + out[i] = monitoring.SQSQueueDepth{ + Queue: s.Queue, + Visible: s.Visible, + NotVisible: s.NotVisible, + Delayed: s.Delayed, + } + } + return out +} + // writeConflictMonitorSources extracts the MVCC stores that expose // per-(kind, key_prefix) OCC conflict counters so monitoring can poll // them for the elastickv_store_write_conflict_total metric. Every @@ -1551,6 +1601,13 @@ func (r *runtimeServerRunner) start() error { return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err) } r.sqsServer = sqsServer + // Plug the SQS adapter into the monitoring registry's depth + // observer (see startSQSDepthObserver). nil sqsServer (e.g. + // --sqsAddress empty on this node) is a no-op so single-binary + // tests don't need to construct a fake source. + if r.sqsServer != nil { + startSQSDepthObserver(r.ctx, r.metricsRegistry, r.sqsServer) + } if err := startMetricsServer(r.ctx, r.lc, r.eg, r.metricsAddress, r.metricsToken, r.metricsRegistry.Handler()); err != nil { return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err) } diff --git a/monitoring/grafana/dashboards/elastickv-sqs.json b/monitoring/grafana/dashboards/elastickv-sqs.json new file mode 100644 index 000000000..52e77363d --- /dev/null +++ b/monitoring/grafana/dashboards/elastickv-sqs.json @@ -0,0 +1,477 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "description": "elastickv SQS-compatible queue health: per-queue depth (visible / in-flight / delayed), HT-FIFO partition activity, and the leader-node breakdown. Source metrics: elastickv_sqs_queue_messages (gauges, leader-emitted) and elastickv_sqs_partition_messages_total (counter).", + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "datasource": "$datasource", + "description": "Distinct queues currently reporting depth on the leader. Drops to 0 if the leader can't run its catalog scan.", + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "red", "value": null }, + { "color": "green", "value": 1 } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 4, "w": 6, "x": 0, "y": 0 }, + "id": 1, + "options": { + "colorMode": "value", + "graphMode": "area", + "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false }, + "textMode": "auto" + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "count(count by (queue) (elastickv_sqs_queue_messages))", + "legendFormat": "queues", + "refId": "A" + } + ], + "title": "Queues observed", + "type": "stat" + }, + { + "datasource": "$datasource", + "description": "Sum across every queue of visible + in-flight + delayed messages. Use for capacity planning and as a proxy for cluster-wide work backlog.", + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1000 }, + { "color": "red", "value": 10000 } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 4, "w": 6, "x": 6, "y": 0 }, + "id": 2, + "options": { + "colorMode": "value", + "graphMode": "area", + "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false }, + "textMode": "auto" + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "sum(elastickv_sqs_queue_messages)", + "legendFormat": "messages", + "refId": "A" + } + ], + "title": "Total backlog (all queues, all states)", + "type": "stat" + }, + { + "datasource": "$datasource", + "description": "HT-FIFO partition operations per second across all queues. Sums send + receive + delete; split out below.", + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { "h": 4, "w": 6, "x": 12, "y": 0 }, + "id": 3, + "options": { + "colorMode": "value", + "graphMode": "area", + "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false }, + "textMode": "auto" + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "sum(rate(elastickv_sqs_partition_messages_total[$__rate_interval]))", + "legendFormat": "ops/s", + "refId": "A" + } + ], + "title": "HT-FIFO ops rate", + "type": "stat" + }, + { + "datasource": "$datasource", + "description": "Node currently emitting depth gauges. The leader runs the catalog scan; followers stay silent so this stat reads the {node_id, node_address} pair off any depth series.", + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "none" + }, + "overrides": [] + }, + "gridPos": { "h": 4, "w": 6, "x": 18, "y": 0 }, + "id": 4, + "options": { + "colorMode": "none", + "graphMode": "none", + "reduceOptions": { "calcs": ["lastNotNull"], "fields": "/^node_id$/", "values": true }, + "textMode": "value_and_name" + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "max by (node_id) (elastickv_sqs_queue_messages > 0) * 0 + 1", + "format": "table", + "instant": true, + "legendFormat": "{{node_id}}", + "refId": "A" + } + ], + "title": "Active leader", + "type": "stat" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 4 }, + "id": 50, + "panels": [], + "title": "Per-queue depth", + "type": "row" + }, + { + "datasource": "$datasource", + "description": "Visible (= ApproximateNumberOfMessages) per queue. The number of messages waiting for ReceiveMessage. Sustained growth means consumers can't keep up.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisLabel": "messages", + "fillOpacity": 10, + "gradientMode": "none", + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 2, + "showPoints": "never", + "spanNulls": false, + "stacking": { "mode": "none" }, + "thresholdsStyle": { "mode": "off" } + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 5 }, + "id": 5, + "options": { + "legend": { "calcs": ["lastNotNull", "max"], "displayMode": "table", "placement": "right", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "elastickv_sqs_queue_messages{queue=~\"$queue\", state=\"visible\"}", + "legendFormat": "{{queue}}", + "refId": "A" + } + ], + "title": "Visible messages", + "type": "timeseries" + }, + { + "datasource": "$datasource", + "description": "In-flight (= ApproximateNumberOfMessagesNotVisible) per queue. Messages received by a consumer but not yet deleted; pinned high = consumer crashed or visibility timeout too long.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisLabel": "messages", + "fillOpacity": 10, + "gradientMode": "none", + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 2, + "showPoints": "never", + "spanNulls": false, + "stacking": { "mode": "none" } + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 5 }, + "id": 6, + "options": { + "legend": { "calcs": ["lastNotNull", "max"], "displayMode": "table", "placement": "right", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "elastickv_sqs_queue_messages{queue=~\"$queue\", state=\"not_visible\"}", + "legendFormat": "{{queue}}", + "refId": "A" + } + ], + "title": "In-flight messages", + "type": "timeseries" + }, + { + "datasource": "$datasource", + "description": "Delayed (= ApproximateNumberOfMessagesDelayed) per queue. Messages with DelaySeconds not yet expired. Usually small unless callers explicitly schedule.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisLabel": "messages", + "fillOpacity": 10, + "gradientMode": "none", + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 2, + "showPoints": "never", + "spanNulls": false, + "stacking": { "mode": "none" } + }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 13 }, + "id": 7, + "options": { + "legend": { "calcs": ["lastNotNull", "max"], "displayMode": "table", "placement": "right", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "elastickv_sqs_queue_messages{queue=~\"$queue\", state=\"delayed\"}", + "legendFormat": "{{queue}}", + "refId": "A" + } + ], + "title": "Delayed messages", + "type": "timeseries" + }, + { + "datasource": "$datasource", + "description": "Top queues by total backlog (visible + in-flight + delayed). Quick way to find the queue that's accumulating most work right now.", + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "custom": { + "align": "auto", + "cellOptions": { "type": "auto" }, + "inspect": false + }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }] }, + "unit": "short" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 13 }, + "id": 8, + "options": { + "cellHeight": "sm", + "footer": { "countRows": false, "fields": "", "reducer": ["sum"], "show": false }, + "showHeader": true, + "sortBy": [{ "desc": true, "displayName": "Backlog" }] + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "topk(20, sum by (queue) (elastickv_sqs_queue_messages{queue=~\"$queue\"}))", + "format": "table", + "instant": true, + "legendFormat": "{{queue}}", + "refId": "A" + } + ], + "title": "Top 20 queues by backlog", + "transformations": [ + { + "id": "organize", + "options": { + "excludeByName": { "Time": true, "node_id": true, "node_address": true, "__name__": true }, + "indexByName": {}, + "renameByName": { "Value": "Backlog", "queue": "Queue" } + } + } + ], + "type": "table" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 21 }, + "id": 51, + "panels": [], + "title": "HT-FIFO partition activity", + "type": "row" + }, + { + "datasource": "$datasource", + "description": "HT-FIFO partition operation rate, broken down by action. Counter is non-zero only for queues with PartitionCount > 1.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisLabel": "ops/s", + "fillOpacity": 10, + "gradientMode": "none", + "lineInterpolation": "linear", + "lineWidth": 1, + "showPoints": "never", + "stacking": { "mode": "normal" } + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 22 }, + "id": 9, + "options": { + "legend": { "calcs": ["lastNotNull", "max"], "displayMode": "table", "placement": "right", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "sum by (action) (rate(elastickv_sqs_partition_messages_total{queue=~\"$queue\"}[$__rate_interval]))", + "legendFormat": "{{action}}", + "refId": "A" + } + ], + "title": "Ops rate by action", + "type": "timeseries" + }, + { + "datasource": "$datasource", + "description": "Per-(queue, partition) operation rate. Use to spot uneven MessageGroupId distribution across partitions: ideal hash spreads roughly evenly, hot partitions show as spikes.", + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisLabel": "ops/s", + "fillOpacity": 10, + "gradientMode": "none", + "lineInterpolation": "linear", + "lineWidth": 1, + "showPoints": "never", + "stacking": { "mode": "none" } + }, + "unit": "ops" + }, + "overrides": [] + }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 22 }, + "id": 10, + "options": { + "legend": { "calcs": ["lastNotNull", "max"], "displayMode": "table", "placement": "right", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "10.4.0", + "targets": [ + { + "datasource": "$datasource", + "expr": "sum by (queue, partition) (rate(elastickv_sqs_partition_messages_total{queue=~\"$queue\"}[$__rate_interval]))", + "legendFormat": "{{queue}}/p{{partition}}", + "refId": "A" + } + ], + "title": "Ops rate by (queue, partition)", + "type": "timeseries" + } + ], + "refresh": "30s", + "schemaVersion": 39, + "tags": ["elastickv", "sqs"], + "templating": { + "list": [ + { + "current": { "selected": false, "text": "Prometheus", "value": "Prometheus" }, + "hide": 0, + "includeAll": false, + "label": "Datasource", + "multi": false, + "name": "datasource", + "options": [], + "query": "prometheus", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + }, + { + "allValue": ".*", + "current": { "selected": false, "text": "All", "value": "$__all" }, + "datasource": "$datasource", + "definition": "label_values(elastickv_sqs_queue_messages, queue)", + "hide": 0, + "includeAll": true, + "label": "Queue", + "multi": true, + "name": "queue", + "options": [], + "query": { + "query": "label_values(elastickv_sqs_queue_messages, queue)", + "refId": "Prometheus-queue-Variable-Query" + }, + "refresh": 2, + "regex": "", + "skipUrlSync": false, + "sort": 1, + "type": "query" + } + ] + }, + "time": { "from": "now-1h", "to": "now" }, + "timepicker": {}, + "timezone": "", + "title": "Elastickv SQS", + "uid": "elastickv-sqs", + "version": 1, + "weekStart": "" +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 8ce4e15d1..2dbb9f24f 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -22,6 +22,7 @@ type Registry struct { pebble *PebbleMetrics writeConflict *WriteConflictMetrics sqs *SQSMetrics + sqsObserver *SQSObserver } // NewRegistry builds a registry with constant labels that identify the local node. @@ -45,6 +46,7 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry { r.pebble = newPebbleMetrics(registerer) r.writeConflict = newWriteConflictMetrics(registerer) r.sqs = newSQSMetrics(registerer) + r.sqsObserver = newSQSObserver(r.sqs) return r } @@ -172,6 +174,17 @@ func (r *Registry) SQSPartitionObserver() SQSPartitionObserver { return r.sqs } +// SQSObserver returns the queue-depth gauge observer backed by +// this registry. Same shape as RaftObserver / RedisObserver: callers +// pull it via the registry, then drive Start(ctx, source, interval) +// from main.go's startMonitoringCollectors. +func (r *Registry) SQSObserver() *SQSObserver { + if r == nil { + return nil + } + return r.sqsObserver +} + // WriteConflictCollector returns a collector that polls each MVCC // store's per-(kind, key_prefix) OCC conflict counters and mirrors // them into the elastickv_store_write_conflict_total Prometheus diff --git a/monitoring/sqs.go b/monitoring/sqs.go index 7616b124c..884ba66e4 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -1,8 +1,10 @@ package monitoring import ( + "context" "strconv" "sync" + "time" "github.com/prometheus/client_golang/prometheus" ) @@ -43,16 +45,53 @@ type SQSPartitionObserver interface { ObservePartitionMessage(queue string, partition uint32, action string) } -// SQSMetrics owns the Prometheus counter for HT-FIFO partition -// operations. Mirrors DynamoDBMetrics' shape: per-Registry -// instance, label-cardinality-bounded by sqsMaxTrackedQueues. +// SQSDepthSource is the contract a per-tick queue-depth source must +// satisfy. Implemented by *adapter.SQSServer; SQSObserver.Start +// calls SnapshotQueueDepths on every interval and writes the +// returned slice to the elastickv_sqs_queue_messages gauges. +// +// Mirrors the Raft observer's StatusReader / ConfigReader pattern +// (monitoring/raft.go): the source returns ready-to-use snapshots +// and the observer owns the gauge state machine (forget-on-disappear, +// cardinality cap). Implementations return an empty slice — not an +// error — when this node is a follower, so the dashboard's gauge +// set always mirrors what the leader's catalog scan would report. +type SQSDepthSource interface { + SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth +} + +// SQSQueueDepth is one queue's depth-attribute snapshot. Mirrors +// adapter.SQSQueueDepth byte-for-byte and is re-declared here to +// keep the monitoring package free of an adapter import. A drift +// between the two definitions surfaces as a compile error at the +// SQSObserver call site. +type SQSQueueDepth struct { + Queue string + Visible int64 + NotVisible int64 + Delayed int64 +} + +// SQSMetrics owns the Prometheus collectors for the SQS adapter. +// Mirrors DynamoDBMetrics' shape: per-Registry instance, label- +// cardinality-bounded by sqsMaxTrackedQueues, and split between +// counters (HT-FIFO partition activity) and gauges (queue depth). type SQSMetrics struct { partitionMessages *prometheus.CounterVec + queueDepth *prometheus.GaugeVec mu sync.Mutex trackedQueues map[string]struct{} } +// SQS depth gauge state-label values. Stable so dashboards / alerts +// can hard-code state="visible" et al. +const ( + sqsQueueStateVisible = "visible" + sqsQueueStateNotVisible = "not_visible" + sqsQueueStateDelayed = "delayed" +) + func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics { m := &SQSMetrics{ partitionMessages: prometheus.NewCounterVec( @@ -62,9 +101,17 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics { }, []string{"queue", "partition", "action"}, ), + queueDepth: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "elastickv_sqs_queue_messages", + Help: "Approximate number of messages in each SQS queue, broken down by state. visible = ApproximateNumberOfMessages, not_visible = ApproximateNumberOfMessagesNotVisible, delayed = ApproximateNumberOfMessagesDelayed. Updated periodically by the leader's queue-depth scraper; followers report no value.", + }, + []string{"queue", "state"}, + ), trackedQueues: map[string]struct{}{}, } registerer.MustRegister(m.partitionMessages) + registerer.MustRegister(m.queueDepth) return m } @@ -97,6 +144,45 @@ func (m *SQSMetrics) ObservePartitionMessage(queue string, partition uint32, act ).Inc() } +// ObserveQueueDepth implements SQSDepthObserver. Updates the three +// state-labelled gauges for queue. Negative values are clamped to 0 +// so a transient scan failure (returning -1 sentinel from a future +// caller) cannot blast a fake backlog onto the dashboard. +func (m *SQSMetrics) ObserveQueueDepth(queue string, visible, notVisible, delayed int64) { + if m == nil { + return + } + if queue == "" { + return + } + queueLabel := m.queueLabelForCardinalityBudget(queue) + m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateVisible).Set(float64(maxInt64(0, visible))) + m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateNotVisible).Set(float64(maxInt64(0, notVisible))) + m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateDelayed).Set(float64(maxInt64(0, delayed))) +} + +// ForgetQueue implements SQSDepthObserver: drops the three gauge +// series for a queue. Called when DeleteQueue / Purge has wiped a +// queue so dashboards stop showing a frozen backlog. The (queue, +// partition, action) counter series stays — it's cumulative-by- +// design and disappearing it would mask retention-period activity. +func (m *SQSMetrics) ForgetQueue(queue string) { + if m == nil || queue == "" { + return + } + queueLabel := m.queueLabelForCardinalityBudget(queue) + m.queueDepth.DeleteLabelValues(queueLabel, sqsQueueStateVisible) + m.queueDepth.DeleteLabelValues(queueLabel, sqsQueueStateNotVisible) + m.queueDepth.DeleteLabelValues(queueLabel, sqsQueueStateDelayed) +} + +func maxInt64(a, b int64) int64 { + if a > b { + return a + } + return b +} + // queueLabelForCardinalityBudget returns queue if the metric has // already emitted a series for it OR there is room in the // tracked-queues set; returns sqsQueueOverflow otherwise. The @@ -126,3 +212,88 @@ func sqsValidPartitionAction(action string) bool { } return false } + +// sqsDepthObserveInterval is the default tick cadence for +// SQSObserver. 30 s mirrors sqsReaperInterval — dashboards are +// rarely refreshed faster, and tighter ticks just add catalog-scan +// load on the leader for no observable benefit. +const sqsDepthObserveInterval = 30 * time.Second + +// SQSObserver polls a SQSDepthSource on a fixed cadence and writes +// the result into the SQSMetrics gauge. Same shape as RaftObserver +// (monitoring/raft.go): the observer owns the state machine +// (current-vs-previous queue diff for ForgetQueue) and the source +// just returns ready-to-use snapshots. The observer is nil-tolerant +// at every entrypoint so test fixtures and metrics-disabled +// deployments can no-op without a defensive nil check. +type SQSObserver struct { + metrics *SQSMetrics + + mu sync.Mutex + lastSeen map[string]struct{} +} + +func newSQSObserver(metrics *SQSMetrics) *SQSObserver { + return &SQSObserver{ + metrics: metrics, + lastSeen: map[string]struct{}{}, + } +} + +// Start kicks off a background ticker that polls source every +// interval (defaulting to sqsDepthObserveInterval when zero) until +// ctx is canceled. The first observation runs synchronously so +// /metrics has fresh data on the first scrape; subsequent ticks +// run on the goroutine. +func (o *SQSObserver) Start(ctx context.Context, source SQSDepthSource, interval time.Duration) { + if o == nil || source == nil { + return + } + if interval <= 0 { + interval = sqsDepthObserveInterval + } + o.observeOnce(ctx, source) + ticker := time.NewTicker(interval) + go func() { + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + o.observeOnce(ctx, source) + } + } + }() +} + +// ObserveOnce captures the latest depth snapshot synchronously. +// Mirrors RaftObserver.ObserveOnce; intended for tests that want +// deterministic single-tick behaviour without spinning up a ticker. +func (o *SQSObserver) ObserveOnce(source SQSDepthSource) { + o.observeOnce(context.Background(), source) +} + +func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) { + if o == nil || o.metrics == nil || source == nil { + return + } + snaps := source.SnapshotQueueDepths(ctx) + current := make(map[string]struct{}, len(snaps)) + for _, snap := range snaps { + o.metrics.ObserveQueueDepth(snap.Queue, snap.Visible, snap.NotVisible, snap.Delayed) + current[snap.Queue] = struct{}{} + } + // Diff against the previous tick: any queue that disappeared + // (DeleteQueue, tombstoned cohort fully drained, leader stepped + // down — source returned []) gets its gauge series dropped so + // dashboards don't show a frozen backlog. + o.mu.Lock() + for prev := range o.lastSeen { + if _, ok := current[prev]; !ok { + o.metrics.ForgetQueue(prev) + } + } + o.lastSeen = current + o.mu.Unlock() +} diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index 5c3784782..ad0ce6de5 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -1,9 +1,11 @@ package monitoring import ( + "context" "strconv" "strings" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" @@ -139,3 +141,215 @@ func TestSQSMetrics_RegistryWiring(t *testing.T) { require.True(t, found, "elastickv_sqs_partition_messages_total must be registered on the public Registry") } + +// TestSQSMetrics_ObserveQueueDepth_EmitsThreeStates pins that one +// ObserveQueueDepth call produces three series (visible / not_visible +// / delayed) under elastickv_sqs_queue_messages keyed on queue name. +func TestSQSMetrics_ObserveQueueDepth_EmitsThreeStates(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + m.ObserveQueueDepth("orders.fifo", 5, 2, 1) + + require.InDelta(t, 5.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateVisible)), 0.001) + require.InDelta(t, 2.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateNotVisible)), 0.001) + require.InDelta(t, 1.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateDelayed)), 0.001) +} + +// TestSQSMetrics_ObserveQueueDepth_ClampsNegativeToZero pins the +// defensive clamp: a future caller that passes a -1 sentinel from a +// failed scan must not blast a fake backlog onto dashboards. +func TestSQSMetrics_ObserveQueueDepth_ClampsNegativeToZero(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + m.ObserveQueueDepth("orders.fifo", -7, -1, -100) + + for _, state := range []string{sqsQueueStateVisible, sqsQueueStateNotVisible, sqsQueueStateDelayed} { + require.InDelta(t, 0.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", state)), 0.001, + "state %s must be clamped to 0 on negative input", state) + } +} + +// TestSQSMetrics_ObserveQueueDepth_DropsEmptyQueue pins the +// defensive empty-name drop: same rule as the partition counter, +// preventing all callers from collapsing onto the empty-label +// series under a future bug. +func TestSQSMetrics_ObserveQueueDepth_DropsEmptyQueue(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + m.ObserveQueueDepth("", 1, 1, 1) + + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, 0, count, "empty queue name must not emit any series") +} + +// TestSQSMetrics_ForgetQueue_DropsThreeSeries pins that ForgetQueue +// removes all three state-labelled series so a deleted queue stops +// reporting a frozen backlog. The (queue, partition, action) counter +// is intentionally untouched (cumulative-by-design). +func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + m.ObserveQueueDepth("orders.fifo", 3, 0, 0) + m.ObservePartitionMessage("orders.fifo", 0, SQSPartitionActionSend) + + m.ForgetQueue("orders.fifo") + + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, 0, count, "ForgetQueue must drop every state series for the queue") + require.InDelta(t, 1.0, testutil.ToFloat64(m.partitionMessages.WithLabelValues("orders.fifo", "0", SQSPartitionActionSend)), 0.001, + "counter must survive ForgetQueue (cumulative metric)") +} + +// TestSQSMetrics_DepthNilReceiverIsSafe mirrors the partition test: +// a typed-nil *SQSMetrics handed to the adapter must not panic on +// either ObserveQueueDepth or ForgetQueue calls. +func TestSQSMetrics_DepthNilReceiverIsSafe(t *testing.T) { + t.Parallel() + var m *SQSMetrics + require.NotPanics(t, func() { + m.ObserveQueueDepth("q.fifo", 1, 2, 3) + m.ForgetQueue("q.fifo") + }) +} + +// TestSQSMetrics_DepthRegistryWiring pins that the depth gauge +// surfaces under the documented metric name on the public Registry +// after a snapshot is observed via the registry's SQSMetrics. The +// SQSObserver path is exercised separately in +// TestSQSObserver_ObserveOnce_EmitsAndForgets below. +func TestSQSMetrics_DepthRegistryWiring(t *testing.T) { + t.Parallel() + r := NewRegistry("node-test", "127.0.0.1:50051") + require.NotNil(t, r.SQSObserver(), "registry must expose a non-nil SQSObserver") + require.NotNil(t, r.sqs, "registry must wire the SQSMetrics") + + r.sqs.ObserveQueueDepth("q.fifo", 4, 1, 0) + + mfs, err := r.Gatherer().Gather() + require.NoError(t, err) + var found bool + for _, mf := range mfs { + if !strings.HasPrefix(mf.GetName(), "elastickv_sqs_queue_messages") { + continue + } + found = true + require.Len(t, mf.GetMetric(), 3, + "three state labels (visible / not_visible / delayed) for one queue") + } + require.True(t, found, + "elastickv_sqs_queue_messages must be registered on the public Registry") +} + +// fakeDepthSource lets the SQSObserver tests script the per-tick +// snapshot without standing up a real SQS adapter + coordinator. +type fakeDepthSource struct { + snapshots [][]SQSQueueDepth + calls int +} + +func (f *fakeDepthSource) SnapshotQueueDepths(_ context.Context) []SQSQueueDepth { + if f.calls >= len(f.snapshots) { + f.calls++ + return nil + } + out := f.snapshots[f.calls] + f.calls++ + return out +} + +// TestSQSObserver_ObserveOnce_EmitsAndForgets pins the observer's +// state machine: queues present in the current tick get gauges +// emitted, queues that disappeared since the last tick get +// ForgetQueue'd. Without that diff the dashboard would show a +// frozen backlog after DeleteQueue. +func TestSQSObserver_ObserveOnce_EmitsAndForgets(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + obs := newSQSObserver(m) + + source := &fakeDepthSource{ + snapshots: [][]SQSQueueDepth{ + // tick 1: two queues + { + {Queue: "orders.fifo", Visible: 5, NotVisible: 2, Delayed: 0}, + {Queue: "audio.fifo", Visible: 0, NotVisible: 0, Delayed: 3}, + }, + // tick 2: orders.fifo disappeared + { + {Queue: "audio.fifo", Visible: 1, NotVisible: 0, Delayed: 0}, + }, + }, + } + + obs.ObserveOnce(source) + require.InDelta(t, 5.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateVisible)), 0.001) + require.InDelta(t, 3.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("audio.fifo", sqsQueueStateDelayed)), 0.001) + + obs.ObserveOnce(source) + // orders.fifo gauges should be gone. + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, 3, count, "after second tick: only audio.fifo's three state gauges remain") + require.InDelta(t, 1.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("audio.fifo", sqsQueueStateVisible)), 0.001) +} + +// TestSQSObserver_ObserveOnce_LeaderStepDownClearsAll pins the +// leader-step-down behaviour: when the source returns an empty +// slice (the adapter's not-leader branch), every gauge from the +// previous tick must be cleared so a stepped-down leader doesn't +// leave stale numbers on the dashboard until the new leader runs +// its first tick. +func TestSQSObserver_ObserveOnce_LeaderStepDownClearsAll(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + obs := newSQSObserver(m) + + source := &fakeDepthSource{ + snapshots: [][]SQSQueueDepth{ + { + {Queue: "orders.fifo", Visible: 5}, + {Queue: "audio.fifo", Visible: 1}, + }, + nil, // leader stepped down + }, + } + obs.ObserveOnce(source) + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, 6, count, "tick 1: 2 queues × 3 states = 6 series") + + obs.ObserveOnce(source) + count, err = testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, 0, count, "tick 2 (leader step-down): all gauges cleared") +} + +// TestSQSObserver_NilTolerant pins that nil observer / nil source +// don't panic — the same nil-tolerant contract Raft / Redis +// observers carry, so a metrics-disabled deployment can pass nil +// through without defensive code at every call site. +func TestSQSObserver_NilTolerant(t *testing.T) { + t.Parallel() + var nilObs *SQSObserver + require.NotPanics(t, func() { + nilObs.Start(context.Background(), &fakeDepthSource{}, time.Millisecond) + nilObs.ObserveOnce(&fakeDepthSource{}) + }) + obs := newSQSObserver(newSQSMetrics(prometheus.NewRegistry())) + require.NotPanics(t, func() { + obs.Start(context.Background(), nil, time.Millisecond) + obs.ObserveOnce(nil) + }) +} From 1911cc1314ef497fae260d0d4a30c6f60fcc2171 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 04:19:43 +0900 Subject: [PATCH 02/10] monitoring/sqs: ForgetQueue frees cardinality slot + dashboard fix (PR #743 r1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three findings from the round-1 review on PR #743, plus the suggested test additions: Medium (cardinality leak): - ForgetQueue used to call DeleteLabelValues but never shrank trackedQueues, so the 512-entry budget filled with stale slots and post-cap new queues silently collapsed to _other forever. Fix: take m.mu, delete from trackedQueues, release, then DeleteLabelValues — but only when the queue was tracked. Overflow-mapped queues (queues that hit the cap and got the _other label) skip DeleteLabelValues so we don't tear down the shared _other series for unrelated queues. Low / style: - maxInt64 helper replaced with the Go 1.21+ builtin max[T]. Module already requires Go 1.25 (go.mod) so the builtin is available; the helper was redundant. - Dashboard 'Active leader' stat: the previous expr filtered on 'elastickv_sqs_queue_messages > 0' which returns no data on an idle cluster (every depth = 0). Switched to 'count by (node_id) (elastickv_sqs_queue_messages)' which surfaces the leader regardless of queue contents. - observeOnce: added single-writer-contract comment so a future caller doesn't accidentally introduce concurrent ticks; the CounterVec / GaugeVec writes are individually atomic, but symmetric documentation prevents drift. Tests: - TestSQSMetrics_ForgetQueue_DropsThreeSeries strengthened: now asserts trackedQueues was emptied AND that a post-forget Observe re-emits under the real queue name (not _other). Without the budget cleanup this assertion would have caught the leak. - TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp pins the overflow-collision safety: ForgetQueue on an _other-mapped queue must not delete the shared _other series and must not shrink the budget. Caller-audit per the standing semantic-change rule: - ForgetQueue's only production caller is SQSObserver.observeOnce. observeOnce calls it from the lastSeen-vs-current diff, only for queues that had been observed in the previous tick — so the not-tracked branch (added with this fix) only fires for overflow queues, never for regular ones, never for never- observed ones. No caller needs an update. --- .../grafana/dashboards/elastickv-sqs.json | 2 +- monitoring/sqs.go | 66 ++++++++++++++----- monitoring/sqs_test.go | 54 ++++++++++++++- 3 files changed, 100 insertions(+), 22 deletions(-) diff --git a/monitoring/grafana/dashboards/elastickv-sqs.json b/monitoring/grafana/dashboards/elastickv-sqs.json index 52e77363d..0ac59896a 100644 --- a/monitoring/grafana/dashboards/elastickv-sqs.json +++ b/monitoring/grafana/dashboards/elastickv-sqs.json @@ -157,7 +157,7 @@ "targets": [ { "datasource": "$datasource", - "expr": "max by (node_id) (elastickv_sqs_queue_messages > 0) * 0 + 1", + "expr": "count by (node_id) (elastickv_sqs_queue_messages)", "format": "table", "instant": true, "legendFormat": "{{node_id}}", diff --git a/monitoring/sqs.go b/monitoring/sqs.go index 884ba66e4..273659606 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -156,31 +156,52 @@ func (m *SQSMetrics) ObserveQueueDepth(queue string, visible, notVisible, delaye return } queueLabel := m.queueLabelForCardinalityBudget(queue) - m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateVisible).Set(float64(maxInt64(0, visible))) - m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateNotVisible).Set(float64(maxInt64(0, notVisible))) - m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateDelayed).Set(float64(maxInt64(0, delayed))) + m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateVisible).Set(float64(max(int64(0), visible))) + m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateNotVisible).Set(float64(max(int64(0), notVisible))) + m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateDelayed).Set(float64(max(int64(0), delayed))) } -// ForgetQueue implements SQSDepthObserver: drops the three gauge -// series for a queue. Called when DeleteQueue / Purge has wiped a -// queue so dashboards stop showing a frozen backlog. The (queue, -// partition, action) counter series stays — it's cumulative-by- -// design and disappearing it would mask retention-period activity. +// ForgetQueue drops the three gauge series for a queue and frees +// its cardinality-budget slot so a long-running deployment that +// regularly creates and deletes queues (CI workloads, ephemeral +// per-job queues) doesn't permanently wedge the 512-entry budget. +// Without the trackedQueues cleanup, post-cap new queues would +// silently collapse onto the _other label even after their +// predecessors had been deleted. +// +// The (queue, partition, action) counter series stays — +// cumulative-by-design and disappearing it would mask retention- +// period activity. Queues that hit the cap and mapped to _other +// have no individual series to delete; we detect the not-tracked +// case and skip the DeleteLabelValues calls so we don't tear down +// the shared _other series for an unrelated queue. +// +// Caller-audit per the standing semantic-change rule: only +// SQSObserver.observeOnce calls this (registry plumbing aside), +// and it's invoked exactly when a queue is observed in the +// previous tick but not the current one — symmetric with +// ObserveQueueDepth's add side, so there is no path that calls +// ForgetQueue without a matching prior tracked-queue insert. func (m *SQSMetrics) ForgetQueue(queue string) { if m == nil || queue == "" { return } - queueLabel := m.queueLabelForCardinalityBudget(queue) - m.queueDepth.DeleteLabelValues(queueLabel, sqsQueueStateVisible) - m.queueDepth.DeleteLabelValues(queueLabel, sqsQueueStateNotVisible) - m.queueDepth.DeleteLabelValues(queueLabel, sqsQueueStateDelayed) -} - -func maxInt64(a, b int64) int64 { - if a > b { - return a + m.mu.Lock() + _, tracked := m.trackedQueues[queue] + if tracked { + delete(m.trackedQueues, queue) + } + m.mu.Unlock() + if !tracked { + // Queue was either never observed or had been collapsed onto + // the _other overflow label. Either way: no per-queue series + // to remove, and we must NOT delete the _other series here + // because other overflow queues may still be sharing it. + return } - return b + m.queueDepth.DeleteLabelValues(queue, sqsQueueStateVisible) + m.queueDepth.DeleteLabelValues(queue, sqsQueueStateNotVisible) + m.queueDepth.DeleteLabelValues(queue, sqsQueueStateDelayed) } // queueLabelForCardinalityBudget returns queue if the metric has @@ -274,6 +295,15 @@ func (o *SQSObserver) ObserveOnce(source SQSDepthSource) { o.observeOnce(context.Background(), source) } +// observeOnce assumes a single-writer contract: in production the +// only caller is the goroutine launched from Start, and tests use +// ObserveOnce serially. ObserveQueueDepth runs unlocked because the +// CounterVec / GaugeVec writes are individually atomic; the +// trackedQueues mutation inside is guarded by m.mu. Concurrent +// observeOnce invocations would race only on the lastSeen diff +// (held briefly under o.mu below), so a future caller that +// violates the single-writer rule would at worst double-emit a +// gauge — never a panic. func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) { if o == nil || o.metrics == nil || source == nil { return diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index ad0ce6de5..cc680944c 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -190,9 +190,11 @@ func TestSQSMetrics_ObserveQueueDepth_DropsEmptyQueue(t *testing.T) { } // TestSQSMetrics_ForgetQueue_DropsThreeSeries pins that ForgetQueue -// removes all three state-labelled series so a deleted queue stops -// reporting a frozen backlog. The (queue, partition, action) counter -// is intentionally untouched (cumulative-by-design). +// (a) removes all three state-labelled series so a deleted queue +// stops reporting a frozen backlog, (b) frees the cardinality- +// budget slot so a churn-heavy deployment doesn't permanently +// exhaust the 512-entry budget, and (c) leaves the +// (queue, partition, action) counter alone (cumulative-by-design). func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) { t.Parallel() reg := prometheus.NewRegistry() @@ -200,14 +202,60 @@ func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) { m.ObserveQueueDepth("orders.fifo", 3, 0, 0) m.ObservePartitionMessage("orders.fifo", 0, SQSPartitionActionSend) + require.Len(t, m.trackedQueues, 1, "queue must have been added to the cardinality budget on Observe") m.ForgetQueue("orders.fifo") count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") require.NoError(t, err) require.Equal(t, 0, count, "ForgetQueue must drop every state series for the queue") + require.Empty(t, m.trackedQueues, "ForgetQueue must free the cardinality-budget slot") require.InDelta(t, 1.0, testutil.ToFloat64(m.partitionMessages.WithLabelValues("orders.fifo", "0", SQSPartitionActionSend)), 0.001, "counter must survive ForgetQueue (cumulative metric)") + + // Observe again post-forget: the new series must be keyed on + // the real name (slot was freed), NOT on the _other overflow + // label. Without the trackedQueues cleanup this would silently + // collapse to _other once the budget eventually filled. + m.ObserveQueueDepth("orders.fifo", 7, 0, 0) + require.InDelta(t, 7.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateVisible)), 0.001, + "post-forget Observe must re-emit under the real queue name") +} + +// TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp pins the +// overflow-collision safety: a queue that hit the cardinality cap +// and got collapsed onto the _other label has no individual series +// to delete. ForgetQueue must NOT call DeleteLabelValues with the +// _other label because that series is shared with every other +// overflow queue — tearing it down would zero-out unrelated data. +func TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + // Saturate the budget with sqsMaxTrackedQueues real queues so + // any further observation collapses to _other. + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ObserveQueueDepth("real-"+strconv.Itoa(i)+".fifo", 1, 0, 0) + } + require.Len(t, m.trackedQueues, sqsMaxTrackedQueues) + + // Two overflow queues — both share the _other label. + m.ObserveQueueDepth("overflow-a.fifo", 100, 0, 0) + m.ObserveQueueDepth("overflow-b.fifo", 200, 0, 0) + overflowVisible := testutil.ToFloat64(m.queueDepth.WithLabelValues(sqsQueueOverflow, sqsQueueStateVisible)) + require.InDelta(t, 200.0, overflowVisible, 0.001, + "second observe should overwrite the shared _other gauge") + + // ForgetQueue on an overflow queue must leave the _other series + // (and the budget) untouched. + m.ForgetQueue("overflow-a.fifo") + require.Len(t, m.trackedQueues, sqsMaxTrackedQueues, + "ForgetQueue on an overflow queue must not change the budget") + require.InDelta(t, 200.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues(sqsQueueOverflow, sqsQueueStateVisible)), + 0.001, + "_other series must still carry the latest value (overflow-b)") } // TestSQSMetrics_DepthNilReceiverIsSafe mirrors the partition test: From 33dca0c700481308d9068dcba2ddc7d3f7e81417 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 04:26:26 +0900 Subject: [PATCH 03/10] adapter/sqs: single readTS per depth-snapshot tick (PR #743 r2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 review: optional cleanup. Move nextTxnReadTS out of the per-queue loop in snapshotOneQueueDepth and into the SnapshotQueueDepths caller so every queue in one tick shares the same MVCC read timestamp. Two wins: - Consistency. With per-queue ts the first queue's depth can reflect catalog state the last queue's depth can't (catalog mutation between calls). The Approximate* metric names already signal advisory-only, so this isn't a correctness bug — but one ts per tick gives the dashboard a coherent point-in-time snapshot that matches what AdminListQueues + AdminDescribeQueue would return at the same instant. - HLC pressure. Each nextTxnReadTS call advances the leader's HLC. On a 500-queue deployment this drops 499 unnecessary ticks per scrape interval. Caller audit per the standing semantic-change rule: - snapshotOneQueueDepth signature gained one parameter (readTS uint64). The only caller is SnapshotQueueDepths in the same file; updated to pass the per-tick value through. No other caller exists to audit. --- adapter/sqs_depth_source.go | 26 ++++++++++++++++---------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/adapter/sqs_depth_source.go b/adapter/sqs_depth_source.go index 2d8580135..32e14fe6d 100644 --- a/adapter/sqs_depth_source.go +++ b/adapter/sqs_depth_source.go @@ -41,12 +41,19 @@ func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth { slog.Warn("sqs depth snapshot: scanQueueNames failed", "err", err) return nil } + // Take a single read timestamp for the whole tick so all queues + // in this snapshot share the same MVCC view. With per-queue + // nextTxnReadTS the first queue's read could see a state the + // last queue's read can't (catalog mutation between calls), and + // every call burns an HLC tick on the leader. One ts per tick + // is both more consistent and lighter on the leader's HLC. + readTS := s.nextTxnReadTS(ctx) out := make([]SQSQueueDepth, 0, len(names)) for _, name := range names { if err := ctx.Err(); err != nil { return out } - if snap, ok := s.snapshotOneQueueDepth(ctx, name); ok { + if snap, ok := s.snapshotOneQueueDepth(ctx, name, readTS); ok { out = append(out, snap) } } @@ -54,15 +61,14 @@ func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth { } // snapshotOneQueueDepth runs the per-queue catalog read pair -// (loadQueueMetaAt + scanApproxCounters) and returns the resulting -// snapshot. Pulled out of the loop body so SnapshotQueueDepths -// stays under the cyclop budget; ok=false means "skip this queue -// from this tick" (queue gone, transient catalog read failure). -// Per-queue scan errors are logged and the offending queue is -// dropped from this tick's snapshot rather than aborting the -// entire pass. -func (s *SQSServer) snapshotOneQueueDepth(ctx context.Context, name string) (SQSQueueDepth, bool) { - readTS := s.nextTxnReadTS(ctx) +// (loadQueueMetaAt + scanApproxCounters) at the caller-supplied +// readTS and returns the resulting snapshot. Pulled out of the +// loop body so SnapshotQueueDepths stays under the cyclop budget; +// ok=false means "skip this queue from this tick" (queue gone, +// transient catalog read failure). Per-queue scan errors are +// logged and the offending queue is dropped from this tick's +// snapshot rather than aborting the entire pass. +func (s *SQSServer) snapshotOneQueueDepth(ctx context.Context, name string, readTS uint64) (SQSQueueDepth, bool) { meta, exists, err := s.loadQueueMetaAt(ctx, name, readTS) if err != nil || !exists { return SQSQueueDepth{}, false From e40ad3bc9fbe381d41eef4b2b4805f55ce900d3c Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 06:00:35 +0900 Subject: [PATCH 04/10] monitoring/sqs: split counter / depth cardinality budgets (PR #743 r3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 (Codex review): ForgetQueue regresses counter cardinality cap. trackedQueues was a single shared map for both partitionMessages (CounterVec) and queueDepth (GaugeVec) cardinality budgets. The r1 fix made ForgetQueue delete the queue's slot so a churn-heavy deployment (CI workloads, ephemeral queues, leader step-down clearing the observer's lastSeen) could reuse budget for new queue names — correct behaviour for the gauge side, since DeleteLabelValues drops the underlying series too. But the counter side is cumulative-by-design: Prometheus can't delete a counter series without losing its observed value, so ObservePartitionMessage never calls DeleteLabelValues. Reclaiming the slot for a forgotten queue therefore lets a NEW queue be admitted under a real label while the OLD queue's counter series is still alive in Prometheus. Repeating this pushes counter cardinality past sqsMaxTrackedQueues without bound. Fix: split trackedQueues into two independent maps — trackedCounterQueues (one-way; ObservePartitionMessage admits, ForgetQueue ignores) and trackedDepthQueues (reclaimable; ForgetQueue frees the slot together with DeleteLabelValues). Two helpers replace queueLabelForCardinalityBudget: - admitForCounterBudget — used by ObservePartitionMessage - admitForDepthBudget — used by ObserveQueueDepth Caller audit per the standing semantic-change rule: ForgetQueue: only caller is SQSObserver.observeOnce. The caller's contract ("drop gauges for queue that disappeared from this tick's snapshot") is preserved. The narrowed scope — counter budget no longer reclaimed — is invisible to the observer because it only touches gauges; counters are fed via ObservePartitionMessage from a different code path. ObservePartitionMessage / ObserveQueueDepth: behaviour from the caller's POV is unchanged (admit-or-_other on a 512-slot cap). The only diff is which map backs the cap. No external contract change. queueLabelForCardinalityBudget: removed. Grep confirms no callers remain. Regression test: TestSQSMetrics_ForgetQueue_DoesNotReclaimCounterBudget saturates the counter budget with sqsMaxTrackedQueues distinct queues, ForgetQueue's all of them, then asserts a fresh queue collapses to _other (would FAIL pre-fix — verified the test fails on the previous tip and passes on this commit). TestSQSMetrics_ForgetQueue_StillReclaimsDepthBudget pins the converse: gauge budget is still reclaimable, so a 513th distinct queue post-forget gets emitted under its real name. Existing ForgetQueue tests updated for the split-map shape. --- monitoring/sqs.go | 124 +++++++++++++++++++++++++++++------------ monitoring/sqs_test.go | 92 ++++++++++++++++++++++++++++-- 2 files changed, 176 insertions(+), 40 deletions(-) diff --git a/monitoring/sqs.go b/monitoring/sqs.go index 273659606..a72c11118 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -76,12 +76,36 @@ type SQSQueueDepth struct { // Mirrors DynamoDBMetrics' shape: per-Registry instance, label- // cardinality-bounded by sqsMaxTrackedQueues, and split between // counters (HT-FIFO partition activity) and gauges (queue depth). +// +// The cardinality budget is split into two independent maps because +// the two metrics have different deletion semantics: +// +// - partitionMessages is a CounterVec. Counters are cumulative; +// deleting a series throws away its observed-since-process-start +// value, so we never call DeleteLabelValues on it. The counter +// budget therefore only ever grows — once a queue is admitted to +// trackedCounterQueues it stays admitted, and ForgetQueue does +// NOT touch this map. +// - queueDepth is a GaugeVec. Gauges have no cumulative state, so +// DeleteLabelValues is safe (and necessary, otherwise a deleted +// queue keeps reporting a frozen backlog on the dashboard). +// ForgetQueue both removes the gauge series and frees the +// queue's slot in trackedDepthQueues so a churn-heavy deployment +// can reuse the budget. +// +// Sharing one map across the two metrics regresses the counter cap: +// ForgetQueue would free a slot, a new queue would be admitted, and +// the previous queue's counter series would still occupy a real-name +// label in Prometheus — letting cardinality grow without bound under +// queue churn (or after a leader step-down clears the observer's +// lastSeen). This was the P1 finding on PR #743. type SQSMetrics struct { partitionMessages *prometheus.CounterVec queueDepth *prometheus.GaugeVec - mu sync.Mutex - trackedQueues map[string]struct{} + mu sync.Mutex + trackedCounterQueues map[string]struct{} + trackedDepthQueues map[string]struct{} } // SQS depth gauge state-label values. Stable so dashboards / alerts @@ -108,7 +132,8 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics { }, []string{"queue", "state"}, ), - trackedQueues: map[string]struct{}{}, + trackedCounterQueues: map[string]struct{}{}, + trackedDepthQueues: map[string]struct{}{}, } registerer.MustRegister(m.partitionMessages) registerer.MustRegister(m.queueDepth) @@ -132,7 +157,7 @@ func (m *SQSMetrics) ObservePartitionMessage(queue string, partition uint32, act // poisoned data. return } - queueLabel := m.queueLabelForCardinalityBudget(queue) + queueLabel := m.admitForCounterBudget(queue) // WithLabelValues avoids the prometheus.Labels map allocation // on every observe call. Label order matches the // NewCounterVec declaration: queue, partition, action. @@ -155,48 +180,59 @@ func (m *SQSMetrics) ObserveQueueDepth(queue string, visible, notVisible, delaye if queue == "" { return } - queueLabel := m.queueLabelForCardinalityBudget(queue) + queueLabel := m.admitForDepthBudget(queue) m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateVisible).Set(float64(max(int64(0), visible))) m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateNotVisible).Set(float64(max(int64(0), notVisible))) m.queueDepth.WithLabelValues(queueLabel, sqsQueueStateDelayed).Set(float64(max(int64(0), delayed))) } // ForgetQueue drops the three gauge series for a queue and frees -// its cardinality-budget slot so a long-running deployment that -// regularly creates and deletes queues (CI workloads, ephemeral -// per-job queues) doesn't permanently wedge the 512-entry budget. -// Without the trackedQueues cleanup, post-cap new queues would -// silently collapse onto the _other label even after their -// predecessors had been deleted. +// its slot in the depth-side cardinality budget so a long-running +// deployment that regularly creates and deletes queues (CI +// workloads, ephemeral per-job queues) doesn't permanently wedge +// the 512-entry depth budget. Without the trackedDepthQueues +// cleanup, post-cap new queues would silently collapse onto the +// _other label even after their predecessors had been deleted. // // The (queue, partition, action) counter series stays — -// cumulative-by-design and disappearing it would mask retention- -// period activity. Queues that hit the cap and mapped to _other -// have no individual series to delete; we detect the not-tracked -// case and skip the DeleteLabelValues calls so we don't tear down -// the shared _other series for an unrelated queue. +// cumulative-by-design — and the queue's slot in +// trackedCounterQueues stays consumed. Reclaiming the counter slot +// would let a later queue be admitted under its real name while the +// original counter series still sat in Prometheus, which would let +// counter cardinality grow past sqsMaxTrackedQueues under churn or +// after a leader step-down clears the observer's lastSeen (the P1 +// finding on PR #743). +// +// Queues that hit the depth cap and mapped to _other have no +// individual gauge series to delete; we detect the not-tracked case +// and skip the DeleteLabelValues calls so we don't tear down the +// shared _other series for an unrelated queue. // // Caller-audit per the standing semantic-change rule: only // SQSObserver.observeOnce calls this (registry plumbing aside), // and it's invoked exactly when a queue is observed in the -// previous tick but not the current one — symmetric with -// ObserveQueueDepth's add side, so there is no path that calls -// ForgetQueue without a matching prior tracked-queue insert. +// previous tick but not the current one. The caller's contract — +// "drop gauges for a queue that disappeared so dashboards don't +// show frozen backlog" — is preserved. The narrowed scope (counter +// budget no longer reclaimed) is invisible to the observer because +// observeOnce only writes gauges; the counter side is fed by +// ObservePartitionMessage from a different code path entirely. func (m *SQSMetrics) ForgetQueue(queue string) { if m == nil || queue == "" { return } m.mu.Lock() - _, tracked := m.trackedQueues[queue] + _, tracked := m.trackedDepthQueues[queue] if tracked { - delete(m.trackedQueues, queue) + delete(m.trackedDepthQueues, queue) } m.mu.Unlock() if !tracked { - // Queue was either never observed or had been collapsed onto - // the _other overflow label. Either way: no per-queue series - // to remove, and we must NOT delete the _other series here - // because other overflow queues may still be sharing it. + // Queue was either never depth-observed or had been collapsed + // onto the _other overflow label. Either way: no per-queue + // gauge series to remove, and we must NOT delete the _other + // series here because other overflow queues may still be + // sharing it. return } m.queueDepth.DeleteLabelValues(queue, sqsQueueStateVisible) @@ -204,22 +240,40 @@ func (m *SQSMetrics) ForgetQueue(queue string) { m.queueDepth.DeleteLabelValues(queue, sqsQueueStateDelayed) } -// queueLabelForCardinalityBudget returns queue if the metric has -// already emitted a series for it OR there is room in the -// tracked-queues set; returns sqsQueueOverflow otherwise. The -// cap-and-collapse pattern mirrors DynamoDBMetrics.tableLabel -// so a misbehaving caller cannot exhaust the Prometheus -// cardinality budget. -func (m *SQSMetrics) queueLabelForCardinalityBudget(queue string) string { +// admitForCounterBudget returns the canonical label for queue: the +// real name when the counter budget has room, sqsQueueOverflow once +// it is saturated. The counter budget never shrinks (counter series +// are cumulative — Prometheus has no semantically-clean delete), so +// this is a one-way admission: once a queue gets in, it stays. +// Mirrors DynamoDBMetrics.tableLabel. +func (m *SQSMetrics) admitForCounterBudget(queue string) string { + m.mu.Lock() + defer m.mu.Unlock() + if _, ok := m.trackedCounterQueues[queue]; ok { + return queue + } + if len(m.trackedCounterQueues) >= sqsMaxTrackedQueues { + return sqsQueueOverflow + } + m.trackedCounterQueues[queue] = struct{}{} + return queue +} + +// admitForDepthBudget mirrors admitForCounterBudget for the gauge +// budget. Unlike the counter side, slots here can be freed via +// ForgetQueue, so a deployment that creates and deletes queues over +// time can reuse budget for new queue names instead of permanently +// exhausting it. +func (m *SQSMetrics) admitForDepthBudget(queue string) string { m.mu.Lock() defer m.mu.Unlock() - if _, ok := m.trackedQueues[queue]; ok { + if _, ok := m.trackedDepthQueues[queue]; ok { return queue } - if len(m.trackedQueues) >= sqsMaxTrackedQueues { + if len(m.trackedDepthQueues) >= sqsMaxTrackedQueues { return sqsQueueOverflow } - m.trackedQueues[queue] = struct{}{} + m.trackedDepthQueues[queue] = struct{}{} return queue } diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index cc680944c..965804646 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -202,14 +202,16 @@ func TestSQSMetrics_ForgetQueue_DropsThreeSeries(t *testing.T) { m.ObserveQueueDepth("orders.fifo", 3, 0, 0) m.ObservePartitionMessage("orders.fifo", 0, SQSPartitionActionSend) - require.Len(t, m.trackedQueues, 1, "queue must have been added to the cardinality budget on Observe") + require.Len(t, m.trackedDepthQueues, 1, "queue must have been admitted to the depth budget on ObserveQueueDepth") + require.Len(t, m.trackedCounterQueues, 1, "queue must have been admitted to the counter budget on ObservePartitionMessage") m.ForgetQueue("orders.fifo") count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") require.NoError(t, err) require.Equal(t, 0, count, "ForgetQueue must drop every state series for the queue") - require.Empty(t, m.trackedQueues, "ForgetQueue must free the cardinality-budget slot") + require.Empty(t, m.trackedDepthQueues, "ForgetQueue must free the depth-side cardinality slot") + require.Len(t, m.trackedCounterQueues, 1, "ForgetQueue must NOT free the counter-side slot (counters are cumulative)") require.InDelta(t, 1.0, testutil.ToFloat64(m.partitionMessages.WithLabelValues("orders.fifo", "0", SQSPartitionActionSend)), 0.001, "counter must survive ForgetQueue (cumulative metric)") @@ -238,7 +240,7 @@ func TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp(t *testing.T) { for i := 0; i < sqsMaxTrackedQueues; i++ { m.ObserveQueueDepth("real-"+strconv.Itoa(i)+".fifo", 1, 0, 0) } - require.Len(t, m.trackedQueues, sqsMaxTrackedQueues) + require.Len(t, m.trackedDepthQueues, sqsMaxTrackedQueues) // Two overflow queues — both share the _other label. m.ObserveQueueDepth("overflow-a.fifo", 100, 0, 0) @@ -250,14 +252,94 @@ func TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp(t *testing.T) { // ForgetQueue on an overflow queue must leave the _other series // (and the budget) untouched. m.ForgetQueue("overflow-a.fifo") - require.Len(t, m.trackedQueues, sqsMaxTrackedQueues, - "ForgetQueue on an overflow queue must not change the budget") + require.Len(t, m.trackedDepthQueues, sqsMaxTrackedQueues, + "ForgetQueue on an overflow queue must not change the depth budget") require.InDelta(t, 200.0, testutil.ToFloat64(m.queueDepth.WithLabelValues(sqsQueueOverflow, sqsQueueStateVisible)), 0.001, "_other series must still carry the latest value (overflow-b)") } +// TestSQSMetrics_ForgetQueue_DoesNotReclaimCounterBudget pins the +// P1 found in PR #743 review: ForgetQueue must NOT free a counter- +// side cardinality budget slot. The (queue, partition, action) +// counter series is cumulative and cannot be deleted from +// Prometheus without losing its observed value, so once a queue +// has been admitted to the counter budget the slot must stay +// permanently consumed — even if the queue is dropped from the +// gauge side. +// +// Pre-fix bug: trackedQueues was a single shared map. ForgetQueue +// would delete the entry, freeing the slot. A subsequent +// ObservePartitionMessage on a NEW queue then got admitted with a +// real label, even though the old queue's counter series was still +// alive in Prometheus. Repeating this (e.g. leader step-down clears +// lastSeen → ForgetQueue called for every queue → 512 fresh queues +// arrive and get admitted) lets counter cardinality grow unbounded. +// +// Post-fix expectation: the counter budget is one-way. Once +// saturated, every later queue collapses to _other regardless of +// how many ForgetQueue calls have run. +func TestSQSMetrics_ForgetQueue_DoesNotReclaimCounterBudget(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + // Saturate the counter budget with sqsMaxTrackedQueues distinct + // queue names — every one is admitted as a real label. + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ObservePartitionMessage("orig-"+strconv.Itoa(i)+".fifo", 0, SQSPartitionActionSend) + } + + // Drop every queue via ForgetQueue. This used to free the + // budget slot for every queue, including ones that had only + // counter activity (depth side never touched them — pre-fix + // the check was non-distinguishing). + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ForgetQueue("orig-" + strconv.Itoa(i) + ".fifo") + } + + // New queue arrives. Pre-fix this would be admitted as a real + // label because the slot looked free. Post-fix the counter + // budget is still saturated (counters were never deleted), so + // the new queue must collapse to _other. + m.ObservePartitionMessage("new-after-forget.fifo", 0, SQSPartitionActionSend) + + require.InDelta(t, 0.0, + testutil.ToFloat64(m.partitionMessages.WithLabelValues("new-after-forget.fifo", "0", SQSPartitionActionSend)), + 0.001, + "counter for the new queue must not exist under its real name — budget should be exhausted") + require.InDelta(t, 1.0, + testutil.ToFloat64(m.partitionMessages.WithLabelValues(sqsQueueOverflow, "0", SQSPartitionActionSend)), + 0.001, + "new queue past the counter cap must collapse to _other") +} + +// TestSQSMetrics_ForgetQueue_StillReclaimsDepthBudget pins the +// converse: the gauge side IS reclaimable, because gauges can be +// deleted from Prometheus without losing semantically-meaningful +// state. A churn-heavy deployment that observes depth for 512 +// distinct queues, then forgets all of them, must be able to +// observe a 513th distinct queue under its real name. +func TestSQSMetrics_ForgetQueue_StillReclaimsDepthBudget(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ObserveQueueDepth("orig-"+strconv.Itoa(i)+".fifo", 1, 0, 0) + } + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ForgetQueue("orig-" + strconv.Itoa(i) + ".fifo") + } + m.ObserveQueueDepth("fresh.fifo", 42, 0, 0) + + require.InDelta(t, 42.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues("fresh.fifo", sqsQueueStateVisible)), + 0.001, + "gauge budget must reclaim slots so a churn-heavy deployment can keep emitting real labels") +} + // TestSQSMetrics_DepthNilReceiverIsSafe mirrors the partition test: // a typed-nil *SQSMetrics handed to the adapter must not panic on // either ObserveQueueDepth or ForgetQueue calls. From 9738f75c4e8e5ba5d93bbf0542c1bb48e6bc2783 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 06:10:28 +0900 Subject: [PATCH 05/10] monitoring/sqs: drop _other gauge when overflow set empties (PR #743 r4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 (Codex review): ForgetQueue leaves phantom _other backlog after overflow queues vanish. When the depth budget saturates, ObserveQueueDepth collapses extra queues onto the shared sqsQueueOverflow ('_other') label. ForgetQueue early-returned for these because they were not in trackedDepthQueues — which lumps two distinct cases together: - never depth-observed (no series to delete; correct no-op) - currently mapped to _other (series exists, shared, must drop only when no overflow queue is still reporting into it) In a churn scenario with >sqsMaxTrackedQueues distinct queues the final overflow queue's last value sits on the dashboard until another overflow queue overwrites it — phantom backlog for queues that no longer exist. Fix: track overflow membership in a third map (overflowDepthQueues) so ForgetQueue can ref-count overflow queues. Two-branch behaviour: - real-name queue: drop the three state series + free the trackedDepthQueues slot (unchanged) - overflow queue: remove from overflowDepthQueues; if the set is now empty, drop the three _other state series too; otherwise leave _other alone so other overflow queues keep reporting into it Counter side has no equivalent map. Counters are cumulative — the _other counter legitimately reflects total operations even from queues that have since been deleted, so there's no phantom-data problem to solve there. Caller audit per the standing semantic-change rule: ForgetQueue: only caller is SQSObserver.observeOnce. The observer's contract ('drop gauges for queue that disappeared from this tick's snapshot') is preserved and consistently extended — overflow queues, previously a silent no-op, now also stop pinning the shared gauge once the last one disappears. The branching is internal to ForgetQueue; the caller passes the queue name verbatim and is unaware of which map it lived in. admitForDepthBudget: signature unchanged. New side effect of populating overflowDepthQueues when collapsing to _other. Only caller is ObserveQueueDepth; the side effect is invisible from the caller's POV (still receives the same string label). Regression test: TestSQSMetrics_ForgetQueue_LastOverflowClearsOtherGauge saturates the depth budget, observes two overflow queues, ForgetQueue's both, and asserts the _other series is gone (verified: pre-fix the gauge GatherAndCount returns 1539, post-fix it returns 1536 — exactly the 3 _other series dropped). Existing TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp continues to pass — ForgetQueue on one of N>=2 overflow queues still preserves the _other series for the remaining ones, which was the original invariant. --- monitoring/sqs.go | 99 +++++++++++++++++++++++++++++++----------- monitoring/sqs_test.go | 63 +++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 25 deletions(-) diff --git a/monitoring/sqs.go b/monitoring/sqs.go index a72c11118..4a7873263 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -106,6 +106,21 @@ type SQSMetrics struct { mu sync.Mutex trackedCounterQueues map[string]struct{} trackedDepthQueues map[string]struct{} + // overflowDepthQueues is the set of queue names whose depth + // gauge is currently collapsed onto the shared sqsQueueOverflow + // label. Tracked separately from trackedDepthQueues (which + // only holds real-name admissions) so ForgetQueue can ref-count + // overflow queues and drop the shared _other gauge once the + // last overflow queue disappears. Without this, churn-heavy + // deployments with >sqsMaxTrackedQueues distinct queues leave + // the dashboard pinned at whichever overflow queue last + // reported, even after every overflow queue has been deleted. + // Counter side has no equivalent map: counter series are + // cumulative (never deleted), so the _other counter can't + // produce phantom data — it correctly reflects the cumulative + // count of all overflow operations, even from queues that no + // longer exist. + overflowDepthQueues map[string]struct{} } // SQS depth gauge state-label values. Stable so dashboards / alerts @@ -134,6 +149,7 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics { ), trackedCounterQueues: map[string]struct{}{}, trackedDepthQueues: map[string]struct{}{}, + overflowDepthQueues: map[string]struct{}{}, } registerer.MustRegister(m.partitionMessages) registerer.MustRegister(m.queueDepth) @@ -190,33 +206,47 @@ func (m *SQSMetrics) ObserveQueueDepth(queue string, visible, notVisible, delaye // its slot in the depth-side cardinality budget so a long-running // deployment that regularly creates and deletes queues (CI // workloads, ephemeral per-job queues) doesn't permanently wedge -// the 512-entry depth budget. Without the trackedDepthQueues -// cleanup, post-cap new queues would silently collapse onto the -// _other label even after their predecessors had been deleted. +// the 512-entry depth budget. // -// The (queue, partition, action) counter series stays — -// cumulative-by-design — and the queue's slot in +// Three cases, by membership at call time: +// +// 1. Queue is in trackedDepthQueues (admitted under its real name): +// drop the three state-labelled series and free the budget slot. +// 2. Queue is in overflowDepthQueues (admitted, then collapsed +// onto the shared sqsQueueOverflow / "_other" label because the +// budget was saturated): remove from the overflow set. If that +// leaves the overflow set empty, drop the three _other series +// too — there's no remaining queue reporting into them, so the +// gauge would otherwise pin phantom backlog. While the overflow +// set is still non-empty the _other series stays put: tearing +// it down would zero out values that other overflow queues are +// legitimately maintaining. +// 3. Queue is in neither map (never depth-observed): no-op. +// +// The (queue, partition, action) counter series stays in all three +// cases — cumulative-by-design — and the queue's slot in // trackedCounterQueues stays consumed. Reclaiming the counter slot // would let a later queue be admitted under its real name while the // original counter series still sat in Prometheus, which would let // counter cardinality grow past sqsMaxTrackedQueues under churn or // after a leader step-down clears the observer's lastSeen (the P1 -// finding on PR #743). -// -// Queues that hit the depth cap and mapped to _other have no -// individual gauge series to delete; we detect the not-tracked case -// and skip the DeleteLabelValues calls so we don't tear down the -// shared _other series for an unrelated queue. +// finding on PR #743). Counter-side _other has no equivalent of +// the gauge phantom-backlog problem either, because cumulative +// counters legitimately reflect total operations from queues that +// have since been deleted. // // Caller-audit per the standing semantic-change rule: only // SQSObserver.observeOnce calls this (registry plumbing aside), // and it's invoked exactly when a queue is observed in the // previous tick but not the current one. The caller's contract — // "drop gauges for a queue that disappeared so dashboards don't -// show frozen backlog" — is preserved. The narrowed scope (counter -// budget no longer reclaimed) is invisible to the observer because -// observeOnce only writes gauges; the counter side is fed by -// ObservePartitionMessage from a different code path entirely. +// show frozen backlog" — is preserved AND extended consistently: +// overflow queues, previously a silent no-op, now also stop +// pinning the shared gauge once the last one disappears. The +// narrowed-but-still-correct scope (counter budget never +// reclaimed) is invisible to observeOnce because the observer +// only writes gauges; counters are fed via ObservePartitionMessage +// from a different code path entirely. func (m *SQSMetrics) ForgetQueue(queue string) { if m == nil || queue == "" { return @@ -226,18 +256,31 @@ func (m *SQSMetrics) ForgetQueue(queue string) { if tracked { delete(m.trackedDepthQueues, queue) } + // Ref-count the overflow set. The shared _other gauge series is + // safe to drop only when no queue is still reporting into it; + // while one or more overflow queues remain, the gauge carries + // real (last-write-wins) data for those queues. + _, overflow := m.overflowDepthQueues[queue] + if overflow { + delete(m.overflowDepthQueues, queue) + } + overflowSetEmpty := overflow && len(m.overflowDepthQueues) == 0 m.mu.Unlock() - if !tracked { - // Queue was either never depth-observed or had been collapsed - // onto the _other overflow label. Either way: no per-queue - // gauge series to remove, and we must NOT delete the _other - // series here because other overflow queues may still be - // sharing it. - return + if tracked { + m.queueDepth.DeleteLabelValues(queue, sqsQueueStateVisible) + m.queueDepth.DeleteLabelValues(queue, sqsQueueStateNotVisible) + m.queueDepth.DeleteLabelValues(queue, sqsQueueStateDelayed) + } + if overflowSetEmpty { + // Last overflow queue forgotten — drop the shared _other + // series so dashboards stop showing phantom backlog for + // queues that no longer exist. Safe because no remaining + // queue is mapped to this label; any future overflow + // queue will re-create the series via ObserveQueueDepth. + m.queueDepth.DeleteLabelValues(sqsQueueOverflow, sqsQueueStateVisible) + m.queueDepth.DeleteLabelValues(sqsQueueOverflow, sqsQueueStateNotVisible) + m.queueDepth.DeleteLabelValues(sqsQueueOverflow, sqsQueueStateDelayed) } - m.queueDepth.DeleteLabelValues(queue, sqsQueueStateVisible) - m.queueDepth.DeleteLabelValues(queue, sqsQueueStateNotVisible) - m.queueDepth.DeleteLabelValues(queue, sqsQueueStateDelayed) } // admitForCounterBudget returns the canonical label for queue: the @@ -264,6 +307,11 @@ func (m *SQSMetrics) admitForCounterBudget(queue string) string { // ForgetQueue, so a deployment that creates and deletes queues over // time can reuse budget for new queue names instead of permanently // exhausting it. +// +// Queues that hit the cap are recorded in overflowDepthQueues so +// ForgetQueue can ref-count them and tear down the shared _other +// gauge once the last one disappears (otherwise the dashboard +// pins phantom backlog for queues that no longer exist). func (m *SQSMetrics) admitForDepthBudget(queue string) string { m.mu.Lock() defer m.mu.Unlock() @@ -271,6 +319,7 @@ func (m *SQSMetrics) admitForDepthBudget(queue string) string { return queue } if len(m.trackedDepthQueues) >= sqsMaxTrackedQueues { + m.overflowDepthQueues[queue] = struct{}{} return sqsQueueOverflow } m.trackedDepthQueues[queue] = struct{}{} diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index 965804646..b07121beb 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -260,6 +260,69 @@ func TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp(t *testing.T) { "_other series must still carry the latest value (overflow-b)") } +// TestSQSMetrics_ForgetQueue_LastOverflowClearsOtherGauge pins the +// P2 found in PR #743 review: when the depth budget is saturated +// and queues collapse onto the shared sqsQueueOverflow ("_other") +// label, ForgetQueue must drop the _other gauge series once the +// LAST overflow queue disappears. Otherwise a churn-heavy +// deployment with >512 queues leaves the dashboard pinned at +// whichever overflow queue last reported, even after every +// overflow queue has been deleted — phantom backlog for queues +// that no longer exist. +// +// Pre-fix bug: ForgetQueue early-returned for any queue not in +// trackedDepthQueues, which lumps together "never observed" (no +// series to delete) with "currently collapsed onto _other" +// (series exists, shared, must drop only when ref-count hits 0). +// +// Post-fix expectation: track overflow queue membership +// separately from trackedDepthQueues so ForgetQueue can +// ref-count overflow queues and clear the shared _other gauge +// the moment the set becomes empty. +// +// The single-overflow-not-empty case (don't tear down _other +// while other overflow queues still report into it) is pinned +// by TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp below. +func TestSQSMetrics_ForgetQueue_LastOverflowClearsOtherGauge(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + // Saturate the depth budget with sqsMaxTrackedQueues real queues + // so subsequent observations collapse onto _other. + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ObserveQueueDepth("real-"+strconv.Itoa(i)+".fifo", 1, 0, 0) + } + + // Two overflow queues — both share the _other label. + m.ObserveQueueDepth("overflow-a.fifo", 100, 0, 0) + m.ObserveQueueDepth("overflow-b.fifo", 200, 0, 0) + require.InDelta(t, 200.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues(sqsQueueOverflow, sqsQueueStateVisible)), + 0.001, + "sanity: _other gauge carries the latest overflow value") + + // Forget the first overflow queue. _other must remain — the + // other overflow queue is still reporting into it. + m.ForgetQueue("overflow-a.fifo") + require.InDelta(t, 200.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues(sqsQueueOverflow, sqsQueueStateVisible)), + 0.001, + "_other must survive ForgetQueue while at least one overflow queue is still alive") + + // Forget the last overflow queue. _other must now be gone: + // no overflow queue exists, so the gauge would only be showing + // phantom data for queues that no longer exist. + m.ForgetQueue("overflow-b.fifo") + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + // 3 state-labelled series per queue. Only the original tracked + // queues should remain — _other (visible/not_visible/delayed) + // must be dropped now that the overflow set is empty. + require.Equal(t, sqsMaxTrackedQueues*3, count, + "only the 3 state series per real queue should remain after the last overflow queue is forgotten") +} + // TestSQSMetrics_ForgetQueue_DoesNotReclaimCounterBudget pins the // P1 found in PR #743 review: ForgetQueue must NOT free a counter- // side cardinality budget slot. The (queue, partition, action) From 7b5ab13b6a69b27d312b6eb10681598e86484242 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 06:22:05 +0900 Subject: [PATCH 06/10] monitoring/sqs: clean overflow set on real-label promotion (PR #743 r5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second P2 (Codex review): admitForDepthBudget leaves stale entries in overflowDepthQueues when a queue gets promoted off the _other label. Scenario: budget saturates, queue X collapses to _other. Later a slot opens (some other queue is ForgetQueue'd). On X's next observation admitForDepthBudget finds room and admits X under its real name — but the previous fix only added X to trackedDepthQueues. The stale entry in overflowDepthQueues was never cleaned up, so: 1. ForgetQueue's overflow ref-count is permanently off by one per stale entry — _other can persist after every live queue has been promoted off it. 2. The _other series itself still carries X's last-overflow value while X is now reporting under its real label. Dashboards see double-counted backlog. Fix: in admitForDepthBudget's slot-available branch, also remove the queue from overflowDepthQueues. If that drains the overflow set entirely, drop the three _other state series too — mirroring ForgetQueue's existing 'last overflow gone' branch. Refactor: factor the three DeleteLabelValues calls into dropGaugeStatesFor since they now appear in three places (ForgetQueue tracked-branch, ForgetQueue overflow-empty-branch, and the new admitForDepthBudget promotion-drains-overflow branch). Pure readability — Prometheus's GaugeVec already does its own per-vector locking so the wrapper holds no extra lock. Caller audit per the standing semantic-change rule: admitForDepthBudget: signature unchanged. New side effect — drop _other gauge on a promotion that drains the overflow set — invisible to the only caller (ObserveQueueDepth) which still gets back the same string label. ObserveQueueDepth callers: unaffected (ObserveQueueDepth's contract is unchanged). dropGaugeStatesFor: new internal helper, no external callers. ForgetQueue: external semantics preserved; only the DeleteLabelValues calls were collapsed into the helper. Regression test: TestSQSMetrics_AdmitForDepthBudget_PromotionClearsOverflow saturates the budget, observes overflow X, ForgetQueue's a real queue to free a slot, re-observes X, and asserts the _other series is gone. Pre-fix: 1539 series (511 real + X real + _other phantom × 3 states). Post-fix: 1536 (the 3 _other phantom series dropped by the new promotion cleanup). --- monitoring/sqs.go | 41 +++++++++++++++++++++++----- monitoring/sqs_test.go | 61 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 7 deletions(-) diff --git a/monitoring/sqs.go b/monitoring/sqs.go index 4a7873263..684184ee4 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -267,9 +267,7 @@ func (m *SQSMetrics) ForgetQueue(queue string) { overflowSetEmpty := overflow && len(m.overflowDepthQueues) == 0 m.mu.Unlock() if tracked { - m.queueDepth.DeleteLabelValues(queue, sqsQueueStateVisible) - m.queueDepth.DeleteLabelValues(queue, sqsQueueStateNotVisible) - m.queueDepth.DeleteLabelValues(queue, sqsQueueStateDelayed) + m.dropGaugeStatesFor(queue) } if overflowSetEmpty { // Last overflow queue forgotten — drop the shared _other @@ -277,9 +275,7 @@ func (m *SQSMetrics) ForgetQueue(queue string) { // queues that no longer exist. Safe because no remaining // queue is mapped to this label; any future overflow // queue will re-create the series via ObserveQueueDepth. - m.queueDepth.DeleteLabelValues(sqsQueueOverflow, sqsQueueStateVisible) - m.queueDepth.DeleteLabelValues(sqsQueueOverflow, sqsQueueStateNotVisible) - m.queueDepth.DeleteLabelValues(sqsQueueOverflow, sqsQueueStateDelayed) + m.dropGaugeStatesFor(sqsQueueOverflow) } } @@ -312,20 +308,51 @@ func (m *SQSMetrics) admitForCounterBudget(queue string) string { // ForgetQueue can ref-count them and tear down the shared _other // gauge once the last one disappears (otherwise the dashboard // pins phantom backlog for queues that no longer exist). +// +// Promotion path: when budget pressure eases (a slot opened up via +// ForgetQueue) a previously-overflowed queue gets admitted under +// its real name on the next observation. The stale overflow entry +// must be cleaned up at the same moment — otherwise ForgetQueue's +// ref-count is permanently off and the _other gauge can persist +// after every live queue has been promoted off it. If the +// promotion drains the overflow set entirely, drop the three +// _other state series too: they're carrying this queue's last +// overflow value but no queue maps to them anymore. func (m *SQSMetrics) admitForDepthBudget(queue string) string { m.mu.Lock() - defer m.mu.Unlock() if _, ok := m.trackedDepthQueues[queue]; ok { + m.mu.Unlock() return queue } if len(m.trackedDepthQueues) >= sqsMaxTrackedQueues { m.overflowDepthQueues[queue] = struct{}{} + m.mu.Unlock() return sqsQueueOverflow } m.trackedDepthQueues[queue] = struct{}{} + _, wasOverflow := m.overflowDepthQueues[queue] + if wasOverflow { + delete(m.overflowDepthQueues, queue) + } + overflowSetEmpty := wasOverflow && len(m.overflowDepthQueues) == 0 + m.mu.Unlock() + if overflowSetEmpty { + m.dropGaugeStatesFor(sqsQueueOverflow) + } return queue } +// dropGaugeStatesFor removes the three state-labelled series for +// label (a real queue name or sqsQueueOverflow) from the depth +// gauge. Single-line caller-readability wrapper — prometheus +// GaugeVec.DeleteLabelValues already does its own per-vector +// locking, so this is safe to call without holding m.mu. +func (m *SQSMetrics) dropGaugeStatesFor(label string) { + m.queueDepth.DeleteLabelValues(label, sqsQueueStateVisible) + m.queueDepth.DeleteLabelValues(label, sqsQueueStateNotVisible) + m.queueDepth.DeleteLabelValues(label, sqsQueueStateDelayed) +} + // sqsValidPartitionAction returns true iff action is one of the // stable label values. Keeps a typo at the call site (e.g. // "Send" vs "send") from polluting the metric. diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index b07121beb..cdf6d0503 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -260,6 +260,67 @@ func TestSQSMetrics_ForgetQueue_OverflowQueueIsNoOp(t *testing.T) { "_other series must still carry the latest value (overflow-b)") } +// TestSQSMetrics_AdmitForDepthBudget_PromotionClearsOverflow pins +// the second P2 found in PR #743 review: admitForDepthBudget must +// remove a queue from overflowDepthQueues when promoting it from +// the shared _other label to a real-name label, AND drop the +// _other gauge series if the overflow set becomes empty as a +// result. +// +// Pre-fix bug: when budget pressure eased (a slot opened up via +// ForgetQueue) and a previously-overflowed queue was re-observed, +// admitForDepthBudget added it to trackedDepthQueues but left a +// stale entry in overflowDepthQueues. Two consequences: +// +// 1. ForgetQueue's overflow ref-count was permanently off by one +// per stale entry — the _other gauge could persist even when +// no live queue mapped to it. +// 2. The _other series itself still carried the queue's last +// value as overflow, even though that queue was now reporting +// under its real name. Dashboards see double-counted backlog. +// +// Post-fix expectation: a successful promotion both removes the +// queue from overflowDepthQueues and (if that drained the set) +// drops the three _other state series, mirroring ForgetQueue's +// "last overflow gone" branch. +func TestSQSMetrics_AdmitForDepthBudget_PromotionClearsOverflow(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + + // Saturate the depth budget. + for i := 0; i < sqsMaxTrackedQueues; i++ { + m.ObserveQueueDepth("real-"+strconv.Itoa(i)+".fifo", 1, 0, 0) + } + + // Overflow queue X collapses to _other. + m.ObserveQueueDepth("overflow-x.fifo", 100, 0, 0) + require.InDelta(t, 100.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues(sqsQueueOverflow, sqsQueueStateVisible)), + 0.001, "sanity: _other carries X's value while X is mapped to overflow") + + // Free a slot so X can be promoted on the next observation. + m.ForgetQueue("real-0.fifo") + + // Re-observe X. With room in the budget admitForDepthBudget + // must promote it to its real label AND clean up the stale + // overflowDepthQueues entry. + m.ObserveQueueDepth("overflow-x.fifo", 200, 0, 0) + require.InDelta(t, 200.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues("overflow-x.fifo", sqsQueueStateVisible)), + 0.001, "X must now report under its real label after promotion") + + // 3 state series per queue: + // 511 remaining tracked queues (real-1..real-511) + X promoted = 512 × 3 = 1536. + // Pre-fix the _other series persists with X's stale 100 value + // (1539 total). Post-fix the promotion drains the overflow set + // and drops _other. + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, sqsMaxTrackedQueues*3, count, + "_other series must be dropped when promotion drains the overflow set") +} + // TestSQSMetrics_ForgetQueue_LastOverflowClearsOtherGauge pins the // P2 found in PR #743 review: when the depth budget is saturated // and queues collapse onto the shared sqsQueueOverflow ("_other") From 1651db7305ebdd76331250b6a673e4e12ff8e33d Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 06:58:28 +0900 Subject: [PATCH 07/10] monitoring/sqs: distinguish skip-tick from empty snapshot (PR #743 r6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 (Codex review): SnapshotQueueDepths returns nil on scanQueueNames failure, observeOnce treats nil as 'no queues exist now' and ForgetQueue's every previously-seen queue. A single transient catalog-read error therefore wipes every depth gauge and dashboard-renders as a false 'all queues drained' event for the duration of the failure. Two empty-snapshot cases were conflated: - ok=true with empty/nil — legitimate empty (follower because the node stepped down, leader with zero queues configured). Observer should diff/ForgetQueue normally. - scan failure / ctx cancel — observer should NOT diff. Existing gauges and lastSeen must stay intact so the next successful tick can still diff against the previous good state. Fix: extend SQSDepthSource.SnapshotQueueDepths from `[]X` to `([]X, bool)` so the source can signal 'skip this tick' distinctly from 'empty but valid'. Caller audit per the standing semantic-change rule: monitoring.SQSDepthSource (interface) — sig changed; doc enumerates the three return shapes. monitoring.SQSObserver.observeOnce — added `if !ok { return }` short-circuit before the gauge writes / diff. Other branches unchanged. adapter.SQSServer.SnapshotQueueDepths — three explicit returns: (nil, true) for nil receiver / follower (legitimate 'this node should not emit'); (nil, false) for scanQueueNames error or ctx cancel mid-scan; (snaps, true) for the success path. Per-queue snapshotOneQueueDepth failures still drop only that queue from this tick (ok stays true) — the observer ForgetQueue's just that one gauge. main.sqsDepthSourceAdapter.SnapshotQueueDepths — bridge propagates the new ok flag verbatim through the adapter→monitoring type translation. monitoring.fakeDepthSource (test helper) — replaced `snapshots [][]X` with `ticks []fakeDepthTick` where each tick scripts (snaps, ok). Existing tests migrated; trailing ticks past the scripted slice still default to (nil, true) so leader-step-down clears gauges as before. Regression test: TestSQSObserver_ObserveOnce_TransientScanErrorPreservesGauges scripts three ticks: success → ok=false (scan fail) → success-recovered. Asserts gauges keep tick-1 values through the failed tick AND that recovery on tick 3 still diffs correctly. Pre-fix shape would wipe both queues on tick 2. --- adapter/sqs_depth_source.go | 51 ++++++++++------ main.go | 17 ++++-- monitoring/sqs.go | 38 ++++++++++-- monitoring/sqs_test.go | 114 ++++++++++++++++++++++++++++++------ 4 files changed, 174 insertions(+), 46 deletions(-) diff --git a/adapter/sqs_depth_source.go b/adapter/sqs_depth_source.go index 32e14fe6d..f375b58c3 100644 --- a/adapter/sqs_depth_source.go +++ b/adapter/sqs_depth_source.go @@ -18,28 +18,39 @@ type SQSQueueDepth struct { } // SnapshotQueueDepths satisfies monitoring.SQSDepthSource. The -// observer Start loop calls this on every tick; the SQSServer -// returns one entry per known queue when this node is the verified -// Raft leader, or an empty slice on followers. Leader-only -// emission keeps the dashboard's queue-depth gauges consistent -// with what AdminListQueues / AdminDescribeQueue would return at -// the same instant — followers that scanned the catalog at the -// same time would race the leader's writes and emit conflicting -// values for the same series. +// observer Start loop calls this on every tick. // -// Per-queue scan errors are logged and the offending queue is -// dropped from this tick's snapshot. The observer detects the -// disappearance and ForgetQueue's the gauges so the dashboard -// surfaces "scrape failed" as a missing series rather than as a -// pinned stale backlog. -func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth { +// Returns: +// +// - (snaps, true) — leader, scrape OK. Observer writes snaps to +// the gauges and diffs against the previous tick (forgetting +// any queue that disappeared from this snapshot). +// - (nil, true) — this node is a follower (leader-only emission +// keeps gauges consistent with AdminListQueues / AdminDescribeQueue +// at the same instant — follower scans would race the leader's +// writes). Empty-but-OK so the observer ForgetQueue's any +// gauges this node was emitting before stepping down. +// - (nil, false) — leader, but scrape failed (transient +// catalog-read error or ctx cancel mid-scan). Tells the +// observer to skip this tick: leave existing gauges in place +// rather than wiping every depth series — a single failed +// scrape would otherwise dashboard-render as a false "all +// queues drained" event until the next successful tick. +// +// Per-queue scan errors (loadQueueMetaAt / scanApproxCounters) +// remain handled in-line by snapshotOneQueueDepth: the offending +// queue is dropped from this tick's snapshot but ok stays true, +// so the observer ForgetQueue's just that one queue's gauges. +// Only a top-level scanQueueNames failure (which would silently +// turn into "no queues anywhere") flips ok to false. +func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) ([]SQSQueueDepth, bool) { if s == nil || s.coordinator == nil || s.store == nil || !s.coordinator.IsLeader() { - return nil + return nil, true } names, err := s.scanQueueNames(ctx) if err != nil { slog.Warn("sqs depth snapshot: scanQueueNames failed", "err", err) - return nil + return nil, false } // Take a single read timestamp for the whole tick so all queues // in this snapshot share the same MVCC view. With per-queue @@ -51,13 +62,17 @@ func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth { out := make([]SQSQueueDepth, 0, len(names)) for _, name := range names { if err := ctx.Err(); err != nil { - return out + // ctx cancel mid-iteration: partial snapshot is + // useless because the observer would diff against it + // and ForgetQueue everything we hadn't reached yet. + // Signal skip-tick instead. + return nil, false } if snap, ok := s.snapshotOneQueueDepth(ctx, name, readTS); ok { out = append(out, snap) } } - return out + return out, true } // snapshotOneQueueDepth runs the per-queue catalog read pair diff --git a/main.go b/main.go index 201ba1755..ffa6b1842 100644 --- a/main.go +++ b/main.go @@ -1156,13 +1156,20 @@ type sqsDepthSourceAdapter struct { inner *adapter.SQSServer } -func (a sqsDepthSourceAdapter) SnapshotQueueDepths(ctx context.Context) []monitoring.SQSQueueDepth { +func (a sqsDepthSourceAdapter) SnapshotQueueDepths(ctx context.Context) ([]monitoring.SQSQueueDepth, bool) { if a.inner == nil { - return nil + // Empty-but-OK: nothing to emit. Mirrors the + // follower / nil-receiver case of the underlying source. + return nil, true + } + snaps, ok := a.inner.SnapshotQueueDepths(ctx) + if !ok { + // Propagate skip-tick verbatim so the observer leaves + // existing gauges alone on a transient scan failure. + return nil, false } - snaps := a.inner.SnapshotQueueDepths(ctx) if len(snaps) == 0 { - return nil + return nil, true } out := make([]monitoring.SQSQueueDepth, len(snaps)) for i, s := range snaps { @@ -1173,7 +1180,7 @@ func (a sqsDepthSourceAdapter) SnapshotQueueDepths(ctx context.Context) []monito Delayed: s.Delayed, } } - return out + return out, true } // writeConflictMonitorSources extracts the MVCC stores that expose diff --git a/monitoring/sqs.go b/monitoring/sqs.go index 684184ee4..fa8a49700 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -53,11 +53,29 @@ type SQSPartitionObserver interface { // Mirrors the Raft observer's StatusReader / ConfigReader pattern // (monitoring/raft.go): the source returns ready-to-use snapshots // and the observer owns the gauge state machine (forget-on-disappear, -// cardinality cap). Implementations return an empty slice — not an -// error — when this node is a follower, so the dashboard's gauge -// set always mirrors what the leader's catalog scan would report. +// cardinality cap). +// +// Two distinct empty-snapshot states, signalled by ok: +// +// - ok=true with an empty/nil slice — "this source legitimately +// has no queues this tick". Triggers when the node is a +// follower (leader-only emission) or the leader genuinely has +// zero queues configured. The observer diffs against the +// previous tick and ForgetQueue's any queue that disappeared +// so a former leader's gauges are cleared on step-down. +// +// - ok=false (regardless of the slice contents) — "the source +// could not produce a snapshot this tick" (transient catalog- +// read failure on the leader, context cancelled mid-scan). +// The observer must skip the diff entirely: leave existing +// gauges in place AND leave lastSeen untouched so the next +// successful tick can still diff against the previous good +// state. Without this branch a single failed scrape would +// wipe every depth gauge and produce a false "all queues +// drained" event on the dashboard until the next successful +// tick. type SQSDepthSource interface { - SnapshotQueueDepths(ctx context.Context) []SQSQueueDepth + SnapshotQueueDepths(ctx context.Context) (snaps []SQSQueueDepth, ok bool) } // SQSQueueDepth is one queue's depth-attribute snapshot. Mirrors @@ -438,7 +456,17 @@ func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) { if o == nil || o.metrics == nil || source == nil { return } - snaps := source.SnapshotQueueDepths(ctx) + snaps, ok := source.SnapshotQueueDepths(ctx) + if !ok { + // Source signalled "skip this tick" (transient catalog-scan + // failure on the leader, ctx cancel mid-scan). Leave + // existing gauges + lastSeen untouched so the dashboard + // keeps the last successful snapshot rather than rendering + // a false "all queues drained" event for the duration of + // the failure. The next successful tick will diff against + // the same lastSeen we leave behind here. + return + } current := make(map[string]struct{}, len(snaps)) for _, snap := range snaps { o.metrics.ObserveQueueDepth(snap.Queue, snap.Visible, snap.NotVisible, snap.Delayed) diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index cdf6d0503..1b5f94a51 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -504,21 +504,32 @@ func TestSQSMetrics_DepthRegistryWiring(t *testing.T) { "elastickv_sqs_queue_messages must be registered on the public Registry") } +// fakeDepthTick is one scripted return from fakeDepthSource. ok +// mirrors the SQSDepthSource contract: false models a transient +// scan failure that should NOT cause the observer to wipe gauges. +type fakeDepthTick struct { + snaps []SQSQueueDepth + ok bool +} + // fakeDepthSource lets the SQSObserver tests script the per-tick // snapshot without standing up a real SQS adapter + coordinator. +// Past the scripted ticks, returns the empty-OK sentinel so trailing +// observeOnce calls drain to a clean state rather than blowing up +// with an out-of-range index. type fakeDepthSource struct { - snapshots [][]SQSQueueDepth - calls int + ticks []fakeDepthTick + calls int } -func (f *fakeDepthSource) SnapshotQueueDepths(_ context.Context) []SQSQueueDepth { - if f.calls >= len(f.snapshots) { +func (f *fakeDepthSource) SnapshotQueueDepths(_ context.Context) ([]SQSQueueDepth, bool) { + if f.calls >= len(f.ticks) { f.calls++ - return nil + return nil, true } - out := f.snapshots[f.calls] + t := f.ticks[f.calls] f.calls++ - return out + return t.snaps, t.ok } // TestSQSObserver_ObserveOnce_EmitsAndForgets pins the observer's @@ -533,16 +544,17 @@ func TestSQSObserver_ObserveOnce_EmitsAndForgets(t *testing.T) { obs := newSQSObserver(m) source := &fakeDepthSource{ - snapshots: [][]SQSQueueDepth{ - // tick 1: two queues - { + ticks: []fakeDepthTick{ + // tick 1: two queues, scrape OK. + {snaps: []SQSQueueDepth{ {Queue: "orders.fifo", Visible: 5, NotVisible: 2, Delayed: 0}, {Queue: "audio.fifo", Visible: 0, NotVisible: 0, Delayed: 3}, - }, - // tick 2: orders.fifo disappeared - { + }, ok: true}, + // tick 2: orders.fifo disappeared from the snapshot, + // scrape OK. + {snaps: []SQSQueueDepth{ {Queue: "audio.fifo", Visible: 1, NotVisible: 0, Delayed: 0}, - }, + }, ok: true}, }, } @@ -571,12 +583,15 @@ func TestSQSObserver_ObserveOnce_LeaderStepDownClearsAll(t *testing.T) { obs := newSQSObserver(m) source := &fakeDepthSource{ - snapshots: [][]SQSQueueDepth{ - { + ticks: []fakeDepthTick{ + {snaps: []SQSQueueDepth{ {Queue: "orders.fifo", Visible: 5}, {Queue: "audio.fifo", Visible: 1}, - }, - nil, // leader stepped down + }, ok: true}, + // Leader stepped down: empty snapshot, but ok=true so + // the observer ForgetQueue's the gauges from tick 1 + // rather than skipping the diff. + {snaps: nil, ok: true}, }, } obs.ObserveOnce(source) @@ -590,6 +605,69 @@ func TestSQSObserver_ObserveOnce_LeaderStepDownClearsAll(t *testing.T) { require.Equal(t, 0, count, "tick 2 (leader step-down): all gauges cleared") } +// TestSQSObserver_ObserveOnce_TransientScanErrorPreservesGauges +// pins the P2 fix from PR #743 r6: when the source returns +// ok=false (transient catalog-scan failure on the leader, ctx +// cancel mid-scan), the observer must NOT diff against the +// previous tick or call ForgetQueue. The existing gauges should +// keep their last successful values until a later successful +// scrape — otherwise a single failed catalog read renders a +// false "all queues drained" event on the dashboard for the +// entire duration of the failure. +// +// Pre-fix shape: SnapshotQueueDepths returned bare nil on scan +// failure, observeOnce treated bare nil as "no queues exist now" +// and ForgetQueue'd every previously-seen queue. Post-fix shape: +// the two cases are distinguishable via the ok return; observer +// short-circuits when ok=false and leaves both the gauges and +// the lastSeen map untouched. +func TestSQSObserver_ObserveOnce_TransientScanErrorPreservesGauges(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + obs := newSQSObserver(m) + + source := &fakeDepthSource{ + ticks: []fakeDepthTick{ + // tick 1: two queues, scrape OK. + {snaps: []SQSQueueDepth{ + {Queue: "orders.fifo", Visible: 5}, + {Queue: "audio.fifo", Visible: 1}, + }, ok: true}, + // tick 2: scan failed — observer must skip. + {snaps: nil, ok: false}, + // tick 3: scan recovered, both queues still present. + {snaps: []SQSQueueDepth{ + {Queue: "orders.fifo", Visible: 7}, + {Queue: "audio.fifo", Visible: 2}, + }, ok: true}, + }, + } + + obs.ObserveOnce(source) + require.InDelta(t, 5.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateVisible)), 0.001) + require.InDelta(t, 1.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("audio.fifo", sqsQueueStateVisible)), 0.001) + + obs.ObserveOnce(source) + // Pre-fix: orders/audio gauges would be wiped via ForgetQueue. + // Post-fix: gauges still carry tick-1 values. + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, 6, count, "transient scan failure must preserve existing gauges (2 queues × 3 states = 6)") + require.InDelta(t, 5.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateVisible)), 0.001, + "orders.fifo gauge must keep its tick-1 value while the scan is failing") + require.InDelta(t, 1.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("audio.fifo", sqsQueueStateVisible)), 0.001, + "audio.fifo gauge must keep its tick-1 value while the scan is failing") + + obs.ObserveOnce(source) + // Tick 3 recovery: both queues still present so neither is + // forgotten; gauges update to the new values. The lastSeen + // map kept its tick-1 contents through the failed tick, so + // no spurious ForgetQueue happens here either. + require.InDelta(t, 7.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("orders.fifo", sqsQueueStateVisible)), 0.001) + require.InDelta(t, 2.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("audio.fifo", sqsQueueStateVisible)), 0.001) +} + // TestSQSObserver_NilTolerant pins that nil observer / nil source // don't panic — the same nil-tolerant contract Raft / Redis // observers carry, so a metrics-disabled deployment can pass nil From dd8a3c3095ffd556353fe318ecb453f0589fd04b Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 07:06:43 +0900 Subject: [PATCH 08/10] monitoring/sqs: share one readTS across depth-scan + counter reads (PR #743 r7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 (Codex review): SnapshotQueueDepths takes its readTS AFTER scanQueueNames returns. scanQueueNames internally calls nextTxnReadTS, so membership lands on one MVCC view while the per-queue counter reads run at a later ts. A queue created or deleted in that microsecond window can be silently missed (or reported with stale counters) for the tick — and the observer's 'current vs previous' diff loop then ForgetQueue's it spuriously, dashboard-rendering as a phantom drop or zero backlog under normal queue churn. Fix: take readTS once at the top of SnapshotQueueDepths and thread it through both scans. Refactor scanQueueNames into a 1-line wrapper around the new scanQueueNamesAt(ctx, readTS) so the depth-source path passes its ts in while the existing 3 callers (AdminListQueues, the reaper, the catalog walk) keep using the fresh-ts shape they already have — none need cross-call consistency. Caller audit per the standing semantic-change rule: scanQueueNames(ctx): external contract unchanged. The fresh-ts wrapper still acquires its own readTS via s.nextTxnReadTS internally, so admin / catalog / reaper callers see no observable difference. scanQueueNamesAt(ctx, readTS): new package-private. Only caller is SnapshotQueueDepths; no observable behavior elsewhere. SnapshotQueueDepths: signature and return semantics unchanged. Internal change tightens an MVCC consistency property — the only externally-visible effect is that the bug above stops happening, which is strictly an improvement to the observer's diff contract. No regression test added: the failure window is microseconds between two HLC ticks, and a deterministic reproduction would need a fault-injecting fake store wired into the catalog read path — meaningful test scaffolding that doesn't exist today and is disproportionate to a mechanical thread-through fix that's self-evidently correct from code reading. The property the fix establishes (membership scan and per-queue reads use the same readTS) is enforced at the scanQueueNamesAt signature: any future code path here must take readTS as a parameter, making backsliding a compile error rather than a runtime race. --- adapter/sqs_catalog.go | 14 +++++++++++++- adapter/sqs_depth_source.go | 21 ++++++++++++--------- 2 files changed, 25 insertions(+), 10 deletions(-) diff --git a/adapter/sqs_catalog.go b/adapter/sqs_catalog.go index 14b87c126..3d5cff159 100644 --- a/adapter/sqs_catalog.go +++ b/adapter/sqs_catalog.go @@ -1219,10 +1219,22 @@ func resolveListQueuesStart(names []string, token string) int { } func (s *SQSServer) scanQueueNames(ctx context.Context) ([]string, error) { + return s.scanQueueNamesAt(ctx, s.nextTxnReadTS(ctx)) +} + +// scanQueueNamesAt is scanQueueNames with the read timestamp passed +// in. SnapshotQueueDepths uses this so the membership scan and the +// per-queue counter reads share one MVCC view — without it, a queue +// created or deleted between the two HLC ticks would either be +// reported with stale counters or silently missed for the tick, +// producing spurious ForgetQueue calls in the observer's diff loop. +// Other callers (AdminListQueues, the reaper, the catalog walk) +// don't need cross-call consistency and continue to use the +// fresh-ts wrapper above. +func (s *SQSServer) scanQueueNamesAt(ctx context.Context, readTS uint64) ([]string, error) { prefix := []byte(SqsQueueMetaPrefix) end := prefixScanEnd(prefix) start := bytes.Clone(prefix) - readTS := s.nextTxnReadTS(ctx) var names []string for { kvs, err := s.store.ScanAt(ctx, start, end, sqsQueueScanPageLimit, readTS) diff --git a/adapter/sqs_depth_source.go b/adapter/sqs_depth_source.go index f375b58c3..f18fe51a6 100644 --- a/adapter/sqs_depth_source.go +++ b/adapter/sqs_depth_source.go @@ -47,18 +47,21 @@ func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) ([]SQSQueueDepth, b if s == nil || s.coordinator == nil || s.store == nil || !s.coordinator.IsLeader() { return nil, true } - names, err := s.scanQueueNames(ctx) + // Take ONE read timestamp up front and pass it through both the + // membership scan and the per-queue counter scans. With separate + // timestamps the membership read and the per-queue reads land on + // different MVCC views, so a queue created or deleted between + // them would be silently missed (or reported with stale + // counters) for one tick — and the observer's "current vs + // previous" diff would then ForgetQueue it spuriously, dashboard- + // rendering as a phantom drop. One ts per tick is also lighter + // on the leader's HLC than two. + readTS := s.nextTxnReadTS(ctx) + names, err := s.scanQueueNamesAt(ctx, readTS) if err != nil { - slog.Warn("sqs depth snapshot: scanQueueNames failed", "err", err) + slog.Warn("sqs depth snapshot: scanQueueNamesAt failed", "err", err) return nil, false } - // Take a single read timestamp for the whole tick so all queues - // in this snapshot share the same MVCC view. With per-queue - // nextTxnReadTS the first queue's read could see a state the - // last queue's read can't (catalog mutation between calls), and - // every call burns an HLC tick on the leader. One ts per tick - // is both more consistent and lighter on the leader's HLC. - readTS := s.nextTxnReadTS(ctx) out := make([]SQSQueueDepth, 0, len(names)) for _, name := range names { if err := ctx.Err(); err != nil { From fd3a3b040c0e897dafb8e6dcc220a02e3479656d Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 07:18:34 +0900 Subject: [PATCH 09/10] adapter/sqs: log loadQueueMetaAt errors symmetrically (PR #743 r8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit Minor: snapshotOneQueueDepth's first branch `if err != nil || !exists` collapses two distinct cases — a catalog read failure and a queue that legitimately doesn't exist — into the same silent return. The function-level docstring and the SnapshotQueueDepths contract both state that per-queue scan errors are logged; only scanApproxCounters errors actually were. Asymmetric: a flapping catalog read showed up as a dashboard gauge gap with nothing in the logs to explain it. Split the branches: err != nil -> slog.Warn with stable keys (queue, err) !exists -> silent return (the queue genuinely vanished — the observer's diff loop will ForgetQueue the gauge on its own; logging this is just noise) No semantic change to caller-visible behavior: both branches return the same (SQSQueueDepth{}, false) sentinel they did before. Pure logging hygiene + alignment with the documented contract. No caller audit needed since neither return value nor branching changes. --- adapter/sqs_depth_source.go | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/adapter/sqs_depth_source.go b/adapter/sqs_depth_source.go index f18fe51a6..018f96476 100644 --- a/adapter/sqs_depth_source.go +++ b/adapter/sqs_depth_source.go @@ -88,7 +88,21 @@ func (s *SQSServer) SnapshotQueueDepths(ctx context.Context) ([]SQSQueueDepth, b // snapshot rather than aborting the entire pass. func (s *SQSServer) snapshotOneQueueDepth(ctx context.Context, name string, readTS uint64) (SQSQueueDepth, bool) { meta, exists, err := s.loadQueueMetaAt(ctx, name, readTS) - if err != nil || !exists { + if err != nil { + // Log meta-read errors symmetrically with the + // scanApproxCounters branch below — both the function-level + // comment and the SnapshotQueueDepths contract state that + // per-queue scan errors are logged. A flapping catalog read + // would otherwise show up as a dashboard gauge gap with + // nothing in the logs to explain it. + slog.Warn("sqs depth snapshot: loadQueueMetaAt failed", "queue", name, "err", err) + return SQSQueueDepth{}, false + } + if !exists { + // Queue is genuinely gone (DeleteQueue tombstoned, generation + // fully drained). No log: the observer's diff will + // ForgetQueue this gauge on its own and a stable steady-state + // "queue no longer exists" is not an error worth alerting on. return SQSQueueDepth{}, false } counters, err := s.scanApproxCounters(ctx, name, meta.Generation, readTS) From eab20993a0b690a0f169195bd333c1b0daedb982 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Thu, 7 May 2026 07:35:13 +0900 Subject: [PATCH 10/10] monitoring/sqs: reclaim depth slots before admitting new queues (PR #743 r9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2 (Codex review): observeOnce emits ObserveQueueDepth for the current snapshot BEFORE forgetting queues that disappeared from the previous tick. Under high churn near the 512-queue cap (many old queues removed and many new queues added in the same tick), admitForDepthBudget sees the stale names still occupying trackedDepthQueues and silently collapses brand-new queues onto _other for at least one interval — even though the same tick will free their slots a few lines later. Fix: split observeOnce into two phases under the existing single-writer contract: Phase 1: build the current-name set + ForgetQueue any name in lastSeen that's not in current. ForgetQueue is the path that frees depth-budget slots in trackedDepthQueues; running it first under m.mu (taken inside ForgetQueue) makes the slots visible to the second phase's admission. Phase 2: emit ObserveQueueDepth for every snap. admitForDepthBudget now sees the freed slots and admits brand-new queues under their real labels instead of overflow. End-state semantics are unchanged: after observeOnce returns, gauges reflect the current tick and lastSeen mirrors the current name set. Only the intra-call ordering changes — phase 1 runs strictly before phase 2 instead of interleaving with it. The two phases serialise on m.mu (ForgetQueue and admitForDepthBudget both take it), so phase 2's admissions are guaranteed to see phase 1's reclamations. Caller audit per the standing semantic-change rule: observeOnce: callers are Start (production goroutine loop) and ObserveOnce (test entry). Both treat the function as 'process one tick'; neither observes intermediate state. The reorder is invisible to callers — same inputs, same end state, same lock discipline. No call-site update needed. Regression test: TestSQSObserver_ObserveOnce_HighChurnReclaimsBeforeAdmit saturates the budget with 512 queues in tick 1, replaces half of them with brand-new names in tick 2, and asserts: - new-0 / new-(half-1) gauges carry their tick-2 values under their REAL labels (pre-fix they collapsed to _other, returned 0) - total series count is exactly sqsMaxTrackedQueues*3 = 1536, leaving no room for any _other series (pre-fix would land at 256 real × 3 + _other × 3 = 771) Pre-fix verified failing with the new-0 assertion (got 0, want 3); post-fix passes. Test-side note: the count assertion runs BEFORE any per-queue spot check that uses WithLabelValues, because WithLabelValues materialises an absent label combination (creating the gauge child with default value 0 as a side effect). Asserting '_other visible == 0' via WithLabelValues would have inflated the count to 1537. This is a Prometheus client-API quirk that bit me during the fix and is now documented inline so future test authors don't trip over it. --- monitoring/sqs.go | 21 ++++++++--- monitoring/sqs_test.go | 83 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 5 deletions(-) diff --git a/monitoring/sqs.go b/monitoring/sqs.go index fa8a49700..aa1d0a2b4 100644 --- a/monitoring/sqs.go +++ b/monitoring/sqs.go @@ -469,13 +469,16 @@ func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) { } current := make(map[string]struct{}, len(snaps)) for _, snap := range snaps { - o.metrics.ObserveQueueDepth(snap.Queue, snap.Visible, snap.NotVisible, snap.Delayed) current[snap.Queue] = struct{}{} } - // Diff against the previous tick: any queue that disappeared - // (DeleteQueue, tombstoned cohort fully drained, leader stepped - // down — source returned []) gets its gauge series dropped so - // dashboards don't show a frozen backlog. + // Phase 1: forget queues that disappeared since the last tick. + // This MUST run before phase 2 — emitting first would let + // admitForDepthBudget see stale names still occupying + // trackedDepthQueues, so any newly-active queue admitted while + // the budget was full would silently collapse onto _other for + // at least one interval even when capacity opens up the same + // tick. Reclaiming first lets phase 2's admissions reuse the + // freed slots immediately. o.mu.Lock() for prev := range o.lastSeen { if _, ok := current[prev]; !ok { @@ -484,4 +487,12 @@ func (o *SQSObserver) observeOnce(ctx context.Context, source SQSDepthSource) { } o.lastSeen = current o.mu.Unlock() + // Phase 2: emit gauges for the current tick. Slots freed in + // phase 1 are now visible to admitForDepthBudget (both phases + // take the SQSMetrics.mu serially), so a brand-new queue + // landing in the same tick that an old queue disappeared gets + // the freed slot's real label rather than overflow. + for _, snap := range snaps { + o.metrics.ObserveQueueDepth(snap.Queue, snap.Visible, snap.NotVisible, snap.Delayed) + } } diff --git a/monitoring/sqs_test.go b/monitoring/sqs_test.go index 1b5f94a51..65be82202 100644 --- a/monitoring/sqs_test.go +++ b/monitoring/sqs_test.go @@ -668,6 +668,89 @@ func TestSQSObserver_ObserveOnce_TransientScanErrorPreservesGauges(t *testing.T) require.InDelta(t, 2.0, testutil.ToFloat64(m.queueDepth.WithLabelValues("audio.fifo", sqsQueueStateVisible)), 0.001) } +// TestSQSObserver_ObserveOnce_HighChurnReclaimsBeforeAdmit pins +// the P2 fix from PR #743 r9: observeOnce must run the +// disappeared-queue ForgetQueue diff BEFORE emitting the current +// tick's gauges, so that depth-budget slots freed by a +// just-disappeared queue are available for a brand-new queue +// admitted in the same tick. +// +// Pre-fix shape: emit ran first (admitForDepthBudget saw stale +// names still occupying trackedDepthQueues, so any new queue +// admitted while the budget was full collapsed to _other), then +// forget cleared the stale names. End state: half the new queues +// stuck on _other for at least one interval even though +// capacity opened up the same tick. +// +// Post-fix shape: forget runs first (slots reclaim under m.mu), +// emit runs second (admit sees the freed slots and gives every +// new queue a real label). End state: every queue admitted under +// its real name, no _other in the overflow gauge. +// +// Scenario: tick 1 saturates the budget with sqsMaxTrackedQueues +// real queues; tick 2 keeps half and replaces the other half +// with brand-new names — full churn at the cap. +func TestSQSObserver_ObserveOnce_HighChurnReclaimsBeforeAdmit(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + m := newSQSMetrics(reg) + obs := newSQSObserver(m) + + tick1 := make([]SQSQueueDepth, sqsMaxTrackedQueues) + for i := 0; i < sqsMaxTrackedQueues; i++ { + tick1[i] = SQSQueueDepth{Queue: "old-" + strconv.Itoa(i) + ".fifo", Visible: 1} + } + + half := sqsMaxTrackedQueues / 2 + tick2 := make([]SQSQueueDepth, 0, sqsMaxTrackedQueues) + for i := 0; i < half; i++ { + // First half of old queues carry over. + tick2 = append(tick2, SQSQueueDepth{Queue: "old-" + strconv.Itoa(i) + ".fifo", Visible: 2}) + } + for i := 0; i < half; i++ { + // Brand-new queues replacing the disappeared half. + tick2 = append(tick2, SQSQueueDepth{Queue: "new-" + strconv.Itoa(i) + ".fifo", Visible: 3}) + } + + source := &fakeDepthSource{ + ticks: []fakeDepthTick{ + {snaps: tick1, ok: true}, + {snaps: tick2, ok: true}, + }, + } + + obs.ObserveOnce(source) + obs.ObserveOnce(source) + + // Series count first — must come before any per-queue + // WithLabelValues call below, because WithLabelValues + // materialises the labelled gauge as a side effect (creating + // it with default value 0 if absent). Asserting "_other has + // no series" via WithLabelValues would actually CREATE the + // _other series and inflate the count; instead we assert the + // total series count here and let the math prove the absence + // (512 real queues × 3 states = 1536 with no room for any + // _other series; pre-fix would land at 256 real × 3 + _other + // × 3 = 771). + count, err := testutil.GatherAndCount(reg, "elastickv_sqs_queue_messages") + require.NoError(t, err) + require.Equal(t, sqsMaxTrackedQueues*3, count, + "512 queues × 3 states = 1536 series, all real labels — pre-fix would be 256 real × 3 + _other × 3 = 771") + + // Per-queue spot checks: the brand-new queues admitted under + // real labels (these series already exist from tick 2's emit + // phase, so WithLabelValues hits the existing children and + // doesn't perturb the count). + require.InDelta(t, 3.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues("new-0.fifo", sqsQueueStateVisible)), + 0.001, + "new queue must report under its real label — slot freed by old-(half) forget within same tick") + require.InDelta(t, 3.0, + testutil.ToFloat64(m.queueDepth.WithLabelValues("new-"+strconv.Itoa(half-1)+".fifo", sqsQueueStateVisible)), + 0.001, + "last new queue must also report under its real label") +} + // TestSQSObserver_NilTolerant pins that nil observer / nil source // don't panic — the same nil-tolerant contract Raft / Redis // observers carry, so a metrics-disabled deployment can pass nil