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/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 { 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) + } +} 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 }