From 1ec9031f421549fba4dd059684f1f392408a14b3 Mon Sep 17 00:00:00 2001 From: Satyam Trivedi Date: Tue, 13 Oct 2026 00:21:08 +0530 Subject: [PATCH 1/2] fix(aws-sts): report the role's RoleId in AssumedRoleId AssumeRole, AssumeRoleWithWebIdentity and AssumeRoleWithSAML returned the fixed AssumedRoleId prefix AROACLOUDEMU0000000000 for every role. Real STS returns ":", where the role id is the role's unique RoleId. Use the IAM role's RoleId when the role is known and keep the synthetic prefix only when no IAM driver knows it. GetCallerIdentity and the SQS SenderId derive from this id, so two roles with the same session name no longer collapse onto one identity. Part of #1514 --- server/aws/sts/assumed_role_id_test.go | 67 ++++++++++++++++++++++++++ server/aws/sts/operations.go | 26 +++++++--- 2 files changed, 87 insertions(+), 6 deletions(-) create mode 100644 server/aws/sts/assumed_role_id_test.go diff --git a/server/aws/sts/assumed_role_id_test.go b/server/aws/sts/assumed_role_id_test.go new file mode 100644 index 000000000..f580b46e6 --- /dev/null +++ b/server/aws/sts/assumed_role_id_test.go @@ -0,0 +1,67 @@ +package sts_test + +import ( + "context" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + awsiam "github.com/aws/aws-sdk-go-v2/service/iam" + awssts "github.com/aws/aws-sdk-go-v2/service/sts" + + awsserver "github.com/stackshy/cloudemu/v2/server/aws" +) + +// TestSDKAssumedRoleIDUsesRoleID checks that AssumedRoleId is the assumed +// role's own RoleId plus the session name, as real STS returns it, so two +// roles with the same session name get different ids. +func TestSDKAssumedRoleIDUsesRoleID(t *testing.T) { + ts, iamClient := newTrustServer(t) + ctx := context.Background() + client := stsClient(t, ts.URL) + + for _, name := range []string{"role-a", "role-b"} { + role, err := iamClient.CreateRole(ctx, &awsiam.CreateRoleInput{ + RoleName: aws.String(name), + AssumeRolePolicyDocument: aws.String(allowRootTrust), + }) + if err != nil { + t.Fatalf("CreateRole %s: %v", name, err) + } + + out, err := client.AssumeRole(ctx, &awssts.AssumeRoleInput{ + RoleArn: aws.String("arn:aws:iam::" + testAccountID + ":role/" + name), + RoleSessionName: aws.String("shared-session"), + }) + if err != nil { + t.Fatalf("AssumeRole %s: %v", name, err) + } + + want := aws.ToString(role.Role.RoleId) + ":shared-session" + if got := aws.ToString(out.AssumedRoleUser.AssumedRoleId); got != want { + t.Fatalf("AssumedRoleId for %s = %q, want %q", name, got, want) + } + + if !strings.HasPrefix(want, "AROA") { + t.Fatalf("RoleId %q does not have the AROA prefix", want) + } + } +} + +// TestSDKAssumedRoleIDWithoutIAM keeps the synthetic prefix when no IAM driver +// knows the role. +func TestSDKAssumedRoleIDWithoutIAM(t *testing.T) { + ts := newServer(t, awsserver.Drivers{STS: true, AccountID: testAccountID, Region: testRegion}) + + out, err := stsClient(t, ts.URL).AssumeRole(context.Background(), &awssts.AssumeRoleInput{ + RoleArn: aws.String("arn:aws:iam::" + testAccountID + ":role/anything"), + RoleSessionName: aws.String("s"), + }) + if err != nil { + t.Fatalf("AssumeRole: %v", err) + } + + if got := aws.ToString(out.AssumedRoleUser.AssumedRoleId); got != "AROACLOUDEMU0000000000:s" { + t.Fatalf("AssumedRoleId = %q, want the synthetic fallback", got) + } +} diff --git a/server/aws/sts/operations.go b/server/aws/sts/operations.go index adceb5504..7cf9a49f2 100644 --- a/server/aws/sts/operations.go +++ b/server/aws/sts/operations.go @@ -73,7 +73,7 @@ func (h *Handler) assumeRole(w http.ResponseWriter, r *http.Request) { } assumedArn := "arn:aws:sts::" + h.accountID + ":assumed-role/" + roleName + "/" + sessionName - assumedRoleID := assumedRoleIDPrefix + ":" + sessionName + assumedRoleID := h.assumedRoleID(r, roleArn, roleName, sessionName) creds, ok := h.mintCredentials(w, durationFromForm(r), awsidentity.Identity{ARN: assumedArn, UserID: assumedRoleID}, roleOwner(roleName)) @@ -95,11 +95,25 @@ func (h *Handler) assumeRole(w http.ResponseWriter, r *http.Request) { } // assumedRoleIDPrefix is the synthetic AssumedRoleId prefix (real STS uses the -// "AROA" prefix followed by a unique id) reported by every AssumeRole-family -// operation, and used to derive the identity a minted session is recorded -// under for GetCallerIdentity. +// role's unique "AROA..." id) reported by an AssumeRole-family operation when +// the role is not known to the IAM driver, and used to derive the identity a +// minted session is recorded under for GetCallerIdentity. const assumedRoleIDPrefix = "AROACLOUDEMU0000000000" +// assumedRoleID is the AssumedRoleId of a session of the role roleArn names: +// ":", where the role id is the role's unique id as +// IAM reports it (RoleId), as real STS returns it. When no IAM driver is wired +// or the role is unknown it falls back to assumedRoleIDPrefix. +func (h *Handler) assumedRoleID(r *http.Request, roleArn, roleName, sessionName string) string { + prefix := assumedRoleIDPrefix + + if role, ok := h.requestedRole(r, roleArn, roleName); ok && role.ID != "" { + prefix = role.ID + } + + return prefix + ":" + sessionName +} + // trustAllows reports whether the caller may assume the role RoleArn names. // // Under EnforceAuth the gate has authenticated the caller and recorded its @@ -150,7 +164,7 @@ func (h *Handler) assumeRoleWithWebIdentity(w http.ResponseWriter, r *http.Reque provider = "cloudemu.local" } - assumedRoleID := assumedRoleIDPrefix + ":" + sessionName + assumedRoleID := h.assumedRoleID(r, r.Form.Get("RoleArn"), roleName, sessionName) creds, ok := h.mintCredentials(w, durationFromForm(r), awsidentity.Identity{ARN: assumedArn, UserID: assumedRoleID}, roleOwner(roleName)) @@ -185,7 +199,7 @@ func (h *Handler) assumeRoleWithSAML(w http.ResponseWriter, r *http.Request) { roleName := roleNameFromArn(r.Form.Get("RoleArn")) sessionName := "cloudemu-saml-session" assumedArn := "arn:aws:sts::" + h.accountID + ":assumed-role/" + roleName + "/" + sessionName - assumedRoleID := assumedRoleIDPrefix + ":" + sessionName + assumedRoleID := h.assumedRoleID(r, r.Form.Get("RoleArn"), roleName, sessionName) creds, ok := h.mintCredentials(w, durationFromForm(r), awsidentity.Identity{ARN: assumedArn, UserID: assumedRoleID}, roleOwner(roleName)) From 17d134b2be65895abb599454d5530026fe36cb73 Mon Sep 17 00:00:00 2001 From: Satyam Trivedi Date: Tue, 13 Oct 2026 00:21:09 +0530 Subject: [PATCH 2/2] fix(aws-sqs): publish queue gauges every minute and resolve SenderId Publish the AWS/SQS queue-state gauges (ApproximateNumberOfMessagesVisible, ApproximateNumberOfMessagesNotVisible, ApproximateNumberOfMessagesDelayed, ApproximateAgeOfOldestMessage and, for FIFO queues, ApproximateNumberOfGroupsWithInflightMessages) from a Tick on the shared serve scheduler: one sample per active queue per minute of provider-clock time, where a queue is active while it holds messages or was accessed in the last six hours. A queue whose consumers stopped keeps reporting a growing backlog and age, and each minute holds one sample, so Sum/Average/SampleCount match real SQS. The counters stay on the request paths and every metric is published after the queue lock is released. The access and sample times are part of the SQS snapshot. GetQueueAttributes ApproximateNumberOfMessagesNotVisible now counts in-flight messages only, not delayed ones, matching the API reference and the gauge. SenderId is the caller's IAM unique id (user id, or role id:session for an assumed role) resolved by the shared awsidentity.Resolver, matching GetCallerIdentity's UserId; library and service-to-service sends keep the account id. Part of #1514 --- providers/aws/aws.go | 6 +- providers/aws/sqs/metrics.go | 209 ++++++++++ providers/aws/sqs/metrics_test.go | 524 +++++++++++++++++++++++++ providers/aws/sqs/movetask.go | 2 +- providers/aws/sqs/permissions.go | 4 +- providers/aws/sqs/snapshot.go | 12 +- providers/aws/sqs/sqs.go | 108 ++--- providers/aws/sqs/tags.go | 6 +- providers/aws/tickables_test.go | 14 +- server/aws/aws.go | 19 +- server/aws/sqs/handler.go | 22 ++ server/aws/sqs_metrics_sdk_test.go | 260 ++++++++++++ services/messagequeue/driver/driver.go | 9 + 13 files changed, 1118 insertions(+), 77 deletions(-) create mode 100644 providers/aws/sqs/metrics.go create mode 100644 providers/aws/sqs/metrics_test.go create mode 100644 server/aws/sqs_metrics_sdk_test.go diff --git a/providers/aws/aws.go b/providers/aws/aws.go index eb2a1d7b3..8b848a975 100644 --- a/providers/aws/aws.go +++ b/providers/aws/aws.go @@ -637,9 +637,9 @@ func awsDrivers(p *Provider) *resourcediscovery.Drivers { // background ticker calls each one on its interval. New time-driven services // register here. func (p *Provider) Tickables() []config.Tickable { - // EFS runs before CloudWatch so the alarms evaluated in the same tick see - // the AWS/EFS metrics it just published. - return []config.Tickable{p.EFS, p.CloudWatch, p.SSM, p.CloudFormation} + // SQS and EFS run before CloudWatch so the alarms evaluated in the same + // tick see the queue gauges and AWS/EFS metrics they just published. + return []config.Tickable{p.SQS, p.EFS, p.CloudWatch, p.SSM, p.CloudFormation} } // Close tears down any real engines wired into the provider via diff --git a/providers/aws/sqs/metrics.go b/providers/aws/sqs/metrics.go new file mode 100644 index 000000000..f8bc603e9 --- /dev/null +++ b/providers/aws/sqs/metrics.go @@ -0,0 +1,209 @@ +package sqs + +import ( + "context" + "math" + "time" + + mondriver "github.com/stackshy/cloudemu/v2/services/monitoring/driver" +) + +// AWS/SQS queue-state gauges, published on the single QueueName dimension +// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-available-cloudwatch-metrics.html). +// +// Real SQS pushes these once a minute for every active queue: one that holds +// messages, or that an action accessed in the last six hours +// (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/monitoring-using-cloudwatch.html). +// Tick does the same on the shared serve scheduler, so a queue whose consumers +// stopped keeps reporting a growing backlog and age. The counters +// (NumberOfMessagesSent and friends) stay on the request paths. +const ( + metricsNamespace = "AWS/SQS" + unitCount = "Count" + unitSeconds = "Seconds" + + metricVisible = "ApproximateNumberOfMessagesVisible" + metricNotVisible = "ApproximateNumberOfMessagesNotVisible" + metricDelayed = "ApproximateNumberOfMessagesDelayed" + metricOldestAge = "ApproximateAgeOfOldestMessage" + metricGroupsInflight = "ApproximateNumberOfGroupsWithInflightMessages" + + // poisonPillReceiveCount is the receive count from which a standard queue + // moves an undeleted message to the back of the queue and leaves it out of + // ApproximateAgeOfOldestMessage until it is processed. + poisonPillReceiveCount = 3 + + // gaugeInterval is how often an active queue publishes its gauges. + gaugeInterval = time.Minute + // activeWindow is how long an action keeps an empty queue active. + activeWindow = 6 * time.Hour +) + +// queueGauges is one sample of a queue's state gauges. +type queueGauges struct { + visible, notVisible, delayed, groupsInflight int + // oldestAge is the age in whole seconds of the oldest message counted by + // ApproximateAgeOfOldestMessage; hasOldest is false when there is none. + oldestAge float64 + hasOldest bool +} + +// sampleGauges reads the queue's state gauges as of now. Caller holds qd.mu. +func sampleGauges(qd *queueData, now time.Time) queueGauges { + var ( + g queueGauges + oldest time.Time + groups map[string]struct{} + ) + + for _, msg := range qd.messages { + if g.countState(msg, now) && qd.info.FIFO && msg.GroupID != "" { + if groups == nil { + groups = make(map[string]struct{}) + } + + groups[msg.GroupID] = struct{}{} + } + + if countsForAge(qd.info.FIFO, msg) && (!g.hasOldest || msg.SentAt.Before(oldest)) { + oldest = msg.SentAt + g.hasOldest = true + } + } + + g.groupsInflight = len(groups) + + if g.hasOldest { + g.oldestAge = math.Max(0, math.Floor(now.Sub(oldest).Seconds())) + } + + return g +} + +// countState adds msg to the visible, delayed or in-flight count and reports +// whether it is in flight. A message that has never been received and whose +// visibility lies in the future is delayed; one that has been received and is +// still hidden is in flight. +func (g *queueGauges) countState(msg *sqsMessage, now time.Time) (inFlight bool) { + switch { + case !msg.VisibleAt.After(now): + g.visible++ + case msg.ReceiveCount == 0: + g.delayed++ + default: + g.notVisible++ + return true + } + + return false +} + +// countsForAge reports whether msg counts toward ApproximateAgeOfOldestMessage. +// A standard queue leaves out poison-pill messages (received three or more +// times without being deleted); FIFO queues keep order, so every message +// counts. +func countsForAge(fifo bool, msg *sqsMessage) bool { + return fifo || msg.ReceiveCount < poisonPillReceiveCount +} + +// Tick publishes one sample of the state gauges for every active queue whose +// last sample is at least gaugeInterval old at now, and reports whether it +// published anything. It has the config.Tickable signature of the shared serve +// scheduler; library mode never calls it, so tests drive it with a FakeClock. +// Each queue is sampled under its own lock and everything is published after +// all locks are released, because a CloudWatch alarm on a gauge may notify an +// SNS topic that delivers straight back into a queue. +func (m *Mock) Tick(now time.Time) bool { + if m.monitoring == nil { + return false + } + + var data []mondriver.MetricDatum + + for _, qd := range m.queues.SortedValues() { + data = append(data, dueGauges(qd, now)...) + } + + if len(data) == 0 { + return false + } + + _ = m.monitoring.PutMetricData(context.Background(), data) + + return true +} + +// dueGauges samples qd when it is active and its gauges are due at now, and +// records the sample time. It returns nil otherwise. Callers must not hold +// qd.mu. +func dueGauges(qd *queueData, now time.Time) []mondriver.MetricDatum { + qd.mu.Lock() + defer qd.mu.Unlock() + + if !qd.lastSampleAt.IsZero() && now.Sub(qd.lastSampleAt) < gaugeInterval { + return nil + } + + if len(qd.messages) == 0 && now.Sub(qd.lastAccessedAt()) >= activeWindow { + return nil + } + + qd.lastSampleAt = now + + return gaugeData(qd.info.Name, qd.info.FIFO, sampleGauges(qd, now), now) +} + +// gaugeData builds the datums of one gauge sample. +func gaugeData(queue string, fifo bool, g queueGauges, now time.Time) []mondriver.MetricDatum { + dims := map[string]string{"QueueName": queue} + datum := func(metric string, value float64, unit string) mondriver.MetricDatum { + return mondriver.MetricDatum{ + Namespace: metricsNamespace, MetricName: metric, Value: value, Unit: unit, + Dimensions: dims, Timestamp: now, + } + } + + data := []mondriver.MetricDatum{ + datum(metricVisible, float64(g.visible), unitCount), + datum(metricNotVisible, float64(g.notVisible), unitCount), + datum(metricDelayed, float64(g.delayed), unitCount), + } + + // The age is reported only while the queue holds a message it counts. + if g.hasOldest { + data = append(data, datum(metricOldestAge, g.oldestAge, unitSeconds)) + } + + if fifo { + data = append(data, datum(metricGroupsInflight, float64(g.groupsInflight), unitCount)) + } + + return data +} + +// touch records an action on the queue at now. +func (qd *queueData) touch(now time.Time) { + qd.lastAccessed.Store(now.UnixNano()) +} + +// lastAccessedAt is the time of the last action on the queue, or the zero +// time when none was recorded. +func (qd *queueData) lastAccessedAt() time.Time { + n := qd.lastAccessed.Load() + if n == 0 { + return time.Time{} + } + + return time.Unix(0, n).UTC() +} + +// accessQueue looks up the queue at url for an API action and records the +// access, which keeps the queue active for its CloudWatch gauges. +func (m *Mock) accessQueue(url string) (*queueData, bool) { + qd, ok := m.queues.Get(url) + if ok { + qd.touch(m.opts.Clock.Now()) + } + + return qd, ok +} diff --git a/providers/aws/sqs/metrics_test.go b/providers/aws/sqs/metrics_test.go new file mode 100644 index 000000000..12bfbd25d --- /dev/null +++ b/providers/aws/sqs/metrics_test.go @@ -0,0 +1,524 @@ +package sqs + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stackshy/cloudemu/v2/config" + "github.com/stackshy/cloudemu/v2/services/messagequeue/driver" + mondriver "github.com/stackshy/cloudemu/v2/services/monitoring/driver" +) + +// gaugeRecorder is a monitoring backend that keeps every datum it is sent. +// onPut, when set, runs inside PutMetricData, so a test can call back into +// the SQS mock the way a CloudWatch alarm -> SNS -> SQS action would. +type gaugeRecorder struct { + mondriver.Monitoring + + mu sync.Mutex + data []mondriver.MetricDatum + onPut func() +} + +func (r *gaugeRecorder) PutMetricData(_ context.Context, data []mondriver.MetricDatum) error { + r.mu.Lock() + r.data = append(r.data, data...) + onPut := r.onPut + r.mu.Unlock() + + if onPut != nil { + onPut() + } + + return nil +} + +// samples returns every datum of metric for queue, oldest first. +func (r *gaugeRecorder) samples(metric, queue string) []mondriver.MetricDatum { + r.mu.Lock() + defer r.mu.Unlock() + + var out []mondriver.MetricDatum + + for _, d := range r.data { + if d.MetricName == metric && d.Dimensions["QueueName"] == queue { + out = append(out, d) + } + } + + return out +} + +// last returns the most recent datum of metric for queue. +func (r *gaugeRecorder) last(metric, queue string) (mondriver.MetricDatum, bool) { + s := r.samples(metric, queue) + if len(s) == 0 { + return mondriver.MetricDatum{}, false + } + + return s[len(s)-1], true +} + +func (r *gaugeRecorder) reset() { + r.mu.Lock() + r.data = nil + r.mu.Unlock() +} + +func newGaugeMock() (*Mock, *gaugeRecorder, *config.FakeClock) { + m, fc := newTestMock() + rec := &gaugeRecorder{} + m.SetMonitoring(rec) + + return m, rec, fc +} + +func requireGauge(t *testing.T, rec *gaugeRecorder, metric, queue string, want float64, unit string) { + t.Helper() + + d, ok := rec.last(metric, queue) + if !ok { + t.Fatalf("%s for %s was never emitted", metric, queue) + } + + if d.Value != want || d.Unit != unit || d.Namespace != "AWS/SQS" { + t.Fatalf("%s for %s = %v %s (%s), want %v %s (AWS/SQS)", metric, queue, d.Value, d.Unit, d.Namespace, want, unit) + } + + if len(d.Dimensions) != 1 { + t.Fatalf("%s dimensions = %v, want only QueueName", metric, d.Dimensions) + } +} + +func requireNoGauge(t *testing.T, rec *gaugeRecorder, metric, queue string) { + t.Helper() + + if d, ok := rec.last(metric, queue); ok { + t.Fatalf("%s for %s emitted %v, want no datum", metric, queue, d.Value) + } +} + +func TestRequestsPublishCountersButNoGauges(t *testing.T) { + m, rec, _ := newGaugeMock() + ctx := context.Background() + q := createStdQueue(m, "quiet-paths") + + sendN(t, m, q.URL, 3) + + msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL}) + requireNoError(t, err) + requireNoError(t, m.ChangeVisibility(ctx, q.URL, msgs[0].ReceiptHandle, 0)) + requireNoError(t, m.DeleteMessage(ctx, q.URL, msgs[0].ReceiptHandle)) + requireNoError(t, m.PurgeQueue(ctx, q.URL)) + + for _, g := range []string{metricVisible, metricNotVisible, metricDelayed, metricOldestAge} { + requireNoGauge(t, rec, g, "quiet-paths") + } + + if got := len(rec.samples("NumberOfMessagesSent", "quiet-paths")); got != 3 { + t.Fatalf("NumberOfMessagesSent datums = %d, want 3", got) + } +} + +func TestTickPublishesOneSamplePerMinute(t *testing.T) { + m, rec, fc := newGaugeMock() + q := createStdQueue(m, "per-minute") + + // Six sends in the same minute: one sample of 6, not one per send. + sendN(t, m, q.URL, 6) + + if !m.Tick(fc.Now()) { + t.Fatal("first Tick published nothing") + } + + requireGauge(t, rec, metricVisible, "per-minute", 6, unitCount) + + for _, step := range []time.Duration{time.Second, 30 * time.Second, 28 * time.Second} { + fc.Advance(step) + + if m.Tick(fc.Now()) { + t.Fatalf("Tick %v after the sample published again inside the minute", fc.Now()) + } + } + + fc.Advance(time.Second) // a full minute since the first sample + + if !m.Tick(fc.Now()) { + t.Fatal("Tick a minute after the last sample published nothing") + } + + if got := len(rec.samples(metricVisible, "per-minute")); got != 2 { + t.Fatalf("Visible samples = %d, want 2 (one per minute)", got) + } +} + +func TestTickTracksVisibleInFlightDelayedAndAge(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + q := createStdQueue(m, "gauges") + + _, err := m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "a"}) + requireNoError(t, err) + + _, err = m.SendMessage(ctx, driver.SendMessageInput{ + QueueURL: q.URL, Body: "delayed", DelaySeconds: 600, DelaySecondsSet: true, + }) + requireNoError(t, err) + + fc.Advance(10 * time.Second) + + _, err = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "c"}) + requireNoError(t, err) + + msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL, VisibilityTimeout: 300}) + requireNoError(t, err) + assertEqual(t, "a", msgs[0].Body) + + m.Tick(fc.Now()) + + // a is in flight, one message is delayed, c is visible, and the oldest + // message was sent 10s ago. + requireGauge(t, rec, metricVisible, "gauges", 1, unitCount) + requireGauge(t, rec, metricNotVisible, "gauges", 1, unitCount) + requireGauge(t, rec, metricDelayed, "gauges", 1, unitCount) + requireGauge(t, rec, metricOldestAge, "gauges", 10, unitSeconds) + requireNoGauge(t, rec, metricGroupsInflight, "gauges") + + // GetQueueAttributes agrees with the gauges: in flight excludes delayed. + attrs, err := m.GetQueueAttributes(ctx, q.URL) + requireNoError(t, err) + assertEqual(t, 1, attrs.ApproximateMessageCount) + assertEqual(t, 1, attrs.ApproximateNotVisibleCount) + assertEqual(t, 1, attrs.ApproximateDelayedCount) +} + +// TestTickStuckQueueKeepsReportingAndAges is the stopped-consumer case: after +// a single delayed send nothing touches the queue, and the gauges still show +// the message becoming visible and getting older every minute. +func TestTickStuckQueueKeepsReportingAndAges(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + q := createStdQueue(m, "stuck") + + _, err := m.SendMessage(ctx, driver.SendMessageInput{ + QueueURL: q.URL, Body: "late", DelaySeconds: 5, DelaySecondsSet: true, + }) + requireNoError(t, err) + + m.Tick(fc.Now()) + requireGauge(t, rec, metricVisible, "stuck", 0, unitCount) + requireGauge(t, rec, metricDelayed, "stuck", 1, unitCount) + requireGauge(t, rec, metricOldestAge, "stuck", 0, unitSeconds) + + fc.Advance(time.Minute) + m.Tick(fc.Now()) + requireGauge(t, rec, metricVisible, "stuck", 1, unitCount) + requireGauge(t, rec, metricDelayed, "stuck", 0, unitCount) + requireGauge(t, rec, metricOldestAge, "stuck", 60, unitSeconds) + + // A queue holding messages stays active long past the six-hour window. + for i := 0; i < 8*60; i++ { + fc.Advance(time.Minute) + m.Tick(fc.Now()) + } + + requireGauge(t, rec, metricOldestAge, "stuck", (8*60+1)*60, unitSeconds) + + if got := len(rec.samples(metricVisible, "stuck")); got != 8*60+2 { + t.Fatalf("Visible samples = %d, want %d (one per minute)", got, 8*60+2) + } +} + +func TestTickStopsForEmptyQueueIdleSixHours(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + q := createStdQueue(m, "idle") + + if !m.Tick(fc.Now()) { + t.Fatal("a just-created queue is active and should publish") + } + + fc.Advance(activeWindow - time.Minute) + + if !m.Tick(fc.Now()) { + t.Fatal("an empty queue accessed under six hours ago should still publish") + } + + fc.Advance(time.Minute) + rec.reset() + + if m.Tick(fc.Now()) { + t.Fatal("an empty queue idle for six hours should stop publishing") + } + + // Any action wakes it up again. + _, err := m.GetQueueAttributes(ctx, q.URL) + requireNoError(t, err) + + fc.Advance(time.Minute) + + if !m.Tick(fc.Now()) { + t.Fatal("an accessed queue should publish again") + } + + requireGauge(t, rec, metricVisible, "idle", 0, unitCount) + requireNoGauge(t, rec, metricOldestAge, "idle") +} + +func TestTickAgeSkipsStandardPoisonPills(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + q := createStdQueue(m, "poison") + + _, err := m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "poison"}) + requireNoError(t, err) + + fc.Advance(100 * time.Second) + + receive := func() { + t.Helper() + + msgs, rerr := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{ + QueueURL: q.URL, VisibilityTimeout: 0, VisibilityTimeoutSet: true, + }) + requireNoError(t, rerr) + assertEqual(t, 1, len(msgs)) + } + + receive() + receive() + m.Tick(fc.Now()) + requireGauge(t, rec, metricOldestAge, "poison", 100, unitSeconds) + + // The third receive makes it a poison pill: nothing else is queued, so + // there is no age to report. + receive() + rec.reset() + fc.Advance(time.Minute) + m.Tick(fc.Now()) + requireGauge(t, rec, metricVisible, "poison", 1, unitCount) + requireNoGauge(t, rec, metricOldestAge, "poison") +} + +func TestTickFIFOCountsInflightGroups(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + + q, err := m.CreateQueue(ctx, driver.QueueConfig{Name: "orders.fifo", FIFO: true, ContentBasedDeduplication: true}) + requireNoError(t, err) + + for i, g := range []string{"g1", "g1", "g2", "g3"} { + _, err = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: fmt.Sprintf("%s-%d", g, i), GroupID: g}) + requireNoError(t, err) + } + + fc.Advance(30 * time.Second) + + msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL, MaxMessages: 10, VisibilityTimeout: 300}) + requireNoError(t, err) + assertEqual(t, 4, len(msgs)) + + m.Tick(fc.Now()) + requireGauge(t, rec, metricGroupsInflight, "orders.fifo", 3, unitCount) + requireGauge(t, rec, metricNotVisible, "orders.fifo", 4, unitCount) + requireGauge(t, rec, metricOldestAge, "orders.fifo", 30, unitSeconds) + + for _, msg := range msgs[:2] { + requireNoError(t, m.DeleteMessage(ctx, q.URL, msg.ReceiptHandle)) + } + + fc.Advance(time.Minute) + m.Tick(fc.Now()) + requireGauge(t, rec, metricGroupsInflight, "orders.fifo", 2, unitCount) + requireGauge(t, rec, metricNotVisible, "orders.fifo", 2, unitCount) +} + +func TestTickFollowsDeadLetterRedriveAndMoveTask(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + + dlq := createStdQueue(m, "dlq") + src, err := m.CreateQueue(ctx, driver.QueueConfig{Name: "src", RedrivePolicy: redrivePolicy(dlq.ARN, 1)}) + requireNoError(t, err) + + sendN(t, m, src.URL, 1) + fc.Advance(50 * time.Second) + + receive := func() { + t.Helper() + + _, rerr := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: src.URL, VisibilityTimeout: 1}) + requireNoError(t, rerr) + } + + receive() + fc.Advance(2 * time.Second) + receive() // exceeds maxReceiveCount=1: the message moves to the DLQ + + m.Tick(fc.Now()) + requireGauge(t, rec, metricVisible, "src", 0, unitCount) + requireGauge(t, rec, metricVisible, "dlq", 1, unitCount) + // The age resets when a message is moved to a DLQ. + requireGauge(t, rec, metricOldestAge, "dlq", 0, unitSeconds) + + fc.Advance(time.Minute) + requireGaugeAfterMove(t, m, rec, fc, dlq.ARN) +} + +func requireGaugeAfterMove(t *testing.T, m *Mock, rec *gaugeRecorder, fc *config.FakeClock, dlqARN string) { + t.Helper() + + _, err := m.StartMessageMoveTask(context.Background(), dlqARN, "", 0) + requireNoError(t, err) + + m.Tick(fc.Now()) + requireGauge(t, rec, metricVisible, "dlq", 0, unitCount) + requireGauge(t, rec, metricVisible, "src", 1, unitCount) +} + +// TestTickPublishesOutsideQueueLocks calls back into the queues from inside +// PutMetricData, as a CloudWatch alarm whose SNS action delivers to a queue +// would. Publishing while holding a queue lock would deadlock here. +func TestTickPublishesOutsideQueueLocks(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + q := createStdQueue(m, "reentrant") + sendN(t, m, q.URL, 1) + + var called atomic.Bool + + // Only the first publish calls back, or the SendMessage counter it emits + // would call back again forever. + rec.onPut = func() { + if called.CompareAndSwap(false, true) { + _, _ = m.GetQueueAttributes(ctx, q.URL) + _, _ = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "from-alarm"}) + } + } + + done := make(chan struct{}) + + go func() { + defer close(done) + + m.Tick(fc.Now()) + + msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL}) + if err == nil && len(msgs) == 1 { + _ = m.DeleteMessage(ctx, q.URL, msgs[0].ReceiptHandle) + } + + _, _ = m.ReceiveMessagesWithOptions(ctx, q.URL, driver.ReceiveOptions{}) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("publishing a metric deadlocked on a queue lock") + } +} + +func TestTickWithoutMonitoringIsNoOp(t *testing.T) { + m, fc := newTestMock() + q := createStdQueue(m, "no-backend") + sendN(t, m, q.URL, 1) + + if m.Tick(fc.Now()) { + t.Fatal("Tick without a monitoring backend reported a change") + } +} + +func TestSnapshotKeepsGaugeSchedule(t *testing.T) { + ctx := context.Background() + src, rec, fc := newGaugeMock() + createStdQueue(src, "sched") + + src.Tick(fc.Now()) + requireGauge(t, rec, metricVisible, "sched", 0, unitCount) + + data, err := src.Snapshot(ctx, false) + requireNoError(t, err) + + dst, dstRec, _ := newGaugeMock() + requireNoError(t, dst.Restore(ctx, data)) + + // The restored queue sampled at fc.Now(), so it is not due again yet. + if dst.Tick(fc.Now().Add(30 * time.Second)) { + t.Fatal("restored queue lost its last sample time") + } + + // It was accessed at fc.Now(), so it is still active a minute later. + if !dst.Tick(fc.Now().Add(time.Minute)) { + t.Fatal("restored queue lost its last access time") + } + + requireGauge(t, dstRec, metricVisible, "sched", 0, unitCount) + + // Six hours after that access the empty queue is idle. + if dst.Tick(fc.Now().Add(activeWindow + time.Minute)) { + t.Fatal("restored empty queue should be idle after six hours") + } +} + +var errHandlerFailed = errors.New("handler failed") + +// failingInvoker is an event-source mapping whose handler always fails. +type failingInvoker struct{} + +func (failingInvoker) DeliverEventSourceBatch(context.Context, string, []byte) (bool, error) { + return true, errHandlerFailed +} + +func TestTickFollowsLambdaESMDeadLetterRedrive(t *testing.T) { + m, rec, fc := newGaugeMock() + ctx := context.Background() + + dlq := createStdQueue(m, "esm-dlq") + src, err := m.CreateQueue(ctx, driver.QueueConfig{Name: "esm-src", RedrivePolicy: redrivePolicy(dlq.ARN, 2)}) + requireNoError(t, err) + + m.SetEventSourceInvoker(failingInvoker{}) + + sendN(t, m, src.URL, 1) + m.Tick(fc.Now()) + + requireGauge(t, rec, metricVisible, "esm-dlq", 1, unitCount) + requireGauge(t, rec, metricVisible, "esm-src", 0, unitCount) + requireGauge(t, rec, metricNotVisible, "esm-src", 0, unitCount) +} + +func TestSenderIDFromCallerElseAccount(t *testing.T) { + m, _ := newTestMock() + ctx := context.Background() + q := createStdQueue(m, "sender") + + _, err := m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "user", SenderID: "AIDAEXAMPLE123ABC"}) + requireNoError(t, err) + + res, err := m.SendMessageBatch(ctx, q.URL, []driver.BatchSendEntry{ + {ID: "1", Body: "role", SenderID: "AROAEXAMPLE123ABC:session"}, + }) + requireNoError(t, err) + assertEqual(t, 1, len(res.Successful)) + + _, err = m.SendMessage(ctx, driver.SendMessageInput{QueueURL: q.URL, Body: "library"}) + requireNoError(t, err) + + msgs, err := m.ReceiveMessages(ctx, driver.ReceiveMessageInput{QueueURL: q.URL, MaxMessages: 10}) + requireNoError(t, err) + assertEqual(t, 3, len(msgs)) + + got := map[string]string{} + for _, msg := range msgs { + got[msg.Body] = msg.SystemAttributes["SenderId"] + } + + assertEqual(t, "AIDAEXAMPLE123ABC", got["user"]) + assertEqual(t, "AROAEXAMPLE123ABC:session", got["role"]) + assertEqual(t, m.opts.AccountID, got["library"]) +} diff --git a/providers/aws/sqs/movetask.go b/providers/aws/sqs/movetask.go index 8cdf84a03..59e45daec 100644 --- a/providers/aws/sqs/movetask.go +++ b/providers/aws/sqs/movetask.go @@ -50,7 +50,7 @@ func (m *Mock) StartMessageMoveTask(_ context.Context, sourceARN, destARN string } sourceURL := m.urlForARN(sourceARN) - if sourceURL == "" { + if _, ok := m.accessQueue(sourceURL); !ok { return "", errors.Newf(errors.NotFound, "no queue found for source arn %q", sourceARN) } diff --git a/providers/aws/sqs/permissions.go b/providers/aws/sqs/permissions.go index 92655940b..5d00f2982 100644 --- a/providers/aws/sqs/permissions.go +++ b/providers/aws/sqs/permissions.go @@ -11,7 +11,7 @@ import ( // the given SQS actions on the queue, mutating the stored access policy. It is // AWS-specific and used directly by the SQS wire handler. func (m *Mock) AddPermission(_ context.Context, queueURL, label string, accountIDs, actions []string) error { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queueURL) } @@ -50,7 +50,7 @@ func (m *Mock) AddPermission(_ context.Context, queueURL, label string, accountI // RemovePermission removes the statement whose Sid equals label. It is // AWS-specific and used directly by the SQS wire handler. func (m *Mock) RemovePermission(_ context.Context, queueURL, label string) error { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queueURL) } diff --git a/providers/aws/sqs/snapshot.go b/providers/aws/sqs/snapshot.go index 8d505759f..6e38a52af 100644 --- a/providers/aws/sqs/snapshot.go +++ b/providers/aws/sqs/snapshot.go @@ -48,6 +48,11 @@ type queueSnapshot struct { DeduplicationIndex map[string]time.Time `json:"deduplicationIndex,omitempty"` DLQConfig *driver.DeadLetterConfig `json:"dlqConfig,omitempty"` SeqCounter uint64 `json:"seqCounter,omitempty"` + // LastAccessedAt and LastSampleAt drive the per-minute CloudWatch gauges + // (metrics.go): whether the queue is still active, and when it last + // published. + LastAccessedAt time.Time `json:"lastAccessedAt"` + LastSampleAt time.Time `json:"lastSampleAt"` } // messageSnapshot mirrors sqsMessage, promoting its one unexported field @@ -132,6 +137,7 @@ func snapshotQueue(qd *queueData) *queueSnapshot { FifoThroughputLim: qd.fifoThroughputLim, CreatedAt: qd.createdAt, LastModifiedAt: qd.lastModifiedAt, DeduplicationIndex: qd.deduplicationIndex, DLQConfig: qd.dlqConfig, SeqCounter: qd.seqCounter.Load(), + LastAccessedAt: qd.lastAccessedAt(), LastSampleAt: qd.lastSampleAt, } if len(qd.messages) > 0 { @@ -190,7 +196,11 @@ func restoreQueue(qs *queueSnapshot) *queueData { sqsManagedSSE: qs.SqsManagedSSE, deduplicationScope: qs.DeduplicationScope, fifoThroughputLim: qs.FifoThroughputLim, createdAt: qs.CreatedAt, lastModifiedAt: qs.LastModifiedAt, deduplicationIndex: qs.DeduplicationIndex, - dlqConfig: qs.DLQConfig, + dlqConfig: qs.DLQConfig, lastSampleAt: qs.LastSampleAt, + } + + if !qs.LastAccessedAt.IsZero() { + qd.touch(qs.LastAccessedAt) } if qd.deduplicationIndex == nil { diff --git a/providers/aws/sqs/sqs.go b/providers/aws/sqs/sqs.go index d2c710159..c8b53223a 100644 --- a/providers/aws/sqs/sqs.go +++ b/providers/aws/sqs/sqs.go @@ -87,6 +87,13 @@ type queueData struct { deduplicationIndex map[string]time.Time dlqConfig *driver.DeadLetterConfig seqCounter atomic.Uint64 + + // lastAccessed is the provider-clock time (Unix nanoseconds) of the last + // action on the queue. With the message count it decides whether the + // queue is active and so publishes its CloudWatch gauges (see metrics.go). + lastAccessed atomic.Int64 + // lastSampleAt is when the gauges were last published. Guarded by mu. + lastSampleAt time.Time } // EventSourceInvoker delivers an SQS event-source-mapping batch to whatever @@ -123,7 +130,7 @@ func (m *Mock) emitMetric(metricName string, value float64, unit string, dims ma } _ = m.monitoring.PutMetricData(context.Background(), []mondriver.MetricDatum{{ - Namespace: "AWS/SQS", MetricName: metricName, Value: value, Unit: unit, + Namespace: metricsNamespace, MetricName: metricName, Value: value, Unit: unit, Dimensions: dims, Timestamp: m.opts.Clock.Now(), }}) } @@ -197,7 +204,7 @@ func (m *Mock) CreateQueue(ctx context.Context, cfg driver.QueueConfig) (*driver // CreateQueue is idempotent: re-creating an existing queue with the exact // same attributes returns the existing URL. QueueNameExists is returned only // when the incoming attributes differ from the stored ones. - if existing, ok := m.queues.Get(url); ok { + if existing, ok := m.accessQueue(url); ok { if sameQueueConfig(existing, &cfg) { result := existing.info return &result, nil @@ -255,6 +262,7 @@ func (m *Mock) CreateQueue(ctx context.Context, cfg driver.QueueConfig) (*driver } } + qd.touch(now) m.queues.Set(url, qd) result := info @@ -443,7 +451,7 @@ func (m *Mock) DeleteQueue(_ context.Context, url string) error { // GetQueueInfo retrieves information about an SQS queue by URL. func (m *Mock) GetQueueInfo(_ context.Context, url string) (*driver.QueueInfo, error) { - qd, ok := m.queues.Get(url) + qd, ok := m.accessQueue(url) if !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", url) } @@ -487,7 +495,7 @@ func (m *Mock) ListQueues(_ context.Context, prefix string) ([]driver.QueueInfo, // //nolint:gocritic // hugeParam: interface method signature cannot be changed. func (m *Mock) SendMessage(ctx context.Context, input driver.SendMessageInput) (*driver.SendMessageOutput, error) { - qd, ok := m.queues.Get(input.QueueURL) + qd, ok := m.accessQueue(input.QueueURL) if !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", input.QueueURL) } @@ -769,13 +777,24 @@ func (m *Mock) buildStoredMessage(qd *queueData, input *driver.SendMessageInput, Attributes: attrs, MessageAttributes: copyMessageAttributes(input.MessageAttributes), SystemAttributes: systemAttributeStrings(input.SystemAttributes), - SenderID: m.opts.AccountID, + SenderID: senderID(input.SenderID, m.opts.AccountID), SequenceNumber: seqNum, VisibleAt: now.Add(time.Duration(delaySeconds) * time.Second), SentAt: now, } } +// senderID is the SenderId recorded on a sent message: the caller's IAM unique +// id when the wire handler resolved one, else the account id (library callers +// and service-to-service deliveries carry no caller principal). +func senderID(caller, accountID string) string { + if caller != "" { + return caller + } + + return accountID +} + // systemAttributeStrings flattens caller-supplied SQS message system attributes // (e.g. AWSTraceHeader) into a name->string map for storage. Only the string // value is retained; system attributes are always String-typed in SQS. @@ -868,7 +887,7 @@ func findDuplicate(qd *queueData, input *driver.SendMessageInput, now time.Time) // When WaitTimeSeconds (or the queue's ReceiveMessageWaitTimeSeconds default) is > 0, it long-polls: // it retries on a short interval until a message is available, the deadline elapses, or ctx is canceled. func (m *Mock) ReceiveMessages(ctx context.Context, input driver.ReceiveMessageInput) ([]driver.Message, error) { - qd, ok := m.queues.Get(input.QueueURL) + qd, ok := m.accessQueue(input.QueueURL) if !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", input.QueueURL) } @@ -1125,7 +1144,7 @@ func (m *Mock) moveToDLQ(dlqURL, sourceURL string, msg *sqsMessage) { // DeleteMessage deletes a message from the specified queue using its receipt handle. func (m *Mock) DeleteMessage(_ context.Context, queueURL, receiptHandle string) error { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queueURL) } @@ -1138,19 +1157,28 @@ func (m *Mock) DeleteMessage(_ context.Context, queueURL, receiptHandle string) } qd.mu.Lock() - defer qd.mu.Unlock() + + deleted := false for i, msg := range qd.messages { if msg.ReceiptHandle == receiptHandle { qd.messages = append(qd.messages[:i], qd.messages[i+1:]...) - m.emitMetric("NumberOfMessagesDeleted", 1, "Count", map[string]string{"QueueName": qd.info.Name}) + deleted = true - return nil + break } } + qd.mu.Unlock() + // DeleteMessage is idempotent: an old/stale (but well-formed) receipt handle - // against an existing queue succeeds without deleting anything. + // against an existing queue succeeds without deleting anything. The metric + // is published after the lock is released, since an alarm on it may + // deliver back into this queue. + if deleted { + m.emitMetric("NumberOfMessagesDeleted", 1, "Count", map[string]string{"QueueName": qd.info.Name}) + } + return nil } @@ -1178,7 +1206,7 @@ func isWellFormedReceiptHandle(handle string) bool { // ChangeVisibility changes the visibility timeout of a message in the specified queue. func (m *Mock) ChangeVisibility(_ context.Context, queueURL, receiptHandle string, timeout int) error { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queueURL) } @@ -1244,6 +1272,7 @@ func batchEntryToSendInput(queue string, entry *driver.BatchSendEntry) driver.Se Attributes: entry.Attributes, MessageAttributes: entry.MessageAttributes, SystemAttributes: entry.SystemAttributes, + SenderID: entry.SenderID, } } @@ -1279,7 +1308,7 @@ func (m *Mock) DeleteMessageBatch( func (m *Mock) ReceiveMessagesWithOptions( _ context.Context, queue string, opts driver.ReceiveOptions, ) ([]driver.Message, error) { - qd, ok := m.queues.Get(queue) + qd, ok := m.accessQueue(queue) if !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", queue) } @@ -1289,7 +1318,6 @@ func (m *Mock) ReceiveMessagesWithOptions( } qd.mu.Lock() - defer qd.mu.Unlock() maxMsgs := defaultMaxMessages(opts.MaxMessages) @@ -1302,17 +1330,14 @@ func (m *Mock) ReceiveMessagesWithOptions( results, toRemove := m.collectVisibleMessages(qd, maxMsgs, visTimeout, now) removeByIndices(qd, toRemove) + qd.mu.Unlock() if results == nil { results = []driver.Message{} } - dims := map[string]string{"QueueName": qd.info.Name} - if len(results) > 0 { - m.emitMetric("NumberOfMessagesReceived", float64(len(results)), "Count", dims) - } else { - m.emitMetric("NumberOfEmptyReceives", 1.0, "Count", dims) - } + // Published after the lock is released, like ReceiveMessages. + m.emitReceiveMetrics(qd, len(results)) return results, nil } @@ -1321,7 +1346,7 @@ func (m *Mock) ReceiveMessagesWithOptions( func (m *Mock) GetQueueAttributes( _ context.Context, queue string, ) (*driver.QueueAttributes, error) { - qd, ok := m.queues.Get(queue) + qd, ok := m.accessQueue(queue) if !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", queue) } @@ -1329,9 +1354,11 @@ func (m *Mock) GetQueueAttributes( qd.mu.Lock() defer qd.mu.Unlock() - now := m.opts.Clock.Now() - visible, notVisible := countMessages(qd, now) - delayed := countDelayedMessages(qd, now) + // ApproximateNumberOfMessagesNotVisible counts in-flight messages only + // (received, not yet deleted or expired); delayed messages are reported + // separately, matching the CloudWatch gauges. + g := sampleGauges(qd, m.opts.Clock.Now()) + visible, notVisible, delayed := g.visible, g.notVisible, g.delayed return &driver.QueueAttributes{ DelaySeconds: qd.delaySeconds, @@ -1357,24 +1384,11 @@ func (m *Mock) GetQueueAttributes( }, nil } -// countDelayedMessages counts messages still within their delay window. -func countDelayedMessages(qd *queueData, now time.Time) int { - delayed := 0 - - for _, msg := range qd.messages { - if msg.ReceiveCount == 0 && msg.VisibleAt.After(now) { - delayed++ - } - } - - return delayed -} - // SetQueueAttributes updates the attributes of the specified queue. func (m *Mock) SetQueueAttributes( _ context.Context, queue string, attrs map[string]int, ) error { - qd, ok := m.queues.Get(queue) + qd, ok := m.accessQueue(queue) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queue) } @@ -1415,7 +1429,7 @@ func applyQueueAttributes(qd *queueData, attrs map[string]int) { // Policy, KmsMasterKeyId) that the numeric SetQueueAttributes path cannot express. // It is AWS-specific and used directly by the SQS wire handler. func (m *Mock) SetQueueAttributesRaw(_ context.Context, queue string, attrs map[string]string) error { - qd, ok := m.queues.Get(queue) + qd, ok := m.accessQueue(queue) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queue) } @@ -1512,7 +1526,7 @@ func parseNumericAttrs(attrs map[string]string) map[string]int { // ListDeadLetterSourceQueues returns the URLs of queues whose RedrivePolicy // targets the given DLQ URL. func (m *Mock) ListDeadLetterSourceQueues(_ context.Context, dlqURL string) ([]string, error) { - if _, ok := m.queues.Get(dlqURL); !ok { + if _, ok := m.accessQueue(dlqURL); !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", dlqURL) } @@ -1533,7 +1547,7 @@ func (m *Mock) ListDeadLetterSourceQueues(_ context.Context, dlqURL string) ([]s // PurgeQueue removes all messages from the specified queue. func (m *Mock) PurgeQueue(_ context.Context, queue string) error { - qd, ok := m.queues.Get(queue) + qd, ok := m.accessQueue(queue) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queue) } @@ -1547,18 +1561,6 @@ func (m *Mock) PurgeQueue(_ context.Context, queue string) error { return nil } -func countMessages(qd *queueData, now time.Time) (visible, notVisible int) { - for _, msg := range qd.messages { - if msg.VisibleAt.After(now) { - notVisible++ - } else { - visible++ - } - } - - return visible, notVisible -} - func removeByIndices(qd *queueData, indices []int) { for i := len(indices) - 1; i >= 0; i-- { idx := indices[i] diff --git a/providers/aws/sqs/tags.go b/providers/aws/sqs/tags.go index 2d3c0f542..598804b61 100644 --- a/providers/aws/sqs/tags.go +++ b/providers/aws/sqs/tags.go @@ -8,7 +8,7 @@ import ( // TagQueue adds or overwrites tags on a queue (SQS TagQueue). func (m *Mock) TagQueue(_ context.Context, queueURL string, tags map[string]string) error { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queueURL) } @@ -29,7 +29,7 @@ func (m *Mock) TagQueue(_ context.Context, queueURL string, tags map[string]stri // UntagQueue removes tags by key from a queue (SQS UntagQueue). func (m *Mock) UntagQueue(_ context.Context, queueURL string, keys []string) error { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return errors.Newf(errors.NotFound, "queue %q not found", queueURL) } @@ -46,7 +46,7 @@ func (m *Mock) UntagQueue(_ context.Context, queueURL string, keys []string) err // ListQueueTags returns a queue's tags (SQS ListQueueTags). func (m *Mock) ListQueueTags(_ context.Context, queueURL string) (map[string]string, error) { - qd, ok := m.queues.Get(queueURL) + qd, ok := m.accessQueue(queueURL) if !ok { return nil, errors.Newf(errors.NotFound, "queue %q not found", queueURL) } diff --git a/providers/aws/tickables_test.go b/providers/aws/tickables_test.go index 08d4b72b0..2347f0c1a 100644 --- a/providers/aws/tickables_test.go +++ b/providers/aws/tickables_test.go @@ -8,15 +8,17 @@ import ( ) // TestTickablesRegistersTimeDrivenServices checks the serve ticker reaches the -// region's own EFS, CloudWatch, SSM and CloudFormation, so EFS metrics, due -// alarms, parameter policies and settling stack operations run without a read. -// EFS comes before CloudWatch so alarms see the metrics published in the same tick. +// region's own SQS, EFS, CloudWatch, SSM and CloudFormation, so queue gauges, +// EFS metrics, due alarms, parameter policies and settling stack operations run +// without a read. SQS and EFS come before CloudWatch so alarms see the metrics +// published in the same tick. func TestTickablesRegistersTimeDrivenServices(t *testing.T) { p := aws.New() got := p.Tickables() - if len(got) != 4 || got[0] != config.Tickable(p.EFS) || got[1] != config.Tickable(p.CloudWatch) || - got[2] != config.Tickable(p.SSM) || got[3] != config.Tickable(p.CloudFormation) { - t.Fatalf("Tickables() = %v, want this provider's EFS, CloudWatch, SSM and CloudFormation", got) + if len(got) != 5 || got[0] != config.Tickable(p.SQS) || got[1] != config.Tickable(p.EFS) || + got[2] != config.Tickable(p.CloudWatch) || got[3] != config.Tickable(p.SSM) || + got[4] != config.Tickable(p.CloudFormation) { + t.Fatalf("Tickables() = %v, want this provider's SQS, EFS, CloudWatch, SSM and CloudFormation", got) } } diff --git a/server/aws/aws.go b/server/aws/aws.go index 14ac0e0f9..ffe0533fe 100644 --- a/server/aws/aws.go +++ b/server/aws/aws.go @@ -623,6 +623,14 @@ type authzSets struct { func newServer(d Drivers) (*server.Server, authzSets) { srv := server.New() + // STS records the sessions it mints here; EKS reads the cluster creator and + // SQS the message SenderId from it, so all of them see the caller the same + // way. + identities := d.Identities + if identities == nil { + identities = awsidentity.New(d.AccountID, d.IAM) + } + // The auth gate reads X-Amz-Target only for the handlers registered // through rpc, which route on that header. jsonRPC := map[server.Handler]bool{} @@ -673,7 +681,9 @@ func newServer(d Drivers) (*server.Server, authzSets) { // prefix (AmazonSQS.* vs DynamoDB_20120810.*); their Matches predicates // are mutually exclusive. if d.SQS != nil { - srv.Register(sqs.New(d.SQS)) + sqsHandler := sqs.New(d.SQS) + sqsHandler.SetIdentities(identities) + srv.Register(sqsHandler) } // Resource Groups Tagging API: X-Amz-Target prefix @@ -1139,13 +1149,6 @@ func newServer(d Drivers) (*server.Server, authzSets) { stsSessions = stssrv.NewSessionStore(authClock) } - // STS records the sessions it mints here, and EKS reads the cluster - // creator from it, so both see the caller the same way. - identities := d.Identities - if identities == nil { - identities = awsidentity.New(d.AccountID, d.IAM) - } - if d.STS { // Pass the IAM driver so AssumeRole can enforce the target role's trust // policy and existence. d.IAM may be nil (standalone STS-only), in which diff --git a/server/aws/sqs/handler.go b/server/aws/sqs/handler.go index 81a1b4b14..5335ff580 100644 --- a/server/aws/sqs/handler.go +++ b/server/aws/sqs/handler.go @@ -18,6 +18,7 @@ import ( cerrors "github.com/stackshy/cloudemu/v2/errors" "github.com/stackshy/cloudemu/v2/server/wire" + "github.com/stackshy/cloudemu/v2/server/wire/awsidentity" mqdriver "github.com/stackshy/cloudemu/v2/services/messagequeue/driver" ) @@ -57,6 +58,9 @@ const ( // driver. type Handler struct { mq mqdriver.MessageQueue + // identities, when set, resolves the caller of a send the way STS + // GetCallerIdentity reports it, for the SenderId message attribute. + identities *awsidentity.Resolver } // New returns an SQS handler backed by mq. @@ -64,6 +68,21 @@ func New(mq mqdriver.MessageQueue) *Handler { return &Handler{mq: mq} } +// SetIdentities shares the resolver STS uses, so a message's SenderId is the +// caller's IAM unique id: the user id for an IAM user, ":" +// for an assumed role. Without it the provider records the account id. +func (h *Handler) SetIdentities(r *awsidentity.Resolver) { h.identities = r } + +// senderID resolves the SenderId of a send request, or "" when no resolver +// is wired. +func (h *Handler) senderID(r *http.Request) string { + if h.identities == nil { + return "" + } + + return h.identities.Resolve(r).UserID +} + // Matches identifies SQS requests by their X-Amz-Target header. SQS shares // the same content-type as DynamoDB (application/x-amz-json-1.0) so the // header prefix is the only reliable discriminator. @@ -311,6 +330,7 @@ func (h *Handler) sendMessage(w http.ResponseWriter, r *http.Request) { DeduplicationID: req.DeduplicationID, MessageAttributes: msgAttrs, SystemAttributes: sysAttrs, + SenderID: h.senderID(r), }) if err != nil { writeErr(w, err) @@ -578,6 +598,7 @@ func (h *Handler) sendMessageBatch(w http.ResponseWriter, r *http.Request) { sysAttrsByID := make(map[string]map[string]mqdriver.MessageAttributeValue, len(req.Entries)) entries := make([]mqdriver.BatchSendEntry, 0, len(req.Entries)) + sender := h.senderID(r) for i := range req.Entries { msgAttrs := toDriverMessageAttributes(req.Entries[i].MessageAttributes) @@ -594,6 +615,7 @@ func (h *Handler) sendMessageBatch(w http.ResponseWriter, r *http.Request) { DeduplicationID: req.Entries[i].MessageDeduplicationID, MessageAttributes: msgAttrs, SystemAttributes: sysAttrs, + SenderID: sender, }) } diff --git a/server/aws/sqs_metrics_sdk_test.go b/server/aws/sqs_metrics_sdk_test.go new file mode 100644 index 000000000..233530af0 --- /dev/null +++ b/server/aws/sqs_metrics_sdk_test.go @@ -0,0 +1,260 @@ +package aws_test + +import ( + "context" + "net/http/httptest" + "slices" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/cloudwatch" + cwtypes "github.com/aws/aws-sdk-go-v2/service/cloudwatch/types" + "github.com/aws/aws-sdk-go-v2/service/iam" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" + "github.com/aws/aws-sdk-go-v2/service/sts" + "github.com/stretchr/testify/require" + + "github.com/stackshy/cloudemu/v2" + "github.com/stackshy/cloudemu/v2/config" + awsserver "github.com/stackshy/cloudemu/v2/server/aws" +) + +// sqsGaugeStat reads one statistic of an AWS/SQS metric for a queue over the +// last few minutes through the real CloudWatch client. +func sqsGaugeStat( + t *testing.T, cw *cloudwatch.Client, metric, queue string, stat cwtypes.Statistic, +) (value float64, unit cwtypes.StandardUnit, found bool) { + t.Helper() + + now := time.Now().UTC() + + out, err := cw.GetMetricStatistics(context.Background(), &cloudwatch.GetMetricStatisticsInput{ + Namespace: aws.String("AWS/SQS"), + MetricName: aws.String(metric), + Dimensions: []cwtypes.Dimension{{Name: aws.String("QueueName"), Value: aws.String(queue)}}, + StartTime: aws.Time(now.Add(-5 * time.Minute)), + EndTime: aws.Time(now.Add(5 * time.Minute)), + Period: aws.Int32(600), + Statistics: []cwtypes.Statistic{stat}, + }) + require.NoError(t, err) + + if len(out.Datapoints) == 0 { + return 0, "", false + } + + dp := out.Datapoints[0] + + switch stat { + case cwtypes.StatisticMaximum: + return aws.ToFloat64(dp.Maximum), dp.Unit, true + case cwtypes.StatisticMinimum: + return aws.ToFloat64(dp.Minimum), dp.Unit, true + case cwtypes.StatisticSum: + return aws.ToFloat64(dp.Sum), dp.Unit, true + default: + return aws.ToFloat64(dp.SampleCount), dp.Unit, true + } +} + +// sqsMetricNames lists the AWS/SQS metric names CloudWatch knows. +func sqsMetricNames(t *testing.T, cw *cloudwatch.Client) []string { + t.Helper() + + metrics, err := cw.ListMetrics(context.Background(), &cloudwatch.ListMetricsInput{Namespace: aws.String("AWS/SQS")}) + require.NoError(t, err) + + names := make([]string, 0, len(metrics.Metrics)) + for _, m := range metrics.Metrics { + names = append(names, aws.ToString(m.MetricName)) + } + + return names +} + +// tickProvider runs one background tick of every time-driven service, the +// way the serve scheduler does. +func tickProvider(p interface{ Tickables() []config.Tickable }) { + now := time.Now() + for _, t := range p.Tickables() { + t.Tick(now) + } +} + +// TestSDKSQSQueueGaugesReachCloudWatch drives SQS with the real SDK and reads +// the queue-state gauges back through CloudWatch, the way a dashboard or an +// alarm on ApproximateNumberOfMessagesVisible would. The gauges come from the +// background tick, not from the requests. +func TestSDKSQSQueueGaugesReachCloudWatch(t *testing.T) { + provider := cloudemu.NewAWS() + ts := httptest.NewServer(awsserver.New(awsserver.Drivers{CloudWatch: provider.CloudWatch, SQS: provider.SQS})) + defer ts.Close() + + cfg := sdkConfig(t, ts.URL) + ctx := context.Background() + sqsClient := sqs.NewFromConfig(cfg) + cw := cloudwatch.NewFromConfig(cfg) + + q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("gauge-q")}) + require.NoError(t, err) + + _, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: q.QueueUrl, MessageBody: aws.String("now")}) + require.NoError(t, err) + + _, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{ + QueueUrl: q.QueueUrl, MessageBody: aws.String("later"), DelaySeconds: 600, + }) + require.NoError(t, err) + + recv, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{QueueUrl: q.QueueUrl, VisibilityTimeout: 300}) + require.NoError(t, err) + require.Len(t, recv.Messages, 1) + + // Requests publish the counters only. + require.False(t, slices.Contains(sqsMetricNames(t, cw), "ApproximateNumberOfMessagesVisible")) + + tickProvider(provider) + + names := sqsMetricNames(t, cw) + for _, want := range []string{ + "ApproximateNumberOfMessagesVisible", "ApproximateNumberOfMessagesNotVisible", + "ApproximateNumberOfMessagesDelayed", "ApproximateAgeOfOldestMessage", + } { + require.True(t, slices.Contains(names, want), "ListMetrics AWS/SQS missing %s (got %v)", want, names) + } + + // One sample: nothing visible, the received message in flight, the + // delayed one delayed. + visible, unit, ok := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "gauge-q", cwtypes.StatisticMaximum) + require.True(t, ok) + require.InDelta(t, 0, visible, 0) + require.Equal(t, cwtypes.StandardUnitCount, unit) + + samples, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "gauge-q", cwtypes.StatisticSampleCount) + require.InDelta(t, 1, samples, 0) + + delayed, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesDelayed", "gauge-q", cwtypes.StatisticMaximum) + require.InDelta(t, 1, delayed, 0) + + inFlight, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesNotVisible", "gauge-q", cwtypes.StatisticMaximum) + require.InDelta(t, 1, inFlight, 0) + + _, ageUnit, ok := sqsGaugeStat(t, cw, "ApproximateAgeOfOldestMessage", "gauge-q", cwtypes.StatisticMaximum) + require.True(t, ok) + require.Equal(t, cwtypes.StandardUnitSeconds, ageUnit) + + // GetQueueAttributes agrees: the delayed message is not counted in flight. + attrs, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: q.QueueUrl, + AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameAll}, + }) + require.NoError(t, err) + require.Equal(t, "1", attrs.Attributes["ApproximateNumberOfMessagesNotVisible"]) + require.Equal(t, "1", attrs.Attributes["ApproximateNumberOfMessagesDelayed"]) + require.Equal(t, "0", attrs.Attributes["ApproximateNumberOfMessages"]) +} + +// TestSDKSQSGaugeSumIsOneSamplePerMinute is the Sum-statistic case: six sends +// in a minute give one Visible datapoint of 6, so Sum is 6, not a running +// total of one sample per send. +func TestSDKSQSGaugeSumIsOneSamplePerMinute(t *testing.T) { + provider := cloudemu.NewAWS() + ts := httptest.NewServer(awsserver.New(awsserver.Drivers{CloudWatch: provider.CloudWatch, SQS: provider.SQS})) + defer ts.Close() + + cfg := sdkConfig(t, ts.URL) + ctx := context.Background() + sqsClient := sqs.NewFromConfig(cfg) + cw := cloudwatch.NewFromConfig(cfg) + + q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("sum-q")}) + require.NoError(t, err) + + for i := 0; i < 6; i++ { + _, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: q.QueueUrl, MessageBody: aws.String("m")}) + require.NoError(t, err) + } + + tickProvider(provider) + tickProvider(provider) // a second tick inside the minute publishes nothing + + sum, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "sum-q", cwtypes.StatisticSum) + require.InDelta(t, 6, sum, 0) + + samples, _, _ := sqsGaugeStat(t, cw, "ApproximateNumberOfMessagesVisible", "sum-q", cwtypes.StatisticSampleCount) + require.InDelta(t, 1, samples, 0) +} + +// receiveSenderID sends body with client and returns the SenderId the queue +// reports for it. +func receiveSenderID(t *testing.T, client *sqs.Client, queueURL *string, body string) string { + t.Helper() + + ctx := context.Background() + + _, err := client.SendMessage(ctx, &sqs.SendMessageInput{QueueUrl: queueURL, MessageBody: aws.String(body)}) + require.NoError(t, err) + + out, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: queueURL, + MessageSystemAttributeNames: []sqstypes.MessageSystemAttributeName{sqstypes.MessageSystemAttributeNameSenderId}, + }) + require.NoError(t, err) + require.Len(t, out.Messages, 1) + require.Equal(t, body, aws.ToString(out.Messages[0].Body)) + + _, err = client.DeleteMessage(ctx, &sqs.DeleteMessageInput{QueueUrl: queueURL, ReceiptHandle: out.Messages[0].ReceiptHandle}) + require.NoError(t, err) + + return out.Messages[0].Attributes[string(sqstypes.MessageSystemAttributeNameSenderId)] +} + +// TestSDKSQSSenderIDMatchesCallerIdentity checks that SenderId is the +// caller's IAM unique id, the same UserId STS GetCallerIdentity reports: the +// user id for long-term keys, ":" for an assumed role. +func TestSDKSQSSenderIDMatchesCallerIdentity(t *testing.T) { + provider := cloudemu.NewAWS() + ts := httptest.NewServer(awsserver.NewFromProvider(provider)) + defer ts.Close() + + cfg := sdkConfig(t, ts.URL) + ctx := context.Background() + sqsClient := sqs.NewFromConfig(cfg) + + q, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("sender-q")}) + require.NoError(t, err) + + who, err := sts.NewFromConfig(cfg).GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) + require.NoError(t, err) + require.NotEqual(t, aws.ToString(who.Account), aws.ToString(who.UserId)) + require.Equal(t, aws.ToString(who.UserId), receiveSenderID(t, sqsClient, q.QueueUrl, "as-user")) + + // An assumed role's messages carry ":". + role, err := iam.NewFromConfig(cfg).CreateRole(ctx, &iam.CreateRoleInput{ + RoleName: aws.String("sender-role"), + AssumeRolePolicyDocument: aws.String(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow",` + + `"Principal":{"AWS":"*"},"Action":"sts:AssumeRole"}]}`), + }) + require.NoError(t, err) + + assumed, err := sts.NewFromConfig(cfg).AssumeRole(ctx, &sts.AssumeRoleInput{ + RoleArn: aws.String("arn:aws:iam::" + aws.ToString(who.Account) + ":role/sender-role"), + RoleSessionName: aws.String("worker-1"), + }) + require.NoError(t, err) + + roleCfg := cfg.Copy() + roleCfg.Credentials = credentials.NewStaticCredentialsProvider( + aws.ToString(assumed.Credentials.AccessKeyId), + aws.ToString(assumed.Credentials.SecretAccessKey), + aws.ToString(assumed.Credentials.SessionToken), + ) + + // AssumedRoleId is the role's own RoleId plus the session name. + roleID := aws.ToString(assumed.AssumedRoleUser.AssumedRoleId) + require.Equal(t, aws.ToString(role.Role.RoleId)+":worker-1", roleID) + require.Equal(t, roleID, receiveSenderID(t, sqs.NewFromConfig(roleCfg), q.QueueUrl, "as-role")) +} diff --git a/services/messagequeue/driver/driver.go b/services/messagequeue/driver/driver.go index 9b21f9ca0..81418e046 100644 --- a/services/messagequeue/driver/driver.go +++ b/services/messagequeue/driver/driver.go @@ -157,6 +157,13 @@ type SendMessageInput struct { // message stays invisible until this instant, measured on the provider's // clock. The zero value means no schedule. Ignored by non-Azure providers. ScheduledEnqueueTime time.Time + // SenderID is the IAM unique id of the caller that sent the message (an + // IAM user id, or ":" for an assumed role), which + // AWS SQS returns as the SenderId message system attribute. The SQS wire + // handler resolves it from the request's credentials; empty means no + // caller principal is known and the account id is used. Ignored by + // non-AWS providers. + SenderID string } // SendMessageOutput is the result of sending a message. @@ -234,6 +241,8 @@ type BatchSendEntry struct { // SystemAttributes are SQS message system attributes (AWS SQS; the only // supported key is AWSTraceHeader). SystemAttributes map[string]MessageAttributeValue + // SenderID has the same meaning as on SendMessageInput. + SenderID string } // BatchSendResult is the result of a batch send.