diff --git a/internal/scheduler/partial_sweep_eviction_test.go b/internal/scheduler/partial_sweep_eviction_test.go new file mode 100644 index 000000000..2b38d98e4 --- /dev/null +++ b/internal/scheduler/partial_sweep_eviction_test.go @@ -0,0 +1,183 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + + "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/pkg/common" + azureprovider "github.com/LeanerCloud/CUDly/providers/azure" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// A partially-swept account must never authorize stale-row eviction. +// +// UpsertRecommendations deletes an account's previous-cycle rows with +// +// DELETE FROM recommendations +// WHERE collected_at < $1 AND (provider, account_key) IN (…) +// +// where the roster comes from the succeeded-account IDs these collect +// functions return. That predicate is keyed on (provider, account_key) -- +// the CUDly account UUID -- but an Azure sweep fails one SUBSCRIPTION at a +// time. A sweep that never queried subscription B returns no rows for B, so +// listing the account as succeeded deletes B's previous rows: its savings +// opportunities vanish and the dashboard reads "nothing to buy" rather than +// "collection incomplete". +// +// The existing account-level tests cannot see this. They assert that a +// collection which returns an *error* is excluded, and a partial sweep +// deliberately returns data plus *PartialSubscriptionFailureError -- the one +// shape that is neither a clean success nor a failure. That is the whole +// defect, and why these tests drive the sub-account granularity directly. +// +// Note the assertions are on the roster, NOT on the recommendations: a +// partial sweep's rows must still be collected and upserted, so the +// subscriptions that did succeed get fresh data. Only the authority to +// DELETE is withheld. Failing the account instead would blank a whole +// account's dashboard over one flaky subscription, which is the opposite +// failure and the invariant SuccessfulCollect exists to protect. + +// partialSweepErr builds the error the Azure fan-out returns when some +// subscriptions could not be queried. Mirrors the real construction site, +// providers/azure/recommendations_multi_subscription.go. +func partialSweepErr(failedSubscriptionIDs ...string) *azureprovider.PartialSubscriptionFailureError { + failures := make([]azureprovider.SubscriptionFailure, 0, len(failedSubscriptionIDs)) + for _, id := range failedSubscriptionIDs { + failures = append(failures, azureprovider.SubscriptionFailure{ + SubscriptionID: id, + Err: errors.New("403 authorization failed"), + }) + } + return &azureprovider.PartialSubscriptionFailureError{ + Attempted: len(failedSubscriptionIDs) + 1, + Succeeded: 1, + Failed: failures, + } +} + +// newPartialSweepScheduler wires a Scheduler whose Azure provider returns +// recs plus a partial-subscription failure, exercising the ambient path +// (no registered accounts + AZURE_SUBSCRIPTION_ID set). +func newPartialSweepScheduler(t *testing.T, sweepErr error) *Scheduler { + t.Helper() + + recClient := new(MockRecommendationsClient) + t.Cleanup(func() { recClient.AssertExpectations(t) }) + recClient.On("GetAllRecommendations", mock.Anything). + Return([]common.Recommendation{{ + Service: common.ServiceEC2, + Region: "westeurope", + ResourceType: "Standard_D2s_v3", + Count: 3, + Term: "1yr", + PaymentOption: "all-upfront", + }}, sweepErr).Once() + + prov := new(MockProvider) + t.Cleanup(func() { prov.AssertExpectations(t) }) + prov.On("GetRecommendationsClient", mock.Anything).Return(recClient, nil).Once() + + factory := new(MockProviderFactory) + t.Cleanup(func() { factory.AssertExpectations(t) }) + factory.On("CreateAndValidateProvider", mock.Anything, "azure", mock.Anything). + Return(prov, nil).Once() + + store := new(MockConfigStore) + t.Cleanup(func() { store.AssertExpectations(t) }) + // No registered Azure accounts -> ambient path. + store.On("ListCloudAccounts", mock.Anything, mock.Anything). + Return([]config.CloudAccount{}, nil).Maybe() + // Ambient account resolution is best-effort; leave it unresolved so the + // roster carries the nil-CloudAccountID sentinel and the assertion is + // about roster membership rather than which ID it holds. + store.On("GetCloudAccountByExternalID", mock.Anything, "azure", mock.Anything). + Return(nil, nil).Maybe() + + return &Scheduler{config: store, providerFactory: factory} +} + +// TestCollectAzure_PartialSweepIsNotEvictionEligible is the regression for +// #1652. It fails on the pre-fix code, where tolerateIncompleteSweep returns +// nil and the account is reported as fully succeeded. +func TestCollectAzure_PartialSweepIsNotEvictionEligible(t *testing.T) { + t.Setenv("AZURE_SUBSCRIPTION_ID", "sub-a") + + s := newPartialSweepScheduler(t, partialSweepErr("sub-b", "sub-c")) + + recs, succeededAccountIDs, err := s.collectAzureRecommendations(context.Background(), &config.GlobalConfig{}) + + require.NoError(t, err, + "a partial sweep must not fail the account: one flaky subscription cannot blank the whole account's dashboard") + assert.NotEmpty(t, recs, + "the subscriptions that DID succeed must still contribute rows to be upserted") + assert.Emptyf(t, succeededAccountIDs, + "a partial sweep must not enter the eviction roster, got %#v: 2 subscription(s) were never "+ + "queried, so their previous-cycle rows would be deleted by the (provider, account_key) DELETE", + succeededAccountIDs) +} + +// TestCollectAzure_CompleteSweepStaysEvictionEligible is the control. The +// fix must withhold eviction ONLY for partial sweeps -- if a clean sweep +// also stopped evicting, stale rows would accumulate forever and rows for +// genuinely-gone resources would never disappear. Under-evicting is a +// quieter bug than over-evicting, so it gets its own assertion. +func TestCollectAzure_CompleteSweepStaysEvictionEligible(t *testing.T) { + t.Setenv("AZURE_SUBSCRIPTION_ID", "sub-a") + + s := newPartialSweepScheduler(t, nil) + + recs, succeededAccountIDs, err := s.collectAzureRecommendations(context.Background(), &config.GlobalConfig{}) + + require.NoError(t, err) + assert.NotEmpty(t, recs) + assert.NotEmpty(t, succeededAccountIDs, + "a complete sweep must still authorize eviction, otherwise stale rows accumulate forever") +} + +// TestFanOutPerAccount_PartialSweepNotInSucceededAccountIDs covers the +// registered-account path, which is the common production shape and does not +// go through the ambient collector the tests above drive. It is the partial +// sibling of TestFanOutPerAccount_FailedCollectionNotInSucceededAccountIDs: +// that one pins the FAILED account, this one pins the account that returned +// data from an incomplete sweep -- neither a clean success nor a failure, and +// the case the account-level tests never exercised. +func TestFanOutPerAccount_PartialSweepNotInSucceededAccountIDs(t *testing.T) { + accounts := []config.CloudAccount{ + {ID: "acct-complete", Name: "acct-complete", ExternalID: "e-ok"}, + {ID: "acct-partial", Name: "acct-partial", ExternalID: "e-partial"}, + } + fn := func(_ context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { + // Both return rows; only one swept every subscription. + return []config.RecommendationRecord{{ID: acct.ID, Provider: "azure"}}, + acct.ID == "acct-complete", nil + } + + recs, outcome := fanOutPerAccount(context.Background(), "Azure", accounts, fn) + + assert.Len(t, recs, 2, + "a partial sweep's rows must still be collected and upserted -- only eviction is withheld") + assert.Equal(t, []string{"acct-complete"}, outcome.SucceededAccountIDs, + "only the fully-swept account may authorize stale-row eviction") + assert.Equal(t, []string{"acct-partial"}, outcome.IncompleteAccountIDs) + assert.Equal(t, 2, outcome.SucceededCount, + "a partial sweep still counts as a success, so one flaky subscription cannot trip the all-accounts-failed guard") + assert.Zero(t, outcome.FailedCount) +} + +// TestCollectAzure_HardSweepErrorStillFailsTheAccount pins that the fix did +// not soften genuine failures into partial ones: a non-partial error must +// still propagate, keeping the account out of the roster via the error path. +func TestCollectAzure_HardSweepErrorStillFailsTheAccount(t *testing.T) { + t.Setenv("AZURE_SUBSCRIPTION_ID", "sub-a") + + s := newPartialSweepScheduler(t, errors.New("429 too many requests")) + + _, succeededAccountIDs, err := s.collectAzureRecommendations(context.Background(), &config.GlobalConfig{}) + + require.Error(t, err, "a hard collection error must still fail the account") + assert.Empty(t, succeededAccountIDs) +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 16e07c28e..f80a32378 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -430,6 +430,22 @@ func (s *Scheduler) collectProviderRecommendations(ctx context.Context, provider } } +// ambientEvictionRoster builds the succeeded-account roster for an ambient +// (unregistered-credential) collection. An incomplete sweep yields an empty +// roster, so UpsertRecommendations evicts nothing for it and the +// subscriptions the sweep never queried keep their previous rows -- see +// tolerateIncompleteSweep and issue #1652. +// +// accountID is "" when the host identity did not resolve to a registered +// cloud_accounts row; that empty-string sentinel is what +// expandSuccessfulCollects maps to a nil CloudAccountID. +func ambientEvictionRoster(complete bool, accountID string) []string { + if !complete { + return nil + } + return []string{accountID} +} + // expandSuccessfulCollects converts the per-account ID slice returned by // each provider collect into the typed []SuccessfulCollect the persist // layer expects. The empty-string sentinel is translated into a nil @@ -459,7 +475,7 @@ func (s *Scheduler) collectAWSRecommendations(ctx context.Context, globalCfg *co // Backward-compatible fallback: no registered accounts → ambient credentials if len(accounts) == 0 { - recs, err := s.collectAWSAmbient(ctx, globalCfg) + recs, complete, err := s.collectAWSAmbient(ctx, globalCfg) if err != nil { return nil, nil, err } @@ -471,12 +487,12 @@ func (s *Scheduler) collectAWSRecommendations(ctx context.Context, globalCfg *co // in those cases. if acctID := s.resolveAmbientHostAccountID(ctx); acctID != "" { recs = s.tagAccount(recs, acctID) - return recs, []string{acctID}, nil + return recs, ambientEvictionRoster(complete, acctID), nil } - return recs, []string{""}, nil + return recs, ambientEvictionRoster(complete, ""), nil } - recs, outcome := fanOutPerAccount(ctx, "AWS", accounts, func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { + recs, outcome := fanOutPerAccount(ctx, "AWS", accounts, func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { return s.collectAWSForAccount(ctx, globalCfg, acct) }) if outcome.FailedCount == len(accounts) && len(accounts) > 0 { @@ -496,15 +512,25 @@ func (s *Scheduler) collectAWSRecommendations(ctx context.Context, globalCfg *co // logs are the operator signal for the failed minority. // // SucceededAccountIDs lists the IDs of accounts whose collection -// completed; the scheduler threads these into the +// completed IN FULL; the scheduler threads these into the // []SuccessfulCollect slice that drives UpsertRecommendations' // account-scoped eviction. Order is not preserved — the eviction // query treats it as a set. +// +// IncompleteAccountIDs lists accounts that returned usable rows from a +// sweep that did not cover every subscription it was asked to. They are +// counted in SucceededCount (their rows are kept and upserted, and they +// must not trip the all-accounts-failed guard) but are deliberately +// absent from SucceededAccountIDs, because eviction is keyed on the +// account while a sweep fails per subscription; evicting on a partial +// sweep deletes the un-queried subscriptions' previous rows. See +// tolerateIncompleteSweep and issue #1652. type accountOutcome struct { - LastErr string - SucceededAccountIDs []string - SucceededCount int - FailedCount int + LastErr string + SucceededAccountIDs []string + IncompleteAccountIDs []string + SucceededCount int + FailedCount int } // fanOutPerAccount runs fn concurrently across accounts, bounded by the @@ -521,7 +547,7 @@ func fanOutPerAccount( ctx context.Context, providerLabel string, accounts []config.CloudAccount, - fn func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error), + fn func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error), ) ([]config.RecommendationRecord, accountOutcome) { g, gctx := errgroup.WithContext(ctx) g.SetLimit(execution.ConcurrencyFromEnv()) @@ -535,7 +561,7 @@ func fanOutPerAccount( for _rvc := range accounts { acct := accounts[_rvc] g.Go(func() error { - recs, err := fn(gctx, acct) + recs, complete, err := fn(gctx, acct) if err != nil { if isAccountPermissionError(providerLabel, err) { // Operator-fixable misconfiguration (missing IAM role @@ -556,8 +582,17 @@ func fanOutPerAccount( } mu.Lock() all = append(all, recs...) + // A partially-swept account counts as succeeded -- its rows are + // kept and upserted, and it must not trip the all-accounts-failed + // guard -- but it is deliberately withheld from + // SucceededAccountIDs, the roster that authorizes stale-row + // eviction. See tolerateIncompleteSweep for why (issue #1652). outcome.SucceededCount++ - outcome.SucceededAccountIDs = append(outcome.SucceededAccountIDs, acct.ID) + if complete { + outcome.SucceededAccountIDs = append(outcome.SucceededAccountIDs, acct.ID) + } else { + outcome.IncompleteAccountIDs = append(outcome.IncompleteAccountIDs, acct.ID) + } mu.Unlock() return nil }) @@ -600,10 +635,10 @@ func errAllAccountsFailed(providerLabel string, outcome accountOutcome) error { providerLabel, outcome.FailedCount, outcome.LastErr) } -func (s *Scheduler) collectAWSAmbient(ctx context.Context, globalCfg *config.GlobalConfig) ([]config.RecommendationRecord, error) { +func (s *Scheduler) collectAWSAmbient(ctx context.Context, globalCfg *config.GlobalConfig) ([]config.RecommendationRecord, bool, error) { prov, err := s.providerFactory.CreateAndValidateProvider(ctx, "aws", nil) if err != nil { - return nil, fmt.Errorf("failed to create AWS provider: %w", err) + return nil, false, fmt.Errorf("failed to create AWS provider: %w", err) } return s.fetchAndConvert(ctx, prov, "aws", nil, globalCfg) } @@ -675,62 +710,62 @@ func (s *Scheduler) resolveAmbientAccountID(ctx context.Context, providerName, e // fallback path in collectAzureRecommendations when no accounts are registered. // subscriptionID is passed explicitly to avoid an unnecessary Azure API round-trip // to auto-discover subscriptions — the caller already resolved it from env. -func (s *Scheduler) collectAzureAmbient(ctx context.Context, subscriptionID string) ([]config.RecommendationRecord, error) { +func (s *Scheduler) collectAzureAmbient(ctx context.Context, subscriptionID string) ([]config.RecommendationRecord, bool, error) { prov, err := s.providerFactory.CreateAndValidateProvider(ctx, "azure", &provider.ProviderConfig{ AzureSubscriptionID: subscriptionID, }) if err != nil { - return nil, fmt.Errorf("create ambient Azure provider: %w", err) + return nil, false, fmt.Errorf("create ambient Azure provider: %w", err) } recClient, err := prov.GetRecommendationsClient(ctx) if err != nil { - return nil, fmt.Errorf("get Azure recommendations client: %w", err) + return nil, false, fmt.Errorf("get Azure recommendations client: %w", err) } recs, err := recClient.GetAllRecommendations(ctx) - err = tolerateIncompleteSweep("azure", err) + complete, err := tolerateIncompleteSweep("azure", err) if err != nil { - return nil, fmt.Errorf("get Azure recommendations: %w", err) + return nil, false, fmt.Errorf("get Azure recommendations: %w", err) } - return s.convertRecommendations(recs, "azure"), nil + return s.convertRecommendations(recs, "azure"), complete, nil } // collectGCPAmbient collects GCP recommendations using Application Default // Credentials. Used by the ambient fallback path in collectGCPRecommendations // when no accounts are registered. -func (s *Scheduler) collectGCPAmbient(ctx context.Context) ([]config.RecommendationRecord, error) { +func (s *Scheduler) collectGCPAmbient(ctx context.Context) ([]config.RecommendationRecord, bool, error) { prov, err := s.providerFactory.CreateAndValidateProvider(ctx, "gcp", nil) if err != nil { - return nil, fmt.Errorf("create ambient GCP provider: %w", err) + return nil, false, fmt.Errorf("create ambient GCP provider: %w", err) } recClient, err := prov.GetRecommendationsClient(ctx) if err != nil { - return nil, fmt.Errorf("get GCP recommendations client: %w", err) + return nil, false, fmt.Errorf("get GCP recommendations client: %w", err) } recs, err := recClient.GetAllRecommendations(ctx) - err = tolerateIncompleteSweep("gcp", err) + complete, err := tolerateIncompleteSweep("gcp", err) if err != nil { - return nil, fmt.Errorf("get GCP recommendations: %w", err) + return nil, false, fmt.Errorf("get GCP recommendations: %w", err) } - return s.convertRecommendations(recs, "gcp"), nil + return s.convertRecommendations(recs, "gcp"), complete, nil } -func (s *Scheduler) collectAWSForAccount(ctx context.Context, globalCfg *config.GlobalConfig, acct config.CloudAccount) ([]config.RecommendationRecord, error) { +func (s *Scheduler) collectAWSForAccount(ctx context.Context, globalCfg *config.GlobalConfig, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { // Self-account (role_arn with no role ARN) or ambient modes use ambient credentials if acct.AWSRoleARN == "" { prov, err := s.providerFactory.CreateAndValidateProvider(ctx, "aws", nil) if err != nil { - return nil, fmt.Errorf("create ambient provider: %w", err) + return nil, false, fmt.Errorf("create ambient provider: %w", err) } return s.fetchAndConvert(ctx, prov, "aws", &acct.ID, globalCfg) } awsCreds, err := credentials.ResolveAWSCredentialProvider(ctx, &acct, s.credStore, s.assumeRoleSTS) if err != nil { - return nil, fmt.Errorf("resolve credentials: %w", err) + return nil, false, fmt.Errorf("resolve credentials: %w", err) } cfg := &provider.ProviderConfig{Name: "aws", AWSCredentialsProvider: awsCreds} prov, err := s.providerFactory.CreateAndValidateProvider(ctx, "aws", cfg) if err != nil { - return nil, fmt.Errorf("create provider: %w", err) + return nil, false, fmt.Errorf("create provider: %w", err) } return s.fetchAndConvert(ctx, prov, "aws", &acct.ID, globalCfg) } @@ -753,15 +788,15 @@ func (s *Scheduler) collectAzureRecommendations(ctx context.Context, _ *config.G logging.Info("No enabled Azure accounts — skipping Azure recommendations") return nil, nil, nil } - recs, err := s.collectAzureAmbient(ctx, subscriptionID) + recs, complete, err := s.collectAzureAmbient(ctx, subscriptionID) if err != nil { return nil, nil, err } if acctID := s.resolveAmbientAccountID(ctx, "azure", subscriptionID); acctID != "" { recs = s.tagAccount(recs, acctID) - return recs, []string{acctID}, nil + return recs, ambientEvictionRoster(complete, acctID), nil } - return recs, []string{""}, nil + return recs, ambientEvictionRoster(complete, ""), nil } recs, outcome := fanOutPerAccount(ctx, "Azure", accounts, s.collectAzureForAccount) @@ -771,7 +806,7 @@ func (s *Scheduler) collectAzureRecommendations(ctx context.Context, _ *config.G return recs, outcome.SucceededAccountIDs, nil } -func (s *Scheduler) collectAzureForAccount(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { +func (s *Scheduler) collectAzureForAccount(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { // Every recommendation returned below is tagged with THIS account's UUID // (see tagAccount at the end of this function), so the provider must be // pinned to this account's subscription. An empty AzureSubscriptionID @@ -780,7 +815,7 @@ func (s *Scheduler) collectAzureForAccount(ctx context.Context, acct config.Clou // other subscriptions' recommendations under this account and expose them // to anyone authorized for it. Fail loud instead; the row is misconfigured. if acct.AzureSubscriptionID == "" { - return nil, fmt.Errorf("cloud account %s has no azure_subscription_id configured", acct.ID) + return nil, false, fmt.Errorf("cloud account %s has no azure_subscription_id configured", acct.ID) } azCred, err := credentials.ResolveAzureTokenCredentialWithOpts(ctx, &acct, s.credStore, credentials.AzureResolveOptions{ @@ -788,24 +823,24 @@ func (s *Scheduler) collectAzureForAccount(ctx context.Context, acct config.Clou IssuerURL: s.oidcIssuerURL, }) if err != nil { - return nil, fmt.Errorf("resolve credentials: %w", err) + return nil, false, fmt.Errorf("resolve credentials: %w", err) } azProv, err := azureprovider.NewAzureProvider(&provider.ProviderConfig{Profile: acct.AzureSubscriptionID}) if err != nil { - return nil, fmt.Errorf("create provider: %w", err) + return nil, false, fmt.Errorf("create provider: %w", err) } azProv.SetCredential(azCred) recClient, err := azProv.GetRecommendationsClient(ctx) if err != nil { - return nil, fmt.Errorf("get recommendations client: %w", err) + return nil, false, fmt.Errorf("get recommendations client: %w", err) } recs, err := recClient.GetAllRecommendations(ctx) - err = tolerateIncompleteSweep("azure", err) + complete, err := tolerateIncompleteSweep("azure", err) if err != nil { - return nil, fmt.Errorf("get recommendations: %w", err) + return nil, false, fmt.Errorf("get recommendations: %w", err) } - return s.tagAccount(s.convertRecommendations(recs, "azure"), acct.ID), nil + return s.tagAccount(s.convertRecommendations(recs, "azure"), acct.ID), complete, nil } // collectGCPRecommendations fans out across all enabled GCP accounts, @@ -826,15 +861,15 @@ func (s *Scheduler) collectGCPRecommendations(ctx context.Context, _ *config.Glo logging.Info("No enabled GCP accounts — skipping GCP recommendations") return nil, nil, nil } - recs, err := s.collectGCPAmbient(ctx) + recs, complete, err := s.collectGCPAmbient(ctx) if err != nil { return nil, nil, err } if acctID := s.resolveAmbientAccountID(ctx, "gcp", projectID); acctID != "" { recs = s.tagAccount(recs, acctID) - return recs, []string{acctID}, nil + return recs, ambientEvictionRoster(complete, acctID), nil } - return recs, []string{""}, nil + return recs, ambientEvictionRoster(complete, ""), nil } recs, outcome := fanOutPerAccount(ctx, "GCP", accounts, s.collectGCPForAccount) @@ -844,13 +879,13 @@ func (s *Scheduler) collectGCPRecommendations(ctx context.Context, _ *config.Glo return recs, outcome.SucceededAccountIDs, nil } -func (s *Scheduler) collectGCPForAccount(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { +func (s *Scheduler) collectGCPForAccount(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { gcpTS, err := credentials.ResolveGCPTokenSourceWithOpts(ctx, &acct, s.credStore, credentials.GCPResolveOptions{ Signer: s.oidcSigner, IssuerURL: s.oidcIssuerURL, }) if err != nil { - return nil, fmt.Errorf("resolve credentials: %w", err) + return nil, false, fmt.Errorf("resolve credentials: %w", err) } var prov provider.Provider @@ -860,21 +895,21 @@ func (s *Scheduler) collectGCPForAccount(ctx context.Context, acct config.CloudA // ADC mode (application_default): use ambient credentials created, createErr := s.providerFactory.CreateAndValidateProvider(ctx, "gcp", nil) if createErr != nil { - return nil, fmt.Errorf("create ambient GCP provider: %w", createErr) + return nil, false, fmt.Errorf("create ambient GCP provider: %w", createErr) } prov = created } recClient, err := prov.GetRecommendationsClient(ctx) if err != nil { - return nil, fmt.Errorf("get recommendations client: %w", err) + return nil, false, fmt.Errorf("get recommendations client: %w", err) } recs, err := recClient.GetAllRecommendations(ctx) - err = tolerateIncompleteSweep("gcp", err) + complete, err := tolerateIncompleteSweep("gcp", err) if err != nil { - return nil, fmt.Errorf("get recommendations: %w", err) + return nil, false, fmt.Errorf("get recommendations: %w", err) } - return s.tagAccount(s.convertRecommendations(recs, "gcp"), acct.ID), nil + return s.tagAccount(s.convertRecommendations(recs, "gcp"), acct.ID), complete, nil } // enabledAccounts returns all enabled cloud accounts for the given provider. @@ -909,32 +944,59 @@ func (s *Scheduler) enabledAccounts(ctx context.Context, providerName string) [] // other error is returned unchanged, preserving the existing fail-loud // behavior. // -// NOTE: this records the incompleteness in the log only. Surfacing it in the -// state table's last_collection_error (so the dashboard shows the sweep as -// partial) needs a partial-note threaded through collectProviderRecommendations -// and its three per-provider implementations, which is left as follow-up work. -func tolerateIncompleteSweep(providerName string, err error) error { +// The complete return value is what stops a tolerated sweep from being +// mistaken for a whole one. It is false only for a partial sweep, and callers +// must keep such an account OUT of the succeeded-account roster: that roster +// drives UpsertRecommendations' stale-row eviction, which deletes with +// +// DELETE FROM recommendations +// WHERE collected_at < $1 AND (provider, account_key) IN (…) +// +// keyed on the CUDly account UUID, while a sweep fails one SUBSCRIPTION at a +// time. Reporting a partial sweep as succeeded therefore deletes the +// previous-cycle rows of every subscription the sweep never queried, and +// their savings silently read as "nothing to buy" rather than "collection +// incomplete" (issue #1652). That is data loss, not a display gap. +// +// Withholding eviction is deliberately all this does. The rows that WERE +// collected are still upserted by natural key, and the account is not marked +// failed -- failing it would blank the whole account's dashboard over one +// flaky subscription, the opposite failure and the one the succeeded-account +// roster exists to prevent. +// +// NOTE: the incompleteness still reaches the state table only as a log line; +// UpsertRecommendations continues to clear last_collection_error. That column +// is a single global row (recommendations_state WHERE id = 1), not per +// account, so representing one account's partial sweep in it is a separate +// design decision and is left as follow-up work. With eviction withheld the +// residue is observability, not lost rows. +func tolerateIncompleteSweep(providerName string, err error) (complete bool, _ error) { partial := azureprovider.AsPartialSubscriptionFailure(err) if partial == nil { - return err + return err == nil, err } logging.Warnf( - "%s recommendations incomplete: %d of %d subscriptions succeeded; not queried: %s (keeping the %d that did)", + "%s recommendations incomplete: %d of %d subscriptions succeeded; not queried: %s (keeping the %d that did, "+ + "and holding back stale-row eviction so the un-queried subscriptions' previous rows survive)", providerName, partial.Succeeded, partial.Attempted, strings.Join(partial.FailedSubscriptionIDs(), ", "), partial.Succeeded) - return nil + return false, nil } // fetchAndConvert is a convenience for the AWS ambient path. -func (s *Scheduler) fetchAndConvert(ctx context.Context, prov provider.Provider, providerName string, accountID *string, globalCfg *config.GlobalConfig) ([]config.RecommendationRecord, error) { +// +// The bool reports whether the sweep covered everything it was asked to; see +// tolerateIncompleteSweep for why an incomplete one must not authorize +// stale-row eviction. +func (s *Scheduler) fetchAndConvert(ctx context.Context, prov provider.Provider, providerName string, accountID *string, globalCfg *config.GlobalConfig) ([]config.RecommendationRecord, bool, error) { recClient, err := prov.GetRecommendationsClient(ctx) if err != nil { - return nil, fmt.Errorf("failed to get %s recommendations client: %w", providerName, err) + return nil, false, fmt.Errorf("failed to get %s recommendations client: %w", providerName, err) } recs, err := recClient.GetAllRecommendations(ctx) - err = tolerateIncompleteSweep(providerName, err) + complete, err := tolerateIncompleteSweep(providerName, err) if err != nil { - return nil, fmt.Errorf("failed to get %s recommendations: %w", providerName, err) + return nil, false, fmt.Errorf("failed to get %s recommendations: %w", providerName, err) } if len(recs) == 0 && globalCfg != nil { lookbackDays := globalCfg.RecommendationsLookbackDays @@ -948,20 +1010,25 @@ func (s *Scheduler) fetchAndConvert(ctx context.Context, prov provider.Provider, } var recErr error recs, recErr = recClient.GetRecommendations(ctx, ¶ms) - recErr = tolerateIncompleteSweep(providerName, recErr) + fallbackComplete, recErr := tolerateIncompleteSweep(providerName, recErr) if recErr != nil { // Fail loud: a misconfigured DefaultPayment/DefaultTerm or a CE // failure on this fallback must surface to the operator instead // of silently presenting as "zero recommendations". - return nil, fmt.Errorf("failed to get %s recommendations with default term/payment fallback (term=%s, payment=%s, lookback=%s): %w", + return nil, false, fmt.Errorf("failed to get %s recommendations with default term/payment fallback (term=%s, payment=%s, lookback=%s): %w", providerName, params.Term, params.PaymentOption, params.LookbackPeriod, recErr) } + // AND, not assignment: the fallback re-queries the same scope, so an + // incomplete first sweep stays incomplete even if the retry happens to + // come back whole. Overwriting here would let the retry launder away + // the first sweep's missing subscriptions and re-authorize eviction. + complete = complete && fallbackComplete } result := s.convertRecommendations(recs, providerName) if accountID != nil { result = s.tagAccount(result, *accountID) } - return result, nil + return result, complete, nil } // tagAccount sets CloudAccountID on each recommendation record. diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 7603f7bcd..317415118 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -753,13 +753,15 @@ func TestScheduler_FetchAndConvert_KeepsPartialSweepData(t *testing.T) { // globalCfg nil so the zero-results fallback branch stays out of the way; // collected is non-empty anyway. - recs, err := s.fetchAndConvert(ctx, prov, "azure", nil, nil) + recs, complete, err := s.fetchAndConvert(ctx, prov, "azure", nil, nil) require.NoError(t, err, "a partial multi-subscription sweep must not fail the whole collection") require.Len(t, recs, 1, "the subscriptions that succeeded must still be persisted, not discarded") assert.Equal(t, 3, recs[0].Count, "the surviving subscription's recommendation must be intact") + assert.False(t, complete, + "a partial sweep must report itself incomplete so the caller withholds stale-row eviction (#1652)") } // The tolerance must be narrow: any error that is NOT a partial-subscription @@ -777,29 +779,42 @@ func TestScheduler_FetchAndConvert_RealErrorStillFailsLoud(t *testing.T) { s := &Scheduler{config: new(MockConfigStore)} - recs, err := s.fetchAndConvert(ctx, prov, "azure", nil, nil) + recs, complete, err := s.fetchAndConvert(ctx, prov, "azure", nil, nil) require.Error(t, err, "a genuine error must still fail the collection") assert.Contains(t, err.Error(), "credentials expired") assert.Nil(t, recs) + assert.False(t, complete, "a failed sweep is never eviction-eligible") } func TestTolerateIncompleteSweep(t *testing.T) { - t.Run("partial failure is swallowed so the caller keeps its data", func(t *testing.T) { + // Swallowing the error is only half the contract. The other half is + // reporting complete=false, which is what keeps the account out of the + // eviction roster; asserting only NoError here would encode the #1652 + // hazard as intended behavior. + t.Run("partial failure is swallowed but reported as incomplete", func(t *testing.T) { partial := &azureprovider.PartialSubscriptionFailureError{ Attempted: 2, Succeeded: 1, Failed: []azureprovider.SubscriptionFailure{{SubscriptionID: "sub-1", Err: errors.New("boom")}}, } - assert.NoError(t, tolerateIncompleteSweep("azure", partial)) + complete, err := tolerateIncompleteSweep("azure", partial) + assert.NoError(t, err, "the collected subscriptions' data must be kept") + assert.False(t, complete, + "an incomplete sweep must not authorize stale-row eviction for the account (#1652)") }) t.Run("other errors pass through unchanged", func(t *testing.T) { boom := errors.New("boom") - assert.Same(t, boom, tolerateIncompleteSweep("azure", boom)) + complete, err := tolerateIncompleteSweep("azure", boom) + assert.Same(t, boom, err) + assert.False(t, complete, "a failed sweep is never eviction-eligible") }) - t.Run("nil stays nil", func(t *testing.T) { - assert.NoError(t, tolerateIncompleteSweep("azure", nil)) + t.Run("nil stays nil and is complete", func(t *testing.T) { + complete, err := tolerateIncompleteSweep("azure", nil) + assert.NoError(t, err) + assert.True(t, complete, + "a clean sweep must stay eviction-eligible, otherwise stale rows accumulate forever") }) } @@ -1116,13 +1131,13 @@ func TestFanOutPerAccount_RespectsParallelismLimit(t *testing.T) { } } - fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { + fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { cur := inflight.Add(1) updatePeak(cur) // Small sleep so concurrent workers genuinely overlap. time.Sleep(5 * time.Millisecond) inflight.Add(-1) - return []config.RecommendationRecord{{ID: acct.ID}}, nil + return []config.RecommendationRecord{{ID: acct.ID}}, true, nil } out, outcome := fanOutPerAccount(context.Background(), "Test", accounts, fn) @@ -1144,8 +1159,8 @@ func TestFanOutPerAccount_AllAccountsFail(t *testing.T) { {ID: "acct-2", Name: "acct-2", ExternalID: "ext-2"}, {ID: "acct-3", Name: "acct-3", ExternalID: "ext-3"}, } - fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { - return nil, fmt.Errorf("cred error for %s", acct.ID) + fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { + return nil, false, fmt.Errorf("cred error for %s", acct.ID) } recs, outcome := fanOutPerAccount(context.Background(), "Test", accounts, fn) @@ -1168,11 +1183,11 @@ func TestFanOutPerAccount_PartialSuccess(t *testing.T) { {ID: "acct-ok-2", Name: "acct-ok-2", ExternalID: "e2"}, {ID: "acct-bad", Name: "acct-bad", ExternalID: "ebad"}, } - fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { + fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { if acct.ID == "acct-bad" { - return nil, fmt.Errorf("transient") + return nil, false, fmt.Errorf("transient") } - return []config.RecommendationRecord{{ID: acct.ID, Provider: "test"}}, nil + return []config.RecommendationRecord{{ID: acct.ID, Provider: "test"}}, true, nil } recs, outcome := fanOutPerAccount(context.Background(), "Test", accounts, fn) @@ -1195,11 +1210,11 @@ func TestFanOutPerAccount_FailedCollectionNotInSucceededAccountIDs(t *testing.T) {ID: "acct-ok", Name: "acct-ok", ExternalID: "e-ok"}, {ID: "acct-all-services-failed", Name: "acct-all-services-failed", ExternalID: "e-bad"}, } - fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { + fn := func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { if acct.ID == "acct-all-services-failed" { - return nil, errors.New("all 6 Azure recommendation services failed: 429 too many requests") + return nil, false, errors.New("all 6 Azure recommendation services failed: 429 too many requests") } - return []config.RecommendationRecord{{ID: acct.ID, Provider: "azure"}}, nil + return []config.RecommendationRecord{{ID: acct.ID, Provider: "azure"}}, true, nil } _, outcome := fanOutPerAccount(context.Background(), "Azure", accounts, fn) @@ -1214,9 +1229,9 @@ func TestFanOutPerAccount_FailedCollectionNotInSucceededAccountIDs(t *testing.T) // correctly skips the all-failed error path. func TestFanOutPerAccount_ZeroAccounts(t *testing.T) { recs, outcome := fanOutPerAccount(context.Background(), "Test", nil, - func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, error) { + func(ctx context.Context, acct config.CloudAccount) ([]config.RecommendationRecord, bool, error) { t.Fatalf("fn must not be called for zero-accounts input") - return nil, nil + return nil, false, nil }) assert.Empty(t, recs) assert.Zero(t, outcome.SucceededCount) @@ -1762,7 +1777,7 @@ func TestScheduler_CollectAzureForAccount_MissingSubscriptionIDFailsLoud(t *test ctx := context.Background() scheduler := &Scheduler{config: new(MockConfigStore)} - recs, err := scheduler.collectAzureForAccount(ctx, config.CloudAccount{ + recs, complete, err := scheduler.collectAzureForAccount(ctx, config.CloudAccount{ ID: "az-no-sub", Provider: "azure", AzureAuthMode: "managed_identity", @@ -1772,6 +1787,7 @@ func TestScheduler_CollectAzureForAccount_MissingSubscriptionIDFailsLoud(t *test require.Error(t, err) assert.Nil(t, recs) + assert.False(t, complete, "a rejected account is never eviction-eligible") assert.Contains(t, err.Error(), "no azure_subscription_id configured", "an unpinned Azure account must be rejected by name, not fall through to an unscoped org-wide collection") assert.Contains(t, err.Error(), "az-no-sub", "the error must identify the misconfigured account")