diff --git a/internal/purchase/azure_redis_routing_test.go b/internal/purchase/azure_redis_routing_test.go index 9f2b6c867..bbde0f52b 100644 --- a/internal/purchase/azure_redis_routing_test.go +++ b/internal/purchase/azure_redis_routing_test.go @@ -24,6 +24,7 @@ func purchaseCacheRecThrough(t *testing.T, prov *MockProvider, providerName, reg store.On("SavePurchaseHistory", ctx, mock.Anything).Return(nil) factory.On("CreateAndValidateProvider", mock.Anything, providerName, mock.Anything).Return(prov, nil) prov.On("GetServiceClient", mock.Anything, want, region).Return(provider.ServiceClient(svc), nil) + expectNoExistingCommitments(svc) svc.On("PurchaseCommitment", mock.Anything, mock.MatchedBy(func(rec common.Recommendation) bool { return rec.Service == common.ServiceCache }), mock.Anything).Return(common.PurchaseResult{Success: true, CommitmentID: "order-1"}, nil) diff --git a/internal/purchase/coverage_extra_test.go b/internal/purchase/coverage_extra_test.go index 05ea829b9..a7d9f0caa 100644 --- a/internal/purchase/coverage_extra_test.go +++ b/internal/purchase/coverage_extra_test.go @@ -535,6 +535,7 @@ func TestManager_ExecuteSinglePurchase_PurchaseNotSuccessful(t *testing.T) { mockStore.On("GetPurchasePlan", ctx, "plan-not-success").Return(plan, nil) mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil) mockProvider.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), common.ServiceElastiCache, "ap-southeast-1").Return(mockServiceClient, nil) + mockServiceClient.On("GetExistingCommitments", mock.Anything).Return([]common.Commitment(nil), nil) mockServiceClient.On("PurchaseCommitment", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")).Return( common.PurchaseResult{Success: false}, nil, ) @@ -1006,6 +1007,7 @@ func TestManager_ExecuteSinglePurchase_DetailsByService(t *testing.T) { mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil) mockProvider.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), tc.serviceType, tc.region).Return(mockServiceClient, nil) + expectNoExistingCommitments(mockServiceClient) mockServiceClient.On( "PurchaseCommitment", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), @@ -1145,6 +1147,7 @@ func TestManager_ExecuteSinglePurchase_LegacyEmptyDetails(t *testing.T) { mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil) mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProvider, nil) mockProvider.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), tc.serviceType, tc.region).Return(mockServiceClient, nil) + expectNoExistingCommitments(mockServiceClient) mockServiceClient.On( "PurchaseCommitment", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), diff --git a/internal/purchase/elasticache_recent_check.go b/internal/purchase/elasticache_recent_check.go index a97bf9a1e..fb8c4b883 100644 --- a/internal/purchase/elasticache_recent_check.go +++ b/internal/purchase/elasticache_recent_check.go @@ -124,3 +124,40 @@ func elastiCacheRecentRefusal(ctx context.Context, client provider.ServiceClient ctx, []common.Recommendation{elastiCacheDedupeCopy(rec)}, elastiCacheRecentDedupeClient{ServiceClient: client, own: ownIDs, self: self, sawSelf: &sawSelf}) return elastiCacheRefusalFor(passed, filtered, err, sawSelf, rec) } + +// purchaseOwnNames carries the commitment names this execution's selected recs +// will create, per provider, so a retry is told apart from another lineage's +// purchase. A nil pointer reads as no own names. +type purchaseOwnNames struct { + gcp, elastiCache map[string]bool +} + +func newPurchaseOwnNames(exec *config.PurchaseExecution, selected []int) *purchaseOwnNames { + return &purchaseOwnNames{ + gcp: gcpOwnCommitmentNames(exec, selected), + elastiCache: elastiCacheOwnReservationIDs(exec, selected), + } +} + +func (o *purchaseOwnNames) gcpNames() map[string]bool { + if o == nil { + return nil + } + return o.gcp +} + +func (o *purchaseOwnNames) elastiCacheNames() map[string]bool { + if o == nil { + return nil + } + return o.elastiCache +} + +// ownElastiCacheID is the reservation ID this rec's purchase call would create, +// or "" when the call carries no idempotency token. +func ownElastiCacheID(opts common.PurchaseOptions) string { + if opts.IdempotencyToken == "" { + return "" + } + return common.IdempotentReservationID(elastiCacheReservationIDPrefix, opts.IdempotencyToken) +} diff --git a/internal/purchase/elasticache_recent_check_wiring_test.go b/internal/purchase/elasticache_recent_check_wiring_test.go new file mode 100644 index 000000000..defdeaa2f --- /dev/null +++ b/internal/purchase/elasticache_recent_check_wiring_test.go @@ -0,0 +1,155 @@ +package purchase + +// #1863 S2: the ElastiCache recent-reservation check wired into the real +// executor path (the real elasticache.Client over the shared fake API). + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" + "github.com/aws/aws-sdk-go-v2/service/elasticache/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// expectNoExistingCommitments is the shared expectation for tests that drive an +// AWS ElastiCache rec through executeSinglePurchase with a MockServiceClient: +// the check lists live reservations first, and none exist. Maybe, because the +// same table also holds recs that never list. +func expectNoExistingCommitments(svc *MockServiceClient) { + svc.On("GetExistingCommitments", mock.Anything).Return([]common.Commitment(nil), nil).Maybe() +} + +// runECExecution runs one AWS ElastiCache rec through the real fan-out. The +// client is returned for the rec's own region, and the rec's region is the one +// recorded in ecRegion, so the listing and the recommendation agree. +func runECExecution(t *testing.T, api *fakeElastiCacheAPI, recs ...config.RecommendationRecord) (*config.PurchaseExecution, []string, bool) { + t.Helper() + ctx := context.Background() + store, sender, factory, prov := new(MockConfigStore), new(MockEmailSender), new(MockProviderFactory), new(MockProvider) + assertMocksOnCleanup(t, store, sender, factory, prov) + store.SavePurchaseExecutionFn = func(context.Context, *config.PurchaseExecution) error { return nil } + store.On("SavePurchaseHistory", ctx, mock.Anything).Return(nil).Maybe() + factory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(prov, nil) + prov.On("GetServiceClient", mock.Anything, common.ServiceElastiCache, ecRegion).Return(provider.ServiceClient(newFakeElastiCacheClient(api)), nil) + + if len(recs) == 0 { + recs = []config.RecommendationRecord{ecCacheRec(t, "elasticache")} + } + for i := range recs { + recs[i].ID = fmt.Sprintf("rec-1863-%d", i) + } + exec := &config.PurchaseExecution{ExecutionID: "exec-1863", IdempotencyKey: "lineage-1863-wiring", Recommendations: recs} + m := &Manager{config: store, email: sender, providerFactory: factory} + _, _, errs, allRefused, err := m.processPurchaseRecommendations(ctx, exec, &config.PurchasePlan{Name: "p"}, "acct", (*provider.Config)(nil), "") + require.NoError(t, err) + return exec, errs, allRefused +} + +func TestElastiCacheWiring_RecentSamePoolReservationRefusesBeforePurchase(t *testing.T) { + api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld("other-lineage", "redis", 2, time.Hour)}} + + exec, errs, allRefused := runECExecution(t, api) + + assert.Empty(t, api.purchases(), "the purchase call must never be made") + assert.True(t, allRefused, "the row ends as a before-submission refusal") + require.Len(t, errs, 1) + assert.Contains(t, errs[0], elastiCacheRecentReservationNote) + assert.False(t, exec.Recommendations[0].Purchased) +} + +// Regression guard (passes on main too): the library's by-ID guard adopts the +// execution's own listed reservation, so the check must not refuse it. +func TestElastiCacheWiring_OwnListedReservationReachesPurchaseAndIsAdopted(t *testing.T) { + own := common.IdempotentReservationID(elastiCacheReservationIDPrefix, common.DeriveIdempotencyToken("lineage-1863-wiring", 0)) + api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld(own, "redis", 2, time.Minute)}} + + exec, errs, allRefused := runECExecution(t, api) + + assert.Empty(t, errs) + assert.False(t, allRefused) + assert.True(t, exec.Recommendations[0].Purchased, "the own reservation is adopted, not refused") +} + +func ecOwnID(i int) string { + return common.IdempotentReservationID(elastiCacheReservationIDPrefix, common.DeriveIdempotencyToken("lineage-1863-wiring", i)) +} + +// Sibling recs of one execution never refuse each other: the sibling's listed +// reservation is in the own-ID set. +func TestElastiCacheWiring_SiblingOwnReservationDoesNotRefuse(t *testing.T) { + api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld(ecOwnID(1), "redis", 2, time.Minute)}} + second := ecCacheRec(t, "elasticache") + second.ResourceType = "cache.r6g.xlarge" + + _, errs, allRefused := runECExecution(t, api, ecCacheRec(t, "elasticache"), second) + + assert.Empty(t, errs) + assert.False(t, allRefused) +} + +// A lost-response retry beside another lineage's same-pool reservation: only the +// self ID (sawSelf) rescues it from the refusal. +func TestElastiCacheWiring_RetryWithForeignReservationIsAdopted(t *testing.T) { + api := &fakeElastiCacheAPI{held: []types.ReservedCacheNode{ + ecHeld(ecOwnID(0), "redis", 2, time.Minute), ecHeld("other-lineage", "redis", 2, time.Hour)}} + + exec, errs, allRefused := runECExecution(t, api) + + assert.Empty(t, errs) + assert.False(t, allRefused) + assert.True(t, exec.Recommendations[0].Purchased) +} + +func TestPreSubmitChecks_NilOwnNamesDoesNotPanic(t *testing.T) { + svc := new(MockServiceClient) + expectNoExistingCommitments(svc) + ecRecord := ecCacheRec(t, "cache") + gcpRecord := config.RecommendationRecord{Provider: "gcp", Service: "compute", Region: "us-central1", ResourceType: "n2-standard-4", Count: 1} + gcpRec := common.Recommendation{Provider: common.ProviderGCP, Service: common.ServiceCompute, Region: "us-central1", ResourceType: "n2-standard-4", Count: 1} + + assert.NotPanics(t, func() { + _, err := preSubmitChecks(context.Background(), svc, gcpRecord, gcpRec, common.PurchaseOptions{}, nil) + assert.NoError(t, err) + _, err = preSubmitChecks(context.Background(), svc, ecRecord, ecRec("redis", 2), common.PurchaseOptions{}, nil) + assert.NoError(t, err) + }) +} + +func TestPreSubmitChecks_ElastiCacheRefusesBeforeDispatchTimeIsStamped(t *testing.T) { + held := newFakeElastiCacheClient(&fakeElastiCacheAPI{held: []types.ReservedCacheNode{ecHeld("other", "redis", 2, time.Hour)}}) + clean := newFakeElastiCacheClient(&fakeElastiCacheAPI{}) + record := ecCacheRec(t, "cache") + + opts, err := preSubmitChecks(context.Background(), held, record, ecRec("redis", 2), common.PurchaseOptions{}, nil) + requireRefusal(t, err, elastiCacheRecentReservationNote) + assert.True(t, opts.RequestedAt.IsZero(), "a refusal returns before withDispatchTime") + + opts, err = preSubmitChecks(context.Background(), clean, record, ecRec("redis", 2), common.PurchaseOptions{}, nil) + require.NoError(t, err) + assert.False(t, opts.RequestedAt.IsZero(), "a pass continues to withDispatchTime") +} + +func TestPreSubmitChecks_OnlyAWSElastiCacheListsReservations(t *testing.T) { + cases := map[string]common.Recommendation{ + "aws ec2": {Provider: common.ProviderAWS, Service: common.ServiceCompute}, + "aws rds": {Provider: common.ProviderAWS, Service: common.ServiceRelationalDB}, + "aws memorydb": {Provider: common.ProviderAWS, Service: common.ServiceMemoryDB}, + "azure cache": {Provider: common.ProviderAzure, Service: common.ServiceCache}, + } + for name, rec := range cases { + t.Run(name, func(t *testing.T) { + rec.Count = 1 + svc := new(MockServiceClient) // no GetExistingCommitments expectation: a call panics + _, err := preSubmitChecks(context.Background(), svc, config.RecommendationRecord{}, rec, common.PurchaseOptions{}, nil) + assert.NoError(t, err) + }) + } +} diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 5106db205..e2b3a333f 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -794,7 +794,7 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf // Build the list of selected indices once so the fan-out closure only // has to look up rec[i] (no second pass over the full slice). selected := selectedIndices(exec.Recommendations) - ownGCPNames := gcpOwnCommitmentNames(exec, selected) + ownNames := newPurchaseOwnNames(exec, selected) if len(selected) == 0 { logging.Infof("purchase[%s]: no selected recommendations, nothing to execute (account=%s plan=%q)", exec.ExecutionID, accountID, plan.Name) @@ -847,7 +847,7 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf recOpts := opts recOpts.IdempotencyToken = common.DeriveIdempotencyToken(idempotencyLineageKey(exec), i) recOpts.AttemptJournal = m.attemptJournal - purchaseResult, err := m.executeSinglePurchase(ctx, rec, provCfg, recOpts, ownGCPNames) + purchaseResult, err := m.executeSinglePurchase(ctx, rec, provCfg, recOpts, ownNames) return recPurchaseOutcome{index: i, purchase: purchaseResult, err: err}, nil }, getMaxAccountParallelism()) @@ -1261,7 +1261,7 @@ func logRecCtxErr(executionID, recTuple string, elapsed time.Duration, recCtxErr // provCfg carries optional per-account credentials; pass nil to use ambient credentials. // opts carries execution-level metadata (the source surface) that providers stamp // onto the commitment they create. -func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.RecommendationRecord, provCfg *provider.Config, opts common.PurchaseOptions, ownGCPNames map[string]bool) (common.PurchaseResult, error) { +func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.RecommendationRecord, provCfg *provider.Config, opts common.PurchaseOptions, ownNames *purchaseOwnNames) (common.PurchaseResult, error) { // Per-purchase Info logs tagged with the owning execution ID so a // CloudWatch filter on the execution UUID surfaces every step of the // purchase attempt -- provider construction, service-client lookup, @@ -1359,7 +1359,7 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen // CloudWatch shows whether a failure came from the SDK call itself // (which the AWS SDK retries up to 3× × 30s by default) versus // something earlier in the flow — issue #667. - opts, err = preSubmitChecks(recCtx, serviceClient, rec, recommendation, opts, ownGCPNames) + opts, err = preSubmitChecks(recCtx, serviceClient, rec, recommendation, opts, ownNames) if err != nil { return common.PurchaseResult{}, err } @@ -1393,13 +1393,17 @@ func (m *Manager) executeSinglePurchase(ctx context.Context, rec config.Recommen return result, nil } -// preSubmitChecks runs what must hold before PurchaseCommitment: the GCP +// preSubmitChecks runs what must hold before PurchaseCommitment: the AWS +// ElastiCache recent-reservation refusal (platform#1863), the GCP // recent-commitment refusal (platform#682, a beforeSubmission error) and go's // Redshift dispatch time, which bounds its DescribeReservedNodes match window // by RequestedAt and must be the lineage's first dispatch time, never this // call's. -func preSubmitChecks(ctx context.Context, client provider.ServiceClient, rec config.RecommendationRecord, recommendation common.Recommendation, opts common.PurchaseOptions, ownGCPNames map[string]bool) (common.PurchaseOptions, error) { - if err := gcpRecentCommitmentRefusal(ctx, client, recommendation, ownGCPNames, ownCommitmentName(opts)); err != nil { +func preSubmitChecks(ctx context.Context, client provider.ServiceClient, rec config.RecommendationRecord, recommendation common.Recommendation, opts common.PurchaseOptions, ownNames *purchaseOwnNames) (common.PurchaseOptions, error) { + if err := elastiCacheRecentRefusal(ctx, client, recommendation, ownNames.elastiCacheNames(), ownElastiCacheID(opts)); err != nil { + return opts, err + } + if err := gcpRecentCommitmentRefusal(ctx, client, recommendation, ownNames.gcpNames(), ownCommitmentName(opts)); err != nil { return opts, err } return withDispatchTime(rec, opts)