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 new file mode 100644 index 000000000..018f96476 --- /dev/null +++ b/adapter/sqs_depth_source.go @@ -0,0 +1,119 @@ +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. +// +// 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, true + } + // 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: scanQueueNamesAt failed", "err", err) + return nil, false + } + out := make([]SQSQueueDepth, 0, len(names)) + for _, name := range names { + if err := ctx.Err(); err != nil { + // 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, true +} + +// snapshotOneQueueDepth runs the per-queue catalog read pair +// (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 { + // 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) + 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..ffa6b1842 100644 --- a/main.go +++ b/main.go @@ -1126,6 +1126,63 @@ 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, bool) { + if a.inner == 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 + } + if len(snaps) == 0 { + return nil, true + } + 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, true +} + // 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 +1608,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..0ac59896a --- /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": "count by (node_id) (elastickv_sqs_queue_messages)", + "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..aa1d0a2b4 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,110 @@ 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). +// +// 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) (snaps []SQSQueueDepth, ok bool) +} + +// 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). +// +// 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{} + // 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 +// 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 +158,19 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics { }, []string{"queue", "partition", "action"}, ), - trackedQueues: map[string]struct{}{}, + 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"}, + ), + trackedCounterQueues: map[string]struct{}{}, + trackedDepthQueues: map[string]struct{}{}, + overflowDepthQueues: map[string]struct{}{}, } registerer.MustRegister(m.partitionMessages) + registerer.MustRegister(m.queueDepth) return m } @@ -85,7 +191,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. @@ -97,25 +203,174 @@ func (m *SQSMetrics) ObservePartitionMessage(queue string, partition uint32, act ).Inc() } -// 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 { +// 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.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 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. +// +// 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). 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 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 + } + m.mu.Lock() + _, tracked := m.trackedDepthQueues[queue] + 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 { + m.dropGaugeStatesFor(queue) + } + 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.dropGaugeStatesFor(sqsQueueOverflow) + } +} + +// 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.trackedQueues[queue]; ok { + if _, ok := m.trackedCounterQueues[queue]; ok { return queue } - if len(m.trackedQueues) >= sqsMaxTrackedQueues { + if len(m.trackedCounterQueues) >= sqsMaxTrackedQueues { return sqsQueueOverflow } - m.trackedQueues[queue] = struct{}{} + 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. +// +// 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). +// +// 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() + 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. @@ -126,3 +381,118 @@ 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) +} + +// 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 + } + 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 { + current[snap.Queue] = struct{}{} + } + // 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 { + o.metrics.ForgetQueue(prev) + } + } + 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 5c3784782..65be82202 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,630 @@ 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 +// (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() + m := newSQSMetrics(reg) + + m.ObserveQueueDepth("orders.fifo", 3, 0, 0) + m.ObservePartitionMessage("orders.fifo", 0, SQSPartitionActionSend) + 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.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)") + + // 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.trackedDepthQueues, 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.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_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") +// 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) +// 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. +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") +} + +// 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 { + ticks []fakeDepthTick + calls int +} + +func (f *fakeDepthSource) SnapshotQueueDepths(_ context.Context) ([]SQSQueueDepth, bool) { + if f.calls >= len(f.ticks) { + f.calls++ + return nil, true + } + t := f.ticks[f.calls] + f.calls++ + return t.snaps, t.ok +} + +// 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{ + 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}, + }, 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}, + }, + } + + 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{ + ticks: []fakeDepthTick{ + {snaps: []SQSQueueDepth{ + {Queue: "orders.fifo", Visible: 5}, + {Queue: "audio.fifo", Visible: 1}, + }, 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) + 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_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_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 +// 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) + }) +}