sqs: add the admin purge/peek audit line and counters (§3.6)

adapter/sqs.go

@@ -214,6 +214,10 @@ type SQSServer struct {
 	// nil on non-monitored fixtures; observeThrottleDecision is
 	// nil-safe so the request path pays one branch when unwired.
 	throttleObserver SQSThrottleObserver
+	// adminObserver records the §3.6 admin purge / peek counters.
+	// nil on non-monitored fixtures; the increment helpers are
+	// nil-safe so an unwired server pays one branch.
+	adminObserver SQSAdminObserver
 }
 
 // SQSPartitionObserver is the metrics-package interface
@@ -224,6 +228,14 @@ type SQSPartitionObserver interface {
 	ObservePartitionMessage(queue string, partition uint32, action string)
 }
 
+// SQSAdminObserver is the metrics-package interface
+// (monitoring.SQSMetrics) re-declared here so the adapter does not
+// import monitoring at the package boundary.
+type SQSAdminObserver interface {
+	ObserveAdminPurgeQueue(queue string, outcome string)
+	ObserveAdminPeekQueue(queue string, outcome string)
+}
+
 // SQSThrottleObserver is the metrics-package interface
 // (monitoring.SQSThrottleObserver) re-declared here so the adapter
 // does not import monitoring at the package boundary.
@@ -278,6 +290,15 @@ func WithSQSLeaderMap(m map[string]string) SQSServerOption {
 	}
 }
 
+// WithSQSAdminObserver installs the §3.6 admin purge / peek counters.
+func WithSQSAdminObserver(o SQSAdminObserver) SQSServerOption {
+	return func(s *SQSServer) {
+		if o != nil {
+			s.adminObserver = o
+		}
+	}
+}
+
 // WithSQSPartitionObserver installs the
 // elastickv_sqs_partition_messages_total counter observer on the
 // SQS server. Pass nil (the default) on non-monitored test

adapter/sqs_admin.go

