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

adapter/sqs.go

@@ -3,6 +3,7 @@ package adapter
 import (
 	"context"
 	"io"
+	"log/slog"
 	"net"
 	"net/http"
 	"strconv"
@@ -214,6 +215,16 @@ 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
+	// adminAuditLogger is where the §3.6 admin audit lines go. Defaults
+	// to slog.Default(); production installs the component="admin" child
+	// logger so these records land in the same destination, with the same
+	// attributes, as every other admin audit entry rather than bypassing
+	// it through the process-wide default.
+	adminAuditLogger *slog.Logger
 }
 
 // SQSPartitionObserver is the metrics-package interface
@@ -224,6 +235,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 +297,25 @@ 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
+		}
+	}
+}
+
+// WithSQSAdminAuditLogger routes the §3.6 admin audit lines to the
+// supplied logger instead of slog.Default(). No-ops on nil.
+func WithSQSAdminAuditLogger(l *slog.Logger) SQSServerOption {
+	return func(s *SQSServer) {
+		if l != nil {
+			s.adminAuditLogger = l
+		}
+	}
+}
+
 // 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,26 @@ 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.recordAdminPurge(ctx, principal, name, adminOutcomeForbidden)
 		return AdminPurgeResult{}, ErrAdminForbidden
 	}
 	if !isVerifiedSQSLeader(ctx, s.coordinator) {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeNotLeader)
 		return AdminPurgeResult{}, ErrAdminNotLeader
 	}
 	if strings.TrimSpace(name) == "" {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeValidation)
 		return AdminPurgeResult{}, ErrAdminSQSValidation
 	}
 	oldGen, newGen, err := s.purgeQueueWithRetry(ctx, name)
 	if err != nil {
+		// Every refusal gets the operation-specific audit record, not
+		// just the counter. A repeated purge inside the 60-second window
+		// used to return from here with no admin.sqs.purge_queue line at
+		// all, leaving only the generic HTTP audit middleware's status
+		// and path -- which cannot say WHY it was refused, and so cannot
+		// answer the question the §3.6 signal exists for.
+		s.recordAdminPurge(ctx, principal, name, adminPurgeOutcomeForError(err))
 		var rateLimit *purgeRateLimitedError
 		if errors.As(err, &rateLimit) {
 			return AdminPurgeResult{}, &PurgeInProgressError{RetryAfter: rateLimit.remaining}
@@ -335,9 +346,109 @@ func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipa
 		}
 		return AdminPurgeResult{}, errors.Wrap(err, "admin purge queue")
 	}
+	s.recordAdminPurge(ctx, principal, name, adminOutcomeOK,
+		slog.Uint64("generation_before", oldGen),
+		slog.Uint64("generation_after", newGen))
 	return AdminPurgeResult{GenerationBefore: oldGen, GenerationAfter: newGen}, nil
 }
 
+// recordAdminPurge emits the §3.6 audit line and bumps the counter for one
+// purge outcome.
+//
+// Both in one place so an exit path cannot record the metric and skip the
+// audit, which is how the purge-in-progress refusal ended up counted but
+// never audited. Generation attributes are passed only by the success path:
+// a refusal has no committed generation pair, and inventing one would put a
+// state that never existed into the audit trail.
+func (s *SQSServer) recordAdminPurge(
+	ctx context.Context,
+	principal AdminPrincipal,
+	name string,
+	outcome string,
+	extra ...slog.Attr,
+) {
+	s.observeAdminPurge(name, outcome)
+	// access_key, role, queue, outcome.
+	const baseAuditAttrs = 4
+	attrs := make([]any, 0, len(extra)+baseAuditAttrs)
+	attrs = append(attrs,
+		// 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.String("outcome", outcome))
+	for _, attr := range extra {
+		attrs = append(attrs, attr)
+	}
+	s.adminLogger().InfoContext(ctx, "admin.sqs.purge_queue", attrs...)
+}
+
+// adminLogger returns the configured audit destination, falling back to the
+// process default so a server built without the option still audits.
+func (s *SQSServer) adminLogger() *slog.Logger {
+	if s == nil || s.adminAuditLogger == nil {
+		return slog.Default()
+	}
+	return s.adminAuditLogger
+}
+
+// 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
+	}
+}
+
+// AdminObserver exposes the §3.6 counters so the admin HTTP handler can
+// record the rejections it serves before reaching this adapter. Returns nil
+// on a server built without an observer.
+func (s *SQSServer) AdminObserver() SQSAdminObserver {
+	if s == nil {
+		return nil
+	}
+	return s.adminObserver
+}
+
+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_audit_test.go

