From 87f62aa0afd9fb46ad9b44e9662f7422aaea5508 Mon Sep 17 00:00:00 2001 From: Satyam Trivedi Date: Thu, 22 Oct 2026 15:38:08 +0530 Subject: [PATCH 1/3] fix(aws): skip resources deleted mid-walk in the RGT walkers that bypass project The Kafka, Kinesis, GuardDuty and Glue walkers and the OpenSearch walker read tags in a loop of their own instead of going through project, so a resource deleted between the list call and its tag read failed the whole GetResources call for every service. Paginated walkers now collect every page with listAll and then go through project, which skips a NotFound tag read and propagates any other error. - Kafka, Kinesis and OpenSearch tag reads return NotFound for a deleted resource, so project skips them as is. - GetDetector reports a missing detector as BadRequestException, as real GuardDuty does, so the GuardDuty row re-lists detectors when GetDetector fails and reports NotFound only when the detector is gone. - Glue GetTags never fails for a deleted job, so the walk never aborted, but it emitted a row for the deleted job. The Glue row now checks the job with GetJob first. --- providers/aws/rgt_discovery.go | 209 +++++++++++++++++---------------- 1 file changed, 108 insertions(+), 101 deletions(-) diff --git a/providers/aws/rgt_discovery.go b/providers/aws/rgt_discovery.go index cd2d207e4..41be8e688 100644 --- a/providers/aws/rgt_discovery.go +++ b/providers/aws/rgt_discovery.go @@ -2,6 +2,7 @@ package aws import ( "context" + "slices" cerrors "github.com/stackshy/cloudemu/v2/errors" "github.com/stackshy/cloudemu/v2/internal/idgen" @@ -14,9 +15,11 @@ import ( guarddutydriver "github.com/stackshy/cloudemu/v2/services/guardduty/driver" kafkadriver "github.com/stackshy/cloudemu/v2/services/kafka/driver" ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver" + kinesisdriver "github.com/stackshy/cloudemu/v2/services/kinesis/driver" kmsdriver "github.com/stackshy/cloudemu/v2/services/kms/driver" mdbdriver "github.com/stackshy/cloudemu/v2/services/memorydb/driver" nfdriver "github.com/stackshy/cloudemu/v2/services/networkfirewall/driver" + opensearchdriver "github.com/stackshy/cloudemu/v2/services/opensearch/driver" ssmdriver "github.com/stackshy/cloudemu/v2/services/parameterstore/driver" "github.com/stackshy/cloudemu/v2/services/resourcediscovery" r53rdriver "github.com/stackshy/cloudemu/v2/services/route53resolver/driver" @@ -85,21 +88,14 @@ func (d taggedDiscovery) discoverOpenSearch(ctx context.Context) ([]resourcedisc return nil, err } - out := make([]resourcediscovery.DiscoveredResource, 0, len(domains)) - - for i := range domains { - name := domains[i].DomainName - arn := idgen.AWSARN("es", d.p.Region, d.p.AccountID, "domain/"+name) - - tags, err := d.p.OpenSearch.ListTags(ctx, arn) - if err != nil { - return nil, err - } + return project(ctx, domains, d.openSearchRow) +} - out = append(out, row("es", "domain", name, arn, tags)) - } +func (d taggedDiscovery) openSearchRow(ctx context.Context, dom opensearchdriver.DomainInfo) (resourcediscovery.DiscoveredResource, error) { + arn := idgen.AWSARN("es", d.p.Region, d.p.AccountID, "domain/"+dom.DomainName) + tags, err := d.p.OpenSearch.ListTags(ctx, arn) - return out, nil + return row("es", "domain", dom.DomainName, arn, tags), err } func (d taggedDiscovery) discoverSFN(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) { @@ -272,124 +268,111 @@ func (d taggedDiscovery) discoverWAFv2(ctx context.Context) ([]resourcediscovery } func (d taggedDiscovery) discoverGlue(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) { - var ( - out []resourcediscovery.DiscoveredResource - token string - ) - - for { - names, next, err := d.p.Glue.ListJobs(ctx, gluedriver.TablePagination{NextToken: token, MaxResults: pageLimit}) - if err != nil { - return nil, err - } - - for _, name := range names { - arn := idgen.AWSARN("glue", d.p.Region, d.p.AccountID, "job/"+name) + names, err := listAll(func(token string) ([]string, string, error) { + return d.p.Glue.ListJobs(ctx, gluedriver.TablePagination{NextToken: token, MaxResults: pageLimit}) + }) + if err != nil { + return nil, err + } - tags, err := d.p.Glue.GetTags(ctx, arn) - if err != nil { - return nil, err - } + return project(ctx, names, d.glueJobRow) +} - out = append(out, row("glue", "job", name, arn, tags)) - } +// glueJobRow reads a Glue job's tags. GetTags reads an ARN-keyed store and +// returns no error for a job that no longer exists, so GetJob runs first: a job +// deleted after ListJobs then reports NotFound and is skipped, instead of +// surfacing as a row for a resource that is gone. +func (d taggedDiscovery) glueJobRow(ctx context.Context, name string) (resourcediscovery.DiscoveredResource, error) { + if _, err := d.p.Glue.GetJob(ctx, name); err != nil { + return resourcediscovery.DiscoveredResource{}, err + } - if next == "" { - return out, nil - } + arn := idgen.AWSARN("glue", d.p.Region, d.p.AccountID, "job/"+name) + tags, err := d.p.Glue.GetTags(ctx, arn) - token = next - } + return row("glue", "job", name, arn, tags), err } func (d taggedDiscovery) discoverKafka(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) { - var ( - out []resourcediscovery.DiscoveredResource - token string - ) - - for { - clusters, next, err := d.p.Kafka.ListClusters(ctx, "", kafkadriver.Page{NextToken: token, MaxResults: pageLimit}) - if err != nil { - return nil, err - } - - for i := range clusters { - tags, err := d.p.Kafka.ListTagsForResource(ctx, clusters[i].ClusterARN) - if err != nil { - return nil, err - } + clusters, err := listAll(func(token string) ([]kafkadriver.Cluster, string, error) { + return d.p.Kafka.ListClusters(ctx, "", kafkadriver.Page{NextToken: token, MaxResults: pageLimit}) + }) + if err != nil { + return nil, err + } - out = append(out, row("kafka", "cluster", clusters[i].ClusterName, clusters[i].ClusterARN, tags)) - } + return project(ctx, clusters, d.kafkaRow) +} - if next == "" { - return out, nil - } +//nolint:gocritic // hugeParam: the signature is project's by-value per-item callback. +func (d taggedDiscovery) kafkaRow(ctx context.Context, c kafkadriver.Cluster) (resourcediscovery.DiscoveredResource, error) { + tags, err := d.p.Kafka.ListTagsForResource(ctx, c.ClusterARN) - token = next - } + return row("kafka", "cluster", c.ClusterName, c.ClusterARN, tags), err } func (d taggedDiscovery) discoverKinesis(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) { - var ( - out []resourcediscovery.DiscoveredResource - token string - ) - - for { + streams, err := listAll(func(token string) ([]kinesisdriver.StreamSummary, string, error) { res, err := d.p.Kinesis.ListStreams(ctx, token, "", pageLimit) if err != nil { - return nil, err + return nil, "", err } - for i := range res.StreamSummaries { - s := res.StreamSummaries[i] + if !res.HasMoreStreams { + return res.StreamSummaries, "", nil + } - tags, err := d.p.Kinesis.ListTagsForResource(ctx, s.StreamARN) - if err != nil { - return nil, err - } + return res.StreamSummaries, res.NextToken, nil + }) + if err != nil { + return nil, err + } - out = append(out, row("kinesis", "stream", s.StreamName, s.StreamARN, tags)) - } + return project(ctx, streams, d.kinesisRow) +} - if !res.HasMoreStreams || res.NextToken == "" { - return out, nil - } +//nolint:gocritic // hugeParam: the signature is project's by-value per-item callback. +func (d taggedDiscovery) kinesisRow(ctx context.Context, s kinesisdriver.StreamSummary) (resourcediscovery.DiscoveredResource, error) { + tags, err := d.p.Kinesis.ListTagsForResource(ctx, s.StreamARN) - token = res.NextToken - } + return row("kinesis", "stream", s.StreamName, s.StreamARN, tags), err } func (d taggedDiscovery) discoverGuardDuty(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) { - var ( - out []resourcediscovery.DiscoveredResource - token string - ) - - for { - ids, next, err := d.p.GuardDuty.ListDetectors(ctx, guarddutydriver.Page{NextToken: token, MaxResults: pageLimit}) - if err != nil { - return nil, err - } + ids, err := d.listGuardDutyDetectors(ctx) + if err != nil { + return nil, err + } - for _, id := range ids { - det, err := d.p.GuardDuty.GetDetector(ctx, id) - if err != nil { - return nil, err - } + return project(ctx, ids, d.guardDutyRow) +} - arn := idgen.AWSARN("guardduty", d.p.Region, d.p.AccountID, "detector/"+id) - out = append(out, row("guardduty", "detector", id, arn, det.Tags)) - } +func (d taggedDiscovery) listGuardDutyDetectors(ctx context.Context) ([]string, error) { + return listAll(func(token string) ([]string, string, error) { + return d.p.GuardDuty.ListDetectors(ctx, guarddutydriver.Page{NextToken: token, MaxResults: pageLimit}) + }) +} - if next == "" { - return out, nil +// guardDutyRow reads a detector's tags through GetDetector. Like real GuardDuty, +// GetDetector reports a detector that does not exist as BadRequestException, +// not NotFound, so the error alone can't tell a deleted detector from a real +// failure. When GetDetector fails, a fresh ListDetectors decides: if the +// detector is no longer listed it was deleted after the walk listed it, and the +// row is reported as NotFound so project skips it. Otherwise the original error +// propagates. +func (d taggedDiscovery) guardDutyRow(ctx context.Context, id string) (resourcediscovery.DiscoveredResource, error) { + det, err := d.p.GuardDuty.GetDetector(ctx, id) + if err != nil { + if ids, lerr := d.listGuardDutyDetectors(ctx); lerr == nil && !slices.Contains(ids, id) { + return resourcediscovery.DiscoveredResource{}, cerrors.Newf(cerrors.NotFound, "detector %s was deleted", id) } - token = next + return resourcediscovery.DiscoveredResource{}, err } + + arn := idgen.AWSARN("guardduty", d.p.Region, d.p.AccountID, "detector/"+id) + + return row("guardduty", "detector", id, arn, det.Tags), nil } func (d taggedDiscovery) discoverConfig(ctx context.Context) ([]resourcediscovery.DiscoveredResource, error) { @@ -447,6 +430,30 @@ func project[T any](ctx context.Context, items []T, return out, nil } +// listAll follows a NextToken-paginated list call to its last page and returns +// every item, so a paginated walker can hand the full list to project. +func listAll[T any](list func(token string) ([]T, string, error)) ([]T, error) { + var ( + all []T + token string + ) + + for { + items, next, err := list(token) + if err != nil { + return nil, err + } + + all = append(all, items...) + + if next == "" { + return all, nil + } + + token = next + } +} + // row builds a DiscoveredResource, dropping an empty tag map so a resource with // no tags surfaces with nil Tags (matching the other walkers). func row(service, typ, id, arn string, tags map[string]string) resourcediscovery.DiscoveredResource { From 49238e0bd576a21fdf35fb5c45370b82843a9f75 Mon Sep 17 00:00:00 2001 From: Satyam Trivedi Date: Thu, 22 Oct 2026 15:38:08 +0530 Subject: [PATCH 2/3] fix(aws): skip resources deleted mid-walk in SageMaker and Redshift discovery sagemakerDiscovery and the Redshift part of rdsDiscovery list resources and then read each one's tags in a separate call. Both tag reads return NotFound for a resource that is gone, which failed the whole discovery walk (Resource Groups Tagging API GetResources and Resource Explorer) when a delete landed in between. Skip that resource, the same as project does. Any other error still propagates. --- providers/aws/rds_discovery.go | 7 +++++++ providers/aws/sagemaker_discovery.go | 8 ++++++++ 2 files changed, 15 insertions(+) diff --git a/providers/aws/rds_discovery.go b/providers/aws/rds_discovery.go index 866a0cd73..544620552 100644 --- a/providers/aws/rds_discovery.go +++ b/providers/aws/rds_discovery.go @@ -133,6 +133,13 @@ func (a rdsDiscovery) discoverRedshift(ctx context.Context) ([]resourcediscovery c := clusters[i] tags, err := a.redshift.DescribeTags(ctx, c.ARN) + // NotFound means the cluster was deleted between DescribeClusters and + // this tag read. Skip it, the same as one that was never listed. Any + // other error propagates. + if cerrors.IsNotFound(err) { + continue + } + if err != nil { return nil, err } diff --git a/providers/aws/sagemaker_discovery.go b/providers/aws/sagemaker_discovery.go index 2f5b15380..20158f02b 100644 --- a/providers/aws/sagemaker_discovery.go +++ b/providers/aws/sagemaker_discovery.go @@ -3,6 +3,7 @@ package aws import ( "context" + cerrors "github.com/stackshy/cloudemu/v2/errors" "github.com/stackshy/cloudemu/v2/services/resourcediscovery" sagemakerdriver "github.com/stackshy/cloudemu/v2/services/sagemaker/driver" ) @@ -169,6 +170,13 @@ func (a sagemakerDiscovery) DiscoverResources( for _, it := range items { tags, err := a.liveTags(ctx, it.arn) + // NotFound means the resource was deleted between its list call and + // this tag read. Skip it, the same as one that was never listed. Any + // other error propagates. + if cerrors.IsNotFound(err) { + continue + } + if err != nil { return nil, err } From dfc59642b0eb3cd18b04304758168532253810ae Mon Sep 17 00:00:00 2001 From: Satyam Trivedi Date: Thu, 22 Oct 2026 15:38:08 +0530 Subject: [PATCH 3/3] test(aws): cover deletes racing every discovery walker that reads tags separately - TestDiscoverySurvivesConcurrentDelete runs each walker 2000 times next to a create/delete loop for Kafka, Kinesis, OpenSearch, GuardDuty, Glue, SageMaker and Redshift. Before the fix every case except Glue failed within the first 512 walks. - TestTaggedWalkersSkipResourceDeletedAfterList deletes a resource after listing and projects the stale list through each walker's row function. - TestSageMakerDiscoverySkipsVanishedResource and TestRedshiftDiscoverySkipsVanishedCluster check that NotFound is skipped and any other tag-read error propagates. --- providers/aws/rgt_discovery_test.go | 502 ++++++++++++++++++++++++++++ 1 file changed, 502 insertions(+) diff --git a/providers/aws/rgt_discovery_test.go b/providers/aws/rgt_discovery_test.go index 3a1255891..a25925250 100644 --- a/providers/aws/rgt_discovery_test.go +++ b/providers/aws/rgt_discovery_test.go @@ -3,12 +3,20 @@ package aws import ( "context" "errors" + "strings" "sync" "testing" cerrors "github.com/stackshy/cloudemu/v2/errors" + gluedriver "github.com/stackshy/cloudemu/v2/services/glue/driver" + guarddutydriver "github.com/stackshy/cloudemu/v2/services/guardduty/driver" + kafkadriver "github.com/stackshy/cloudemu/v2/services/kafka/driver" ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver" + kinesisdriver "github.com/stackshy/cloudemu/v2/services/kinesis/driver" + opensearchdriver "github.com/stackshy/cloudemu/v2/services/opensearch/driver" + rdbdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" "github.com/stackshy/cloudemu/v2/services/resourcediscovery" + sagemakerdriver "github.com/stackshy/cloudemu/v2/services/sagemaker/driver" ) // A resource deleted between the list call and its tag read makes the tag read @@ -103,3 +111,497 @@ func TestKeyspacesDiscoverySurvivesConcurrentDelete(t *testing.T) { } } } + +// discoveryChurnCase drives one discovery walker while another goroutine keeps +// creating and deleting a resource of the same service. +type discoveryChurnCase struct { + name string + // setup creates the long-lived resource the walk must keep reporting. It + // returns that resource's ID, or "" for a service that allows only one + // resource (GuardDuty has one detector per account and Region). + setup func(ctx context.Context, t *testing.T, p *Provider) string + // churn creates and deletes one short-lived resource. + churn func(ctx context.Context, p *Provider) + walk func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) +} + +func discoveryChurnCases() []discoveryChurnCase { + tagged := map[string]string{"env": "prod"} + + return []discoveryChurnCase{ + { + name: "kafka", + setup: func(ctx context.Context, t *testing.T, p *Provider) string { + t.Helper() + + if _, err := p.Kafka.CreateCluster(ctx, kafkaCluster("stable", tagged)); err != nil { + t.Fatalf("CreateCluster: %v", err) + } + + return "stable" + }, + churn: func(ctx context.Context, p *Provider) { + if c, err := p.Kafka.CreateCluster(ctx, kafkaCluster("churn", nil)); err == nil { + _, _, _ = p.Kafka.DeleteCluster(ctx, c.ClusterARN, "") + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + return taggedDiscovery{p}.discoverKafka(ctx) + }, + }, + { + name: "kinesis", + setup: func(ctx context.Context, t *testing.T, p *Provider) string { + t.Helper() + + err := p.Kinesis.CreateStream(ctx, kinesisdriver.CreateStreamInput{StreamName: "stable", ShardCount: 1, Tags: tagged}) + if err != nil { + t.Fatalf("CreateStream: %v", err) + } + + return "stable" + }, + churn: func(ctx context.Context, p *Provider) { + if p.Kinesis.CreateStream(ctx, kinesisdriver.CreateStreamInput{StreamName: "churn", ShardCount: 1}) == nil { + _ = p.Kinesis.DeleteStream(ctx, "churn", "", true) + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + return taggedDiscovery{p}.discoverKinesis(ctx) + }, + }, + { + name: "opensearch", + setup: func(ctx context.Context, t *testing.T, p *Provider) string { + t.Helper() + + if _, err := p.OpenSearch.CreateDomain(ctx, openSearchDomain("stable", tagged)); err != nil { + t.Fatalf("CreateDomain: %v", err) + } + + return "stable" + }, + churn: func(ctx context.Context, p *Provider) { + if _, err := p.OpenSearch.CreateDomain(ctx, openSearchDomain("churn", nil)); err == nil { + _, _ = p.OpenSearch.DeleteDomain(ctx, "churn") + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + return taggedDiscovery{p}.discoverOpenSearch(ctx) + }, + }, + { + name: "guardduty", + setup: func(context.Context, *testing.T, *Provider) string { + return "" + }, + churn: func(ctx context.Context, p *Provider) { + if det, err := p.GuardDuty.CreateDetector(ctx, guarddutydriver.CreateDetectorInput{Enable: true, Tags: tagged}); err == nil { + _ = p.GuardDuty.DeleteDetector(ctx, det.ID) + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + return taggedDiscovery{p}.discoverGuardDuty(ctx) + }, + }, + { + name: "glue", + setup: func(ctx context.Context, t *testing.T, p *Provider) string { + t.Helper() + + if _, err := p.Glue.CreateJob(ctx, gluedriver.Job{Name: "stable", Role: "r", Tags: tagged}); err != nil { + t.Fatalf("CreateJob: %v", err) + } + + return "stable" + }, + churn: func(ctx context.Context, p *Provider) { + if _, err := p.Glue.CreateJob(ctx, gluedriver.Job{Name: "churn", Role: "r"}); err == nil { + _, _ = p.Glue.DeleteJob(ctx, "churn") + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + return taggedDiscovery{p}.discoverGlue(ctx) + }, + }, + { + name: "sagemaker", + setup: func(ctx context.Context, t *testing.T, p *Provider) string { + t.Helper() + + if _, err := p.SageMaker.CreateModel(ctx, sagemakerdriver.ModelConfig{ + ModelName: "stable", Tags: []sagemakerdriver.Tag{{Key: "env", Value: "prod"}}, + }); err != nil { + t.Fatalf("CreateModel: %v", err) + } + + return "stable" + }, + churn: func(ctx context.Context, p *Provider) { + if _, err := p.SageMaker.CreateModel(ctx, sagemakerdriver.ModelConfig{ModelName: "churn"}); err == nil { + _ = p.SageMaker.DeleteModel(ctx, "churn") + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + return sagemakerDiscovery{p.SageMaker}.DiscoverResources(ctx) + }, + }, + { + name: "redshift", + setup: func(ctx context.Context, t *testing.T, p *Provider) string { + t.Helper() + + if _, err := p.Redshift.CreateCluster(ctx, rdbdriver.ClusterConfig{ID: "stable", Tags: tagged}); err != nil { + t.Fatalf("CreateCluster: %v", err) + } + + return "stable" + }, + churn: func(ctx context.Context, p *Provider) { + if _, err := p.Redshift.CreateCluster(ctx, rdbdriver.ClusterConfig{ID: "churn"}); err == nil { + _ = p.Redshift.DeleteCluster(ctx, "churn") + } + }, + walk: func(ctx context.Context, p *Provider) ([]resourcediscovery.DiscoveredResource, error) { + dbs, err := rdsDiscovery{m: p.RDS, redshift: p.Redshift}.discoverRedshift(ctx) + + out := make([]resourcediscovery.DiscoveredResource, 0, len(dbs)) + for _, db := range dbs { + out = append(out, resourcediscovery.DiscoveredResource{ID: db.Name, ARN: db.ARN, Tags: db.Tags}) + } + + return out, err + }, + }, + } +} + +// A delete racing a discovery walk must not fail it, for every walker that +// reads tags in a call separate from its list call. The long-lived resource +// must keep surfacing with its tags on every walk. +func TestDiscoverySurvivesConcurrentDelete(t *testing.T) { + for _, tc := range discoveryChurnCases() { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + p := New() + stable := tc.setup(ctx, t, p) + + stop := make(chan struct{}) + + var wg sync.WaitGroup + + wg.Add(1) + + go func() { + defer wg.Done() + + for { + select { + case <-stop: + return + default: + } + + tc.churn(ctx, p) + } + }() + + defer func() { + close(stop) + wg.Wait() + }() + + for i := 0; i < 2000; i++ { + rows, err := tc.walk(ctx, p) + if err != nil { + t.Fatalf("walk %d: %v", i, err) + } + + if stable != "" && !hasTaggedRow(rows, stable) { + t.Fatalf("walk %d: %s missing or untagged: %+v", i, stable, rows) + } + } + }) + } +} + +func hasTaggedRow(rows []resourcediscovery.DiscoveredResource, id string) bool { + for _, r := range rows { + if r.ID == id && r.Tags["env"] == "prod" { + return true + } + } + + return false +} + +func kafkaCluster(name string, tags map[string]string) kafkadriver.CreateClusterInput { + return kafkadriver.CreateClusterInput{ + ClusterName: name, KafkaVersion: "3.6.0", NumberOfBrokerNodes: 2, + BrokerNodeGroupInfo: &kafkadriver.BrokerNodeGroupInfo{ + ClientSubnets: []string{"subnet-1", "subnet-2"}, InstanceType: "kafka.m5.large", + }, + Tags: tags, + } +} + +func openSearchDomain(name string, tags map[string]string) opensearchdriver.CreateDomainInput { + return opensearchdriver.CreateDomainInput{ + DomainName: name, EngineVersion: "OpenSearch_2.11", + ClusterConfig: opensearchdriver.ClusterConfig{InstanceType: "t3.small.search", InstanceCount: 1}, + Tags: tags, + } +} + +// Each paginated or non-project walker reads tags with its own row function. +// These tests take the list the walker would see, delete one resource, and +// then project that stale list through the walker's row function. The deleted +// resource must be skipped and the live one must still surface with its tags. +func TestTaggedWalkersSkipResourceDeletedAfterList(t *testing.T) { + ctx := context.Background() + tagged := map[string]string{"env": "prod"} + + t.Run("kafka", func(t *testing.T) { + p := New() + d := taggedDiscovery{p} + + if _, err := p.Kafka.CreateCluster(ctx, kafkaCluster("live", tagged)); err != nil { + t.Fatal(err) + } + + gone, err := p.Kafka.CreateCluster(ctx, kafkaCluster("gone", nil)) + mustNoErr(t, err) + + listed, _, err := p.Kafka.ListClusters(ctx, "", kafkadriver.Page{MaxResults: pageLimit}) + mustNoErr(t, err) + + _, _, err = p.Kafka.DeleteCluster(ctx, gone.ClusterARN, "") + mustNoErr(t, err) + + rows, err := project(ctx, listed, d.kafkaRow) + wantOnlyTagged(t, rows, err, "live") + }) + + t.Run("kinesis", func(t *testing.T) { + p := New() + d := taggedDiscovery{p} + + mustNoErr(t, p.Kinesis.CreateStream(ctx, kinesisdriver.CreateStreamInput{StreamName: "live", ShardCount: 1, Tags: tagged})) + mustNoErr(t, p.Kinesis.CreateStream(ctx, kinesisdriver.CreateStreamInput{StreamName: "gone", ShardCount: 1})) + + listed, err := p.Kinesis.ListStreams(ctx, "", "", pageLimit) + mustNoErr(t, err) + mustNoErr(t, p.Kinesis.DeleteStream(ctx, "gone", "", true)) + + rows, err := project(ctx, listed.StreamSummaries, d.kinesisRow) + wantOnlyTagged(t, rows, err, "live") + }) + + t.Run("opensearch", func(t *testing.T) { + p := New() + d := taggedDiscovery{p} + + if _, err := p.OpenSearch.CreateDomain(ctx, openSearchDomain("live", tagged)); err != nil { + t.Fatal(err) + } + + if _, err := p.OpenSearch.CreateDomain(ctx, openSearchDomain("gone", nil)); err != nil { + t.Fatal(err) + } + + listed, err := p.OpenSearch.ListDomainNames(ctx, "") + mustNoErr(t, err) + + if _, err := p.OpenSearch.DeleteDomain(ctx, "gone"); err != nil { + t.Fatal(err) + } + + rows, err := project(ctx, listed, d.openSearchRow) + wantOnlyTagged(t, rows, err, "live") + }) + + t.Run("glue", func(t *testing.T) { + p := New() + d := taggedDiscovery{p} + + if _, err := p.Glue.CreateJob(ctx, gluedriver.Job{Name: "live", Role: "r", Tags: tagged}); err != nil { + t.Fatal(err) + } + + if _, err := p.Glue.CreateJob(ctx, gluedriver.Job{Name: "gone", Role: "r"}); err != nil { + t.Fatal(err) + } + + listed, _, err := p.Glue.ListJobs(ctx, gluedriver.TablePagination{MaxResults: pageLimit}) + mustNoErr(t, err) + + if _, err := p.Glue.DeleteJob(ctx, "gone"); err != nil { + t.Fatal(err) + } + + rows, err := project(ctx, listed, d.glueJobRow) + wantOnlyTagged(t, rows, err, "live") + }) + + // GuardDuty allows one detector per account and Region, so there is no + // live detector to keep. The deleted one must be skipped even though + // GetDetector reports it as BadRequest, not NotFound. + t.Run("guardduty", func(t *testing.T) { + p := New() + d := taggedDiscovery{p} + + det, err := p.GuardDuty.CreateDetector(ctx, guarddutydriver.CreateDetectorInput{Enable: true, Tags: tagged}) + mustNoErr(t, err) + + listed, err := d.listGuardDutyDetectors(ctx) + mustNoErr(t, err) + + rows, err := project(ctx, listed, d.guardDutyRow) + wantOnlyTagged(t, rows, err, det.ID) + + mustNoErr(t, p.GuardDuty.DeleteDetector(ctx, det.ID)) + + if _, err := p.GuardDuty.GetDetector(ctx, det.ID); err == nil || cerrors.IsNotFound(err) { + t.Fatalf("GetDetector on a deleted detector: err = %v, want a non-NotFound error", err) + } + + rows, err = project(ctx, listed, d.guardDutyRow) + if err != nil || len(rows) != 0 { + t.Fatalf("rows = %+v, err = %v, want no rows and no error", rows, err) + } + }) +} + +// The SageMaker walker must skip a resource whose tag read reports NotFound and +// still fail on any other tag-read error. +func TestSageMakerDiscoverySkipsVanishedResource(t *testing.T) { + ctx := context.Background() + boom := errors.New("boom") + + for _, tc := range []struct { + name string + readErr error + wantErr error + }{ + {name: "not found", readErr: cerrors.Newf(cerrors.NotFound, "model gone")}, + {name: "other error", readErr: boom, wantErr: boom}, + } { + t.Run(tc.name, func(t *testing.T) { + p := New() + + for _, name := range []string{"live", "gone"} { + _, err := p.SageMaker.CreateModel(ctx, sagemakerdriver.ModelConfig{ + ModelName: name, Tags: []sagemakerdriver.Tag{{Key: "env", Value: "prod"}}, + }) + mustNoErr(t, err) + } + + m := failingTagRead{smMock: p.SageMaker, failARN: "model/gone", err: tc.readErr} + rows, err := sagemakerDiscovery{m}.DiscoverResources(ctx) + + if tc.wantErr != nil { + if !errors.Is(err, tc.wantErr) { + t.Fatalf("err = %v, want %v", err, tc.wantErr) + } + + return + } + + wantOnlyTagged(t, rows, err, "live") + }) + } +} + +// failingTagRead wraps the SageMaker mock and fails ListTags for one ARN, as if +// that resource were deleted after the walk listed it. +type failingTagRead struct { + smMock + + failARN string + err error +} + +func (f failingTagRead) ListTags(ctx context.Context, arn string) ([]sagemakerdriver.Tag, error) { + if strings.HasSuffix(arn, f.failARN) { + return nil, f.err + } + + return f.smMock.ListTags(ctx, arn) +} + +// The Redshift walker must skip a cluster whose tag read reports NotFound and +// still fail on any other tag-read error. +func TestRedshiftDiscoverySkipsVanishedCluster(t *testing.T) { + ctx := context.Background() + boom := errors.New("boom") + + for _, tc := range []struct { + name string + readErr error + wantErr error + }{ + {name: "not found", readErr: cerrors.Newf(cerrors.NotFound, "cluster gone")}, + {name: "other error", readErr: boom, wantErr: boom}, + } { + t.Run(tc.name, func(t *testing.T) { + p := New() + + for _, id := range []string{"live", "gone"} { + _, err := p.Redshift.CreateCluster(ctx, rdbdriver.ClusterConfig{ID: id, Tags: map[string]string{"env": "prod"}}) + mustNoErr(t, err) + } + + rs := failingRedshiftTags{redshiftClusters: p.Redshift, failARN: "cluster:gone", err: tc.readErr} + dbs, err := rdsDiscovery{m: p.RDS, redshift: rs}.discoverRedshift(ctx) + + if tc.wantErr != nil { + if !errors.Is(err, tc.wantErr) { + t.Fatalf("err = %v, want %v", err, tc.wantErr) + } + + return + } + + if err != nil || len(dbs) != 1 || dbs[0].Name != "live" || dbs[0].Tags["env"] != "prod" { + t.Fatalf("dbs = %+v, err = %v, want only the tagged live cluster", dbs, err) + } + }) + } +} + +// failingRedshiftTags wraps the Redshift mock and fails DescribeTags for one +// ARN, as if that cluster were deleted after DescribeClusters. +type failingRedshiftTags struct { + redshiftClusters + + failARN string + err error +} + +func (f failingRedshiftTags) DescribeTags(ctx context.Context, arn string) (map[string]string, error) { + if strings.HasSuffix(arn, f.failARN) { + return nil, f.err + } + + return f.redshiftClusters.DescribeTags(ctx, arn) +} + +func wantOnlyTagged(t *testing.T, rows []resourcediscovery.DiscoveredResource, err error, id string) { + t.Helper() + + if err != nil { + t.Fatalf("walk: %v", err) + } + + if len(rows) != 1 || !hasTaggedRow(rows, id) { + t.Fatalf("rows = %+v, want only %s tagged env=prod", rows, id) + } +} + +func mustNoErr(t *testing.T, err error) { + t.Helper() + + if err != nil { + t.Fatal(err) + } +}