@@ -3,6 +3,7 @@ package adapter
 import (
 	"bytes"
 	"context"
+	"log/slog"
 	"sort"
 	"strconv"
 	"strings"
@@ -316,16 +317,20 @@ func (e *PurgeInProgressError) Is(target error) bool {
 //   - ErrAdminSQSValidation      — empty / whitespace name
 func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipal, name string) (AdminPurgeResult, error) {
 	if !principal.Role.canWrite() {
+		s.observeAdminPurge(name, adminOutcomeForbidden)
 		return AdminPurgeResult{}, ErrAdminForbidden
 	}
 	if !isVerifiedSQSLeader(ctx, s.coordinator) {
+		s.observeAdminPurge(name, adminOutcomeNotLeader)
 		return AdminPurgeResult{}, ErrAdminNotLeader
 	}
 	if strings.TrimSpace(name) == "" {
+		s.observeAdminPurge(name, adminOutcomeValidation)
 		return AdminPurgeResult{}, ErrAdminSQSValidation
 	}
 	oldGen, newGen, err := s.purgeQueueWithRetry(ctx, name)
 	if err != nil {
+		s.observeAdminPurge(name, adminPurgeOutcomeForError(err))
 		var rateLimit *purgeRateLimitedError
 		if errors.As(err, &rateLimit) {
 			return AdminPurgeResult{}, &PurgeInProgressError{RetryAfter: rateLimit.remaining}
@@ -335,9 +340,69 @@ func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipa
 		}
 		return AdminPurgeResult{}, errors.Wrap(err, "admin purge queue")
 	}
+	s.observeAdminPurge(name, adminOutcomeOK)
+	// §3.6 audit line. Deliberately lean: subject, role, queue and the
+	// two generations are everything needed to reconstruct
+	// who-purged-what-when. The generations come from the committed
+	// OCC round rather than a pre/post read, so they cannot report a
+	// pair of values that never existed as one consistent state.
+	slog.InfoContext(ctx, "admin.sqs.purge_queue",
+		// AdminPrincipal carries AccessKey, not the design's
+		// "subject": the access key ID is the identity the admin
+		// surface authenticates, and it is an identifier rather than
+		// a secret (the signing key never appears here).
+		slog.String("access_key", principal.AccessKey),
+		slog.String("role", string(principal.Role)),
+		slog.String("queue", name),
+		slog.Uint64("generation_before", oldGen),
+		slog.Uint64("generation_after", newGen))
 	return AdminPurgeResult{GenerationBefore: oldGen, GenerationAfter: newGen}, nil
 }
 
+// Outcome labels for the §3.6 admin counters, mirrored from
+// monitoring so the adapter does not import it at this boundary.
+const (
+	adminOutcomeOK              = "ok"
+	adminOutcomeForbidden       = "forbidden"
+	adminOutcomeNotLeader       = "not_leader"
+	adminOutcomeNotFound        = "not_found"
+	adminOutcomeValidation      = "validation"
+	adminOutcomePurgeInProgress = "purge_in_progress"
+	// NOTE: no "throttled" here — the admin peek throttle is a
+	// separate deferred follow-up, and declaring the label before a
+	// call site exists would imply coverage the code does not have.
+	adminOutcomeInternalError = "internal_error"
+)
+
+// adminPurgeOutcomeForError classifies a purge failure by SENTINEL,
+// never by message text: an error-string label would let one recurring
+// failure grow the series set without bound.
+func adminPurgeOutcomeForError(err error) string {
+	var rateLimit *purgeRateLimitedError
+	switch {
+	case errors.As(err, &rateLimit):
+		return adminOutcomePurgeInProgress
+	case isSQSAdminQueueDoesNotExist(err):
+		return adminOutcomeNotFound
+	default:
+		return adminOutcomeInternalError
+	}
+}
+
+func (s *SQSServer) observeAdminPurge(queue, outcome string) {
+	if s == nil || s.adminObserver == nil {
+		return
+	}
+	s.adminObserver.ObserveAdminPurgeQueue(queue, outcome)
+}
+
+func (s *SQSServer) observeAdminPeek(queue, outcome string) {
+	if s == nil || s.adminObserver == nil {
+		return
+	}
+	s.adminObserver.ObserveAdminPeekQueue(queue, outcome)
+}
+
 // AdminSetQueueAttributes is the SigV4-bypass counterpart to
 // SetQueueAttributes. It is intentionally generic rather than
 // DLQ-specific so the admin SPA can edit RedrivePolicy and

adapter/sqs_admin_peek.go

@@ -215,43 +215,69 @@ func (s *SQSServer) AdminPeekQueue(
 	opts AdminPeekMessageOptions,
 ) ([]AdminPeekedMessage, string, error) {
 	if !principal.Role.canRead() {
+		s.observeAdminPeek(name, adminOutcomeForbidden)
 		return nil, "", ErrAdminForbidden
 	}
 	if !isVerifiedSQSLeader(ctx, s.coordinator) {
+		s.observeAdminPeek(name, adminOutcomeNotLeader)
 		return nil, "", ErrAdminNotLeader
 	}
 	if strings.TrimSpace(name) == "" {
+		s.observeAdminPeek(name, adminOutcomeValidation)
 		return nil, "", ErrAdminSQSValidation
 	}
 	limit := clampPeekLimit(opts.Limit)
 	bodyMaxBytes := clampPeekBodyBytes(opts.BodyMaxBytes)
 	cursor, err := decodePeekCursor(opts.Cursor)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
 	readTS := s.nextTxnReadTS(ctx)
 	meta, exists, err := s.loadQueueMetaAt(ctx, name, readTS)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", errors.WithStack(err)
 	}
 	if !exists {
+		s.observeAdminPeek(name, adminOutcomeNotFound)
 		return nil, "", ErrAdminSQSNotFound
 	}
 	cursor, err = preparePeekCursor(cursor, meta, name, s.peekStartPartition(name, cursor, meta))
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
 	rows, nextCursor, err := s.walkPeek(ctx, name, meta, readTS, cursor, limit, bodyMaxBytes)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
 	encoded, err := encodePeekCursor(nextCursor)
 	if err != nil {
+		s.observeAdminPeek(name, adminPeekOutcomeForError(err))
 		return nil, "", err
 	}
+	s.observeAdminPeek(name, adminOutcomeOK)
 	return rows, encoded, nil
 }
 
+// adminPeekOutcomeForError classifies a peek failure by SENTINEL,
+// never by message text — an error-string label would let one
+// recurring failure grow the series set without bound.
+func adminPeekOutcomeForError(err error) string {
+	switch {
+	case errors.Is(err, ErrAdminSQSValidation):
+		return adminOutcomeValidation
+	case errors.Is(err, ErrAdminSQSNotFound):
+		return adminOutcomeNotFound
+	case isSQSAdminQueueDoesNotExist(err):
+		return adminOutcomeNotFound
+	default:
+		return adminOutcomeInternalError
+	}
+}
+
 // peekStartPartition returns the starting partition for a fresh peek
 // walk, or 0 when a starting partition is not needed (continuation
 // pages already carry one in the cursor; non-partitioned queues only

