diff --git a/providers/aws/aws.go b/providers/aws/aws.go index 3657d8bf5..99b5ee8a7 100644 --- a/providers/aws/aws.go +++ b/providers/aws/aws.go @@ -409,6 +409,14 @@ func wireBedrockAgentRuntime(p *Provider) { p.BedrockAgentRuntime.SetObjectStore(p.S3) } +// wireRDS connects RDS to its cross-service dependencies: DB subnet groups +// resolve their subnets through VPC, and event subscriptions deliver matching +// RDS events to their SNS topic. +func wireRDS(p *Provider) { + p.RDS.SetSubnetResolver(p.VPC) + p.RDS.SetSNSPublisher(p.SNS) +} + // newProvider builds a provider, using the supplied shared global services when // non-nil (multi-region) or a fresh bundle when nil (standalone). The global // struct fields are assigned BEFORE the cross-service Set* wiring runs, so every @@ -450,7 +458,7 @@ func newProvider(o *config.Options, shared *GlobalServices) *Provider { p.Athena.SetMonitoring(p.CloudWatch) p.Athena.SetCatalog(p.Glue) wireBedrockAgentRuntime(p) - p.RDS.SetSubnetResolver(p.VPC) + wireRDS(p) p.ElastiCache.SetSubnetResolver(p.VPC) p.EC2.SetSubnetResolver(p.VPC) // RunInstances materializes the instance's primary (eth0) ENI in the VPC, and diff --git a/providers/aws/rds/advanced_restore.go b/providers/aws/rds/advanced_restore.go index 308f95e73..fdd1d9ce0 100644 --- a/providers/aws/rds/advanced_restore.go +++ b/providers/aws/rds/advanced_restore.go @@ -108,6 +108,8 @@ func (m *Mock) RestoreDBInstanceToPointInTime( return nil, cerrors.New(cerrors.InvalidArgument, "TargetDBInstanceIdentifier is required") } + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -158,6 +160,8 @@ func (m *Mock) RestoreDBInstanceToPointInTime( m.instances.Set(inst.ID, inst) m.rootPasswords[inst.ID] = restoreCfg.MasterUserPassword + m.recordEventLocked(eventSource{sourceDBInstance, inst.ID, inst.ARN}, evInstancePITR, + input.SourceInstanceID, inst.ID) m.emitInstanceMetrics(inst.ID, inst.Engine, cpuMetricRunning, connectionsRunning) diff --git a/providers/aws/rds/aurora.go b/providers/aws/rds/aurora.go index c060e9686..6a670fdcf 100644 --- a/providers/aws/rds/aurora.go +++ b/providers/aws/rds/aurora.go @@ -205,6 +205,8 @@ func (m *Mock) DeleteDBClusterEndpoint(_ context.Context, endpointID string) (*r // ---- failover ---- func (m *Mock) FailoverDBCluster(_ context.Context, clusterID, targetInstanceID string) (*rdsdriver.Cluster, error) { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -236,6 +238,7 @@ func (m *Mock) FailoverDBCluster(_ context.Context, clusterID, targetInstanceID } m.clusters.Set(clusterID, cluster) + m.recordEventLocked(eventSource{sourceDBCluster, clusterID, cluster.ARN}, evClusterFailover, cluster.Members[0]) out := cluster diff --git a/providers/aws/rds/dbproxy.go b/providers/aws/rds/dbproxy.go index fbe6764c3..550272202 100644 --- a/providers/aws/rds/dbproxy.go +++ b/providers/aws/rds/dbproxy.go @@ -36,6 +36,8 @@ func (m *Mock) CreateDBProxy(ctx context.Context, cfg rdsdriver.DBProxyConfig) ( return nil, cerrors.New(cerrors.InvalidArgument, "EngineFamily is required") } + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -60,6 +62,7 @@ func (m *Mock) CreateDBProxy(ctx context.Context, cfg rdsdriver.DBProxyConfig) ( CreatedAt: m.opts.Clock.Now().UTC(), } m.proxies.Set(cfg.Name, proxy) + m.recordEventLocked(eventSource{sourceDBProxy, cfg.Name, proxy.ARN}, evProxyCreated, cfg.Name) out := proxy @@ -108,6 +111,8 @@ func cloneProxy(p rdsdriver.DBProxy) rdsdriver.DBProxy { } func (m *Mock) ModifyDBProxy(_ context.Context, name string, input rdsdriver.ModifyDBProxyInput) (*rdsdriver.DBProxy, error) { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -133,6 +138,7 @@ func (m *Mock) ModifyDBProxy(_ context.Context, name string, input rdsdriver.Mod } m.proxies.Set(name, p) + m.recordEventLocked(eventSource{sourceDBProxy, name, p.ARN}, evProxyModified, name) out := p @@ -140,6 +146,8 @@ func (m *Mock) ModifyDBProxy(_ context.Context, name string, input rdsdriver.Mod } func (m *Mock) DeleteDBProxy(_ context.Context, name string) (*rdsdriver.DBProxy, error) { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -151,6 +159,7 @@ func (m *Mock) DeleteDBProxy(_ context.Context, name string) (*rdsdriver.DBProxy p.Status = "deleting" m.proxies.Delete(name) + m.recordEventLocked(eventSource{sourceDBProxy, name, p.ARN}, evProxyDeleted, name) out := p diff --git a/providers/aws/rds/default_group_upgrade_test.go b/providers/aws/rds/default_group_upgrade_test.go new file mode 100644 index 000000000..5f24b13e8 --- /dev/null +++ b/providers/aws/rds/default_group_upgrade_test.go @@ -0,0 +1,324 @@ +package rds + +import ( + "context" + "errors" + "testing" + + cerrors "github.com/stackshy/cloudemu/v2/errors" + rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" +) + +// TestModifyInstanceEngineUpgradeMovesDefaultParamGroup: a major version +// upgrade moves an instance on its default group to the new family's default +// group; a minor upgrade, a deferred upgrade and an explicit new-family group +// on the same request are left as the caller expects. +func TestModifyInstanceEngineUpgradeMovesDefaultParamGroup(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + create := func(id, group string) { + t.Helper() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ + ID: id, Engine: "postgres", EngineVersion: "15.4", DBParameterGroupName: group, + }) + requireNoError(t, err) + } + + modify := func(id string, in rdsdriver.ModifyInstanceInput) *rdsdriver.Instance { + t.Helper() + + out, err := m.ModifyInstance(ctx, id, in) + requireNoError(t, err) + + return out + } + + create("major", "") + got := modify("major", rdsdriver.ModifyInstanceInput{EngineVersion: "16.2", ApplyImmediately: true}) + assertEqual(t, "16.2", got.EngineVersion) + assertEqual(t, "default.postgres16", got.DBParameterGroupName) + + described, err := m.DescribeInstances(ctx, []string{"major"}) + requireNoError(t, err) + assertEqual(t, "default.postgres16", described[0].DBParameterGroupName) + + create("minor", "") + got = modify("minor", rdsdriver.ModifyInstanceInput{EngineVersion: "15.5", ApplyImmediately: true}) + assertEqual(t, "default.postgres15", got.DBParameterGroupName) + + create("deferred", "") + got = modify("deferred", rdsdriver.ModifyInstanceInput{EngineVersion: "16.2"}) + assertEqual(t, "15.4", got.EngineVersion) + assertEqual(t, "default.postgres15", got.DBParameterGroupName) + + _, err = m.CreateDBParameterGroup(ctx, rdsdriver.ParameterGroupConfig{Name: "my16", Family: "postgres16"}) + requireNoError(t, err) + + create("explicit", "") + got = modify("explicit", rdsdriver.ModifyInstanceInput{ + EngineVersion: "16.2", ApplyImmediately: true, DBParameterGroupName: "my16", + }) + assertEqual(t, "16.2", got.EngineVersion) + assertEqual(t, "my16", got.DBParameterGroupName) +} + +// TestModifyInstanceRejectsParamGroupFamilyMismatch: RDS rejects, with +// InvalidParameterCombination, a major upgrade that would leave the instance +// on a custom group of the old family, and any request that puts a group of +// another family on the instance. A custom group survives a minor upgrade. +func TestModifyInstanceRejectsParamGroupFamilyMismatch(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + for _, g := range []rdsdriver.ParameterGroupConfig{ + {Name: "my15", Family: "postgres15"}, {Name: "my16", Family: "postgres16"}, + } { + _, err := m.CreateDBParameterGroup(ctx, g) + requireNoError(t, err) + } + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ + ID: "db1", Engine: "postgres", EngineVersion: "15.4", DBParameterGroupName: "my15", + }) + requireNoError(t, err) + + _, err = m.ModifyInstance(ctx, "db1", rdsdriver.ModifyInstanceInput{EngineVersion: "16.2", ApplyImmediately: true}) + requireCode(t, err, cerrors.IsInvalidArgument) + assertEqual(t, "The current DB instance parameter group my15 is custom. You must explicitly specify a new DB "+ + "instance parameter group, either default or custom, for the engine version upgrade.", errMsg(err)) + + _, err = m.ModifyInstance(ctx, "db1", rdsdriver.ModifyInstanceInput{ + EngineVersion: "17.2", ApplyImmediately: true, DBParameterGroupName: "my16", + }) + requireCode(t, err, cerrors.IsInvalidArgument) + assertEqual(t, "The parameter group my16 with DBParameterGroupFamily postgres16 can't be used for this "+ + "instance. Use a parameter group with DBParameterGroupFamily postgres17.", errMsg(err)) + + _, err = m.ModifyInstance(ctx, "db1", rdsdriver.ModifyInstanceInput{DBParameterGroupName: "default.postgres16"}) + requireCode(t, err, cerrors.IsInvalidArgument) + + // A rejected request changes nothing. + described, err := m.DescribeInstances(ctx, []string{"db1"}) + requireNoError(t, err) + assertEqual(t, "15.4", described[0].EngineVersion) + assertEqual(t, "my15", described[0].DBParameterGroupName) + + got, err := m.ModifyInstance(ctx, "db1", rdsdriver.ModifyInstanceInput{EngineVersion: "15.5", ApplyImmediately: true}) + requireNoError(t, err) + assertEqual(t, "my15", got.DBParameterGroupName) + + got, err = m.ModifyInstance(ctx, "db1", rdsdriver.ModifyInstanceInput{ + EngineVersion: "16.2", ApplyImmediately: true, DBParameterGroupName: "my16", + }) + requireNoError(t, err) + assertEqual(t, "16.2", got.EngineVersion) + assertEqual(t, "my16", got.DBParameterGroupName) +} + +// TestModifyClusterMajorUpgradeMovesMembers: an Aurora major upgrade upgrades +// every member instance; members on the default instance group move to the new +// family's default, or to DBInstanceParameterGroupName when it is given. A +// member on a custom old-family group needs DBInstanceParameterGroupName, which +// is itself valid only on a major upgrade. +func TestModifyClusterMajorUpgradeMovesMembers(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + newCluster := func(id string, groups ...string) { + t.Helper() + + _, err := m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: id, Engine: "aurora-postgresql", EngineVersion: "15.4"}) + requireNoError(t, err) + + for i, g := range groups { + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ + ID: id + "m" + string(rune('1'+i)), Engine: "aurora-postgresql", ClusterID: id, + DBParameterGroupName: g, + }) + requireNoError(t, err) + } + } + + member := func(id string) rdsdriver.Instance { + t.Helper() + + got, err := m.DescribeInstances(ctx, []string{id}) + requireNoError(t, err) + + return got[0] + } + + newCluster("c1", "", "") + assertEqual(t, "default.aurora-postgresql15", member("c1m1").DBParameterGroupName) + + _, err := m.ModifyCluster(ctx, "c1", rdsdriver.ModifyInstanceInput{EngineVersion: "16.2", ApplyImmediately: true}) + requireNoError(t, err) + + for _, id := range []string{"c1m1", "c1m2"} { + assertEqual(t, "16.2", member(id).EngineVersion) + assertEqual(t, "default.aurora-postgresql16", member(id).DBParameterGroupName) + } + + // A minor upgrade moves the version only. + _, err = m.ModifyCluster(ctx, "c1", rdsdriver.ModifyInstanceInput{EngineVersion: "16.3"}) + requireNoError(t, err) + assertEqual(t, "16.3", member("c1m1").EngineVersion) + assertEqual(t, "default.aurora-postgresql16", member("c1m1").DBParameterGroupName) + + _, err = m.ModifyCluster(ctx, "c1", rdsdriver.ModifyInstanceInput{ + EngineVersion: "16.4", DBInstanceParameterGroupName: "default.aurora-postgresql16", + }) + requireCode(t, err, cerrors.IsInvalidArgument) + + for _, g := range []rdsdriver.ParameterGroupConfig{ + {Name: "apg15", Family: "aurora-postgresql15"}, {Name: "apg16", Family: "aurora-postgresql16"}, + } { + _, err = m.CreateDBParameterGroup(ctx, g) + requireNoError(t, err) + } + + newCluster("c2", "apg15") + + _, err = m.ModifyCluster(ctx, "c2", rdsdriver.ModifyInstanceInput{EngineVersion: "16.2", ApplyImmediately: true}) + requireCode(t, err, cerrors.IsInvalidArgument) + assertEqual(t, "15.4", member("c2m1").EngineVersion) + + clusters, err := m.DescribeClusters(ctx, []string{"c2"}) + requireNoError(t, err) + assertEqual(t, "15.4", clusters[0].EngineVersion) + + _, err = m.ModifyCluster(ctx, "c2", rdsdriver.ModifyInstanceInput{ + EngineVersion: "16.2", DBInstanceParameterGroupName: "apg15", + }) + requireCode(t, err, cerrors.IsInvalidArgument) + + _, err = m.ModifyCluster(ctx, "c2", rdsdriver.ModifyInstanceInput{ + EngineVersion: "16.2", DBInstanceParameterGroupName: "apg16", + }) + requireNoError(t, err) + assertEqual(t, "16.2", member("c2m1").EngineVersion) + assertEqual(t, "apg16", member("c2m1").DBParameterGroupName) +} + +// TestUnknownDefaultFamilyIsNotFound: only "default." names of a real +// family exist. "default.bogus" is DBParameterGroupNotFound on describe, +// modify and tagging, not a synthesized or immutable default group. +func TestUnknownDefaultFamilyIsNotFound(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + params := []rdsdriver.Parameter{{Name: "max_connections", Value: "10"}} + + for _, name := range []string{"default.bogus", "default.postgres16.2", "default.postgres-16", "default."} { + _, err := m.ModifyDBParameterGroup(ctx, name, params) + requireCode(t, err, cerrors.IsNotFound) + + _, err = m.DescribeDBParameterGroups(ctx, []string{name}) + requireCode(t, err, cerrors.IsNotFound) + + _, err = m.ListTagsForResource(ctx, parameterGroupARN("us-east-1", "123456789012", name)) + requireCode(t, err, cerrors.IsNotFound) + } + + _, err := m.ModifyDBClusterParameterGroup(ctx, "default.bogus-family", params) + requireCode(t, err, cerrors.IsNotFound) + + for _, family := range []string{ + "postgres16", "postgres9.6", "mysql8.0", "mariadb10.11", "aurora-mysql8.0", "aurora-postgresql15", + "docdb5.0", "neptune1", "neptune1.3", "oracle-ee-19", "oracle-se2-cdb-21", "sqlserver-ee-15.0", "db2-se", + } { + if !isKnownFamily(family) { + t.Errorf("isKnownFamily(%q) = false, want true", family) + } + } +} + +// errMsg returns the human message of a cloudemu error. +func errMsg(err error) string { + var e *cerrors.Error + if errors.As(err, &e) { + return e.Message + } + + return err.Error() +} + +// TestModifyClusterEngineUpgradeMovesDefaultClusterParamGroup mirrors the +// instance case for DBClusterParameterGroupName. +func TestModifyClusterEngineUpgradeMovesDefaultClusterParamGroup(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + _, err := m.CreateCluster(ctx, rdsdriver.ClusterConfig{ + ID: "c1", Engine: "aurora-postgresql", EngineVersion: "15.4", + }) + requireNoError(t, err) + + got, err := m.ModifyCluster(ctx, "c1", rdsdriver.ModifyInstanceInput{EngineVersion: "16.2", ApplyImmediately: true}) + requireNoError(t, err) + assertEqual(t, "default.aurora-postgresql16", got.DBClusterParameterGroupName) + + got, err = m.ModifyCluster(ctx, "c1", rdsdriver.ModifyInstanceInput{ + EngineVersion: "16.3", DBClusterParameterGroupName: "mine", + }) + requireNoError(t, err) + assertEqual(t, "mine", got.DBClusterParameterGroupName) +} + +// TestCreateClusterResolvesDefaultEngineVersion: a cluster created without an +// EngineVersion reports the resolved default version, consistent with its +// version-derived default cluster parameter group. +func TestCreateClusterResolvesDefaultEngineVersion(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + got, err := m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: "c1", Engine: "aurora-mysql"}) + requireNoError(t, err) + + want := defaultEngineVersion("aurora-mysql") + if want == "" { + t.Fatal("no catalog default version for aurora-mysql") + } + + assertEqual(t, want, got.EngineVersion) + assertEqual(t, "default."+parameterGroupFamily("aurora-mysql", want), got.DBClusterParameterGroupName) + + described, err := m.DescribeClusters(ctx, []string{"c1"}) + requireNoError(t, err) + assertEqual(t, want, described[0].EngineVersion) +} + +// TestModifyDefaultParameterGroupRejected: a listed default group cannot be +// modified or reset, and says so (InvalidArgument), instead of NotFound. +func TestModifyDefaultParameterGroupRejected(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + params := []rdsdriver.Parameter{{Name: "max_connections", Value: "10"}} + + _, err := m.ModifyDBParameterGroup(ctx, "default.postgres15", params) + requireCode(t, err, cerrors.IsInvalidArgument) + + _, err = m.ResetDBParameterGroup(ctx, "default.postgres15", nil, true) + requireCode(t, err, cerrors.IsInvalidArgument) + + _, err = m.ModifyDBClusterParameterGroup(ctx, "default.aurora-postgresql15", params) + requireCode(t, err, cerrors.IsInvalidArgument) + + _, err = m.ResetDBClusterParameterGroup(ctx, "default.aurora-postgresql15", nil, true) + requireCode(t, err, cerrors.IsInvalidArgument) + + _, err = m.ModifyDBParameterGroup(ctx, "nope", params) + requireCode(t, err, cerrors.IsNotFound) + + _, err = m.ModifyDBClusterParameterGroup(ctx, "nope", params) + requireCode(t, err, cerrors.IsNotFound) +} + +func requireCode(t *testing.T, err error, is func(error) bool) { + t.Helper() + + if err == nil || !is(err) { + t.Fatalf("unexpected error %v", err) + } +} diff --git a/providers/aws/rds/engine_upgrade.go b/providers/aws/rds/engine_upgrade.go new file mode 100644 index 000000000..1130c912e --- /dev/null +++ b/providers/aws/rds/engine_upgrade.go @@ -0,0 +1,150 @@ +package rds + +import ( + cerrors "github.com/stackshy/cloudemu/v2/errors" + rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" +) + +// errGroupFamilyMismatch is the InvalidParameterCombination RDS returns when a +// DB parameter group of one family is put on an instance whose engine version +// needs another (an explicit postgres15 group on a 16.2 upgrade). +func errGroupFamilyMismatch(group, groupFamily, want string) error { + return cerrors.Newf(cerrors.InvalidArgument, + "The parameter group %s with DBParameterGroupFamily %s can't be used for this instance. "+ + "Use a parameter group with DBParameterGroupFamily %s.", group, groupFamily, want) +} + +// errCustomGroupNeedsReplacement is the InvalidParameterCombination RDS +// returns for a major version upgrade of an instance on a custom group of the +// old family when the request names no new group. +func errCustomGroupNeedsReplacement(group string) error { + return cerrors.Newf(cerrors.InvalidArgument, + "The current DB instance parameter group %s is custom. You must explicitly specify a new DB instance "+ + "parameter group, either default or custom, for the engine version upgrade.", group) +} + +// errInstanceGroupNeedsMajorUpgrade rejects ModifyDBCluster +// DBInstanceParameterGroupName outside a major version upgrade, the only case +// RDS accepts it in. +func errInstanceGroupNeedsMajorUpgrade() error { + return cerrors.New(cerrors.InvalidArgument, + "DBInstanceParameterGroupName is valid in combination with AllowMajorVersionUpgrade for a major version upgrade only.") +} + +// checkGroupForVersion checks that an instance of engine moving to version +// ends on a DB parameter group of that version's family. explicit is the group +// the request names ("" for none) and must match the family; with none, a +// default group follows the upgrade (upgradedDefaultGroup) while a custom +// group of another family is rejected, as RDS does. Groups of unknown family +// and engines with no known family are not checked. The caller holds m.mu. +func (m *Mock) checkGroupForVersion(engine, version, current, explicit string) error { + want := parameterGroupFamily(engine, version) + if want == "" { + return nil + } + + if explicit != "" { + if family, ok := m.dbParameterGroupFamily(explicit); ok && family != want { + return errGroupFamilyMismatch(explicit, family, want) + } + + return nil + } + + if _, isDefault := defaultGroupFamily(current); isDefault { + return nil + } + + if family, ok := m.dbParameterGroupFamily(current); ok && family != want { + return errCustomGroupNeedsReplacement(current) + } + + return nil +} + +// checkInstanceModifyGroup validates the parameter group a ModifyDBInstance +// request leaves the instance on against the engine version it asks for (or +// the current version when it asks for none). The caller holds m.mu. +func (m *Mock) checkInstanceModifyGroup(inst *rdsdriver.Instance, input *rdsdriver.ModifyInstanceInput) error { + if input.EngineVersion == "" && input.DBParameterGroupName == "" { + return nil + } + + version := input.EngineVersion + if version == "" { + version = inst.EngineVersion + } + + return m.checkGroupForVersion(inst.Engine, version, inst.DBParameterGroupName, input.DBParameterGroupName) +} + +// isMajorUpgrade reports whether moving engine from version from to version +// to changes the parameter group family, which is how RDS tells a major +// version upgrade from a minor one. +func isMajorUpgrade(engine, from, to string) bool { + if to == "" { + return false + } + + next := parameterGroupFamily(engine, to) + + return next != "" && next != parameterGroupFamily(engine, from) +} + +// clusterMemberUpgrades returns the member instances of cluster as they stand +// after the ModifyDBCluster request: in Aurora every instance runs the cluster +// version, so an engine version change moves each member to it, and on a major +// upgrade each member moves to DBInstanceParameterGroupName when given, or +// from a default group to the new version's default group. A custom member +// group of the old family with no DBInstanceParameterGroupName fails the +// request, as in RDS. Nothing is stored; the caller holds m.mu. +func (m *Mock) clusterMemberUpgrades( + cluster *rdsdriver.Cluster, input *rdsdriver.ModifyInstanceInput, +) ([]rdsdriver.Instance, error) { + major := isMajorUpgrade(cluster.Engine, cluster.EngineVersion, input.EngineVersion) + if input.DBInstanceParameterGroupName != "" && !major { + return nil, errInstanceGroupNeedsMajorUpgrade() + } + + if input.EngineVersion == "" { + return nil, nil + } + + out := make([]rdsdriver.Instance, 0, len(cluster.Members)) + + for _, id := range cluster.Members { + inst, ok := m.instances.Get(id) + if !ok { + continue + } + + if major { + if err := m.upgradeMemberGroup(&inst, input); err != nil { + return nil, err + } + } + + inst.EngineVersion = input.EngineVersion + out = append(out, inst) + } + + return out, nil +} + +// upgradeMemberGroup moves one cluster member's DB parameter group for a major +// version upgrade (see clusterMemberUpgrades). The caller holds m.mu. +func (m *Mock) upgradeMemberGroup(inst *rdsdriver.Instance, input *rdsdriver.ModifyInstanceInput) error { + explicit := input.DBInstanceParameterGroupName + if err := m.checkGroupForVersion(inst.Engine, input.EngineVersion, inst.DBParameterGroupName, explicit); err != nil { + return err + } + + if explicit != "" { + inst.DBParameterGroupName = explicit + return nil + } + + inst.DBParameterGroupName = upgradedDefaultGroup(inst.DBParameterGroupName, inst.Engine, input.EngineVersion) + + return nil +} diff --git a/providers/aws/rds/event_log.go b/providers/aws/rds/event_log.go new file mode 100644 index 000000000..ecb1d9253 --- /dev/null +++ b/providers/aws/rds/event_log.go @@ -0,0 +1,425 @@ +package rds + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "time" + + cerrors "github.com/stackshy/cloudemu/v2/errors" + snsdriver "github.com/stackshy/cloudemu/v2/services/notification/driver" + rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" +) + +// RDS event source types (the DescribeEvents / CreateEventSubscription +// SourceType values). +const ( + sourceDBInstance = "db-instance" + sourceDBCluster = "db-cluster" + sourceDBSnapshot = "db-snapshot" + sourceDBClusterSnapshot = "db-cluster-snapshot" + sourceDBParameterGroup = "db-parameter-group" + sourceDBProxy = "db-proxy" + sourceDBSecurityGroup = "db-security-group" +) + +// RDS event categories ("Amazon RDS event categories and event messages"). +const ( + catAvailability = "availability" + catBackup = "backup" + catConfigurationChange = "configuration change" + catCreation = "creation" + catDeletion = "deletion" + catFailover = "failover" + catFailure = "failure" + catMaintenance = "maintenance" + catNotification = "notification" + catRecovery = "recovery" + catRestoration = "restoration" +) + +// Event-log bounds. Real RDS returns events for the past 14 days and, absent a +// time window, the past hour (DescribeEvents API reference). maxRetainedEvents +// additionally caps the in-memory log so a long-running emulator under heavy +// churn stays bounded even within the 14-day window. +const ( + eventRetention = 14 * 24 * time.Hour + defaultEventWindowMins = 60 + maxRetainedEvents = 10000 + rdsNotificationSubject = "RDS Notification Message" + rdsEventDocURLPrefix = "http://docs.amazonwebservices.com/AmazonRDS/latest/UserGuide/USER_Events.html#" + rdsNotificationTimeFmt = "2006-01-02 15:04:05.000" + snsStringAttributeType = "String" + rdsConsoleInstanceLink = "https://console.aws.amazon.com/rds/home?region=%s#dbinstance:id=%s" + rdsEventAttributeID = "EventID" + rdsEventAttributeSource = "Resource" +) + +// eventSpec is one entry of the "Amazon RDS event categories and event +// messages" catalog: the RDS event ID, its category, and its message (a +// fmt format when the documented message carries {{name}} placeholders). +// Messages follow the catalog wording without the trailing period, matching +// the Message text RDS returns from DescribeEvents and in notifications. +type eventSpec struct { + id string + category string + message string +} + +// The lifecycle events the emulator emits, verbatim from +// https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_Events.Messages.html +// and https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_Events.Messages.html. +// +//nolint:gochecknoglobals // static event catalog +var ( + evInstanceDeleted = eventSpec{"RDS-EVENT-0003", catDeletion, "DB instance deleted"} + evInstanceCreated = eventSpec{"RDS-EVENT-0005", catCreation, "DB instance created"} + evInstanceRestarted = eventSpec{"RDS-EVENT-0006", catAvailability, "DB instance restarted"} + evInstancePITR = eventSpec{"RDS-EVENT-0019", catRestoration, "Restored from DB instance %s to %s"} + evInstanceStopped = eventSpec{"RDS-EVENT-0087", catNotification, "DB instance stopped"} + evInstanceStarted = eventSpec{"RDS-EVENT-0088", catNotification, "DB instance started"} + + evParameterUpdated = eventSpec{"RDS-EVENT-0037", catConfigurationChange, "Updated parameter %s to %s with apply method %s"} + + evSnapshotCreating = eventSpec{"RDS-EVENT-0040", catCreation, "Creating manual snapshot"} + evSnapshotDeleted = eventSpec{"RDS-EVENT-0041", catDeletion, "Deleted user snapshot"} + evSnapshotCreated = eventSpec{"RDS-EVENT-0042", catCreation, "Manual snapshot created"} + + evClusterFailover = eventSpec{"RDS-EVENT-0071", catFailover, "Completed failover to DB instance: %s"} + evClusterStopped = eventSpec{"RDS-EVENT-0150", catNotification, "DB cluster stopped"} + evClusterStarted = eventSpec{"RDS-EVENT-0151", catNotification, "DB cluster started"} + evClusterCreated = eventSpec{"RDS-EVENT-0170", catCreation, "DB cluster created"} + evClusterDeleted = eventSpec{"RDS-EVENT-0171", catDeletion, "DB cluster deleted"} + + evClusterSnapshotCreating = eventSpec{"RDS-EVENT-0074", catBackup, "Creating manual cluster snapshot"} + evClusterSnapshotCreated = eventSpec{"RDS-EVENT-0075", catBackup, "Manual cluster snapshot created"} + + evProxyCreated = eventSpec{"RDS-EVENT-0203", catCreation, "RDS created DB proxy %s"} + evProxyModified = eventSpec{"RDS-EVENT-0204", catConfigurationChange, "RDS modified DB proxy %s"} + evProxyDeleted = eventSpec{"RDS-EVENT-0205", catDeletion, "RDS deleted DB proxy %s"} +) + +// SNSPublisher publishes a message to an SNS topic. The SNS mock satisfies it, +// so an RDS event matching an enabled event subscription is delivered to the +// subscription's topic (and from there to its SQS/Lambda subscribers). +type SNSPublisher interface { + Publish(ctx context.Context, input snsdriver.PublishInput) (*snsdriver.PublishOutput, error) +} + +// SetSNSPublisher wires the SNS backend that event subscriptions deliver to. +// Without it events are still recorded for DescribeEvents but not delivered. +func (m *Mock) SetSNSPublisher(p SNSPublisher) { + m.mu.Lock() + defer m.mu.Unlock() + + m.sns = p +} + +// eventNotification is one queued delivery of an event to a subscription's +// SNS topic. Deliveries are queued under the lock and published after it is +// released (flushEventNotifications), so an SNS subscriber that calls back +// into RDS can never deadlock on the provider lock. +type eventNotification struct { + topicARN string + input snsdriver.PublishInput +} + +// eventSource names the resource an event is about. +type eventSource struct { + sourceType string + id string + arn string +} + +// recordEventLocked appends an event to the log and queues a notification for +// every enabled subscription that matches it. The caller holds m.mu (write) +// and must call flushEventNotifications after releasing it. +func (m *Mock) recordEventLocked(src eventSource, spec eventSpec, args ...any) { + msg := spec.message + if len(args) > 0 { + msg = fmt.Sprintf(spec.message, args...) + } + + ev := rdsdriver.Event{ + SourceIdentifier: src.id, + SourceType: src.sourceType, + SourceARN: src.arn, + Message: msg, + EventCategories: []string{spec.category}, + Date: m.opts.Clock.Now().UTC(), + EventID: spec.id, + } + + m.events = append(pruneEvents(m.events, ev.Date), ev) + if over := len(m.events) - maxRetainedEvents; over > 0 { + m.events = slices.Clone(m.events[over:]) + } + + if m.sns == nil { + return + } + + subs := m.eventSubs.SortedValues() + for i := range subs { + if !subscriptionMatches(&subs[i], &ev) { + continue + } + + m.outbox = append(m.outbox, eventNotification{ + topicARN: subs[i].SnsTopicARN, + input: m.notificationFor(subs[i].SnsTopicARN, &ev), + }) + } +} + +// pruneEvents drops events older than the 14-day retention window. +func pruneEvents(events []rdsdriver.Event, now time.Time) []rdsdriver.Event { + cutoff := now.Add(-eventRetention) + + i := 0 + for i < len(events) && events[i].Date.Before(cutoff) { + i++ + } + + if i == 0 { + return events + } + + return slices.Clone(events[i:]) +} + +// subscriptionMatches applies an event subscription's filters: it must be +// enabled, and its SourceType, SourceIds and EventCategories (each "all" when +// empty) must admit the event. +func subscriptionMatches(sub *rdsdriver.EventSubscription, ev *rdsdriver.Event) bool { + if !sub.Enabled { + return false + } + + if sub.SourceType != "" && sub.SourceType != ev.SourceType { + return false + } + + if len(sub.SourceIDs) > 0 && !slices.Contains(sub.SourceIDs, ev.SourceIdentifier) { + return false + } + + if len(sub.EventCategories) == 0 { + return true + } + + for _, c := range ev.EventCategories { + if slices.Contains(sub.EventCategories, c) { + return true + } + } + + return false +} + +// rdsNotificationBody is the JSON message RDS publishes to a subscription's +// SNS topic. Field names and order follow the RDS-to-SNS sample in +// https://docs.aws.amazon.com/lambda/latest/dg/services-rds.html. +type rdsNotificationBody struct { + EventSource string `json:"Event Source"` + EventTime string `json:"Event Time"` + IdentifierLink string `json:"Identifier Link,omitempty"` + SourceID string `json:"Source ID"` + EventID string `json:"Event ID"` + EventMessage string `json:"Event Message"` +} + +// notificationFor builds the SNS publish for an event: the JSON body, the +// "RDS Notification Message" subject, and the EventID / Resource message +// attributes RDS attaches (USER_Events.TagsAttributesForFiltering). +func (m *Mock) notificationFor(topicARN string, ev *rdsdriver.Event) snsdriver.PublishInput { + body := rdsNotificationBody{ + EventSource: ev.SourceType, + EventTime: ev.Date.UTC().Format(rdsNotificationTimeFmt), + SourceID: ev.SourceIdentifier, + EventID: rdsEventDocURLPrefix + ev.EventID, + EventMessage: ev.Message, + } + + // The documented sample only shows the console link of a DB instance, so + // other source types omit it rather than invent a URL shape. + if ev.SourceType == sourceDBInstance { + body.IdentifierLink = fmt.Sprintf(rdsConsoleInstanceLink, arnRegion(ev.SourceARN, m.opts.Region), ev.SourceIdentifier) + } + + raw, _ := json.Marshal(body) //nolint:errchkjson // a struct of strings always marshals. + + attrs := map[string]string{rdsEventAttributeID: ev.EventID} + entries := map[string]snsdriver.MessageAttribute{ + rdsEventAttributeID: {DataType: snsStringAttributeType, Value: ev.EventID}, + } + + if ev.SourceARN != "" { + attrs[rdsEventAttributeSource] = ev.SourceARN + entries[rdsEventAttributeSource] = snsdriver.MessageAttribute{DataType: snsStringAttributeType, Value: ev.SourceARN} + } + + return snsdriver.PublishInput{ + TopicID: arnLastField(topicARN), + Subject: rdsNotificationSubject, + Message: string(raw), + Attributes: attrs, + AttributeEntries: entries, + } +} + +// arnLastField returns the segment after the last colon of an ARN (the SNS +// topic name for a topic ARN). +func arnLastField(arn string) string { + for i := len(arn) - 1; i >= 0; i-- { + if arn[i] == ':' { + return arn[i+1:] + } + } + + return arn +} + +// flushEventNotifications publishes every queued event notification. It must +// be called WITHOUT m.mu held. Delivery is best-effort, like real RDS whose +// notifications are asynchronous: a missing topic or failed publish never +// fails the RDS operation that raised the event. +func (m *Mock) flushEventNotifications() { + m.mu.Lock() + pending := m.outbox + m.outbox = nil + sns := m.sns + m.mu.Unlock() + + if sns == nil { + return + } + + for i := range pending { + _, _ = sns.Publish(context.Background(), pending[i].input) + } +} + +// recordEvent records an event (taking the lock) and delivers its +// notifications. For call sites that do not already hold m.mu. +func (m *Mock) recordEvent(src eventSource, spec eventSpec, args ...any) { + m.mu.Lock() + m.recordEventLocked(src, spec, args...) + m.mu.Unlock() + + m.flushEventNotifications() +} + +// eventSourceTypes are DescribeEvents' valid SourceType values (DescribeEvents +// API reference, SourceType "Valid Values"). +// +//nolint:gochecknoglobals // static lookup table +var eventSourceTypes = map[string]struct{}{ + sourceDBInstance: {}, sourceDBParameterGroup: {}, sourceDBSecurityGroup: {}, sourceDBSnapshot: {}, + sourceDBCluster: {}, sourceDBClusterSnapshot: {}, "custom-engine-version": {}, sourceDBProxy: {}, + "blue-green-deployment": {}, "db-shard-group": {}, "zero-etl": {}, +} + +// errSourceIdentifierWithoutType is the message for a DescribeEvents +// SourceIdentifier given without a SourceType. The wire handler maps an +// InvalidArgument carrying the "must also be provided" wording to +// InvalidParameterCombination, the fault real RDS returns. +const errSourceIdentifierWithoutType = "If SourceIdentifier is supplied, SourceType must also be provided." + +// validateEventSource checks DescribeEvents' SourceType / SourceIdentifier: the +// type must be one of the documented values, and "If SourceIdentifier is +// supplied, SourceType must also be provided" (DescribeEvents API reference). +func validateEventSource(sourceType, sourceID string) error { + if sourceType != "" { + if _, ok := eventSourceTypes[sourceType]; !ok { + return cerrors.New(cerrors.InvalidArgument, "Invalid source type: "+sourceType) + } + } + + if sourceID != "" && sourceType == "" { + return cerrors.New(cerrors.InvalidArgument, errSourceIdentifierWithoutType) + } + + return nil +} + +// DescribeEvents returns the recorded events matching the input's filters, +// oldest first. The time window defaults to the past 60 minutes (or Duration +// minutes) ending at EndTime (default now), and never reaches back further +// than the 14-day retention window. An unknown SourceType, or a +// SourceIdentifier without a SourceType, is rejected with InvalidArgument. +// +//nolint:gocritic // input matches the driver interface signature. +func (m *Mock) DescribeEvents(_ context.Context, input rdsdriver.DescribeEventsInput) ([]rdsdriver.Event, error) { + if err := validateEventSource(input.SourceType, input.SourceIdentifier); err != nil { + return nil, err + } + + m.mu.RLock() + defer m.mu.RUnlock() + + now := m.opts.Clock.Now() + + end := input.EndTime + if end.IsZero() { + end = now + } + + start := input.StartTime + if start.IsZero() { + mins := input.Duration + if mins <= 0 { + mins = defaultEventWindowMins + } + + start = end.Add(-time.Duration(mins) * time.Minute) + } + + if floor := now.Add(-eventRetention); start.Before(floor) { + start = floor + } + + out := make([]rdsdriver.Event, 0) + + for i := range m.events { + ev := &m.events[i] + if ev.Date.Before(start) || ev.Date.After(end) { + continue + } + + if !eventMatchesFilter(ev, &input) { + continue + } + + c := *ev + c.EventCategories = cloneSlice(ev.EventCategories) + out = append(out, c) + } + + return out, nil +} + +// eventMatchesFilter applies DescribeEvents' SourceType, SourceIdentifier and +// EventCategories filters. +func eventMatchesFilter(ev *rdsdriver.Event, input *rdsdriver.DescribeEventsInput) bool { + if input.SourceType != "" && ev.SourceType != input.SourceType { + return false + } + + if input.SourceIdentifier != "" && ev.SourceIdentifier != input.SourceIdentifier { + return false + } + + if len(input.EventCategories) == 0 { + return true + } + + for _, c := range ev.EventCategories { + if slices.Contains(input.EventCategories, c) { + return true + } + } + + return false +} diff --git a/providers/aws/rds/event_log_test.go b/providers/aws/rds/event_log_test.go new file mode 100644 index 000000000..7dc332b81 --- /dev/null +++ b/providers/aws/rds/event_log_test.go @@ -0,0 +1,341 @@ +package rds + +import ( + "context" + "encoding/json" + "slices" + "sync" + "testing" + "time" + + "github.com/stackshy/cloudemu/v2/config" + cerrors "github.com/stackshy/cloudemu/v2/errors" + snsdriver "github.com/stackshy/cloudemu/v2/services/notification/driver" + rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" +) + +// fakeSNS records every publish it receives. +type fakeSNS struct { + mu sync.Mutex + msgs []snsdriver.PublishInput +} + +//nolint:gocritic // matches the SNSPublisher signature. +func (f *fakeSNS) Publish(_ context.Context, in snsdriver.PublishInput) (*snsdriver.PublishOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + + f.msgs = append(f.msgs, in) + + return &snsdriver.PublishOutput{MessageID: "m"}, nil +} + +func (f *fakeSNS) published() []snsdriver.PublishInput { + f.mu.Lock() + defer f.mu.Unlock() + + return append([]snsdriver.PublishInput(nil), f.msgs...) +} + +func newEventTestMock() (*Mock, *config.FakeClock) { + fc := config.NewFakeClock(time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)) + opts := config.NewOptions( + config.WithClock(fc), + config.WithRegion("us-east-1"), + config.WithAccountID("123456789012"), + ) + + return New(opts), fc +} + +func eventIDs(events []rdsdriver.Event) []string { + out := make([]string, 0, len(events)) + for i := range events { + out = append(out, events[i].EventID) + } + + return out +} + +func TestDescribeEventsRecordsInstanceLifecycle(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "db1", Engine: "postgres", EngineVersion: "15.4"}) + requireNoError(t, err) + requireNoError(t, m.StopInstance(ctx, "db1")) + requireNoError(t, m.StartInstance(ctx, "db1")) + requireNoError(t, m.RebootInstance(ctx, "db1")) + requireNoError(t, m.DeleteInstance(ctx, "db1")) + + events, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{SourceType: "db-instance", SourceIdentifier: "db1"}) + requireNoError(t, err) + + want := []string{"RDS-EVENT-0005", "RDS-EVENT-0087", "RDS-EVENT-0088", "RDS-EVENT-0006", "RDS-EVENT-0003"} + assertStrings(t, want, eventIDs(events)) + + first := events[0] + assertEqual(t, "DB instance created", first.Message) + assertStrings(t, []string{"creation"}, first.EventCategories) + assertEqual(t, "arn:aws:rds:us-east-1:123456789012:db:db1", first.SourceARN) + assertEqual(t, "db-instance", first.SourceType) +} + +func TestDescribeEventsFilters(t *testing.T) { + ctx := context.Background() + m, fc := newEventTestMock() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "a", Engine: "mysql"}) + requireNoError(t, err) + + fc.Advance(2 * time.Hour) + + _, err = m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "b", Engine: "mysql"}) + requireNoError(t, err) + _, err = m.CreateSnapshot(ctx, rdsdriver.SnapshotConfig{ID: "s1", InstanceID: "b"}) + requireNoError(t, err) + + // Default window is the past hour: instance "a" (2h ago) is excluded. + got, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{}) + requireNoError(t, err) + assertStrings(t, []string{"RDS-EVENT-0005", "RDS-EVENT-0040", "RDS-EVENT-0042"}, eventIDs(got)) + + // Duration widens the window. + got, err = m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{Duration: 180, SourceType: "db-instance"}) + requireNoError(t, err) + assertEqual(t, 2, len(got)) + + // StartTime/EndTime bound the window explicitly. + start := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) + got, err = m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{StartTime: start, EndTime: start.Add(time.Minute)}) + requireNoError(t, err) + assertEqual(t, 1, len(got)) + assertEqual(t, "a", got[0].SourceIdentifier) + + // Category filter. + got, err = m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{EventCategories: []string{"creation"}, SourceType: "db-snapshot"}) + requireNoError(t, err) + assertEqual(t, 2, len(got)) + + got, err = m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{EventCategories: []string{"deletion"}}) + requireNoError(t, err) + assertEqual(t, 0, len(got)) +} + +func TestEventLogRetention(t *testing.T) { + ctx := context.Background() + m, fc := newEventTestMock() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "old", Engine: "mysql"}) + requireNoError(t, err) + + fc.Advance(15 * 24 * time.Hour) + + _, err = m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "new", Engine: "mysql"}) + requireNoError(t, err) + + // A window reaching back past 14 days never returns the expired event. + got, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{Duration: 30 * 24 * 60}) + requireNoError(t, err) + assertEqual(t, 1, len(got)) + assertEqual(t, "new", got[0].SourceIdentifier) + + // And the expired event was pruned from the stored log. + m.mu.RLock() + n := len(m.events) + m.mu.RUnlock() + assertEqual(t, 1, n) +} + +func TestEventSubscriptionDeliversToSNS(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + sns := &fakeSNS{} + m.SetSNSPublisher(sns) + + topic := "arn:aws:sns:us-east-1:123456789012:rds-events" + + _, err := m.CreateEventSubscription(ctx, rdsdriver.EventSubscriptionConfig{ + Name: "all-db1", SnsTopicARN: topic, SourceType: "db-instance", + SourceIDs: []string{"db1"}, Enabled: true, + }) + requireNoError(t, err) + + _, err = m.CreateEventSubscription(ctx, rdsdriver.EventSubscriptionConfig{ + Name: "deletions", SnsTopicARN: topic, EventCategories: []string{"deletion"}, Enabled: true, + }) + requireNoError(t, err) + + _, err = m.CreateEventSubscription(ctx, rdsdriver.EventSubscriptionConfig{ + Name: "disabled", SnsTopicARN: topic, Enabled: false, + }) + requireNoError(t, err) + + _, err = m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "db1", Engine: "mysql"}) + requireNoError(t, err) + _, err = m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "other", Engine: "mysql"}) + requireNoError(t, err) + requireNoError(t, m.DeleteInstance(ctx, "other")) + + msgs := sns.published() + // db1 created -> "all-db1"; other created -> none; other deleted -> "deletions". + assertEqual(t, 2, len(msgs)) + + first := msgs[0] + assertEqual(t, "rds-events", first.TopicID) + assertEqual(t, "RDS Notification Message", first.Subject) + assertEqual(t, "RDS-EVENT-0005", first.Attributes["EventID"]) + assertEqual(t, "arn:aws:rds:us-east-1:123456789012:db:db1", first.Attributes["Resource"]) + + var body map[string]string + requireNoError(t, json.Unmarshal([]byte(first.Message), &body)) + assertEqual(t, "db-instance", body["Event Source"]) + assertEqual(t, "2025-01-01 00:00:00.000", body["Event Time"]) + assertEqual(t, "db1", body["Source ID"]) + assertEqual(t, "http://docs.amazonwebservices.com/AmazonRDS/latest/UserGuide/USER_Events.html#RDS-EVENT-0005", body["Event ID"]) + assertEqual(t, "DB instance created", body["Event Message"]) + assertEqual(t, "https://console.aws.amazon.com/rds/home?region=us-east-1#dbinstance:id=db1", body["Identifier Link"]) + + var second map[string]string + requireNoError(t, json.Unmarshal([]byte(msgs[1].Message), &second)) + assertEqual(t, "other", second["Source ID"]) + assertEqual(t, "DB instance deleted", second["Event Message"]) +} + +func TestEventsForClusterSnapshotParamGroupProxy(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + _, err := m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: "c1", Engine: "aurora-postgresql"}) + requireNoError(t, err) + _, err = m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "c1-a", Engine: "aurora-postgresql", ClusterID: "c1"}) + requireNoError(t, err) + _, err = m.FailoverDBCluster(ctx, "c1", "c1-a") + requireNoError(t, err) + requireNoError(t, m.StopCluster(ctx, "c1")) + requireNoError(t, m.StartCluster(ctx, "c1")) + _, err = m.CreateClusterSnapshot(ctx, rdsdriver.ClusterSnapshotConfig{ID: "cs1", ClusterID: "c1"}) + requireNoError(t, err) + + _, err = m.CreateDBParameterGroup(ctx, rdsdriver.ParameterGroupConfig{Name: "pg1", Family: "postgres15", Description: "d"}) + requireNoError(t, err) + _, err = m.ModifyDBParameterGroup(ctx, "pg1", []rdsdriver.Parameter{{Name: "work_mem", Value: "8192", ApplyMethod: "immediate"}}) + requireNoError(t, err) + + _, err = m.CreateDBProxy(ctx, rdsdriver.DBProxyConfig{Name: "px", EngineFamily: "POSTGRESQL"}) + requireNoError(t, err) + _, err = m.DeleteDBProxy(ctx, "px") + requireNoError(t, err) + + check := func(sourceType string, want []string) []rdsdriver.Event { + t.Helper() + + got, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{SourceType: sourceType}) + requireNoError(t, err) + assertStrings(t, want, eventIDs(got)) + + return got + } + + cl := check("db-cluster", []string{"RDS-EVENT-0170", "RDS-EVENT-0071", "RDS-EVENT-0150", "RDS-EVENT-0151"}) + assertEqual(t, "Completed failover to DB instance: c1-a", cl[1].Message) + + check("db-cluster-snapshot", []string{"RDS-EVENT-0074", "RDS-EVENT-0075"}) + + pg := check("db-parameter-group", []string{"RDS-EVENT-0037"}) + assertEqual(t, "Updated parameter work_mem to 8192 with apply method immediate", pg[0].Message) + + px := check("db-proxy", []string{"RDS-EVENT-0203", "RDS-EVENT-0205"}) + assertEqual(t, "RDS created DB proxy px", px[0].Message) +} + +func TestEventLogSurvivesSnapshotRestore(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "db1", Engine: "mysql"}) + requireNoError(t, err) + + raw, err := m.Snapshot(ctx, false) + requireNoError(t, err) + + restored, _ := newEventTestMock() + requireNoError(t, restored.Restore(ctx, raw)) + + got, err := restored.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{}) + requireNoError(t, err) + assertStrings(t, []string{"RDS-EVENT-0005"}, eventIDs(got)) +} + +func assertStrings(t *testing.T, want, got []string) { + t.Helper() + + if !slices.Equal(want, got) { + t.Fatalf("got %v, want %v", got, want) + } +} + +// TestTagsOnEventSubscriptionAndDefaultGroup: the "es:" ARN of an event +// subscription and the ARN of a synthesized default parameter group are +// taggable, as Terraform's aws_db_event_subscription requires. +func TestTagsOnEventSubscriptionAndDefaultGroup(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + sub, err := m.CreateEventSubscription(ctx, rdsdriver.EventSubscriptionConfig{ + Name: "tfsub", SnsTopicARN: "arn:aws:sns:us-east-1:123456789012:t", Enabled: true, + Tags: map[string]string{"env": "dev"}, + }) + requireNoError(t, err) + + tags, err := m.ListTagsForResource(ctx, sub.ARN) + requireNoError(t, err) + assertEqual(t, "dev", tags["env"]) + + requireNoError(t, m.AddTagsToResource(ctx, sub.ARN, map[string]string{"team": "db"})) + requireNoError(t, m.RemoveTagsFromResource(ctx, sub.ARN, []string{"env"})) + + tags, err = m.ListTagsForResource(ctx, sub.ARN) + requireNoError(t, err) + assertEqual(t, 1, len(tags)) + assertEqual(t, "db", tags["team"]) + + _, err = m.DeleteEventSubscription(ctx, "tfsub") + requireNoError(t, err) + + _, err = m.ListTagsForResource(ctx, sub.ARN) + requireCode(t, err, cerrors.IsNotFound) + + // A re-created subscription of the same name starts untagged. + _, err = m.CreateEventSubscription(ctx, rdsdriver.EventSubscriptionConfig{ + Name: "tfsub", SnsTopicARN: "arn:aws:sns:us-east-1:123456789012:t", Enabled: true, + }) + requireNoError(t, err) + + tags, err = m.ListTagsForResource(ctx, sub.ARN) + requireNoError(t, err) + assertEqual(t, 0, len(tags)) + + pgARN := parameterGroupARN("us-east-1", "123456789012", "default.postgres15") + + tags, err = m.ListTagsForResource(ctx, pgARN) + requireNoError(t, err) + assertEqual(t, 0, len(tags)) +} + +// TestDescribeEventsSourceRulesInProvider: a library caller gets the same +// SourceType / SourceIdentifier validation the wire returns. +func TestDescribeEventsSourceRulesInProvider(t *testing.T) { + ctx := context.Background() + m, _ := newEventTestMock() + + _, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{SourceIdentifier: "db1"}) + requireCode(t, err, cerrors.IsInvalidArgument) + assertEqual(t, errSourceIdentifierWithoutType, err.(*cerrors.Error).Message) //nolint:errorlint // cerrors.New returns *Error. + + _, err = m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{SourceType: "db-bogus"}) + requireCode(t, err, cerrors.IsInvalidArgument) + + _, err = m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{SourceType: "db-instance", SourceIdentifier: "db1"}) + requireNoError(t, err) +} diff --git a/providers/aws/rds/events.go b/providers/aws/rds/events.go index 99d2d737d..c07725a14 100644 --- a/providers/aws/rds/events.go +++ b/providers/aws/rds/events.go @@ -24,16 +24,16 @@ func errEventSubscriptionNotFound(name string) error { // //nolint:gochecknoglobals // static lookup table var eventCategoryCatalog = map[string][]string{ - "db-instance": { - "availability", "backup", "configuration change", "creation", "deletion", - "failover", "failure", "maintenance", "notification", "recovery", "restoration", + sourceDBInstance: { + catAvailability, catBackup, catConfigurationChange, catCreation, catDeletion, + catFailover, catFailure, catMaintenance, catNotification, catRecovery, catRestoration, }, - "db-cluster": {"configuration change", "creation", "deletion", "failover", "failure", "maintenance", "notification"}, - "db-snapshot": {"creation", "deletion", "notification", "restoration"}, - "db-cluster-snapshot": {"backup", "notification"}, - "db-parameter-group": {"configuration change"}, - "db-security-group": {"configuration change", "failure"}, - "db-proxy": {"configuration change", "creation", "deletion"}, + sourceDBCluster: {catConfigurationChange, catCreation, catDeletion, catFailover, catFailure, catMaintenance, catNotification}, + sourceDBSnapshot: {catCreation, catDeletion, catNotification, catRestoration}, + sourceDBClusterSnapshot: {catBackup, catNotification}, + sourceDBParameterGroup: {catConfigurationChange}, + sourceDBSecurityGroup: {catConfigurationChange, catFailure}, + sourceDBProxy: {catConfigurationChange, catCreation, catDeletion}, } //nolint:gocritic // cfg matches the driver interface signature. @@ -66,6 +66,10 @@ func (m *Mock) CreateEventSubscription(ctx context.Context, cfg rdsdriver.EventS CreatedAt: m.opts.Clock.Now().UTC(), } m.eventSubs.Set(cfg.Name, sub) + // The subscription's tags live beside it, keyed by its ARN, so + // Add/Remove/ListTagsForResource on the "es:" ARN work (Terraform's + // aws_db_event_subscription reads them back after create). + m.setGroupTags(sub.ARN, copyTags(cfg.Tags)) out := sub @@ -155,18 +159,13 @@ func (m *Mock) DeleteEventSubscription(_ context.Context, name string) (*rdsdriv } m.eventSubs.Delete(name) + m.setGroupTags(sub.ARN, nil) out := sub return &out, nil } -// DescribeEvents returns an empty list: the emulator does not retain an event -// timeline, so there are truthfully no events to report for any window. -func (*Mock) DescribeEvents(_ context.Context, _, _ string, _ []string) ([]rdsdriver.Event, error) { - return []rdsdriver.Event{}, nil -} - func (*Mock) DescribeEventCategories(_ context.Context, sourceType string) ([]rdsdriver.EventCategoryGroup, error) { if sourceType != "" { cats, ok := eventCategoryCatalog[sourceType] @@ -194,6 +193,6 @@ func (*Mock) DescribeEventCategories(_ context.Context, sourceType string) ([]rd // //nolint:gochecknoglobals // static lookup table var eventCategorySourceTypes = []string{ - "db-instance", "db-cluster", "db-snapshot", "db-cluster-snapshot", - "db-parameter-group", "db-security-group", "db-proxy", + sourceDBInstance, sourceDBCluster, sourceDBSnapshot, sourceDBClusterSnapshot, + sourceDBParameterGroup, sourceDBSecurityGroup, sourceDBProxy, } diff --git a/providers/aws/rds/events_test.go b/providers/aws/rds/events_test.go index 7bd883dd1..cf686da26 100644 --- a/providers/aws/rds/events_test.go +++ b/providers/aws/rds/events_test.go @@ -62,8 +62,8 @@ func TestDescribeEventsAndCategories(t *testing.T) { ctx := context.Background() m := newTestMock() - // No event timeline is retained, so events are always empty. - events, err := m.DescribeEvents(ctx, "db-instance", "mydb", nil) + // No resource named mydb exists, so it has no events. + events, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{SourceType: "db-instance", SourceIdentifier: "mydb"}) if err != nil || len(events) != 0 { t.Fatalf("DescribeEvents: got %d events, err %v", len(events), err) } diff --git a/providers/aws/rds/instance_config_validation.go b/providers/aws/rds/instance_config_validation.go index 562a4586c..9b5c8b5b0 100644 --- a/providers/aws/rds/instance_config_validation.go +++ b/providers/aws/rds/instance_config_validation.go @@ -11,8 +11,8 @@ import ( // query-protocol wire in cloudemu, so their engine names are accepted here too. // AWS rejects any other value with InvalidParameterValue. var validEngines = map[string]struct{}{ //nolint:gochecknoglobals // static lookup table - "aurora-mysql": {}, - "aurora-postgresql": {}, + engineAuroraMySQL: {}, + engineAuroraPostgreSQL: {}, "custom-oracle-ee": {}, "custom-oracle-ee-cdb": {}, "custom-oracle-se2": {}, @@ -24,13 +24,13 @@ var validEngines = map[string]struct{}{ //nolint:gochecknoglobals // static look "db2-ae": {}, "db2-ce": {}, "db2-se": {}, - "mariadb": {}, - "mysql": {}, + engineMariaDB: {}, + engineMySQL: {}, "oracle-ee": {}, "oracle-ee-cdb": {}, "oracle-se2": {}, "oracle-se2-cdb": {}, - "postgres": {}, + enginePostgres: {}, "sqlserver-dev-ee": {}, "sqlserver-ee": {}, "sqlserver-se": {}, diff --git a/providers/aws/rds/instance_defaults_test.go b/providers/aws/rds/instance_defaults_test.go index 700d6581d..667aafc28 100644 --- a/providers/aws/rds/instance_defaults_test.go +++ b/providers/aws/rds/instance_defaults_test.go @@ -14,12 +14,19 @@ func TestCreateInstanceDefaultParameterGroup(t *testing.T) { tests := []struct { name string engine string + version string paramGroup string want string }{ {name: "mysql default", engine: "mysql", want: "default.mysql8.0"}, - {name: "postgres default", engine: "postgres", want: "default.postgres16"}, + // The default postgres version is 15.4, so the family is postgres15, + // not the family of the newest catalog version. + {name: "postgres default version", engine: "postgres", want: "default.postgres15"}, + {name: "postgres 15.4", engine: "postgres", version: "15.4", want: "default.postgres15"}, + {name: "postgres 16.2", engine: "postgres", version: "16.2", want: "default.postgres16"}, + {name: "aurora-postgresql 15.4", engine: "aurora-postgresql", version: "15.4", want: "default.aurora-postgresql15"}, {name: "aurora-mysql default", engine: "aurora-mysql", want: "default.aurora-mysql8.0"}, + {name: "mariadb 10.11.6", engine: "mariadb", version: "10.11.6", want: "default.mariadb10.11"}, {name: "explicit preserved", engine: "mysql", paramGroup: "custom-pg", want: "custom-pg"}, } @@ -30,6 +37,7 @@ func TestCreateInstanceDefaultParameterGroup(t *testing.T) { inst, err := m.CreateInstance(context.Background(), rdsdriver.InstanceConfig{ ID: "db1", Engine: tc.engine, + EngineVersion: tc.version, DBParameterGroupName: tc.paramGroup, }) requireNoError(t, err) diff --git a/providers/aws/rds/metadata.go b/providers/aws/rds/metadata.go index 7ab8713da..ef1a93201 100644 --- a/providers/aws/rds/metadata.go +++ b/providers/aws/rds/metadata.go @@ -2,6 +2,7 @@ package rds import ( "context" + "strconv" "strings" cerrors "github.com/stackshy/cloudemu/v2/errors" @@ -23,6 +24,7 @@ const ( arnKindClusterParamGroup = "cluster-pg" arnKindOptionGroup = "og" arnKindSubnetGroup = "subgrp" + arnKindEventSubscription = "es" arnFieldCount = 7 ) @@ -32,26 +34,166 @@ const ( // //nolint:gochecknoglobals // static lookup table var engineVersionCatalog = map[string][]string{ - "mysql": {"8.0.35", "8.0.36"}, - "postgres": {"15.4", "16.2"}, - "mariadb": {"10.11.6"}, - "aurora-mysql": {"8.0.mysql_aurora.3.05.2"}, - "aurora-postgresql": {"15.4", "16.1"}, - "docdb": {"5.0.0"}, - "neptune": {"1.3.1.0"}, + engineMySQL: {"8.0.35", "8.0.36"}, + enginePostgres: {"15.4", "16.2"}, + engineMariaDB: {"10.11.6"}, + engineAuroraMySQL: {"8.0.mysql_aurora.3.05.2"}, + engineAuroraPostgreSQL: {"15.4", "16.1"}, + engineDocDB: {"5.0.0"}, + engineNeptune: {"1.3.1.0"}, } -// engineFamily maps an engine to its parameter-group family prefix. +// Engine names whose parameter-group family is derived from the engine version. +const ( + engineMySQL = "mysql" + enginePostgres = "postgres" + engineMariaDB = "mariadb" + engineAuroraMySQL = "aurora-mysql" + engineAuroraPostgreSQL = "aurora-postgresql" +) + +// Version thresholds for the family naming rules below. +const ( + // postgresMajorOnlyFrom is the first PostgreSQL major whose family is named + // by major alone (postgres10, postgres15); earlier majors use major.minor + // (postgres9.6). + postgresMajorOnlyFrom = 10 + // neptuneMinorFamilyFrom is the Neptune 1.x minor (1.2.0.0) from which the + // family carries major.minor (neptune1.2, neptune1.3); earlier releases share + // the "neptune1" family. + neptuneMinorFamilyFrom = 2 + // familyVersionFields is the number of leading version fields a family + // needs (major.minor). + familyVersionFields = 2 +) + +// parameterGroupFamily returns the DB parameter group family real RDS reports +// (DescribeDBEngineVersions.DBParameterGroupFamily) and uses for the default +// "default." group of an engine at a given engine version. The family +// tracks the version, not just the engine: PostgreSQL 15.4 belongs to +// postgres15 while 16.2 belongs to postgres16. The naming per engine follows +// the DBParameterGroupFamily values RDS documents (e.g. postgres14, mysql8.0, +// mariadb10.6, aurora-mysql8.0, aurora-postgresql14, oracle-ee-19, +// oracle-se2-cdb-21, sqlserver-ee-15.0, db2-ae, neptune1.2, docdb5.0): // -//nolint:gochecknoglobals // static lookup table -var engineFamily = map[string]string{ - "mysql": "mysql8.0", - "postgres": "postgres16", - "mariadb": "mariadb10.11", - "aurora-mysql": "aurora-mysql8.0", - "aurora-postgresql": "aurora-postgresql16", - "docdb": "docdb5.0", - "neptune": "neptune1.3", +// postgres / aurora-postgresql: (major >= 10), else . +// mysql / mariadb / docdb: . +// aurora-mysql: aurora-mysql. ("8.0.mysql_aurora.3.05.2" -> 8.0) +// neptune: neptune1 before 1.2.0.0, else neptune. +// oracle-*: - ("19.0.0.0.ru-2024-01.rur-2024-01.r1" -> 19) +// sqlserver-*: -. ("15.00.4365.2.v1" -> 15.0) +// db2-*: (one family per edition) +// +// An empty version resolves to the engine's default version first. An unknown +// engine, or a version that does not parse, yields "" (no known family). +func parameterGroupFamily(engine, version string) string { + if strings.HasPrefix(engine, "db2-") { + return engine + } + + if version == "" { + version = defaultEngineVersion(engine) + } + + major, minor, ok := versionMajorMinor(version) + if !ok { + return "" + } + + return versionedFamily(engine, major, minor) +} + +// versionedFamily names the family of an engine whose family carries its +// version (see parameterGroupFamily), or "" for an unknown engine. +func versionedFamily(engine string, major, minor int) string { + majorStr := strconv.Itoa(major) + majorMinor := majorStr + "." + strconv.Itoa(minor) + + switch { + case engine == enginePostgres || engine == engineAuroraPostgreSQL: + if major < postgresMajorOnlyFrom { + return engine + majorMinor + } + + return engine + majorStr + case engine == engineMySQL, engine == engineMariaDB, engine == engineDocDB, engine == engineAuroraMySQL: + return engine + majorMinor + case engine == engineNeptune: + return neptuneFamily(major, minor) + case strings.HasPrefix(engine, "oracle-"): + return engine + "-" + majorStr + case strings.HasPrefix(engine, "sqlserver-"): + return engine + "-" + majorMinor + default: + return "" + } +} + +// isKnownFamily reports whether family is a DB parameter group family that +// parameterGroupFamily produces for some engine and version: the family names +// a recognized engine followed by a version it could come from (postgres16, +// mysql8.0, aurora-postgresql15, oracle-ee-19, sqlserver-ee-15.0, neptune1, +// db2-se). It rejects names such as "bogus", "postgres16.2" or +// "postgres-16", so "default." groups resolve only for real families. +func isKnownFamily(family string) bool { + for engine := range validEngines { + if familyFromEngine(engine, family) { + return true + } + } + + return false +} + +// familyFromEngine reports whether family is the family of engine at the +// version spelled by the rest of the name ("16" or "16.0" for postgres16). +func familyFromEngine(engine, family string) bool { + if family == engine { + return parameterGroupFamily(engine, "") == family + } + + rest, ok := strings.CutPrefix(family, engine) + if !ok { + return false + } + + rest = strings.TrimPrefix(rest, "-") + if !strings.Contains(rest, ".") { + rest += ".0" + } + + return parameterGroupFamily(engine, rest) == family +} + +// neptuneFamily returns the Neptune family: neptune1 for every release before +// 1.2.0.0, then neptune.. +func neptuneFamily(major, minor int) string { + if major == 1 && minor < neptuneMinorFamilyFrom { + return "neptune1" + } + + return engineNeptune + strconv.Itoa(major) + "." + strconv.Itoa(minor) +} + +// versionMajorMinor parses the leading numeric major.minor fields of an engine +// version ("15.4" -> 15, 4; "8.0.mysql_aurora.3.05.2" -> 8, 0). +func versionMajorMinor(version string) (major, minor int, ok bool) { + parts := strings.Split(version, ".") + if len(parts) < familyVersionFields { + return 0, 0, false + } + + major, err := strconv.Atoi(parts[0]) + if err != nil { + return 0, 0, false + } + + minor, err = strconv.Atoi(parts[1]) + if err != nil { + return 0, 0, false + } + + return major, minor, true } // orderableInstanceClasses is the representative set of instance classes the @@ -92,7 +234,7 @@ func (*Mock) DescribeDBEngineVersions(_ context.Context, engine, engineVersion s Engine: eng, EngineVersion: v, DBEngineDescription: "Amazon RDS " + eng, - DBParameterGroupFamily: engineFamily[eng], + DBParameterGroupFamily: parameterGroupFamily(eng, v), }) } } @@ -133,7 +275,7 @@ func (*Mock) DescribeOrderableDBInstanceOptions( // //nolint:gochecknoglobals // static lookup table var engineCatalogOrder = []string{ - "mysql", "postgres", "mariadb", "aurora-mysql", "aurora-postgresql", "docdb", "neptune", + engineMySQL, enginePostgres, engineMariaDB, engineAuroraMySQL, engineAuroraPostgreSQL, engineDocDB, engineNeptune, } // ---- tagging ---- @@ -227,27 +369,37 @@ func (m *Mock) recordTags(kind, id string) (map[string]string, bool) { return nil, false } -// isGroupKind reports whether an ARN kind is a parameter/option/subnet group. +// isGroupKind reports whether an ARN kind keeps its tags in groupTags (keyed +// by ARN) rather than on its record: parameter/option/subnet groups and event +// subscriptions. func isGroupKind(kind string) bool { switch kind { - case arnKindParamGroup, arnKindClusterParamGroup, arnKindOptionGroup, arnKindSubnetGroup: + case arnKindParamGroup, arnKindClusterParamGroup, arnKindOptionGroup, arnKindSubnetGroup, + arnKindEventSubscription: return true default: return false } } -// groupExists reports whether a parameter/option/subnet group named id exists. +// groupExists reports whether a parameter/option/subnet group or event +// subscription named id exists. A "default." DB / cluster parameter +// group always exists (DescribeDB[Cluster]ParameterGroups resolves it), so it +// is taggable like a stored group. func (m *Mock) groupExists(kind, id string) bool { + _, isDefault := defaultGroupFamily(id) + switch kind { case arnKindParamGroup: - return m.paramGroups.Has(id) + return m.paramGroups.Has(id) || isDefault case arnKindClusterParamGroup: - return m.clusterParamGroups.Has(id) + return m.clusterParamGroups.Has(id) || isDefault case arnKindOptionGroup: return m.optionGroups.Has(id) case arnKindSubnetGroup: return m.subnetGroups.Has(id) + case arnKindEventSubscription: + return m.eventSubs.Has(id) default: return false } diff --git a/providers/aws/rds/option_group.go b/providers/aws/rds/option_group.go index fedec5e35..f0e791c0c 100644 --- a/providers/aws/rds/option_group.go +++ b/providers/aws/rds/option_group.go @@ -52,8 +52,8 @@ func errOptionGroupNotFound(name string) error { // //nolint:gochecknoglobals // static lookup table var optionGroupOptionCatalog = map[string][]string{ - "mysql": {"MARIADB_AUDIT_PLUGIN"}, - "mariadb": {"MARIADB_AUDIT_PLUGIN"}, + engineMySQL: {"MARIADB_AUDIT_PLUGIN"}, + engineMariaDB: {"MARIADB_AUDIT_PLUGIN"}, "oracle-ee": {"APEX", "OEM", "S3_INTEGRATION", "NATIVE_NETWORK_ENCRYPTION"}, "oracle-se2": {"APEX", "OEM", "S3_INTEGRATION"}, "sqlserver-ee": {"TDE", "SQLSERVER_AUDIT"}, diff --git a/providers/aws/rds/parameter_group.go b/providers/aws/rds/parameter_group.go index 9fc7f1fcd..9fd743716 100644 --- a/providers/aws/rds/parameter_group.go +++ b/providers/aws/rds/parameter_group.go @@ -2,6 +2,8 @@ package rds import ( "context" + "maps" + "slices" "strings" cerrors "github.com/stackshy/cloudemu/v2/errors" @@ -54,6 +56,20 @@ func errClusterParameterGroupNotFound(name string) error { return cerrors.Newf(cerrors.NotFound, "DB cluster parameter group %q not found", name) } +// missingGroupErr is the error for a modify/reset of a DB or cluster parameter +// group that is not stored. A "default." group always exists in RDS (and +// DescribeDB[Cluster]ParameterGroups lists it), but its parameters are fixed: +// real RDS answers "Cannot modify a default parameter group." with +// InvalidParameterValue rather than denying the group exists. Any other name +// is genuinely unknown and gets notFound. +func missingGroupErr(name string, notFound error) error { + if _, isDefault := defaultGroupFamily(name); isDefault { + return cerrors.New(cerrors.InvalidArgument, "Cannot modify a default parameter group.") + } + + return notFound +} + func applyMethodOrDefault(m string) string { if m == "" { return "pending-reboot" @@ -120,16 +136,30 @@ func (m *Mock) CreateDBParameterGroup(ctx context.Context, cfg rdsdriver.Paramet return &out, nil } -func (m *Mock) DescribeDBParameterGroups(_ context.Context, names []string) ([]rdsdriver.ParameterGroup, error) { +// DescribeDBParameterGroups lists the stored groups plus the always-present +// "default." groups RDS associates with instances created without one. +// A default group is synthesized (never stored, so it cannot be modified or +// deleted) when an instance references it, or when it is named explicitly, +// mirroring DescribeDBParameters. +func (m *Mock) DescribeDBParameterGroups(ctx context.Context, names []string) ([]rdsdriver.ParameterGroup, error) { m.mu.RLock() defer m.mu.RUnlock() + region := regionctx.RegionOr(ctx, m.opts.Region) + if len(names) == 0 { all := m.paramGroups.SortedValues() for i := range all { all[i].Parameters = copyParams(all[i].Parameters) } + for _, name := range m.referencedDefaultParamGroups() { + family, _ := defaultGroupFamily(name) + all = append(all, m.defaultParameterGroup(region, name, family)) + } + + slices.SortFunc(all, func(a, b rdsdriver.ParameterGroup) int { return strings.Compare(a.Name, b.Name) }) + return all, nil } @@ -138,7 +168,12 @@ func (m *Mock) DescribeDBParameterGroups(_ context.Context, names []string) ([]r for _, name := range names { pg, ok := m.paramGroups.Get(name) if !ok { - return nil, errParameterGroupNotFound(name) + family, isDefault := defaultGroupFamily(name) + if !isDefault { + return nil, errParameterGroupNotFound(name) + } + + pg = m.defaultParameterGroup(region, name, family) } pg.Parameters = copyParams(pg.Parameters) @@ -149,17 +184,25 @@ func (m *Mock) DescribeDBParameterGroups(_ context.Context, names []string) ([]r } func (m *Mock) ModifyDBParameterGroup(_ context.Context, name string, params []rdsdriver.Parameter) (*rdsdriver.ParameterGroup, error) { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() pg, ok := m.paramGroups.Get(name) if !ok { - return nil, errParameterGroupNotFound(name) + return nil, missingGroupErr(name, errParameterGroupNotFound(name)) } pg.Parameters = withParams(pg.Parameters, params) m.paramGroups.Set(name, pg) + // Real RDS raises one RDS-EVENT-0037 per updated parameter. + for _, prm := range params { + m.recordEventLocked(eventSource{sourceDBParameterGroup, name, pg.ARN}, evParameterUpdated, + prm.Name, prm.Value, applyMethodOrDefault(prm.ApplyMethod)) + } + out := pg return &out, nil @@ -207,14 +250,26 @@ func (m *Mock) DescribeDBParameters(_ context.Context, name string) ([]rdsdriver // defaultGroupFamily extracts the family from a "default." group name // (e.g. "default.mysql8.0" -> "mysql8.0"). RDS provides one such always-present -// group per engine family. +// group per engine family, so only a known family counts: "default.bogus" is +// not a group and stays DBParameterGroupNotFound. func defaultGroupFamily(name string) (string, bool) { - const prefix = "default." - if !strings.HasPrefix(name, prefix) { + family, ok := strings.CutPrefix(name, "default.") + if !ok || !isKnownFamily(family) { return "", false } - return strings.TrimPrefix(name, prefix), true + return family, true +} + +// dbParameterGroupFamily returns the family of the DB parameter group named +// name: a stored group's family, or the family of a "default." group. +// It reports false for a name that is neither. The caller holds m.mu. +func (m *Mock) dbParameterGroupFamily(name string) (string, bool) { + if pg, ok := m.paramGroups.Get(name); ok { + return pg.Family, pg.Family != "" + } + + return defaultGroupFamily(name) } //nolint:dupl // structurally mirrors its sibling per-resource block by design. @@ -224,7 +279,7 @@ func (m *Mock) ResetDBParameterGroup(_ context.Context, name string, params []st pg, ok := m.paramGroups.Get(name) if !ok { - return nil, errParameterGroupNotFound(name) + return nil, missingGroupErr(name, errParameterGroupNotFound(name)) } if resetAll { @@ -315,16 +370,30 @@ func (m *Mock) CreateDBClusterParameterGroup( return &out, nil } -func (m *Mock) DescribeDBClusterParameterGroups(_ context.Context, names []string) ([]rdsdriver.ClusterParameterGroup, error) { +// DescribeDBClusterParameterGroups mirrors DescribeDBParameterGroups for +// cluster parameter groups, synthesizing the "default." groups that +// clusters reference or that are named explicitly. +func (m *Mock) DescribeDBClusterParameterGroups( + ctx context.Context, names []string, +) ([]rdsdriver.ClusterParameterGroup, error) { m.mu.RLock() defer m.mu.RUnlock() + region := regionctx.RegionOr(ctx, m.opts.Region) + if len(names) == 0 { all := m.clusterParamGroups.SortedValues() for i := range all { all[i].Parameters = copyParams(all[i].Parameters) } + for _, name := range m.referencedDefaultClusterParamGroups() { + family, _ := defaultGroupFamily(name) + all = append(all, m.defaultClusterParameterGroup(region, name, family)) + } + + slices.SortFunc(all, func(a, b rdsdriver.ClusterParameterGroup) int { return strings.Compare(a.Name, b.Name) }) + return all, nil } @@ -333,7 +402,12 @@ func (m *Mock) DescribeDBClusterParameterGroups(_ context.Context, names []strin for _, name := range names { pg, ok := m.clusterParamGroups.Get(name) if !ok { - return nil, errClusterParameterGroupNotFound(name) + family, isDefault := defaultGroupFamily(name) + if !isDefault { + return nil, errClusterParameterGroupNotFound(name) + } + + pg = m.defaultClusterParameterGroup(region, name, family) } pg.Parameters = copyParams(pg.Parameters) @@ -351,7 +425,7 @@ func (m *Mock) ModifyDBClusterParameterGroup( pg, ok := m.clusterParamGroups.Get(name) if !ok { - return nil, errClusterParameterGroupNotFound(name) + return nil, missingGroupErr(name, errClusterParameterGroupNotFound(name)) } pg.Parameters = withParams(pg.Parameters, params) @@ -409,7 +483,7 @@ func (m *Mock) ResetDBClusterParameterGroup( pg, ok := m.clusterParamGroups.Get(name) if !ok { - return nil, errClusterParameterGroupNotFound(name) + return nil, missingGroupErr(name, errClusterParameterGroupNotFound(name)) } if resetAll { @@ -462,3 +536,61 @@ func (m *Mock) CopyDBClusterParameterGroup( return &out, nil } + +// defaultParameterGroup synthesizes the always-present "default." DB +// parameter group, described as RDS describes it ("Default parameter group for +// "). +func (m *Mock) defaultParameterGroup(region, name, family string) rdsdriver.ParameterGroup { + return rdsdriver.ParameterGroup{ + Name: name, + Family: family, + Description: "Default parameter group for " + family, + ARN: parameterGroupARN(region, m.opts.AccountID, name), + Parameters: map[string]rdsdriver.Parameter{}, + } +} + +// defaultClusterParameterGroup synthesizes the always-present +// "default." DB cluster parameter group. +func (m *Mock) defaultClusterParameterGroup(region, name, family string) rdsdriver.ClusterParameterGroup { + return rdsdriver.ClusterParameterGroup{ + Name: name, + Family: family, + Description: "Default cluster parameter group for " + family, + ARN: clusterParameterGroupARN(region, m.opts.AccountID, name), + Parameters: map[string]rdsdriver.Parameter{}, + } +} + +// referencedDefaultParamGroups returns, sorted and de-duplicated, the +// "default." DB parameter groups referenced by instances that are not +// stored as explicit groups. The caller holds m.mu. +func (m *Mock) referencedDefaultParamGroups() []string { + seen := map[string]struct{}{} + + instances := m.instances.All() + for id := range instances { + name := instances[id].DBParameterGroupName + if _, isDefault := defaultGroupFamily(name); isDefault && !m.paramGroups.Has(name) { + seen[name] = struct{}{} + } + } + + return slices.Sorted(maps.Keys(seen)) +} + +// referencedDefaultClusterParamGroups is referencedDefaultParamGroups for the +// cluster parameter groups clusters reference. The caller holds m.mu. +func (m *Mock) referencedDefaultClusterParamGroups() []string { + seen := map[string]struct{}{} + + clusters := m.clusters.All() + for id := range clusters { + name := clusters[id].DBClusterParameterGroupName + if _, isDefault := defaultGroupFamily(name); isDefault && !m.clusterParamGroups.Has(name) { + seen[name] = struct{}{} + } + } + + return slices.Sorted(maps.Keys(seen)) +} diff --git a/providers/aws/rds/parameter_group_family_test.go b/providers/aws/rds/parameter_group_family_test.go new file mode 100644 index 000000000..bece94a94 --- /dev/null +++ b/providers/aws/rds/parameter_group_family_test.go @@ -0,0 +1,143 @@ +package rds + +import ( + "context" + "testing" + + rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" +) + +// TestParameterGroupFamily verifies the family real RDS derives from an +// engine's version (DescribeDBEngineVersions.DBParameterGroupFamily). +func TestParameterGroupFamily(t *testing.T) { + tests := []struct { + engine, version, want string + }{ + {"postgres", "15.4", "postgres15"}, + {"postgres", "16.2", "postgres16"}, + {"postgres", "9.6.24", "postgres9.6"}, + {"aurora-postgresql", "16.1", "aurora-postgresql16"}, + {"mysql", "8.0.36", "mysql8.0"}, + {"mysql", "5.7.44", "mysql5.7"}, + {"mariadb", "10.11.6", "mariadb10.11"}, + {"aurora-mysql", "8.0.mysql_aurora.3.05.2", "aurora-mysql8.0"}, + {"aurora-mysql", "5.7.mysql_aurora.2.11.2", "aurora-mysql5.7"}, + {"docdb", "5.0.0", "docdb5.0"}, + {"neptune", "1.3.1.0", "neptune1.3"}, + {"neptune", "1.1.1.0", "neptune1"}, + {"oracle-ee", "19.0.0.0.ru-2024-01.rur-2024-01.r1", "oracle-ee-19"}, + {"oracle-se2-cdb", "21.0.0.0.ru-2024-01.rur-2024-01.r1", "oracle-se2-cdb-21"}, + {"sqlserver-ee", "15.00.4365.2.v1", "sqlserver-ee-15.0"}, + {"sqlserver-web", "16.00.4095.4.v1", "sqlserver-web-16.0"}, + {"db2-ae", "11.5.9.0.sb00000000.r1", "db2-ae"}, + {"db2-se", "", "db2-se"}, + {"postgres", "", "postgres15"}, + {"unknown", "1.0", ""}, + {"postgres", "garbage", ""}, + } + + for _, tc := range tests { + t.Run(tc.engine+"/"+tc.version, func(t *testing.T) { + assertEqual(t, tc.want, parameterGroupFamily(tc.engine, tc.version)) + }) + } +} + +// TestDescribeEngineVersionsFamilyPerVersion verifies each catalog version +// reports its own family, not one family per engine. +func TestDescribeEngineVersionsFamilyPerVersion(t *testing.T) { + m := newTestMock() + + got, err := m.DescribeDBEngineVersions(context.Background(), "postgres", "") + requireNoError(t, err) + + families := map[string]string{} + for _, v := range got { + families[v.EngineVersion] = v.DBParameterGroupFamily + } + + assertEqual(t, "postgres15", families["15.4"]) + assertEqual(t, "postgres16", families["16.2"]) +} + +// TestRestoreFromSnapshotDefaultParameterGroup verifies a snapshot restore gets +// the default group of the snapshot's engine-version family. +func TestRestoreFromSnapshotDefaultParameterGroup(t *testing.T) { + ctx := context.Background() + m := newTestMock() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "src", Engine: "postgres", EngineVersion: "15.4"}) + requireNoError(t, err) + + _, err = m.CreateSnapshot(ctx, rdsdriver.SnapshotConfig{ID: "snap", InstanceID: "src"}) + requireNoError(t, err) + + restored, err := m.RestoreInstanceFromSnapshot(ctx, rdsdriver.RestoreInstanceInput{ + NewInstanceID: "restored", SnapshotID: "snap", + }) + requireNoError(t, err) + assertEqual(t, "default.postgres15", restored.DBParameterGroupName) +} + +// TestCreateClusterDefaultClusterParameterGroup verifies a cluster created +// without a cluster parameter group gets the default one of its engine +// version, while an explicit group is preserved. +func TestCreateClusterDefaultClusterParameterGroup(t *testing.T) { + ctx := context.Background() + m := newTestMock() + + c, err := m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: "c1", Engine: "aurora-postgresql", EngineVersion: "15.4"}) + requireNoError(t, err) + assertEqual(t, "default.aurora-postgresql15", c.DBClusterParameterGroupName) + + c, err = m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: "c2", Engine: "aurora-mysql", EngineVersion: "8.0.mysql_aurora.3.05.2"}) + requireNoError(t, err) + assertEqual(t, "default.aurora-mysql8.0", c.DBClusterParameterGroupName) + + c, err = m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: "c3", Engine: "aurora-mysql", DBClusterParameterGroupName: "custom"}) + requireNoError(t, err) + assertEqual(t, "custom", c.DBClusterParameterGroupName) + + _, err = m.CreateClusterSnapshot(ctx, rdsdriver.ClusterSnapshotConfig{ID: "cs1", ClusterID: "c1"}) + requireNoError(t, err) + + restored, err := m.RestoreClusterFromSnapshot(ctx, rdsdriver.RestoreClusterInput{NewClusterID: "c4", SnapshotID: "cs1"}) + requireNoError(t, err) + assertEqual(t, "default.aurora-postgresql15", restored.DBClusterParameterGroupName) +} + +// TestDescribeDefaultParameterGroups verifies the default groups an instance +// or cluster is associated with are describable, by name and in the listing, +// with the family of the resolved engine version. +func TestDescribeDefaultParameterGroups(t *testing.T) { + ctx := context.Background() + m := newTestMock() + + _, err := m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "pg", Engine: "postgres", EngineVersion: "15.4", InstanceClass: "db.t3.micro"}) + requireNoError(t, err) + + _, err = m.CreateCluster(ctx, rdsdriver.ClusterConfig{ID: "apg", Engine: "aurora-postgresql", EngineVersion: "16.2"}) + requireNoError(t, err) + + got, err := m.DescribeDBParameterGroups(ctx, []string{"default.postgres15"}) + requireNoError(t, err) + assertEqual(t, 1, len(got)) + assertEqual(t, "postgres15", got[0].Family) + assertEqual(t, "Default parameter group for postgres15", got[0].Description) + + all, err := m.DescribeDBParameterGroups(ctx, nil) + requireNoError(t, err) + assertEqual(t, 1, len(all)) + assertEqual(t, "default.postgres15", all[0].Name) + + clusterGroups, err := m.DescribeDBClusterParameterGroups(ctx, nil) + requireNoError(t, err) + assertEqual(t, 1, len(clusterGroups)) + assertEqual(t, "default.aurora-postgresql16", clusterGroups[0].Name) + assertEqual(t, "aurora-postgresql16", clusterGroups[0].Family) + + _, err = m.DescribeDBParameterGroups(ctx, []string{"missing"}) + if err == nil { + t.Fatal("want DBParameterGroupNotFound for a missing non-default group") + } +} diff --git a/providers/aws/rds/rds.go b/providers/aws/rds/rds.go index ddbfadd79..e41b22a63 100644 --- a/providers/aws/rds/rds.go +++ b/providers/aws/rds/rds.go @@ -125,9 +125,16 @@ type Mock struct { instSettle *settle.Set snapSettle *settle.Set + // events is the RDS event log surfaced by DescribeEvents (guarded by mu), + // oldest first and bounded to the 14-day retention window. outbox holds the + // event notifications queued for SNS delivery once mu is released. + events []rdsdriver.Event + outbox []eventNotification + opts *config.Options subnetResolver SubnetResolver monitoring mondriver.Monitoring + sns SNSPublisher } // New creates a new AWS RDS mock. @@ -206,7 +213,7 @@ func (m *Mock) emitInstanceMetrics(instanceID, engine string, cpu, connections f func defaultPortFor(engine string) int { switch engine { - case "postgres", "aurora-postgresql": + case enginePostgres, engineAuroraPostgreSQL: return defaultPortPostgres case engineNeptune: return defaultPortNeptune @@ -382,6 +389,7 @@ func (m *Mock) CreateInstance(ctx context.Context, cfg rdsdriver.InstanceConfig) out := m.finalizeInstance(cfg.ID, cfg.ClusterID, inst, plan) m.emitInstanceMetrics(cfg.ID, cfg.Engine, cpuMetricRunning, connectionsRunning) + m.recordEvent(eventSource{sourceDBInstance, cfg.ID, out.ARN}, evInstanceCreated) m.mu.RLock() out.State = m.settleInstanceState(cfg.ID, out.State) @@ -461,7 +469,7 @@ func (m *Mock) newInstance(ctx context.Context, cfg rdsdriver.InstanceConfig) rd // RDS returns rather than an empty element. paramGroup := cfg.DBParameterGroupName if paramGroup == "" { - paramGroup = defaultParameterGroupName(cfg.Engine) + paramGroup = defaultParameterGroupName(cfg.Engine, engineVersion) } // BackupRetentionPeriod is NOT defaulted here: unlike AllocatedStorage/ @@ -517,13 +525,34 @@ func (m *Mock) newInstance(ctx context.Context, cfg rdsdriver.InstanceConfig) rd } } +// upgradedDefaultGroup returns the parameter group a resource is on after its +// engine version changes to newVersion. A resource on a "default." +// group moves to the default group of the new version's family (postgres 15.4 +// on default.postgres15 upgraded to 16.2 lands on default.postgres16), as real +// RDS does for a major version upgrade; a minor upgrade keeps the same family +// and so the same name. A custom group, or an engine with no known family, is +// left unchanged. An explicit DBParameterGroupName on the same request is +// applied afterwards and wins. +func upgradedDefaultGroup(current, engine, newVersion string) string { + if _, isDefault := defaultGroupFamily(current); !isDefault { + return current + } + + if next := defaultParameterGroupName(engine, newVersion); next != "" { + return next + } + + return current +} + // defaultParameterGroupName returns the DB parameter group real RDS assigns to -// an instance whose caller supplied none: "default." + the engine's -// parameter-group family (e.g. "default.mysql8.0"). An unknown engine has no -// known family and so no default group, matching real RDS which only +// an instance whose caller supplied none: "default." + the parameter-group +// family of the instance's resolved engine version (e.g. postgres 15.4 -> +// "default.postgres15", mysql 8.0.35 -> "default.mysql8.0"). An unknown engine +// has no known family and so no default group, matching real RDS which only // auto-assigns one for a recognized engine. -func defaultParameterGroupName(engine string) string { - family := engineFamily[engine] +func defaultParameterGroupName(engine, engineVersion string) string { + family := parameterGroupFamily(engine, engineVersion) if family == "" { return "" } @@ -712,6 +741,12 @@ func (m *Mock) ModifyInstance( return nil, err } + // The parameter group the request leaves the instance on must belong to the + // family of the version it asks for, checked up front as RDS does. + if err := m.checkInstanceModifyGroup(&inst, &input); err != nil { + return nil, err + } + // ApplyImmediately (default false) governs the deferrable resize/version/ // password/storage changes. When set they take effect now and clear any // pending changes; otherwise they are recorded in PendingModifiedValues and @@ -765,6 +800,7 @@ func applyDeferrableMods(inst *rdsdriver.Instance, input *rdsdriver.ModifyInstan if input.EngineVersion != "" { inst.EngineVersion = input.EngineVersion + inst.DBParameterGroupName = upgradedDefaultGroup(inst.DBParameterGroupName, inst.Engine, input.EngineVersion) } if input.MultiAZ != nil { @@ -942,6 +978,8 @@ func (m *Mock) rotateEnginePassword(ctx context.Context, inst *rdsdriver.Instanc // DeleteInstance removes an instance. func (m *Mock) DeleteInstance(ctx context.Context, id string) error { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -987,6 +1025,7 @@ func (m *Mock) DeleteInstance(ctx context.Context, id string) error { m.instances.Delete(id) delete(m.rootPasswords, id) m.instSettle.Clear(id) + m.recordEventLocked(eventSource{sourceDBInstance, id, inst.ARN}, evInstanceDeleted) return nil } @@ -1003,6 +1042,8 @@ func (m *Mock) StopInstance(_ context.Context, id string) error { // RebootInstance cycles an instance through rebooting → available. func (m *Mock) RebootInstance(_ context.Context, id string) error { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1024,11 +1065,14 @@ func (m *Mock) RebootInstance(_ context.Context, id string) error { m.opts.SettleDuration(settle.DefaultDBRebootSettle)) m.emitInstanceMetrics(id, inst.Engine, cpuMetricRunning, connectionsRunning) + m.recordEventLocked(eventSource{sourceDBInstance, id, inst.ARN}, evInstanceRestarted) return nil } func (m *Mock) transitionInstance(id, from, to string, cpu, conns float64, verb string) error { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1054,6 +1098,13 @@ func (m *Mock) transitionInstance(id, from, to string, cpu, conns float64, verb m.emitInstanceMetrics(id, inst.Engine, cpu, conns) + spec := evInstanceStarted + if to == rdsdriver.StateStopped { + spec = evInstanceStopped + } + + m.recordEventLocked(eventSource{sourceDBInstance, id, inst.ARN}, spec) + return nil } @@ -1069,6 +1120,8 @@ func (m *Mock) CreateCluster(ctx context.Context, cfg rdsdriver.ClusterConfig) ( return nil, cerrors.New(cerrors.InvalidArgument, "Engine is required") } + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1106,12 +1159,14 @@ func (m *Mock) CreateCluster(ctx context.Context, cfg rdsdriver.ClusterConfig) ( maintenanceWindow = defaultMaintenanceWindow } + engineVersion, clusterParamGroup := clusterVersionAndParamGroup(&cfg) + region := regionctx.RegionOr(ctx, m.opts.Region) cluster := rdsdriver.Cluster{ ID: cfg.ID, ARN: clusterARN(region, m.opts.AccountID, cfg.ID), Engine: cfg.Engine, - EngineVersion: cfg.EngineVersion, + EngineVersion: engineVersion, MasterUsername: cfg.MasterUsername, DatabaseName: cfg.DatabaseName, Endpoint: endpointFor(cfg.ID, region, "cluster"), @@ -1120,7 +1175,7 @@ func (m *Mock) CreateCluster(ctx context.Context, cfg rdsdriver.ClusterConfig) ( State: rdsdriver.StateAvailable, VPCSecurityGroups: append([]string(nil), cfg.VPCSecurityGroups...), SubnetGroupName: cfg.SubnetGroupName, - DBClusterParameterGroupName: cfg.DBClusterParameterGroupName, + DBClusterParameterGroupName: clusterParamGroup, BackupRetentionPeriod: backupRetention, PreferredBackupWindow: backupWindow, PreferredMaintenanceWindow: maintenanceWindow, @@ -1138,12 +1193,34 @@ func (m *Mock) CreateCluster(ctx context.Context, cfg rdsdriver.ClusterConfig) ( m.clusters.Set(cfg.ID, cluster) m.clusterCreds[cfg.ID] = clusterCred{user: cfg.MasterUsername, pass: cfg.MasterUserPassword} + m.recordEventLocked(eventSource{sourceDBCluster, cfg.ID, cluster.ARN}, evClusterCreated) out := cluster return &out, nil } +// clusterVersionAndParamGroup resolves a new cluster's engine version and DB +// cluster parameter group. An omitted EngineVersion resolves to the engine's +// default version (as newInstance does for instances), so the reported +// EngineVersion and the version-derived default group always agree. An +// omitted group is the default DB cluster parameter group "for the specified +// DB engine and version" (CreateDBCluster API reference, +// DBClusterParameterGroupName). +func clusterVersionAndParamGroup(cfg *rdsdriver.ClusterConfig) (engineVersion, paramGroup string) { + engineVersion = cfg.EngineVersion + if engineVersion == "" { + engineVersion = defaultEngineVersion(cfg.Engine) + } + + paramGroup = cfg.DBClusterParameterGroupName + if paramGroup == "" { + paramGroup = defaultParameterGroupName(cfg.Engine, engineVersion) + } + + return engineVersion, paramGroup +} + // DescribeClusters returns all clusters if ids is empty, else only matching ones. func (m *Mock) DescribeClusters(_ context.Context, ids []string) ([]rdsdriver.Cluster, error) { m.mu.RLock() @@ -1183,6 +1260,8 @@ func (m *Mock) DescribeClusters(_ context.Context, ids []string) ([]rdsdriver.Cl func applyClusterModify(cluster *rdsdriver.Cluster, input rdsdriver.ModifyInstanceInput) { if input.EngineVersion != "" { cluster.EngineVersion = input.EngineVersion + cluster.DBClusterParameterGroupName = upgradedDefaultGroup( + cluster.DBClusterParameterGroupName, cluster.Engine, input.EngineVersion) } if input.DBClusterParameterGroupName != "" { @@ -1235,10 +1314,19 @@ func (m *Mock) ModifyCluster( return nil, err } + members, err := m.clusterMemberUpgrades(&cluster, &input) + if err != nil { + return nil, err + } + applyClusterModify(&cluster, input) m.clusters.Set(id, cluster) + for i := range members { + m.instances.Set(members[i].ID, members[i]) + } + out := cluster return &out, nil @@ -1266,6 +1354,8 @@ func (m *Mock) rotateClusterPassword(ctx context.Context, id, engine, newPasswor // DeleteCluster removes a cluster (only if it has no members). func (m *Mock) DeleteCluster(ctx context.Context, id string) error { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1288,6 +1378,7 @@ func (m *Mock) DeleteCluster(ctx context.Context, id string) error { m.clusters.Delete(id) delete(m.clusterCreds, id) + m.recordEventLocked(eventSource{sourceDBCluster, id, cluster.ARN}, evClusterDeleted) return nil } @@ -1303,6 +1394,8 @@ func (m *Mock) StopCluster(_ context.Context, id string) error { } func (m *Mock) transitionCluster(id, from, to, verb string) error { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1323,6 +1416,13 @@ func (m *Mock) transitionCluster(id, from, to, verb string) error { cluster.State = to m.clusters.Set(id, cluster) + spec := evClusterStarted + if to == rdsdriver.StateStopped { + spec = evClusterStopped + } + + m.recordEventLocked(eventSource{sourceDBCluster, id, cluster.ARN}, spec) + return nil } @@ -1332,6 +1432,8 @@ func (m *Mock) CreateSnapshot(_ context.Context, cfg rdsdriver.SnapshotConfig) ( return nil, cerrors.New(cerrors.InvalidArgument, "DBSnapshotIdentifier is required") } + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1373,6 +1475,10 @@ func (m *Mock) CreateSnapshot(_ context.Context, cfg rdsdriver.SnapshotConfig) ( m.instSettle.Begin(cfg.InstanceID, rdsdriver.StateBackingUp, now, m.opts.SettleDuration(settle.DefaultDBSnapshotSettle)) + snapSrc := eventSource{sourceDBSnapshot, cfg.ID, snap.ARN} + m.recordEventLocked(snapSrc, evSnapshotCreating) + m.recordEventLocked(snapSrc, evSnapshotCreated) + out := snap out.State = m.settleSnapshotState(cfg.ID, out.State) @@ -1413,14 +1519,18 @@ func (m *Mock) DescribeSnapshots( // DeleteSnapshot removes an instance snapshot. func (m *Mock) DeleteSnapshot(_ context.Context, id string) error { + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() - if !m.snapshots.Delete(id) { + snap, ok := m.snapshots.Get(id) + if !ok || !m.snapshots.Delete(id) { return cerrors.Newf(cerrors.NotFound, "DB snapshot %q not found", id) } m.snapSettle.Clear(id) + m.recordEventLocked(eventSource{sourceDBSnapshot, id, snap.ARN}, evSnapshotDeleted) return nil } @@ -1478,7 +1588,7 @@ func (m *Mock) RestoreInstanceFromSnapshot( PubliclyAccessible: input.PubliclyAccessible, DeletionProtection: input.DeletionProtection, SubnetGroupName: input.SubnetGroupName, - DBParameterGroupName: defaultParameterGroupName(snap.Engine), + DBParameterGroupName: defaultParameterGroupName(snap.Engine, snap.EngineVersion), AvailabilityZone: defaultAvailabilityZone("", input.MultiAZ, region), CreatedAt: m.opts.Clock.Now().UTC(), Tags: copyTags(input.Tags), @@ -1626,6 +1736,8 @@ func (m *Mock) CreateClusterSnapshot( return nil, cerrors.New(cerrors.InvalidArgument, "DBClusterSnapshotIdentifier is required") } + defer m.flushEventNotifications() // runs after the unlock below + m.mu.Lock() defer m.mu.Unlock() @@ -1656,6 +1768,10 @@ func (m *Mock) CreateClusterSnapshot( m.clusterSnapshots.Set(cfg.ID, snap) + csSrc := eventSource{sourceDBClusterSnapshot, cfg.ID, snap.ARN} + m.recordEventLocked(csSrc, evClusterSnapshotCreating) + m.recordEventLocked(csSrc, evClusterSnapshotCreated) + out := snap return &out, nil @@ -1758,8 +1874,11 @@ func (m *Mock) RestoreClusterFromSnapshot( EngineMode: engineMode, DBClusterResourceID: resourceID("cluster-", input.NewClusterID), AvailabilityZones: availabilityZones(region), - CreatedAt: m.opts.Clock.Now().UTC(), - Tags: copyTags(input.Tags), + // A restored cluster gets the default cluster parameter group of the + // snapshot's engine version, like a cluster created without one. + DBClusterParameterGroupName: defaultParameterGroupName(snap.Engine, snap.EngineVersion), + CreatedAt: m.opts.Clock.Now().UTC(), + Tags: copyTags(input.Tags), } // Provision the shared database keyed and named by the new cluster id (an diff --git a/providers/aws/rds/snapshot.go b/providers/aws/rds/snapshot.go index 5111eebe8..1910efdeb 100644 --- a/providers/aws/rds/snapshot.go +++ b/providers/aws/rds/snapshot.go @@ -4,8 +4,10 @@ import ( "context" "encoding/json" "fmt" + "maps" "github.com/stackshy/cloudemu/v2/internal/snapshot" + rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" ) var _ snapshot.Snapshottable = (*Mock)(nil) @@ -14,7 +16,8 @@ var _ snapshot.Snapshottable = (*Mock)(nil) // store holds a fully-exported rdsdriver type, so each round-trips through the // generic memstore helper keyed by its resource id. The mu-guarded maps carry // state that lives beside the stores: clusterCreds (unexported value type, so -// promoted), groupTags, and rootPasswords. The in-flight settle overlays +// promoted), groupTags, rootPasswords, and the DescribeEvents event log. The +// queued SNS deliveries (outbox) are transient and not serialized. The in-flight settle overlays // (instSettle/snapSettle) and the wired deps (opts, subnetResolver, monitoring) // are intentionally not serialized. A restored record reports its stored state. type rdsSnapshot struct { @@ -34,6 +37,7 @@ type rdsSnapshot struct { ClusterCreds map[string]clusterCredSnapshot `json:"clusterCreds,omitempty"` GroupTags map[string]map[string]string `json:"groupTags,omitempty"` RootPasswords map[string]string `json:"rootPasswords,omitempty"` + Events []rdsdriver.Event `json:"events,omitempty"` } // clusterCredSnapshot is the exported form of clusterCred (whose fields are @@ -101,8 +105,35 @@ func (m *Mock) snapshotScalarState(snap *rdsSnapshot) { snap.ClusterCreds = creds } - snap.GroupTags = m.groupTags - snap.RootPasswords = m.rootPasswords + // Copy every map and slice while the read lock is held: json.Marshal runs + // after it is released, and writers mutate groupTags / rootPasswords in + // place, so handing out the live values would race with them. + if len(m.groupTags) > 0 { + tags := make(map[string]map[string]string, len(m.groupTags)) + for arn, t := range m.groupTags { + tags[arn] = maps.Clone(t) + } + + snap.GroupTags = tags + } + + snap.RootPasswords = maps.Clone(m.rootPasswords) + snap.Events = cloneEvents(m.events) +} + +// cloneEvents deep-copies the event log (each event's category slice too). +func cloneEvents(events []rdsdriver.Event) []rdsdriver.Event { + if len(events) == 0 { + return nil + } + + out := make([]rdsdriver.Event, len(events)) + for i := range events { + out[i] = events[i] + out[i].EventCategories = cloneSlice(events[i].EventCategories) + } + + return out } // Restore rebuilds the mock's state under the original identities: every @@ -177,4 +208,9 @@ func (m *Mock) restoreScalarState(snap *rdsSnapshot) { if snap.RootPasswords != nil { m.rootPasswords = snap.RootPasswords } + + // The event log is replaced unconditionally: a snapshot taken before any + // event was recorded omits the key, and restoring it must clear the log + // rather than keep events about resources the snapshot does not contain. + m.events = snap.Events } diff --git a/providers/aws/rds/snapshot_test.go b/providers/aws/rds/snapshot_test.go index a93ec8bf5..05cc71f60 100644 --- a/providers/aws/rds/snapshot_test.go +++ b/providers/aws/rds/snapshot_test.go @@ -2,8 +2,10 @@ package rds import ( "context" + "sync" "testing" + "github.com/stackshy/cloudemu/v2/config" rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" ) @@ -82,3 +84,67 @@ func TestSnapshotRoundTripRDS(t *testing.T) { t.Fatalf("restored clusterCreds[cl1] = %+v, ok %v", cred, ok) } } + +// TestRestoreSnapshotWithoutEventsClearsLog: loading a snapshot saved before +// any event was recorded leaves no stale events behind. +func TestRestoreSnapshotWithoutEventsClearsLog(t *testing.T) { + ctx := context.Background() + + empty, _ := newEventTestMock() + + data, err := empty.Snapshot(ctx, false) + requireNoError(t, err) + + m, _ := newEventTestMock() + _, err = m.CreateInstance(ctx, rdsdriver.InstanceConfig{ID: "db1", Engine: "mysql"}) + requireNoError(t, err) + + requireNoError(t, m.Restore(ctx, data)) + + events, err := m.DescribeEvents(ctx, rdsdriver.DescribeEventsInput{}) + requireNoError(t, err) + assertEqual(t, 0, len(events)) +} + +// TestSnapshotConcurrentWithStateWrites runs Snapshot alongside writers of +// the lock-guarded maps and the event log; under -race it fails if Snapshot +// hands json.Marshal a live map instead of a copy taken under the lock. +func TestSnapshotConcurrentWithStateWrites(t *testing.T) { + ctx := context.Background() + m := New(config.NewOptions(config.WithRegion("us-east-1"), config.WithAccountID("123456789012"))) + + _, err := m.CreateDBParameterGroup(ctx, rdsdriver.ParameterGroupConfig{ + Name: "pg1", Family: "mysql8.0", Tags: map[string]string{"a": "1"}, + }) + requireNoError(t, err) + + pgARN := parameterGroupARN("us-east-1", "123456789012", "pg1") + + const rounds = 50 + + var wg sync.WaitGroup + + wg.Add(2) + + go func() { + defer wg.Done() + + for i := range rounds { + _ = m.AddTagsToResource(ctx, pgARN, map[string]string{"k": string(rune('a' + i%26))}) + _, _ = m.ModifyDBParameterGroup(ctx, "pg1", []rdsdriver.Parameter{{Name: "x", Value: "1"}}) + } + }() + + go func() { + defer wg.Done() + + for range rounds { + if _, err := m.Snapshot(ctx, false); err != nil { + t.Errorf("Snapshot: %v", err) + return + } + } + }() + + wg.Wait() +} diff --git a/server/aws/rds/default_group_upgrade_sdk_roundtrip_test.go b/server/aws/rds/default_group_upgrade_sdk_roundtrip_test.go new file mode 100644 index 000000000..ad290a4a4 --- /dev/null +++ b/server/aws/rds/default_group_upgrade_sdk_roundtrip_test.go @@ -0,0 +1,222 @@ +package rds_test + +import ( + "context" + "errors" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + awsrds "github.com/aws/aws-sdk-go-v2/service/rds" + rdstypes "github.com/aws/aws-sdk-go-v2/service/rds/types" + smithy "github.com/aws/smithy-go" +) + +// TestSDKRDSDefaultGroupUpgradeAndImmutability covers, over the wire, a major +// version upgrade moving the instance to the new default group, a modify of a +// listed default group answering InvalidParameterValue (not NotFound), and an +// event subscription's ARN being taggable. +func TestSDKRDSDefaultGroupUpgradeAndImmutability(t *testing.T) { + client := newSDKClient(t) + ctx := context.Background() + + if _, err := client.CreateDBInstance(ctx, &awsrds.CreateDBInstanceInput{ + DBInstanceIdentifier: aws.String("up1"), Engine: aws.String("postgres"), + EngineVersion: aws.String("15.4"), DBInstanceClass: aws.String("db.t3.micro"), + }); err != nil { + t.Fatalf("CreateDBInstance: %v", err) + } + + mod, err := client.ModifyDBInstance(ctx, &awsrds.ModifyDBInstanceInput{ + DBInstanceIdentifier: aws.String("up1"), EngineVersion: aws.String("16.2"), + AllowMajorVersionUpgrade: aws.Bool(true), ApplyImmediately: aws.Bool(true), + }) + if err != nil { + t.Fatalf("ModifyDBInstance: %v", err) + } + + if g := aws.ToString(mod.DBInstance.DBParameterGroups[0].DBParameterGroupName); g != "default.postgres16" { + t.Fatalf("parameter group after upgrade = %q, want default.postgres16", g) + } + + _, err = client.ModifyDBParameterGroup(ctx, &awsrds.ModifyDBParameterGroupInput{ + DBParameterGroupName: aws.String("default.postgres16"), + Parameters: []rdstypes.Parameter{{ + ParameterName: aws.String("max_connections"), ParameterValue: aws.String("10"), + ApplyMethod: rdstypes.ApplyMethodImmediate, + }}, + }) + + var apiErr smithy.APIError + if !errors.As(err, &apiErr) || apiErr.ErrorCode() != "InvalidParameterValue" { + t.Fatalf("modify default group err = %v, want InvalidParameterValue", err) + } + + sub, err := client.CreateEventSubscription(ctx, &awsrds.CreateEventSubscriptionInput{ + SubscriptionName: aws.String("tfsub"), + SnsTopicArn: aws.String("arn:aws:sns:us-east-1:000000000000:t"), + Tags: []rdstypes.Tag{{Key: aws.String("env"), Value: aws.String("dev")}}, + }) + if err != nil { + t.Fatalf("CreateEventSubscription: %v", err) + } + + tags, err := client.ListTagsForResource(ctx, &awsrds.ListTagsForResourceInput{ + ResourceName: sub.EventSubscription.EventSubscriptionArn, + }) + if err != nil || len(tags.TagList) != 1 || aws.ToString(tags.TagList[0].Value) != "dev" { + t.Fatalf("ListTagsForResource(es ARN) = %+v, err %v", tags, err) + } +} + +// TestSDKRDSUpgradeRejectsParamGroupFamilyMismatch: a major upgrade of an +// instance on a custom old-family group, or an explicit group of the wrong +// family, is InvalidParameterCombination over the wire. +func TestSDKRDSUpgradeRejectsParamGroupFamilyMismatch(t *testing.T) { + client := newSDKClient(t) + ctx := context.Background() + + for name, family := range map[string]string{"my15": "postgres15", "my16": "postgres16"} { + if _, err := client.CreateDBParameterGroup(ctx, &awsrds.CreateDBParameterGroupInput{ + DBParameterGroupName: aws.String(name), DBParameterGroupFamily: aws.String(family), + Description: aws.String(name), + }); err != nil { + t.Fatalf("CreateDBParameterGroup %s: %v", name, err) + } + } + + if _, err := client.CreateDBInstance(ctx, &awsrds.CreateDBInstanceInput{ + DBInstanceIdentifier: aws.String("fam1"), Engine: aws.String("postgres"), + EngineVersion: aws.String("15.4"), DBInstanceClass: aws.String("db.t3.micro"), + DBParameterGroupName: aws.String("my15"), + }); err != nil { + t.Fatalf("CreateDBInstance: %v", err) + } + + for _, in := range []*awsrds.ModifyDBInstanceInput{ + {EngineVersion: aws.String("16.2")}, + {EngineVersion: aws.String("17.2"), DBParameterGroupName: aws.String("my16")}, + } { + in.DBInstanceIdentifier = aws.String("fam1") + in.AllowMajorVersionUpgrade = aws.Bool(true) + in.ApplyImmediately = aws.Bool(true) + + _, err := client.ModifyDBInstance(ctx, in) + + var apiErr smithy.APIError + if !errors.As(err, &apiErr) || apiErr.ErrorCode() != "InvalidParameterCombination" { + t.Fatalf("ModifyDBInstance(%s, %s) err = %v, want InvalidParameterCombination", + aws.ToString(in.EngineVersion), aws.ToString(in.DBParameterGroupName), err) + } + } + + mod, err := client.ModifyDBInstance(ctx, &awsrds.ModifyDBInstanceInput{ + DBInstanceIdentifier: aws.String("fam1"), EngineVersion: aws.String("16.2"), + DBParameterGroupName: aws.String("my16"), AllowMajorVersionUpgrade: aws.Bool(true), + ApplyImmediately: aws.Bool(true), + }) + if err != nil { + t.Fatalf("ModifyDBInstance to 16.2 with my16: %v", err) + } + + if g := aws.ToString(mod.DBInstance.DBParameterGroups[0].DBParameterGroupName); g != "my16" { + t.Fatalf("parameter group = %q, want my16", g) + } +} + +// TestSDKRDSAuroraMajorUpgradeMovesMembers: modify-db-cluster with a major +// version moves every member instance to the new version and default group, +// or to DBInstanceParameterGroupName when given. +func TestSDKRDSAuroraMajorUpgradeMovesMembers(t *testing.T) { + client := newSDKClient(t) + ctx := context.Background() + + if _, err := client.CreateDBCluster(ctx, &awsrds.CreateDBClusterInput{ + DBClusterIdentifier: aws.String("ac1"), Engine: aws.String("aurora-postgresql"), + EngineVersion: aws.String("15.4"), MasterUsername: aws.String("admin"), + MasterUserPassword: aws.String("secret123"), + }); err != nil { + t.Fatalf("CreateDBCluster: %v", err) + } + + for _, id := range []string{"ac1w", "ac1r"} { + if _, err := client.CreateDBInstance(ctx, &awsrds.CreateDBInstanceInput{ + DBInstanceIdentifier: aws.String(id), DBClusterIdentifier: aws.String("ac1"), + Engine: aws.String("aurora-postgresql"), DBInstanceClass: aws.String("db.r5.large"), + }); err != nil { + t.Fatalf("CreateDBInstance %s: %v", id, err) + } + } + + if _, err := client.ModifyDBCluster(ctx, &awsrds.ModifyDBClusterInput{ + DBClusterIdentifier: aws.String("ac1"), EngineVersion: aws.String("16.2"), + AllowMajorVersionUpgrade: aws.Bool(true), ApplyImmediately: aws.Bool(true), + }); err != nil { + t.Fatalf("ModifyDBCluster: %v", err) + } + + out, err := client.DescribeDBInstances(ctx, &awsrds.DescribeDBInstancesInput{}) + if err != nil { + t.Fatalf("DescribeDBInstances: %v", err) + } + + for i := range out.DBInstances { + inst := out.DBInstances[i] + g := aws.ToString(inst.DBParameterGroups[0].DBParameterGroupName) + + if aws.ToString(inst.EngineVersion) != "16.2" || g != "default.aurora-postgresql16" { + t.Fatalf("member %s = %s / %s, want 16.2 / default.aurora-postgresql16", + aws.ToString(inst.DBInstanceIdentifier), aws.ToString(inst.EngineVersion), g) + } + } + + if _, err := client.CreateDBParameterGroup(ctx, &awsrds.CreateDBParameterGroupInput{ + DBParameterGroupName: aws.String("apg17"), DBParameterGroupFamily: aws.String("aurora-postgresql17"), + Description: aws.String("apg17"), + }); err != nil { + t.Fatalf("CreateDBParameterGroup: %v", err) + } + + if _, err := client.ModifyDBCluster(ctx, &awsrds.ModifyDBClusterInput{ + DBClusterIdentifier: aws.String("ac1"), EngineVersion: aws.String("17.2"), + AllowMajorVersionUpgrade: aws.Bool(true), ApplyImmediately: aws.Bool(true), + DBInstanceParameterGroupName: aws.String("apg17"), + }); err != nil { + t.Fatalf("ModifyDBCluster with DBInstanceParameterGroupName: %v", err) + } + + got, err := client.DescribeDBInstances(ctx, &awsrds.DescribeDBInstancesInput{DBInstanceIdentifier: aws.String("ac1r")}) + if err != nil { + t.Fatalf("DescribeDBInstances: %v", err) + } + + if g := aws.ToString(got.DBInstances[0].DBParameterGroups[0].DBParameterGroupName); g != "apg17" { + t.Fatalf("member group = %q, want apg17", g) + } +} + +// TestSDKRDSUnknownDefaultGroupNotFound: a "default." name of no real family +// is DBParameterGroupNotFound, not an immutable default group. +func TestSDKRDSUnknownDefaultGroupNotFound(t *testing.T) { + client := newSDKClient(t) + ctx := context.Background() + + _, err := client.ModifyDBParameterGroup(ctx, &awsrds.ModifyDBParameterGroupInput{ + DBParameterGroupName: aws.String("default.bogus"), + Parameters: []rdstypes.Parameter{{ + ParameterName: aws.String("max_connections"), ParameterValue: aws.String("10"), + ApplyMethod: rdstypes.ApplyMethodImmediate, + }}, + }) + + var notFound *rdstypes.DBParameterGroupNotFoundFault + if !errors.As(err, ¬Found) { + t.Fatalf("ModifyDBParameterGroup(default.bogus) err = %v, want DBParameterGroupNotFound", err) + } + + _, err = client.DescribeDBParameterGroups(ctx, &awsrds.DescribeDBParameterGroupsInput{ + DBParameterGroupName: aws.String("default.bogus"), + }) + if !errors.As(err, ¬Found) { + t.Fatalf("DescribeDBParameterGroups(default.bogus) err = %v, want DBParameterGroupNotFound", err) + } +} diff --git a/server/aws/rds/describe_events_sdk_roundtrip_test.go b/server/aws/rds/describe_events_sdk_roundtrip_test.go new file mode 100644 index 000000000..e3a6321df --- /dev/null +++ b/server/aws/rds/describe_events_sdk_roundtrip_test.go @@ -0,0 +1,172 @@ +package rds_test + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsrds "github.com/aws/aws-sdk-go-v2/service/rds" + rdstypes "github.com/aws/aws-sdk-go-v2/service/rds/types" + "github.com/aws/smithy-go" + + "github.com/stackshy/cloudemu/v2" + awsserver "github.com/stackshy/cloudemu/v2/server/aws" +) + +// TestSDKRDSDescribeEventsWindowAndPaging covers DescribeEvents over the wire: +// lifecycle events are recorded, StartTime/EndTime and the filters bound the +// result, and MaxRecords/Marker page through the oldest-first list. +func TestSDKRDSDescribeEventsWindowAndPaging(t *testing.T) { + client := newSDKClient(t) + ctx := context.Background() + + const n = 25 + + for i := range n { + if _, err := client.CreateDBInstance(ctx, &awsrds.CreateDBInstanceInput{ + DBInstanceIdentifier: aws.String(fmt.Sprintf("e%02d", i)), + Engine: aws.String("mysql"), + DBInstanceClass: aws.String("db.t3.micro"), + }); err != nil { + t.Fatalf("CreateDBInstance %d: %v", i, err) + } + } + + all, err := client.DescribeEvents(ctx, &awsrds.DescribeEventsInput{}) + if err != nil { + t.Fatalf("DescribeEvents: %v", err) + } + + if len(all.Events) != n || all.Marker != nil { + t.Fatalf("got %d events (marker %v), want %d on one page", len(all.Events), all.Marker, n) + } + + one, err := client.DescribeEvents(ctx, &awsrds.DescribeEventsInput{ + SourceType: rdstypes.SourceTypeDbInstance, SourceIdentifier: aws.String("e07"), + EventCategories: []string{"creation"}, + }) + if err != nil || len(one.Events) != 1 || aws.ToString(one.Events[0].Message) != "DB instance created" { + t.Fatalf("filtered DescribeEvents: %+v, err %v", one, err) + } + + // A window entirely in the past excludes everything. + past := time.Now().Add(-48 * time.Hour) + + old, err := client.DescribeEvents(ctx, &awsrds.DescribeEventsInput{ + StartTime: aws.Time(past), EndTime: aws.Time(past.Add(time.Hour)), + }) + if err != nil { + t.Fatalf("DescribeEvents(past window): %v", err) + } + + if len(old.Events) != 0 { + t.Fatalf("past window: got %d events, want 0", len(old.Events)) + } + + page, err := client.DescribeEvents(ctx, &awsrds.DescribeEventsInput{MaxRecords: aws.Int32(20)}) + if err != nil { + t.Fatalf("DescribeEvents page 1: %v", err) + } + + if len(page.Events) != 20 || aws.ToString(page.Marker) == "" { + t.Fatalf("page 1: %d events, marker %q", len(page.Events), aws.ToString(page.Marker)) + } + + if got := aws.ToString(page.Events[0].SourceIdentifier); got != "e00" { + t.Fatalf("oldest event first: got %q, want e00", got) + } + + next, err := client.DescribeEvents(ctx, &awsrds.DescribeEventsInput{MaxRecords: aws.Int32(20), Marker: page.Marker}) + if err != nil { + t.Fatalf("DescribeEvents page 2: %v", err) + } + + if len(next.Events) != 5 || aws.ToString(next.Events[0].SourceIdentifier) != "e20" || next.Marker != nil { + t.Fatalf("page 2: %d events, marker %v", len(next.Events), next.Marker) + } + + // MaxRecords outside the documented [20, 100] range is rejected. + for _, bad := range []int32{19, 101} { + _, err := client.DescribeEvents(ctx, &awsrds.DescribeEventsInput{MaxRecords: aws.Int32(bad)}) + + var apiErr smithy.APIError + if !errors.As(err, &apiErr) || apiErr.ErrorCode() != "InvalidParameterValue" { + t.Fatalf("MaxRecords=%d: want InvalidParameterValue, got %v", bad, err) + } + } + // SourceIdentifier without SourceType, and an undocumented SourceType, are + // rejected like real RDS. + cases := []struct { + in *awsrds.DescribeEventsInput + code string + }{ + {&awsrds.DescribeEventsInput{SourceIdentifier: aws.String("e00")}, "InvalidParameterCombination"}, + {&awsrds.DescribeEventsInput{SourceType: rdstypes.SourceType("db-bogus")}, "InvalidParameterValue"}, + } + + for _, tc := range cases { + _, err := client.DescribeEvents(ctx, tc.in) + + var apiErr smithy.APIError + if !errors.As(err, &apiErr) || apiErr.ErrorCode() != tc.code { + t.Fatalf("DescribeEvents %+v: want %s, got %v", tc.in, tc.code, err) + } + } +} + +// TestWireDescribeEventsRejectsMalformedTimes: a raw query client sending a +// timestamp or Duration that does not parse gets InvalidParameterValue, not +// the default window. The SourceIdentifier-without-SourceType rule, now +// enforced by the provider, still surfaces as InvalidParameterCombination. +func TestWireDescribeEventsRejectsMalformedTimes(t *testing.T) { + ts := httptest.NewServer(awsserver.New(awsserver.Drivers{RDS: cloudemu.NewAWS().RDS})) + t.Cleanup(ts.Close) + + cases := []struct { + name string + extra url.Values + wantCode string + }{ + {"bad StartTime", url.Values{"StartTime": {"garbage"}}, "InvalidParameterValue"}, + {"bad EndTime", url.Values{"EndTime": {"2025-13-45"}}, "InvalidParameterValue"}, + {"bad Duration", url.Values{"Duration": {"ten"}}, "InvalidParameterValue"}, + {"id without type", url.Values{"SourceIdentifier": {"db1"}}, "InvalidParameterCombination"}, + {"bogus type", url.Values{"SourceType": {"db-bogus"}}, "InvalidParameterValue"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + form := url.Values{"Action": {"DescribeEvents"}, "Version": {"2014-10-31"}} + for k, v := range tc.extra { + form[k] = v + } + + req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, ts.URL, + strings.NewReader(form.Encode())) + if err != nil { + t.Fatal(err) + } + + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusBadRequest || !strings.Contains(string(body), ""+tc.wantCode+"") { + t.Fatalf("status %d body %s, want 400 %s", resp.StatusCode, body, tc.wantCode) + } + }) + } +} diff --git a/server/aws/rds/events.go b/server/aws/rds/events.go index f5987655e..4a009dbfd 100644 --- a/server/aws/rds/events.go +++ b/server/aws/rds/events.go @@ -4,7 +4,10 @@ import ( "encoding/xml" "net/http" "net/url" + "strconv" + "time" + "github.com/stackshy/cloudemu/v2/internal/pagination" "github.com/stackshy/cloudemu/v2/server/wire/awsquery" rdsdriver "github.com/stackshy/cloudemu/v2/services/relationaldb/driver" ) @@ -27,6 +30,7 @@ type eventSubscriptionXML struct { type eventXML struct { SourceIdentifier string `xml:"SourceIdentifier,omitempty"` SourceType string `xml:"SourceType,omitempty"` + SourceArn string `xml:"SourceArn,omitempty"` Message string `xml:"Message,omitempty"` EventCategories []string `xml:"EventCategories>EventCategory,omitempty"` Date string `xml:"Date,omitempty"` @@ -81,6 +85,7 @@ type describeEventsResponse struct { } type eventsList struct { + Marker string `xml:"Marker,omitempty"` Events []eventXML `xml:"Events>Event"` } @@ -231,19 +236,43 @@ func (h *Handler) describeEvents(w http.ResponseWriter, r *http.Request) { return } - events, err := store.DescribeEvents(r.Context(), - r.Form.Get("SourceType"), r.Form.Get("SourceIdentifier"), - awsquery.ListStrings(r.Form, "EventCategories.EventCategory")) + maxRecords, ok := describeEventsMaxRecords(r.Form.Get("MaxRecords")) + if !ok { + awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterValue", + "MaxRecords must be between 20 and 100") + + return + } + + input, field, ok := describeEventsInput(r.Form) + if !ok { + awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterValue", + "Invalid value for parameter "+field) + + return + } + + events, err := store.DescribeEvents(r.Context(), input) if err != nil { writeErr(w, err) return } - out := make([]eventXML, 0, len(events)) - for _, e := range events { + // Events are already oldest-first; page them in that order (no re-sort). + page, err := pagination.Paginate(events, r.Form.Get("Marker"), maxRecords) + if err != nil { + awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterValue", "invalid Marker") + return + } + + out := make([]eventXML, 0, len(page.Items)) + + for i := range page.Items { + e := &page.Items[i] out = append(out, eventXML{ SourceIdentifier: e.SourceIdentifier, SourceType: e.SourceType, + SourceArn: e.SourceARN, Message: e.Message, EventCategories: e.EventCategories, Date: e.Date.UTC().Format("2006-01-02T15:04:05.000Z"), @@ -252,7 +281,7 @@ func (h *Handler) describeEvents(w http.ResponseWriter, r *http.Request) { awsquery.WriteXMLResponse(w, describeEventsResponse{ Xmlns: Namespace, - Result: eventsList{Events: out}, + Result: eventsList{Marker: page.NextPageToken, Events: out}, Metadata: responseMetadata{RequestID: awsquery.RequestID}, }) } @@ -282,6 +311,91 @@ func (h *Handler) describeEventCategories(w http.ResponseWriter, r *http.Request }) } +// DescribeEvents MaxRecords bounds: "Default: 100. Constraints: Minimum 20, +// maximum 100." (DescribeEvents API reference). +const ( + minEventRecords = 20 + maxEventRecords = 100 + defaultEventRecords = 100 +) + +// describeEventsMaxRecords resolves DescribeEvents' MaxRecords: absent means +// the default 100; a non-integer or a value outside [20, 100] is invalid. +func describeEventsMaxRecords(raw string) (int, bool) { + if raw == "" { + return defaultEventRecords, true + } + + n, err := strconv.Atoi(raw) + if err != nil || n < minEventRecords || n > maxEventRecords { + return 0, false + } + + return n, true +} + +// eventTimeLayouts are the ISO 8601 forms DescribeEvents StartTime/EndTime +// arrive in: the SDKs send full RFC 3339 timestamps, while the API reference +// example uses the minute-precision form "2009-07-08T18:00Z". +// +//nolint:gochecknoglobals // static lookup table +var eventTimeLayouts = []string{time.RFC3339Nano, "2006-01-02T15:04Z07:00", "2006-01-02T15:04", time.DateOnly} + +// describeEventsInput decodes DescribeEvents' filters from the form. It +// returns the name of the first malformed field (a StartTime/EndTime that is +// not an ISO 8601 timestamp, or a non-integer Duration) and false, so the +// caller answers InvalidParameterValue as real RDS does instead of silently +// widening the window. The SourceType / SourceIdentifier rules are business +// rules and are enforced by the provider. +func describeEventsInput(form url.Values) (rdsdriver.DescribeEventsInput, string, bool) { + start, ok := parseEventTime(form.Get("StartTime")) + if !ok { + return rdsdriver.DescribeEventsInput{}, "StartTime", false + } + + end, ok := parseEventTime(form.Get("EndTime")) + if !ok { + return rdsdriver.DescribeEventsInput{}, "EndTime", false + } + + duration := 0 + + if raw := form.Get("Duration"); raw != "" { + n, err := strconv.Atoi(raw) + if err != nil { + return rdsdriver.DescribeEventsInput{}, "Duration", false + } + + duration = n + } + + return rdsdriver.DescribeEventsInput{ + SourceType: form.Get("SourceType"), + SourceIdentifier: form.Get("SourceIdentifier"), + EventCategories: awsquery.ListStrings(form, "EventCategories.EventCategory"), + StartTime: start, + EndTime: end, + Duration: duration, + }, "", true +} + +// parseEventTime parses an optional DescribeEvents timestamp. An absent value +// yields the zero time ("no bound") and true; a value in none of the accepted +// layouts yields false. +func parseEventTime(v string) (time.Time, bool) { + if v == "" { + return time.Time{}, true + } + + for _, layout := range eventTimeLayouts { + if t, err := time.Parse(layout, v); err == nil { + return t, true + } + } + + return time.Time{}, false +} + // enabledDefaultTrue reads the Enabled flag, defaulting to true when the field // is absent (RDS CreateEventSubscription defaults Enabled to true). func enabledDefaultTrue(form url.Values) bool { diff --git a/server/aws/rds/events_sdk_roundtrip_test.go b/server/aws/rds/events_sdk_roundtrip_test.go index f516d8c27..e477368a6 100644 --- a/server/aws/rds/events_sdk_roundtrip_test.go +++ b/server/aws/rds/events_sdk_roundtrip_test.go @@ -2,10 +2,13 @@ package rds_test import ( "context" + "errors" "testing" "github.com/aws/aws-sdk-go-v2/aws" awsrds "github.com/aws/aws-sdk-go-v2/service/rds" + rdstypes "github.com/aws/aws-sdk-go-v2/service/rds/types" + smithy "github.com/aws/smithy-go" ) func TestSDKRDSEventSubscriptionLifecycle(t *testing.T) { @@ -80,3 +83,57 @@ func TestSDKRDSEventSubscriptionLifecycle(t *testing.T) { t.Fatalf("DeleteEventSubscription: %v", err) } } + +// TestSDKRDSEventSubscriptionFaultCodes: a missing or duplicate subscription +// answers with the RDS wire codes SubscriptionNotFound and +// SubscriptionAlreadyExist, so the SDK returns its typed faults. Terraform's +// aws_db_event_subscription delete waiter relies on the typed NotFound. +func TestSDKRDSEventSubscriptionFaultCodes(t *testing.T) { + client := newSDKClient(t) + ctx := context.Background() + + _, err := client.DeleteEventSubscription(ctx, &awsrds.DeleteEventSubscriptionInput{ + SubscriptionName: aws.String("missing"), + }) + + var notFound *rdstypes.SubscriptionNotFoundFault + if !errors.As(err, ¬Found) { + t.Fatalf("DeleteEventSubscription(missing) err = %v, want *types.SubscriptionNotFoundFault", err) + } + + assertAPICode(t, err, "SubscriptionNotFound") + + _, err = client.DescribeEventSubscriptions(ctx, &awsrds.DescribeEventSubscriptionsInput{ + SubscriptionName: aws.String("missing"), + }) + if !errors.As(err, ¬Found) { + t.Fatalf("DescribeEventSubscriptions(missing) err = %v, want *types.SubscriptionNotFoundFault", err) + } + + in := &awsrds.CreateEventSubscriptionInput{ + SubscriptionName: aws.String("dup"), + SnsTopicArn: aws.String("arn:aws:sns:us-east-1:123456789012:rds-events"), + } + if _, err := client.CreateEventSubscription(ctx, in); err != nil { + t.Fatalf("CreateEventSubscription: %v", err) + } + + _, err = client.CreateEventSubscription(ctx, in) + + var exists *rdstypes.SubscriptionAlreadyExistFault + if !errors.As(err, &exists) { + t.Fatalf("duplicate CreateEventSubscription err = %v, want *types.SubscriptionAlreadyExistFault", err) + } + + assertAPICode(t, err, "SubscriptionAlreadyExist") +} + +// assertAPICode checks the wire error code of an SDK error. +func assertAPICode(t *testing.T, err error, want string) { + t.Helper() + + var apiErr smithy.APIError + if !errors.As(err, &apiErr) || apiErr.ErrorCode() != want { + t.Fatalf("error code of %v, want %s", err, want) + } +} diff --git a/server/aws/rds/handler.go b/server/aws/rds/handler.go index 1f85f2255..8aa8149f2 100644 --- a/server/aws/rds/handler.go +++ b/server/aws/rds/handler.go @@ -396,7 +396,7 @@ func writeErr(w http.ResponseWriter, err error) { case cerrors.IsAlreadyExists(err): awsquery.WriteXMLError(w, http.StatusBadRequest, alreadyExistsCode(err), msg) case cerrors.IsInvalidArgument(err): - awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterValue", msg) + awsquery.WriteXMLError(w, http.StatusBadRequest, invalidArgumentCode(err), msg) case cerrors.IsFailedPrecondition(err): awsquery.WriteXMLError(w, http.StatusBadRequest, invalidStateCode(err), msg) default: @@ -450,7 +450,7 @@ var notFoundFaults = []faultMapping{ {"parameter group", "DBParameterGroupNotFound"}, {"option group", "OptionGroupNotFoundFault"}, {"DB proxy", "DBProxyNotFoundFault"}, - {"event subscription", "SubscriptionNotFoundFault"}, + {"event subscription", "SubscriptionNotFound"}, {"DB cluster endpoint", "DBClusterEndpointNotFoundFault"}, {"global cluster", "GlobalClusterNotFoundFault"}, {"DB instance", "DBInstanceNotFound"}, @@ -466,7 +466,7 @@ var alreadyExistsFaults = []faultMapping{ {"parameter group", "DBParameterGroupAlreadyExists"}, {"option group", "OptionGroupAlreadyExistsFault"}, {"DB proxy", "DBProxyAlreadyExistsFault"}, - {"event subscription", "SubscriptionAlreadyExistFault"}, + {"event subscription", "SubscriptionAlreadyExist"}, {"DB cluster endpoint", "DBClusterEndpointAlreadyExistsFault"}, {"global cluster", "GlobalClusterAlreadyExistsFault"}, {"DB instance", "DBInstanceAlreadyExists"}, @@ -491,6 +491,30 @@ var invalidStateFaults = []faultMapping{ {"DB instance", "InvalidDBInstanceState"}, } +// codeInvalidParameterCombination is the RDS fault for parameters that are +// individually valid but cannot be used together. +const codeInvalidParameterCombination = "InvalidParameterCombination" + +// invalidArgumentFaults maps an invalid-argument error to the AWS-shaped fault +// code. A rule that one parameter requires another ("X is supplied, Y must +// also be provided"), or a parameter group whose family does not fit the +// engine version, is InvalidParameterCombination in real RDS; every other +// invalid argument is InvalidParameterValue. +// +//nolint:gochecknoglobals // ordered static lookup table +var invalidArgumentFaults = []faultMapping{ + {"must also be provided", codeInvalidParameterCombination}, + {"can't be used for this instance", codeInvalidParameterCombination}, + {"You must explicitly specify a new DB instance parameter group", codeInvalidParameterCombination}, + {"valid in combination with AllowMajorVersionUpgrade", codeInvalidParameterCombination}, +} + +// invalidArgumentCode picks the AWS-shaped invalid-argument fault from the +// message. +func invalidArgumentCode(err error) string { + return matchFault(err.Error(), invalidArgumentFaults, "InvalidParameterValue") +} + // notFoundCode picks the AWS-shaped NotFound fault from the error message. func notFoundCode(err error) string { return matchFault(err.Error(), notFoundFaults, "ResourceNotFoundFault") diff --git a/server/aws/rds/operations.go b/server/aws/rds/operations.go index d94671ed7..383c35458 100644 --- a/server/aws/rds/operations.go +++ b/server/aws/rds/operations.go @@ -361,7 +361,7 @@ func (h *Handler) deleteDBInstance(w http.ResponseWriter, r *http.Request) { // via ModifyDBInstance first. Real RDS rejects with InvalidParameterCombination // and leaves the instance untouched. if last.DeletionProtection { - awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterCombination", + awsquery.WriteXMLError(w, http.StatusBadRequest, codeInvalidParameterCombination, "Cannot delete protected DB Instance, please disable deletion protection and try again.") return @@ -376,7 +376,7 @@ func (h *Handler) deleteDBInstance(w http.ResponseWriter, r *http.Request) { if last.ClusterID == "" && !formBool(r.Form.Get("SkipFinalSnapshot")) { finalID = r.Form.Get("FinalDBSnapshotIdentifier") if finalID == "" { - awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterCombination", + awsquery.WriteXMLError(w, http.StatusBadRequest, codeInvalidParameterCombination, "FinalDBSnapshotIdentifier is required unless SkipFinalSnapshot is true") return @@ -552,13 +552,14 @@ func (h *Handler) modifyDBCluster(w http.ResponseWriter, r *http.Request) { id := form.Get("DBClusterIdentifier") input := rdsdriver.ModifyInstanceInput{ - EngineVersion: form.Get("EngineVersion"), - MasterUserPassword: form.Get("MasterUserPassword"), - DBClusterParameterGroupName: form.Get("DBClusterParameterGroupName"), - BackupRetentionPeriod: formInt(form.Get("BackupRetentionPeriod")), - PreferredBackupWindow: form.Get("PreferredBackupWindow"), - PreferredMaintenanceWindow: form.Get("PreferredMaintenanceWindow"), - Tags: parseRDSTags(form), + EngineVersion: form.Get("EngineVersion"), + MasterUserPassword: form.Get("MasterUserPassword"), + DBClusterParameterGroupName: form.Get("DBClusterParameterGroupName"), + DBInstanceParameterGroupName: form.Get("DBInstanceParameterGroupName"), + BackupRetentionPeriod: formInt(form.Get("BackupRetentionPeriod")), + PreferredBackupWindow: form.Get("PreferredBackupWindow"), + PreferredMaintenanceWindow: form.Get("PreferredMaintenanceWindow"), + Tags: parseRDSTags(form), } if v := form.Get("DeletionProtection"); v != "" { @@ -604,7 +605,7 @@ func (h *Handler) deleteDBCluster(w http.ResponseWriter, r *http.Request) { // A deletion-protected cluster cannot be deleted; the flag must be cleared via // ModifyDBCluster first. Real RDS rejects with InvalidParameterCombination. if last.DeletionProtection { - awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterCombination", + awsquery.WriteXMLError(w, http.StatusBadRequest, codeInvalidParameterCombination, "Cannot delete protected DB Cluster, please disable deletion protection and try again.") return @@ -616,7 +617,7 @@ func (h *Handler) deleteDBCluster(w http.ResponseWriter, r *http.Request) { if !formBool(r.Form.Get("SkipFinalSnapshot")) { finalID := r.Form.Get("FinalDBSnapshotIdentifier") if finalID == "" { - awsquery.WriteXMLError(w, http.StatusBadRequest, "InvalidParameterCombination", + awsquery.WriteXMLError(w, http.StatusBadRequest, codeInvalidParameterCombination, "FinalDBSnapshotIdentifier is required unless SkipFinalSnapshot is true") return diff --git a/server/aws/rds_events_sdk_test.go b/server/aws/rds_events_sdk_test.go new file mode 100644 index 000000000..de6b2d2a6 --- /dev/null +++ b/server/aws/rds_events_sdk_test.go @@ -0,0 +1,139 @@ +package aws_test + +import ( + "context" + "encoding/json" + "net/http/httptest" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + awsrds "github.com/aws/aws-sdk-go-v2/service/rds" + rdstypes "github.com/aws/aws-sdk-go-v2/service/rds/types" + "github.com/aws/aws-sdk-go-v2/service/sns" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/stackshy/cloudemu/v2" + awsserver "github.com/stackshy/cloudemu/v2/server/aws" +) + +// TestE2ERDSEventSubscriptionDeliversToSNSToSQS drives the RDS event pipeline +// the way a real user does: an RDS event subscription on an SNS topic that an +// SQS queue subscribes to must deliver the RDS lifecycle notification to the +// queue, and DescribeEvents must report the same event. +func TestE2ERDSEventSubscriptionDeliversToSNSToSQS(t *testing.T) { + provider := cloudemu.NewAWS() + srv := awsserver.New(awsserver.Drivers{ + RDS: provider.RDS, + SNS: provider.SNS, + SQS: provider.SQS, + }) + ts := httptest.NewServer(srv) + defer ts.Close() + + cfg := sdkConfig(t, ts.URL) + ctx := context.Background() + + rdsClient := awsrds.NewFromConfig(cfg) + snsClient := sns.NewFromConfig(cfg) + sqsClient := sqs.NewFromConfig(cfg) + + topic, err := snsClient.CreateTopic(ctx, &sns.CreateTopicInput{Name: aws.String("rds-events")}) + require.NoError(t, err) + + queue, err := sqsClient.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("rds-events-q")}) + require.NoError(t, err) + + attrs, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: queue.QueueUrl, + AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameQueueArn}, + }) + require.NoError(t, err) + + _, err = snsClient.Subscribe(ctx, &sns.SubscribeInput{ + TopicArn: topic.TopicArn, + Protocol: aws.String("sqs"), + Endpoint: aws.String(attrs.Attributes[string(sqstypes.QueueAttributeNameQueueArn)]), + Attributes: map[string]string{"RawMessageDelivery": "true"}, + }) + require.NoError(t, err) + + _, err = rdsClient.CreateEventSubscription(ctx, &awsrds.CreateEventSubscriptionInput{ + SubscriptionName: aws.String("db-lifecycle"), + SnsTopicArn: topic.TopicArn, + SourceType: aws.String("db-instance"), + EventCategories: []string{"creation", "deletion"}, + }) + require.NoError(t, err) + + _, err = rdsClient.CreateDBInstance(ctx, &awsrds.CreateDBInstanceInput{ + DBInstanceIdentifier: aws.String("evdb"), + Engine: aws.String("postgres"), + EngineVersion: aws.String("15.4"), + DBInstanceClass: aws.String("db.t3.micro"), + MasterUsername: aws.String("admin"), + MasterUserPassword: aws.String("password123"), + AllocatedStorage: aws.Int32(20), + }) + require.NoError(t, err) + + msgs, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: queue.QueueUrl, MaxNumberOfMessages: 10, + }) + require.NoError(t, err) + require.Len(t, msgs.Messages, 1, "creation event should reach the queue") + + var body map[string]string + require.NoError(t, json.Unmarshal([]byte(aws.ToString(msgs.Messages[0].Body)), &body)) + assert.Equal(t, "db-instance", body["Event Source"]) + assert.Equal(t, "evdb", body["Source ID"]) + assert.Equal(t, "DB instance created", body["Event Message"]) + assert.Equal(t, "http://docs.amazonwebservices.com/AmazonRDS/latest/UserGuide/USER_Events.html#RDS-EVENT-0005", body["Event ID"]) + + events, err := rdsClient.DescribeEvents(ctx, &awsrds.DescribeEventsInput{ + SourceType: rdstypes.SourceTypeDbInstance, + SourceIdentifier: aws.String("evdb"), + }) + require.NoError(t, err) + require.Len(t, events.Events, 1) + + ev := events.Events[0] + assert.Equal(t, "DB instance created", aws.ToString(ev.Message)) + assert.Equal(t, []string{"creation"}, ev.EventCategories) + assert.Equal(t, rdstypes.SourceTypeDbInstance, ev.SourceType) + assert.Equal(t, "arn:aws:rds:us-east-1:123456789012:db:evdb", aws.ToString(ev.SourceArn)) + assert.NotNil(t, ev.Date) + + // The instance's default parameter group follows its 15.4 version family. + inst, err := rdsClient.DescribeDBInstances(ctx, &awsrds.DescribeDBInstancesInput{DBInstanceIdentifier: aws.String("evdb")}) + require.NoError(t, err) + require.Len(t, inst.DBInstances[0].DBParameterGroups, 1) + assert.Equal(t, "default.postgres15", aws.ToString(inst.DBInstances[0].DBParameterGroups[0].DBParameterGroupName)) + + // A disabled subscription stops delivery, while DescribeEvents still logs. + _, err = rdsClient.ModifyEventSubscription(ctx, &awsrds.ModifyEventSubscriptionInput{ + SubscriptionName: aws.String("db-lifecycle"), Enabled: aws.Bool(false), + }) + require.NoError(t, err) + + _, err = rdsClient.DeleteDBInstance(ctx, &awsrds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String("evdb"), SkipFinalSnapshot: aws.Bool(true), + }) + require.NoError(t, err) + + _, err = sqsClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{ + QueueUrl: queue.QueueUrl, ReceiptHandle: msgs.Messages[0].ReceiptHandle, + }) + require.NoError(t, err) + + again, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{QueueUrl: queue.QueueUrl, MaxNumberOfMessages: 10}) + require.NoError(t, err) + assert.Empty(t, again.Messages, "disabled subscription must not deliver") + + deletions, err := rdsClient.DescribeEvents(ctx, &awsrds.DescribeEventsInput{EventCategories: []string{"deletion"}}) + require.NoError(t, err) + require.Len(t, deletions.Events, 1) + assert.Equal(t, "DB instance deleted", aws.ToString(deletions.Events[0].Message)) +} diff --git a/services/relationaldb/driver/driver.go b/services/relationaldb/driver/driver.go index 29e1a5ce5..d9ad49108 100644 --- a/services/relationaldb/driver/driver.go +++ b/services/relationaldb/driver/driver.go @@ -286,7 +286,11 @@ type ModifyInstanceInput struct { DBParameterGroupName string OptionGroupName string DBClusterParameterGroupName string - ElasticPoolID string + // DBInstanceParameterGroupName is the AWS RDS ModifyDBCluster input that + // moves every instance of an Aurora cluster to this DB parameter group on a + // major version upgrade; empty means "no change". Other engines ignore it. + DBInstanceParameterGroupName string + ElasticPoolID string // MinimalTLSVersion / PublicNetworkAccess / RestrictOutboundNetworkAccess // update the Azure SQL logical-server properties on a PATCH; empty means "no // change". Other engines ignore them. @@ -1732,9 +1736,28 @@ type ModifyEventSubscriptionInput struct { type Event struct { SourceIdentifier string SourceType string - Message string + // SourceARN is the ARN of the resource that emitted the event. + SourceARN string + Message string + EventCategories []string + Date time.Time + // EventID is the RDS event identifier (e.g. "RDS-EVENT-0005"). It is not + // part of the DescribeEvents wire shape; it is carried in event + // notifications delivered to an SNS topic. + EventID string +} + +// DescribeEventsInput filters DescribeEvents. Zero values mean "no filter", +// except the time window: with neither StartTime nor Duration set, the window +// is the past 60 minutes ending at EndTime (default now). +type DescribeEventsInput struct { + SourceType string + SourceIdentifier string EventCategories []string - Date time.Time + StartTime time.Time + EndTime time.Time + // Duration is the window length in minutes, used when StartTime is unset. + Duration int } // EventCategoryGroup lists the event categories available for a source type. @@ -1750,7 +1773,7 @@ type EventSubscriptions interface { DescribeEventSubscriptions(ctx context.Context, names []string) ([]EventSubscription, error) ModifyEventSubscription(ctx context.Context, name string, input ModifyEventSubscriptionInput) (*EventSubscription, error) DeleteEventSubscription(ctx context.Context, name string) (*EventSubscription, error) - DescribeEvents(ctx context.Context, sourceType, sourceIdentifier string, categories []string) ([]Event, error) + DescribeEvents(ctx context.Context, input DescribeEventsInput) ([]Event, error) DescribeEventCategories(ctx context.Context, sourceType string) ([]EventCategoryGroup, error) }