@@ -0,0 +1,133 @@
+package adapter
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// captureAdminAudit points the server's audit logger at a buffer and returns a
+// reader for the admin.sqs.purge_queue records it emits.
+func captureAdminAudit(t *testing.T, server *SQSServer) func() []map[string]any {
+	t.Helper()
+
+	var buf bytes.Buffer
+	// A distinguishing attribute, so a record that went out through
+	// slog.Default() instead of this logger is detectable rather than just
+	// absent.
+	server.adminAuditLogger = slog.New(slog.NewJSONHandler(&buf, nil)).
+		With(slog.String("component", "admin"))
+
+	return func() []map[string]any {
+		var records []map[string]any
+		for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {
+			if strings.TrimSpace(line) == "" {
+				continue
+			}
+			var rec map[string]any
+			require.NoError(t, json.Unmarshal([]byte(line), &rec))
+			if rec["msg"] == "admin.sqs.purge_queue" {
+				records = append(records, rec)
+			}
+		}
+		return records
+	}
+}
+
+// TestAdminPurgeQueueAuditsThroughTheConfiguredLogger pins that the §3.6 audit
+// record goes to the configured audit destination.
+//
+// It used to call slog.InfoContext, i.e. the process-wide slog.Default(), so a
+// server built with a dedicated audit sink or the production component="admin"
+// child logger never saw these records and they lost that logger's attributes
+// — unlike every other admin audit entry.
+func TestAdminPurgeQueueAuditsThroughTheConfiguredLogger(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	_ = createSQSQueueForTest(t, node, "audited")
+	records := captureAdminAudit(t, node.sqsServer)
+
+	_, err := node.sqsServer.AdminPurgeQueue(context.Background(), fullAdminPrincipal, "audited")
+	require.NoError(t, err)
+
+	got := records()
+	require.Len(t, got, 1, "exactly one audit record for one purge")
+	require.Equal(t, "admin", got[0]["component"],
+		"the record must carry the configured logger's attributes")
+	require.Equal(t, "audited", got[0]["queue"])
+	require.Equal(t, adminOutcomeOK, got[0]["outcome"])
+	require.Contains(t, got[0], "generation_before")
+	require.Contains(t, got[0], "generation_after")
+}
+
+// TestAdminPurgeQueueAuditsAPurgeInProgressRefusal is the load-bearing case.
+//
+// A second purge inside the 60-second window returned before the only audit
+// call, so the documented signal for repeated rate-limited attempts produced
+// nothing. The generic HTTP audit middleware records status and path, which
+// cannot say WHY the request was refused — the one thing this record exists to
+// answer.
+func TestAdminPurgeQueueAuditsAPurgeInProgressRefusal(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	_ = createSQSQueueForTest(t, node, "repeat-purged")
+	records := captureAdminAudit(t, node.sqsServer)
+	ctx := context.Background()
+
+	_, err := node.sqsServer.AdminPurgeQueue(ctx, fullAdminPrincipal, "repeat-purged")
+	require.NoError(t, err)
+	_, err = node.sqsServer.AdminPurgeQueue(ctx, fullAdminPrincipal, "repeat-purged")
+	require.ErrorIs(t, err, ErrAdminSQSPurgeInProgress)
+
+	got := records()
+	require.Len(t, got, 2, "the refusal must be audited, not only the success")
+	require.Equal(t, adminOutcomeOK, got[0]["outcome"])
+	require.Equal(t, adminOutcomePurgeInProgress, got[1]["outcome"])
+	require.Equal(t, "repeat-purged", got[1]["queue"])
+
+	// No invented generation pair: the refusal committed nothing, and
+	// recording a state that never existed would corrupt the audit trail.
+	require.NotContains(t, got[1], "generation_before")
+	require.NotContains(t, got[1], "generation_after")
+}
+
+// A forbidden principal is refused before any storage work, and must still be
+// audited with its own outcome.
+func TestAdminPurgeQueueAuditsAForbiddenRefusal(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+	node := sqsLeaderNode(t, nodes)
+
+	_ = createSQSQueueForTest(t, node, "guarded")
+	records := captureAdminAudit(t, node.sqsServer)
+
+	_, err := node.sqsServer.AdminPurgeQueue(context.Background(), readOnlyAdminPrincipal, "guarded")
+	require.ErrorIs(t, err, ErrAdminForbidden)
+
+	got := records()
+	require.Len(t, got, 1)
+	require.Equal(t, adminOutcomeForbidden, got[0]["outcome"])
+	require.NotContains(t, got[0], "generation_before")
+}
+
+// A server built without the option must still audit, through the default.
+func TestAdminPurgeQueueFallsBackToTheDefaultLogger(t *testing.T) {
+	t.Parallel()
+
+	var server *SQSServer
+	require.NotNil(t, server.adminLogger(), "a nil server must not panic")
+	require.NotNil(t, (&SQSServer{}).adminLogger(),
+		"a server with no configured audit logger must still have somewhere to audit")
+}