From c67c3d670a4a219db59123d5fd20438965792bdd Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Mon, 3 Aug 2026 16:11:15 +0200 Subject: [PATCH] fix(scheduler): withhold stale-row eviction after a partial sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tolerateIncompleteSweep turned an Azure *PartialSubscriptionFailureError into a nil error so the subscriptions that DID answer keep their data. That much is right, but it discarded the fact that the sweep was incomplete, and the account then landed in the succeeded-account roster that authorizes stale-row eviction. Eviction is account-granular: DELETE FROM recommendations WHERE collected_at < $1 AND (provider, account_key) IN (…) keyed on account_key, the CUDly account UUID. A sweep, however, fails one SUBSCRIPTION at a time. A sweep that never queried subscription B returns no rows for B, so listing the account as succeeded deleted B's previous-cycle rows. Its savings opportunities vanished and read as "nothing to buy" rather than "collection incomplete" — the COR-03 hazard mergeServiceResults already warns about, landing on a money-adjacent table. tolerateIncompleteSweep now also reports whether the sweep was complete, and that flows through the per-provider collect functions into fanOutPerAccount, which keeps a partially-swept account out of SucceededAccountIDs while still counting it in SucceededCount. The narrowness is the point. The rows that were collected are still upserted by natural key, so the subscriptions that answered get fresh data, and the account is NOT marked failed: failing it would blank a whole account's dashboard over one flaky subscription, the opposite failure and the one the succeeded-account roster exists to prevent. Only the authority to DELETE is withheld. This reuses the semantics UpsertRecommendations already documents for failed collections ("their stale rows stay — callers see older data with a banner rather than a blank section"). Scoping eviction to the subscriptions actually queried was considered and rejected: the recommendations table has no subscription column (its identity is (provider, account_key), migration 000030), so it would need a new column, a migration, and the subscription threaded through RecommendationRecord — and would still need this same signal plumbed the same way. Regression coverage drives the sub-account granularity the existing account-level tests never reach: they assert that a collection returning an ERROR is excluded, while a partial sweep returns data plus an error, the one shape that is neither a clean success nor a failure. TestCollectAzure_PartialSweepIsNotEvictionEligible fails on the pre-fix code (roster is []string{""}). Controls pin both other directions: a complete sweep must stay eviction-eligible, or stale rows accumulate forever, and a hard error must still fail the account. TestTolerateIncompleteSweep asserted only NoError on a partial sweep, which encoded this hazard as intended behavior; it now pins the completeness half of the contract too. The function's NOTE framed the discarded signal as cosmetic ("so the dashboard shows the sweep as partial"). The same signal fed the eviction roster, so it was data loss, not display. Rewritten to say so. last_collection_error is still cleared on a partial sweep. That column is a single global row (recommendations_state WHERE id = 1), not per account, so carrying one account's partial sweep in it is a separate design decision; with eviction withheld the residue is observability, not lost rows. Documented at the source and left as follow-up. Note the fan-out is not yet reachable from the collection path: both Azure collect paths pin a subscription (collectAzureAmbient only runs when AZURE_SUBSCRIPTION_ID is set; collectAzureForAccount hard-errors on an empty azure_subscription_id), and GetRecommendationsClient only builds the multi-subscription client when no subscription is named. This lands the guarantee before that changes. Closes #1652 --- .../scheduler/partial_sweep_eviction_test.go | 183 ++++++++++++++++ internal/scheduler/scheduler.go | 199 ++++++++++++------ internal/scheduler/scheduler_test.go | 56 +++-- 3 files changed, 352 insertions(+), 86 deletions(-) create mode 100644 internal/scheduler/partial_sweep_eviction_test.go 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")