docs/design/2026_05_16_implemented_admin_purge_queue.md

@@ -16,7 +16,7 @@
 
 Out-of-scope follow-ups (tracked separately, not gating this rename):
 - Throttle integration (`bucketActionAdminPeek` + dedicated per-queue admin-peek bucket per §3.1)
-- Audit logging + Prometheus counters per §3.6
+- ~~Audit logging + Prometheus counters per §3.6~~ — **implemented**
 - `principalForReadSensitive` live `RoleStore` re-check (Goal 8, blocked on wider RoleStore plumbing)
 - Page-size selector (20 / 50 / 100) + response-size warning
 
@@ -491,7 +491,11 @@ mirroring the existing `deleteQueue` / `describeQueue` shape. `peekQueue` is `si
 
 ### 3.6 Audit and observability
 
-_Not yet implemented in the initial rollout — see "Out-of-scope follow-ups" at the top. Mitigation in absence: the admin handler still emits the standard request-log line with `route` / `subject` / `status_code` for both purge and peek calls, so an operator can correlate "who did what when" against the application logs at audit-review time. The structured `admin.sqs.purge_queue` audit line and the two Prometheus counters land alongside the SPA wiring so the metrics have a real consumer._
+_**Implemented.** The `admin.sqs.purge_queue` audit line and both Prometheus counters are live. Two deviations from the text below, both forced by the code as it stands:_
+
+_1. The audit line logs `access_key`, not `subject`: `AdminPrincipal` carries `AccessKey` and `Role` and has no `Subject` field. The access key ID is the identity the admin surface authenticates and is an identifier rather than a secret — the signing key never appears in the log._
+
+_2. The two outcome sets are deliberately asymmetric. `purge_in_progress` exists only on the purge counter and `throttled` only on peek, because purge signals contention via the generation gate and peek via the throttle. Accepting both on either counter would let the two paths drift into describing one condition two ways. The peek `throttled` outcome is defined but not yet emitted — admin-peek throttle integration remains a separate open follow-up._
 
 New structured log line at `slog.Info` level (matches `AdminDeleteQueue`):
 

main_sqs.go

@@ -43,6 +43,13 @@ func prepareSQSServer(
 	if o, ok := partitionObserver.(adapter.SQSThrottleObserver); ok {
 		throttleObserver = o
 	}
+	// The registry's SQSMetrics satisfies all three observer
+	// interfaces; derive the admin one the same way rather than
+	// widening this function's signature.
+	var adminObserver adapter.SQSAdminObserver
+	if o, ok := partitionObserver.(adapter.SQSAdminObserver); ok {
+		adminObserver = o
+	}
 	sqsServer := adapter.NewSQSServer(
 		sqsL,
 		shardStore,
@@ -53,6 +60,7 @@ func prepareSQSServer(
 		adapter.WithSQSPartitionResolver(partitionResolver),
 		adapter.WithSQSPartitionObserver(partitionObserver),
 		adapter.WithSQSThrottleObserver(throttleObserver),
+		adapter.WithSQSAdminObserver(adminObserver),
 	)
 	return sqsServer, sqsL, nil
 }

monitoring/sqs.go

@@ -149,6 +149,13 @@ type SQSMetrics struct {
 	queueDepth        *prometheus.GaugeVec
 	throttledRequests *prometheus.CounterVec
 	throttleTokens    *prometheus.GaugeVec
+	adminPurgeQueue   *prometheus.CounterVec
+	adminPeekQueue    *prometheus.CounterVec
+	// trackedAdminCounterQueues bounds the queue label on the two
+	// admin counters with the same budget the data-path counters use.
+	// Queue names are operator-supplied, so an unbounded label here
+	// would let queue churn grow the series set without limit.
+	trackedAdminCounterQueues map[string]struct{}
 
 	mu                           sync.Mutex
 	trackedCounterQueues         map[string]struct{}
@@ -215,6 +222,7 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics {
 			[]string{"queue", "action"},
 		),
 		trackedCounterQueues:         map[string]struct{}{},
+		trackedAdminCounterQueues:    map[string]struct{}{},
 		trackedThrottleCounterQueues: map[string]struct{}{},
 		trackedDepthQueues:           map[string]struct{}{},
 		trackedThrottleGaugeQueues:   map[string]map[string]struct{}{},
@@ -223,13 +231,113 @@ func newSQSMetrics(registerer prometheus.Registerer) *SQSMetrics {
 		overflowDepthQueues:          map[string]struct{}{},
 		overflowThrottleGaugeQueues:  map[string]map[string]struct{}{},
 	}
+	m.adminPurgeQueue = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "elastickv_sqs_admin_purge_queue_total",
+			Help: "Total admin PurgeQueue calls by queue and outcome.",
+		},
+		[]string{"queue", "outcome"},
+	)
+	m.adminPeekQueue = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "elastickv_sqs_admin_peek_queue_total",
+			Help: "Total admin PeekQueue calls by queue and outcome.",
+		},
+		[]string{"queue", "outcome"},
+	)
 	registerer.MustRegister(m.partitionMessages)
 	registerer.MustRegister(m.queueDepth)
 	registerer.MustRegister(m.throttledRequests)
 	registerer.MustRegister(m.throttleTokens)
+	registerer.MustRegister(m.adminPurgeQueue)
+	registerer.MustRegister(m.adminPeekQueue)
 	return m
 }
 
+// Outcomes for the two admin SQS counters. Closed sets: the label is
+// derived from a sentinel comparison, never from an error string, so
+// one recurring failure cannot mint new series.
+const (
+	SQSAdminOutcomeOK              = "ok"
+	SQSAdminOutcomeForbidden       = "forbidden"
+	SQSAdminOutcomeNotLeader       = "not_leader"
+	SQSAdminOutcomeNotFound        = "not_found"
+	SQSAdminOutcomeValidation      = "validation"
+	SQSAdminOutcomePurgeInProgress = "purge_in_progress"
+	SQSAdminOutcomeThrottled       = "throttled"
+	SQSAdminOutcomeInternalError   = "internal_error"
+)
+
+// ObserveAdminPurgeQueue counts one AdminPurgeQueue call.
+func (m *SQSMetrics) ObserveAdminPurgeQueue(queue, outcome string) {
+	if m == nil {
+		return
+	}
+	m.adminPurgeQueue.WithLabelValues(
+		m.admitForAdminCounterBudget(queue),
+		normalizeSQSAdminPurgeOutcome(outcome),
+	).Inc()
+}
+
+// ObserveAdminPeekQueue counts one AdminPeekQueue call.
+func (m *SQSMetrics) ObserveAdminPeekQueue(queue, outcome string) {
+	if m == nil {
+		return
+	}
+	m.adminPeekQueue.WithLabelValues(
+		m.admitForAdminCounterBudget(queue),
+		normalizeSQSAdminPeekOutcome(outcome),
+	).Inc()
+}
+
+func (m *SQSMetrics) admitForAdminCounterBudget(queue string) string {
+	if queue == "" {
+		// An empty name reaches here only on a validation rejection,
+		// where there is no queue to attribute the call to.
+		return sqsQueueOverflow
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return admitCounterQueueLocked(queue, m.trackedAdminCounterQueues)
+}
+
+// normalizeSQSAdminPurgeOutcome keeps the purge label inside §3.6's
+// closed set. `throttled` is deliberately absent: purge signals
+// contention as purge_in_progress, and accepting both would let the
+// two paths drift into reporting the same condition differently.
+func normalizeSQSAdminPurgeOutcome(outcome string) string {
+	switch outcome {
+	case SQSAdminOutcomeOK,
+		SQSAdminOutcomeForbidden,
+		SQSAdminOutcomeNotLeader,
+		SQSAdminOutcomeNotFound,
+		SQSAdminOutcomeValidation,
+		SQSAdminOutcomePurgeInProgress,
+		SQSAdminOutcomeInternalError:
+		return outcome
+	default:
+		return SQSAdminOutcomeInternalError
+	}
+}
+
+// normalizeSQSAdminPeekOutcome keeps the peek label inside §3.6's
+// closed set. `purge_in_progress` is absent for the mirror-image
+// reason: peek is throttled, not generation-gated.
+func normalizeSQSAdminPeekOutcome(outcome string) string {
+	switch outcome {
+	case SQSAdminOutcomeOK,
+		SQSAdminOutcomeForbidden,
+		SQSAdminOutcomeNotLeader,
+		SQSAdminOutcomeNotFound,
+		SQSAdminOutcomeValidation,
+		SQSAdminOutcomeThrottled,
+		SQSAdminOutcomeInternalError:
+		return outcome
+	default:
+		return SQSAdminOutcomeInternalError
+	}
+}
+
 // ObservePartitionMessage implements SQSPartitionObserver. The
 // (queue, action) pair is validated and (queue) is collapsed to
 // the overflow label past sqsMaxTrackedQueues distinct names.