diff --git a/.gitallowed b/.gitallowed index 5b99a6993..648653520 100644 --- a/.gitallowed +++ b/.gitallowed @@ -75,7 +75,7 @@ # bracketed letter in PRIVAT[E] for the same self-scan reason as above. ^(SECURITY\.md|\.github/workflows/README\.md):[0-9]+:\|[-| ]+\|$ ^cmd/[a-z_]+\.go:[0-9]+:[[:space:]]*(fmt|AppLogger)\.Println\("-+"\)$ -^cmd/multi_service(_helpers)?\.go:[0-9]+:[[:space:]]*AppLogger\.Printf\("(\\n)?(━)+\\n"\)$ +^cmd/multi_service(_helpers|_csv_run)?\.go:[0-9]+:[[:space:]]*AppLogger\.Printf\("(\\n)?(━)+\\n"\)$ ^cmd/configure_gcp_test\.go:[0-9]+:// --- addMemberToPolicyBinding -+$ ^cmd/configure_test\.go:[0-9]+:[[:space:]]*assert\.Equal\(t, "-----BEGIN PRIVAT[E] KEY-----\\n\.\.\.", creds\.PrivateKey\)$ ^cmd/configure_gcp_store_test\.go:[0-9]+:[[:space:]]*"private_key": "-----BEGIN PRIVAT[E] KEY-----\\nMIIEvQIBADANBg\.\.\.\\n-----END PRIVATE KEY-----\\n",$ diff --git a/cmd/multi_service.go b/cmd/multi_service.go index eba5be7d5..69d528063 100644 --- a/cmd/multi_service.go +++ b/cmd/multi_service.go @@ -6,7 +6,6 @@ import ( "fmt" "log" "sync/atomic" - "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" @@ -380,726 +379,6 @@ func rederiveRDSFamilyCoverage(recs []common.Recommendation, cfg Config, coverag return recommendations.RederiveFamilyCoverage(recs, coverageMap) } -// rankingRule names the ordering --max-instances consumes, so the cap can say -// which rule decided what survived. The two paths rank on different keys -// because they carry different data, and an operator reading the drop list has -// to know which one applied. Typed rather than a bare string so the call sites -// cannot invent a third wording that does not match any implemented ordering. -type rankingRule string - -const ( - // rankBySavingsPercentage is the recommendation-driven path: scorer.Score - // sorts on SavingsPercentage, an intensive rate, before the cap runs. - rankBySavingsPercentage rankingRule = "highest savings-percentage recommendations first" - // rankBySavingsPerInstance is the --input-csv path: the cap is a budget in - // instances, and a CSV may lack the savings percentage (older reports), so - // sortBySavingsPerInstance derives the rate from the EstimatedSavings and - // Count columns every CSV carries. - rankBySavingsPerInstance rankingRule = "highest savings-per-instance rows first (EstimatedSavings / Count, not savings percentage)" -) - -// capBinds reports whether --max-instances will actually remove instances from -// recs, rather than being unset or already satisfied. -// -// applyGlobalInstanceLimit and requireRankingSignal have to agree on this -// exactly: the second refuses precisely the runs whose outcome the first would -// otherwise decide on an ordering that carries no information. If the two -// conditions drift apart, either a run is refused for a cap that changes -// nothing, or an unrankable run reaches the cap after all. -func capBinds(recs []common.Recommendation, cfg Config) bool { - others := withoutSavingsPlans(recs) - return cfg.MaxInstances > 0 && CalculateTotalInstances(others) > int(cfg.MaxInstances) -} - -// applyGlobalInstanceLimit enforces --max-instances once across the entire run. -// -// The flag is documented as a hard cap on the total number of instances -// purchased across all recommendations, so it has to see every service and -// every region together. Applying it inside the per-region fetch instead caps -// each (service, region) pair independently and multiplies the operator's cap -// by the number of pairs. -// -// passed must already be ordered best-first by rule: ApplyInstanceLimit -// consumes the slice in order and drops the tail, so the ordering decides -// which commitments survive. scorer.Score guarantees the -// rankBySavingsPercentage ordering; scoreAndLimitCSVRecs establishes the -// rankBySavingsPerInstance one. -// -// Truncation can push a recommendation under --min-count, which is a hard -// floor rather than advice, so dropTruncatedBelowMinCount removes any such -// recommendation instead of purchasing it short. -// -// Nothing is truncated silently. Every reduced or dropped recommendation is -// named on stdout, and the drops are counted into the end-of-run summary. -func applyGlobalInstanceLimit(passed []common.Recommendation, cfg Config, rule rankingRule, drops *common.DropSummary) []common.Recommendation { - if !capBinds(passed, cfg) { - return passed - } - - // Savings Plans are exempt from the cap and the post-cap floor: their Count - // is a placeholder 1, not an instance count. The report and the floor see - // only the instance-based rows, whose capped result is a prefix of its input. - others := withoutSavingsPlans(passed) - totalBefore := CalculateTotalInstances(others) - limited := ApplyInstanceLimit(others, cfg.MaxInstances) - limited, belowMin := dropTruncatedBelowMinCount(limited, cfg.MinCount) - reportInstanceLimit(others, limited, len(belowMin), totalBefore, cfg.MaxInstances, rule, drops) - reportMinCountDrops(belowMin, cfg.MinCount, drops) - return mergeKeepingOrder(passed, limited) -} - -// dropTruncatedBelowMinCount removes recommendations that the cap truncated to -// fewer instances than --min-count allows, returning the survivors and the -// removed recommendations at their truncated counts. -// -// --min-count is a hard floor everywhere else in the codebase, never advice: -// the scorer rejects recommendations under it outright -// (scorer.filterReason, "count %d below minimum %d"), the scheduler's -// meetsMinCount drops them, and both `docs/cli/filtering.md` and -// `docs/cli/README.md` describe it as dropping recommendations below the -// number. filtering.md applies it to "the adjusted instance count (after -// coverage scaling)", so the floor is meant to gate the *sized* count, and -// truncation by --max-instances is another form of sizing. -// -// Buying a commitment smaller than the operator's stated minimum can be worse -// than buying nothing, which is the whole reason the floor exists, so a -// truncated recommendation is dropped rather than purchased short. The freed -// budget is deliberately not redistributed: the next recommendation would have -// to fit in an even smaller remainder and would fail the same floor. -// -// Removal is always from the tail. ApplyInstanceLimit reduces at most one -// recommendation (the one where the budget runs out, which is the last it -// keeps), and every earlier one still carries the full count that already -// cleared the scorer's floor. Taking only from the tail keeps the result a -// prefix of the input, which reportInstanceLimit relies on. -func dropTruncatedBelowMinCount(limited []common.Recommendation, minCount int) (kept, removed []common.Recommendation) { - if minCount <= 0 { - return limited, nil - } - kept = limited - for len(kept) > 0 && kept[len(kept)-1].Count < minCount { - removed = append(removed, kept[len(kept)-1]) - kept = kept[:len(kept)-1] - } - return kept, removed -} - -// reportMinCountDrops names each recommendation the --min-count floor rejected -// after --max-instances truncated it. It continues the reportInstanceLimit -// listing, where these already appear as dropped, and explains why they were -// not simply purchased at the reduced count. -func reportMinCountDrops(removed []common.Recommendation, minCount int, drops *common.DropSummary) { - for i := range removed { - rec := removed[i] - AppLogger.Printf(" ↳ %s %s %s: the cap left room for only %d instances, below --min-count %d, so it is dropped rather than purchased short\n", - rec.Service, rec.Region, rec.ResourceType, rec.Count, minCount) - } - drops.Add(common.DropMinCountAfterCap, len(removed)) -} - -// reportInstanceLimit prints what --max-instances removed from the run. -// after must be the prefix of before produced by ApplyInstanceLimit, so -// after[i] and before[i] describe the same recommendation. -// -// rule is the ordering the caller sorted before by, and is printed verbatim: -// the drop list is unreadable without knowing which key decided it, and the -// two paths do not rank on the same key. -// -// belowMinCount is how many of the missing entries were removed by the -// --min-count floor rather than by the budget. They are still listed here as -// dropped (they were), but they are attributed to --min-count-after-cap by -// reportMinCountDrops, so excluding them from this tally keeps each dropped -// recommendation counted exactly once in the end-of-run summary. -func reportInstanceLimit(before, after []common.Recommendation, belowMinCount, totalBefore int, maxInstances int32, rule rankingRule, drops *common.DropSummary) { - AppLogger.Printf("\nšŸ”’ --max-instances=%d caps the whole run: the %d recommendations that passed scoring total %d instances.\n", - maxInstances, len(before), totalBefore) - AppLogger.Printf(" Keeping the %s. The following are reduced or dropped:\n", rule) - - // after is a prefix of before, so membership is positional: every entry past - // len(after) was removed, whatever its Count (a zero-Count entry equals its - // own "kept" count of 0 and must still be counted as removed). - reduced, dropped := 0, 0 - for i := range before { - rec := before[i] - if i >= len(after) { - dropped++ - AppLogger.Printf(" • dropped: %s %s %s (%d instances)\n", rec.Service, rec.Region, rec.ResourceType, rec.Count) - continue - } - if kept := after[i].Count; kept != rec.Count { - reduced++ - AppLogger.Printf(" • reduced: %s %s %s %d → %d instances\n", rec.Service, rec.Region, rec.ResourceType, rec.Count, kept) - } - } - - // The floor only removes entries from the tail of what the cap kept, so - // they are a subset of the removed entries. DropSummary.Add panics on a - // negative count; surface the broken invariant with context instead. - budgetDropped := dropped - belowMinCount - if budgetDropped < 0 { - panic(fmt.Sprintf("reportInstanceLimit: %d recommendations attributed to --min-count-after-cap but only %d were removed (before=%d, after=%d)", - belowMinCount, dropped, len(before), len(after))) - } - drops.Add(common.DropMaxInstances, budgetDropped) - AppLogger.Printf(" Proceeding with %d instances across %d recommendations (%d reduced, %d dropped).\n", - CalculateTotalInstances(after), len(after), reduced, dropped) -} - -// sumPassedRecs returns total instance count and total estimated savings for passed recs. -func sumPassedRecs(recs []common.Recommendation) (total int, totalSavings float64) { - for _rvc := range recs { - r := recs[_rvc] - total += r.Count - totalSavings += r.EstimatedSavings - } - return -} - -// executePurchasePipeline purchases each rec in the passed list (or dry-runs) and writes audit records. -// It stops at the first audit failure on a real run and returns errAuditTrailLost. -func executePurchasePipeline(ctx context.Context, awsCfg aws.Config, recs []common.Recommendation, isDryRun bool, runID string, cfg Config) ([]common.PurchaseResult, error) { - results := make([]common.PurchaseResult, 0, len(recs)) - for i := range recs { - rec := recs[i] - if stopRequested(ctx) { - log.Printf("Shutdown requested — skipping %d remaining recommendations", len(recs)-i) - break - } - if err := probeAuditLog(isDryRun, cfg.AuditLog); err != nil { - return results, err - } - // The purchase runs on a context that survives Ctrl-C: canceling the - // request mid-flight can leave AWS having created the commitment - // while the CLI records a failure, and a re-run then buys it again. - // The per-call SDK timeouts still bound it. - result, status := purchaseSingleRec(context.WithoutCancel(ctx), awsCfg, rec, i+1, isDryRun, cfg) - results = append(results, result) - if err := writePurchaseAuditRecord(runID, rec, result, status, isDryRun, cfg.AuditLog); err != nil { - return results, err - } - if !isDryRun && i < len(recs)-1 { - waitBetweenPurchases(ctx) - } - } - return results, nil -} - -// writePurchaseAuditRecord writes a single purchase's audit record. Shared by -// both purchase entry points -- executePurchasePipeline (the main pipeline) -// and processPurchaseLoop (the --input-csv path) -- so every recommendation -// that reaches a purchase attempt, dry-run or real, is recorded to -// cfg.AuditLog regardless of which one produced it. Before #1609, -// processPurchaseLoop never wrote a record at all, so CSV-mode purchases left -// no audit trail: on a partial failure there was no durable, per-recommendation -// record of which rows succeeded, so an operator could only re-run the whole -// file, which is a double purchase for the rows that already succeeded. -// -// A failed write on a real run returns errAuditTrailLost (#2103): the caller -// must stop buying. The record may be missing or written but not durable -// (sync failed); both count as lost. A dry run only warns. -func writePurchaseAuditRecord(runID string, rec common.Recommendation, result common.PurchaseResult, status string, isDryRun bool, auditLogPath string) error { - auditRec := common.NewAuditRecord(runID, rec, result, status, isDryRun, common.PurchaseSourceCLI) - if err := writeAuditRecordFn(auditRec, auditLogPath); err != nil { - if isDryRun { - log.Printf("Warning: failed to write audit record: %v", err) - return nil - } - return fmt.Errorf("%w: write: %w", errAuditTrailLost, err) - } - return nil -} - -// probeAuditLog checks, before a real purchase, that the audit log can still -// be appended to, so a log that broke after the start-of-run check stops the -// run before money moves. Dry runs skip it. -func probeAuditLog(isDryRun bool, auditLogPath string) error { - if isDryRun { - return nil - } - if err := common.CheckAuditLogWritable(auditLogPath); err != nil { - return fmt.Errorf("%w: probe before purchase: %w", errAuditTrailLost, err) - } - return nil -} - -// newServiceClient is the service-client factory used by purchaseSingleRec. -// It is a seam so tests can drive the real purchase path with a fake provider. -var newServiceClient = createServiceClient - -// purchaseSucceeded reports whether a purchase result is a success: the -// provider set Success AND reported no error. A provider may return -// Success=true with a non-nil Error; that is a failure (issue #1900). -func purchaseSucceeded(result common.PurchaseResult) bool { - return result.Success && result.Error == nil -} - -// purchaseAuditStatus derives the audit status for a completed (non-dry-run) -// purchase attempt. Callers handle the dry-run ("skipped") case separately, -// since that never reaches a PurchaseResult from an actual API call. -func purchaseAuditStatus(result common.PurchaseResult) string { - if purchaseSucceeded(result) { - return "success" - } - return "error" -} - -// purchaseSingleRec executes or dry-runs a single purchase and returns the result + audit status. -func purchaseSingleRec(ctx context.Context, awsCfg aws.Config, rec common.Recommendation, index int, isDryRun bool, cfg Config) (purchaseResult common.PurchaseResult, auditStatus string) { - label := purchaseRegionLabel(rec, rec.Region) - AppLogger.Printf(" [%d] %s %s %s (count=%d)\n", index, rec.Service, label, rec.ResourceType, rec.Count) - if err := validatePurchasePreconditions(rec, rec.Region, cfg.CallerAccountID); err != nil { - AppLogger.Printf(" āŒ cannot purchase: %v\n", err) - return preconditionFailure(rec, err, isDryRun), "error" - } - if isDryRun { - result := createDryRunResult(rec, label, index, cfg) - AppLogger.Printf(" [dry-run] %s\n", result.CommitmentID) - return result, "skipped" - } - - clientRegion, err := clientRegionFor(rec.Service, rec.Region) - if err != nil { - AppLogger.Printf(" āŒ %v\n", err) - return common.PurchaseResult{Recommendation: rec, Success: false, Error: err, Timestamp: time.Now()}, "error" - } - regionalCfg := awsCfg.Copy() - regionalCfg.Region = clientRegion - serviceClient := newServiceClient(rec.Service, regionalCfg) - if serviceClient == nil { - AppLogger.Printf(" āš ļø No service client for %s\n", rec.Service) - return common.PurchaseResult{Success: false}, "error" - } - - result := executePurchase(ctx, rec, label, index, serviceClient, cfg) - status := purchaseAuditStatus(result) - if purchaseSucceeded(result) { - AppLogger.Printf(" āœ… %s\n", result.CommitmentID) - } else { - AppLogger.Printf(" āŒ %v\n", result.Error) - } - return result, status -} - -// buildServiceStats computes per-service statistics from a purchase run. -// Results are assumed to be in the same order as recs (1:1 correspondence). -func buildServiceStats(recs []common.Recommendation, results []common.PurchaseResult) map[common.ServiceType]ServiceProcessingStats { - byService := make(map[common.ServiceType][]common.Recommendation) - resultsByService := make(map[common.ServiceType][]common.PurchaseResult) - for i := range recs { - rec := recs[i] - byService[rec.Service] = append(byService[rec.Service], rec) - if i < len(results) { - resultsByService[rec.Service] = append(resultsByService[rec.Service], results[i]) - } - } - stats := make(map[common.ServiceType]ServiceProcessingStats) - for service, serviceRecs := range byService { - stats[service] = calculateServiceStats(service, serviceRecs, resultsByService[service]) - } - return stats -} - -// runCSVPathOrFatal runs the CSV purchase path and exits fatally on error. -// It isolates the error-to-fatal glue from runToolMultiService so that path -// stays under the cyclomatic-complexity budget while runToolFromCSV remains -// unit-testable via its returned error. -func runCSVPathOrFatal(ctx context.Context, cfg Config) { - if err := runToolFromCSV(ctx, cfg); err != nil { - log.Fatalf("%v", err) - } -} - -// prepareCSVPurchaseRun validates and loads everything runToolFromCSV needs -// before the per-service purchase loop: the audit log writability, the CSV -// file, filtering/sizing, the single run-wide purchase confirmation, and the -// AWS config. Extracted to keep runToolFromCSV under the project's gocyclo -// budget. -// -// The audit-log check runs first and before any cloud API call, matching the -// non-CSV path (CheckAuditLogWritable in runToolMultiService). Before #1609 -// this check ran only on the non-CSV path, so a CSV-mode purchase run could -// reach real purchase calls with no way to have written a durable, -// per-recommendation audit record even in principle. -// -// The confirmation runs once against the full post-filter set, as the -// non-CSV path does (confirmPurchaseRun, shared with runPurchaseAndReport). -// Before #1610 it was asked once per (service, region) with only that -// region's totals, and declining canceled only that region. -// -// A nil recs with a nil error means there is nothing to do (already logged): -// filtering left no recommendations, or the user declined the confirmation. -func prepareCSVPurchaseRun(ctx context.Context, cfg Config, csvModeCoverage float64, isDryRun bool) (recs []common.Recommendation, awsCfg aws.Config, runID string, err error) { - if err = CheckAuditLogWritable(cfg.AuditLog); err != nil { - return nil, aws.Config{}, "", fmt.Errorf("cannot write audit log: %w", err) - } - - AppLogger.Printf("šŸ“„ Reading recommendations from CSV: %s\n", cfg.CSVInput) - recs, err = loadCSVRecommendations(cfg.CSVInput, csvThresholdNeedsFor(cfg)) - if err != nil { - return nil, aws.Config{}, "", fmt.Errorf("failed to read CSV file: %w", err) - } - AppLogger.Printf("āœ… Loaded %d recommendations from CSV\n", len(recs)) - - err = rejectDuplicateSavingsPlanRows(recs) - if err != nil { - return nil, aws.Config{}, "", err - } - - recs, err = filterAndAdjustRecommendations(ctx, recs, csvModeCoverage, cfg) - if err != nil { - return nil, aws.Config{}, "", err - } - if len(recs) == 0 { - AppLogger.Println("āš ļø No recommendations to process after filtering") - return nil, aws.Config{}, "", nil - } - if !confirmPurchaseRun(recs, isDryRun) { - AppLogger.Printf("\nāŒ Purchase canceled.\n") - return nil, aws.Config{}, "", nil - } - - awsCfg, err = loadAWSConfig(ctx, cfg) - if err != nil { - return nil, aws.Config{}, "", fmt.Errorf("failed to load AWS config: %w", err) - } - - return recs, awsCfg, uuid.New().String(), nil -} - -// runToolFromCSV processes recommendations from a CSV input file. -// It returns an error instead of exiting so the orchestration glue is -// unit-testable; the caller (runCSVPathOrFatal) turns errors fatal. -func runToolFromCSV(ctx context.Context, cfg Config) error { - isDryRun := effectiveDryRun(cfg) - printRunMode(isDryRun) - - // Register SIGINT handler so an in-flight purchase run can be stopped - // cleanly between regions, matching the non-CSV path - // (runToolMultiService). Before #1610 this path had no SIGINT handling - // at all: runToolMultiService registers it only on the non-CSV branch, - // in code unreachable from CSV mode (the CSV branch returns first). - defer registerShutdownSignalHandler(ctx)() - - csvModeCoverage := determineCSVCoverage(cfg) - - recs, awsCfg, runID, err := prepareCSVPurchaseRun(ctx, cfg, csvModeCoverage, isDryRun) - if err != nil { - return err - } - if len(recs) == 0 { - return nil - } - - cfg, err = withCallerAccount(ctx, awsCfg, cfg) - if err != nil { - return err - } - return runCSVPurchases(ctx, awsCfg, recs, isDryRun, cfg, runID) -} - -// runCSVPurchases groups the confirmed CSV rows by service and region, buys -// them, and writes the report and summary for whatever ran. A non-nil error -// is errAuditTrailLost and/or errPurchasesFailed. -func runCSVPurchases(ctx context.Context, awsCfg aws.Config, recs []common.Recommendation, isDryRun bool, cfg Config, runID string) error { - // Purchase IDs label the sizing via cfg.Coverage (effectiveSizingPct); - // pin it to the coverage the CSV sizing used so the label cannot drift (#2248). - cfg.Coverage = determineCSVCoverage(cfg) - - // Create account alias cache for lookup - accountCache := NewAccountAliasCache(awsCfg) - - // Populate account names from account IDs - populateAccountNames(ctx, recs, accountCache) - - // Group recommendations by service and region - recsByServiceRegion := groupRecommendationsByServiceRegion(recs) - - // Process purchases - allResults := make([]common.PurchaseResult, 0) - var serviceResults []common.PurchaseResult - serviceStats := make(map[common.ServiceType]ServiceProcessingStats) - // allAdjustedRecs accumulates post-dedup recommendations so the final summary - // reflects what was actually processed rather than the pre-dedup input slice. - allAdjustedRecs := make([]common.Recommendation, 0) - var auditErr error // errAuditTrailLost: set by a region, stops the service loop too - - for service, regionRecs := range recsByServiceRegion { - if stopRequested(ctx) { - log.Printf("Shutdown requested; stopping before %s", getServiceDisplayName(service)) - break - } - - AppLogger.Printf("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n") - AppLogger.Printf("šŸŽÆ Processing %s\n", getServiceDisplayName(service)) - AppLogger.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n") - - var serviceRecs []common.Recommendation - serviceRecs, serviceResults, auditErr = processCSVService(ctx, awsCfg, service, regionRecs, isDryRun, cfg, runID) - allAdjustedRecs = append(allAdjustedRecs, serviceRecs...) - - // Add service results to overall results - allResults = append(allResults, serviceResults...) - - // Calculate service statistics (using only this service's results) - stats := calculateServiceStats(service, serviceRecs, serviceResults) - serviceStats[service] = stats - printServiceSummary(service, stats) - if auditErr != nil { - break - } - } - - // Generate CSV filename and write report - finalCSVOutput := generateCSVFilename(isDryRun, cfg) - - // Write CSV report - writeCSVReport(allResults, finalCSVOutput) - - // Print final summary using the post-dedup slice so counts match what was - // actually processed, not the pre-dedup input passed into the outer loop. - printMultiServiceSummary(allAdjustedRecs, allResults, serviceStats, isDryRun) - if stopRequested(ctx) { - printInterruptedSummary(len(allResults), len(recs), allResults, isDryRun) - } - if auditErr != nil { - AppLogger.Print(auditStopMessage(len(allResults), len(recs), allResults, auditErr)) - } - return errors.Join(auditErr, purchaseFailureError(allResults, isDryRun)) -} - -// processCSVRegionPurchases handles a single (service, region) pair within -// the --input-csv purchase loop: builds the regional service client, runs -// the duplicate check, applies the --min-count floor to whatever the -// deduction left, and executes the purchase loop. ok=false means there is -// nothing to add for this region (no service client yet, or the duplicate -// check refused it) and the caller should move on to the next region. -// Extracted out of runToolFromCSV to keep it under the project's gocyclo -// budget. -func processCSVRegionPurchases(ctx context.Context, awsCfg aws.Config, service common.ServiceType, region string, recs []common.Recommendation, isDryRun bool, cfg Config, runID string) (processedRecs []common.Recommendation, results []common.PurchaseResult, ok bool, auditErr error) { - clientRegion, err := clientRegionFor(service, region) - if err != nil { - AppLogger.Printf(" āŒ Cannot purchase %d %s recommendation(s) from the CSV: %v\n", len(recs), getServiceDisplayName(service), err) - return nil, nil, false, nil - } - regionalCfg := awsCfg.Copy() - regionalCfg.Region = clientRegion - serviceClient := createServiceClient(service, regionalCfg) - - if serviceClient == nil { - AppLogger.Printf(" āš ļø Service client not yet implemented for %s\n", getServiceDisplayName(service)) - AppLogger.Printf(" (Skipping purchase phase for this service)\n") - return nil, nil, false, nil - } - - // Check for duplicate RIs to avoid double purchasing. - adjustedRecs, dedupOK := checkDuplicatesForCSVRegion(ctx, recs, serviceClient, service, region, isDryRun) - if !dedupOK { - return nil, nil, false, nil - } - // Deducting existing commitments shrinks Count, which can push a row - // that cleared the floor in filterAndAdjustRecommendations back under it - // (--min-count 5, a row of 6, and 5 matching recent commitments would - // otherwise be purchased at 1). --min-count is a floor on what gets - // bought, so it is re-applied to whatever the deduction left, not only - // to the pre-deduction counts. - processedRecs = applyMinCountFloor(adjustedRecs, cfg.MinCount) - - results, auditErr = processPurchaseLoop(ctx, processedRecs, region, isDryRun, serviceClient, cfg, runID) - return processedRecs, results, true, auditErr -} - -// checkDuplicatesForCSVRegion runs the duplicate check for a single -// (service, region) pair in CSV mode and reports whether the caller should -// still process that region (ok). The duplicate check is the only guard -// between a re-run and a double purchase, so a failed check must not fall -// back to the un-deduplicated counts on a purchase run: that would buy -// reserved capacity the account already owns. A dry run logs a loud warning -// and continues with the un-deduplicated counts (nothing is bought, so -// reporting fidelity wins); a purchase run refuses to spend and returns -// ok=false, mirroring the fail-closed behavior of checkDuplicates on the -// non-CSV pipeline. -func checkDuplicatesForCSVRegion(ctx context.Context, recs []common.Recommendation, serviceClient provider.ServiceClient, service common.ServiceType, region string, isDryRun bool) (adjustedRecs []common.Recommendation, ok bool) { - adjustedRecs, err := adjustRecsForDuplicates(ctx, recs, serviceClient) - if err == nil { - return adjustedRecs, true - } - if ctx.Err() != nil { - // Interrupted: terminal in dry and real runs; runToolFromCSV stops on ctx.Err(). - return nil, false - } - if !isDryRun { - AppLogger.Printf(" āŒ Refusing to purchase %s/%s: could not check for existing RIs (%v). Skipping this region rather than risking a duplicate purchase.\n", getServiceDisplayName(service), region, err) - return nil, false - } - AppLogger.Printf(" āš ļø Warning: Could not check for existing RIs: %v (dry run; continuing with un-deduplicated counts)\n", err) - return recs, true -} - -// filterAndAdjustRecommendations applies filters, coverage, count override, -// the --min-count floor and the run-wide --max-instances cap to -// recommendations loaded from --input-csv. -// -// It returns an error when the cap cannot be enforced honestly; see -// requireRankingSignal. -func filterAndAdjustRecommendations(ctx context.Context, recs []common.Recommendation, csvModeCoverage float64, cfg Config) ([]common.Recommendation, error) { - if err := ctx.Err(); err != nil { - return nil, err - } - engineData, err := engineVersionFetcher(ctx, cfg, !cfg.IncludeExtendedSupport && hasDatabaseRecs(recs)) - if err != nil { - return nil, err - } - if err := requireFilterInputs(recs, cfg, engineData); err != nil { - return nil, err - } - - // Apply filters (empty currentRegion since we're processing from CSV, not iterating regions). - // Drop tracking is skipped on the CSV path (nil drops). - originalCount := len(recs) - recs = applyFilters(recs, &cfg, engineData.instanceVersions, engineData.versionInfo, "", nil) - if len(recs) < originalCount { - AppLogger.Printf("šŸ” After filters: %d recs (filtered out %d)\n", len(recs), originalCount-len(recs)) - } - - // Apply sizing — target-coverage if set, otherwise coverage. - // Coverage 100% is a no-op (early-returned inside ApplyCoverage), but - // --target-coverage always applies even at coverage 100%, so the - // CSV-path short-circuit is conditional on TargetCoverage == 0. - if cfg.TargetCoverage > 0 || csvModeCoverage < 100 { - beforeSize := len(recs) - recs = applySizing(recs, cfg, csvModeCoverage, nil) - if cfg.TargetCoverage > 0 { - AppLogger.Printf("šŸŽÆ Applying %.1f%% target-coverage: %d recs selected (from %d)\n", cfg.TargetCoverage, len(recs), beforeSize) - } else { - AppLogger.Printf("šŸ“ˆ Applying %.1f%% coverage: %d recs selected (from %d)\n", csvModeCoverage, len(recs), beforeSize) - } - } - - // Apply count override if specified - if cfg.OverrideCount > 0 { - recs = ApplyCountOverride(recs, cfg.OverrideCount) - } - - // Enforce --min-count and the run-wide --max-instances cap. - return scoreAndLimitCSVRecs(recs, cfg) -} - -// requireFilterInputs fails when the data the filters need is missing: RDS -// inventory for the extended-support exclusion, and account names for -// name-based account filters. -func requireFilterInputs(recs []common.Recommendation, cfg Config, engineData engineVersionData) error { - if err := requireRegionInventory(recs, cfg, engineData); err != nil { - return err - } - return requireResolvedAccountNames(recs, &cfg) -} - -// processPurchaseLoop processes purchases for a single region (used by CSV -// mode). runID groups every recommendation processed across the whole CSV -// run into one audit trail, matching how executePurchasePipeline (the main -// pipeline) generates one runID per invocation. Confirmation is not asked -// here: prepareCSVPurchaseRun confirms once for the whole run before this -// loop is reached. -func processPurchaseLoop(ctx context.Context, recs []common.Recommendation, region string, isDryRun bool, serviceClient provider.ServiceClient, cfg Config, runID string) ([]common.PurchaseResult, error) { - results := make([]common.PurchaseResult, 0, len(recs)) - - for j := range recs { - if stopRequested(ctx) { - log.Printf("Shutdown requested; skipping %d remaining recommendation(s) in %s", len(recs)-j, region) - break - } - - if err := probeAuditLog(isDryRun, cfg.AuditLog); err != nil { - return results, err - } - - rec := recs[j] - label := purchaseRegionLabel(rec, region) - AppLogger.Printf(" [%d/%d] Processing: %s %s\n", j+1, len(recs), rec.Service, rec.ResourceType) - AppLogger.Printf(" šŸ’³ Purchasing %d instances\n", rec.Count) - - result, status, attempted := runOnePurchase(ctx, rec, label, region, j+1, isDryRun, serviceClient, cfg) - - // Audit first, then the rate-limit delay: a Ctrl-C during the delay - // must not lose the record of a purchase that already completed. - results = append(results, result) - auditErr := writePurchaseAuditRecord(runID, rec, result, status, isDryRun, cfg.AuditLog) - logPurchaseOutcome(result) - if auditErr != nil { - return results, auditErr - } - - // Rate-limit delay between real purchases; returns at once on Ctrl-C. - if attempted && j < len(recs)-1 { - waitBetweenPurchases(ctx) - } - } - - return results, nil -} - -// runOnePurchase validates and then dry-runs or executes one CSV-path -// recommendation. attempted reports whether a real purchase call was made. -func runOnePurchase(ctx context.Context, rec common.Recommendation, label, region string, index int, isDryRun bool, serviceClient provider.ServiceClient, cfg Config) (result common.PurchaseResult, status string, attempted bool) { - if err := validatePurchasePreconditions(rec, region, cfg.CallerAccountID); err != nil { - return preconditionFailure(rec, err, isDryRun), "error", false - } - if isDryRun { - return createDryRunResult(rec, label, index, cfg), "skipped", false - } - // Detached from Ctrl-C for the same reason as in executePurchasePipeline. - result = executePurchase(context.WithoutCancel(ctx), rec, label, index, serviceClient, cfg) - return result, purchaseAuditStatus(result), true -} - -func logPurchaseOutcome(result common.PurchaseResult) { - if purchaseSucceeded(result) { - AppLogger.Printf(" āœ… Success: %s\n", result.CommitmentID) - return - } - errMsg := "unknown error" - if result.Error != nil { - errMsg = result.Error.Error() - } - AppLogger.Printf(" āŒ Failed: %s\n", errMsg) -} - -// processCSVService runs every region of one service. A non-nil error is -// errAuditTrailLost: the partial results are still returned, and the caller -// stops the service loop as well. -func processCSVService(ctx context.Context, awsCfg aws.Config, service common.ServiceType, regionRecs map[string][]common.Recommendation, isDryRun bool, cfg Config, runID string) (serviceRecs []common.Recommendation, serviceResults []common.PurchaseResult, auditErr error) { - for region, recs := range regionRecs { - if stopRequested(ctx) { - log.Printf("Shutdown requested; skipping remaining regions for %s", getServiceDisplayName(service)) - break - } - - AppLogger.Printf("\n šŸ“ Region: %s (%d recommendations)\n", region, len(recs)) - - processedRecs, regionResults, ok, regionErr := processCSVRegionPurchases(ctx, awsCfg, service, region, recs, isDryRun, cfg, runID) - if !ok { - continue - } - serviceRecs = append(serviceRecs, processedRecs...) - serviceResults = append(serviceResults, regionResults...) - if regionErr != nil { - return serviceRecs, serviceResults, regionErr - } - } - return serviceRecs, serviceResults, nil -} - -// hasDatabaseRecs reports whether any recommendation is one the extended-support -// exclusion applies to (the same *common.DatabaseDetails test the adjuster uses). -func hasDatabaseRecs(recs []common.Recommendation) bool { - for i := range recs { - if d, ok := recs[i].Details.(*common.DatabaseDetails); ok && d != nil { - return true - } - } - return false -} - // exitOnExclusionError aborts the run before any purchase when the // extended-support exclusion cannot be applied (issue #2147). func exitOnFetchError(err error) { diff --git a/cmd/multi_service_csv_run.go b/cmd/multi_service_csv_run.go new file mode 100644 index 000000000..8472b8514 --- /dev/null +++ b/cmd/multi_service_csv_run.go @@ -0,0 +1,414 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/google/uuid" +) + +// runCSVPathOrFatal runs the CSV purchase path and exits fatally on error. +// It isolates the error-to-fatal glue from runToolMultiService so that path +// stays under the cyclomatic-complexity budget while runToolFromCSV remains +// unit-testable via its returned error. +func runCSVPathOrFatal(ctx context.Context, cfg Config) { + if err := runToolFromCSV(ctx, cfg); err != nil { + log.Fatalf("%v", err) + } +} + +// prepareCSVPurchaseRun validates and loads everything runToolFromCSV needs +// before the per-service purchase loop: the audit log writability, the CSV +// file, filtering/sizing, the single run-wide purchase confirmation, and the +// AWS config. Extracted to keep runToolFromCSV under the project's gocyclo +// budget. +// +// The audit-log check runs first and before any cloud API call, matching the +// non-CSV path (CheckAuditLogWritable in runToolMultiService). Before #1609 +// this check ran only on the non-CSV path, so a CSV-mode purchase run could +// reach real purchase calls with no way to have written a durable, +// per-recommendation audit record even in principle. +// +// The confirmation runs once against the full post-filter set, as the +// non-CSV path does (confirmPurchaseRun, shared with runPurchaseAndReport). +// Before #1610 it was asked once per (service, region) with only that +// region's totals, and declining canceled only that region. +// +// A nil recs with a nil error means there is nothing to do (already logged): +// filtering left no recommendations, or the user declined the confirmation. +func prepareCSVPurchaseRun(ctx context.Context, cfg Config, csvModeCoverage float64, isDryRun bool) (recs []common.Recommendation, awsCfg aws.Config, runID string, err error) { + if err = CheckAuditLogWritable(cfg.AuditLog); err != nil { + return nil, aws.Config{}, "", fmt.Errorf("cannot write audit log: %w", err) + } + + AppLogger.Printf("šŸ“„ Reading recommendations from CSV: %s\n", cfg.CSVInput) + recs, err = loadCSVRecommendations(cfg.CSVInput, csvThresholdNeedsFor(cfg)) + if err != nil { + return nil, aws.Config{}, "", fmt.Errorf("failed to read CSV file: %w", err) + } + AppLogger.Printf("āœ… Loaded %d recommendations from CSV\n", len(recs)) + + err = rejectDuplicateSavingsPlanRows(recs) + if err != nil { + return nil, aws.Config{}, "", err + } + + recs, err = filterAndAdjustRecommendations(ctx, recs, csvModeCoverage, cfg) + if err != nil { + return nil, aws.Config{}, "", err + } + if len(recs) == 0 { + AppLogger.Println("āš ļø No recommendations to process after filtering") + return nil, aws.Config{}, "", nil + } + if !confirmPurchaseRun(recs, isDryRun) { + AppLogger.Printf("\nāŒ Purchase canceled.\n") + return nil, aws.Config{}, "", nil + } + + awsCfg, err = loadAWSConfig(ctx, cfg) + if err != nil { + return nil, aws.Config{}, "", fmt.Errorf("failed to load AWS config: %w", err) + } + + return recs, awsCfg, uuid.New().String(), nil +} + +// runToolFromCSV processes recommendations from a CSV input file. +// It returns an error instead of exiting so the orchestration glue is +// unit-testable; the caller (runCSVPathOrFatal) turns errors fatal. +func runToolFromCSV(ctx context.Context, cfg Config) error { + isDryRun := effectiveDryRun(cfg) + printRunMode(isDryRun) + + // Register SIGINT handler so an in-flight purchase run can be stopped + // cleanly between regions, matching the non-CSV path + // (runToolMultiService). Before #1610 this path had no SIGINT handling + // at all: runToolMultiService registers it only on the non-CSV branch, + // in code unreachable from CSV mode (the CSV branch returns first). + defer registerShutdownSignalHandler(ctx)() + + csvModeCoverage := determineCSVCoverage(cfg) + + recs, awsCfg, runID, err := prepareCSVPurchaseRun(ctx, cfg, csvModeCoverage, isDryRun) + if err != nil { + return err + } + if len(recs) == 0 { + return nil + } + + cfg, err = withCallerAccount(ctx, awsCfg, cfg) + if err != nil { + return err + } + return runCSVPurchases(ctx, awsCfg, recs, isDryRun, cfg, runID) +} + +// runCSVPurchases groups the confirmed CSV rows by service and region, buys +// them, and writes the report and summary for whatever ran. A non-nil error +// is errAuditTrailLost and/or errPurchasesFailed. +func runCSVPurchases(ctx context.Context, awsCfg aws.Config, recs []common.Recommendation, isDryRun bool, cfg Config, runID string) error { + // Purchase IDs label the sizing via cfg.Coverage (effectiveSizingPct); + // pin it to the coverage the CSV sizing used so the label cannot drift (#2248). + cfg.Coverage = determineCSVCoverage(cfg) + + // Create account alias cache for lookup + accountCache := NewAccountAliasCache(awsCfg) + + // Populate account names from account IDs + populateAccountNames(ctx, recs, accountCache) + + // Group recommendations by service and region + recsByServiceRegion := groupRecommendationsByServiceRegion(recs) + + // Process purchases + allResults := make([]common.PurchaseResult, 0) + var serviceResults []common.PurchaseResult + serviceStats := make(map[common.ServiceType]ServiceProcessingStats) + // allAdjustedRecs accumulates post-dedup recommendations so the final summary + // reflects what was actually processed rather than the pre-dedup input slice. + allAdjustedRecs := make([]common.Recommendation, 0) + var auditErr error // errAuditTrailLost: set by a region, stops the service loop too + + for service, regionRecs := range recsByServiceRegion { + if stopRequested(ctx) { + log.Printf("Shutdown requested; stopping before %s", getServiceDisplayName(service)) + break + } + + AppLogger.Printf("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n") + AppLogger.Printf("šŸŽÆ Processing %s\n", getServiceDisplayName(service)) + AppLogger.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n") + + var serviceRecs []common.Recommendation + serviceRecs, serviceResults, auditErr = processCSVService(ctx, awsCfg, service, regionRecs, isDryRun, cfg, runID) + allAdjustedRecs = append(allAdjustedRecs, serviceRecs...) + + // Add service results to overall results + allResults = append(allResults, serviceResults...) + + // Calculate service statistics (using only this service's results) + stats := calculateServiceStats(service, serviceRecs, serviceResults) + serviceStats[service] = stats + printServiceSummary(service, stats) + if auditErr != nil { + break + } + } + + // Generate CSV filename and write report + finalCSVOutput := generateCSVFilename(isDryRun, cfg) + + // Write CSV report + writeCSVReport(allResults, finalCSVOutput) + + // Print final summary using the post-dedup slice so counts match what was + // actually processed, not the pre-dedup input passed into the outer loop. + printMultiServiceSummary(allAdjustedRecs, allResults, serviceStats, isDryRun) + if stopRequested(ctx) { + printInterruptedSummary(len(allResults), len(recs), allResults, isDryRun) + } + if auditErr != nil { + AppLogger.Print(auditStopMessage(len(allResults), len(recs), allResults, auditErr)) + } + return errors.Join(auditErr, purchaseFailureError(allResults, isDryRun)) +} + +// processCSVRegionPurchases handles a single (service, region) pair within +// the --input-csv purchase loop: builds the regional service client, runs +// the duplicate check, applies the --min-count floor to whatever the +// deduction left, and executes the purchase loop. ok=false means there is +// nothing to add for this region (no service client yet, or the duplicate +// check refused it) and the caller should move on to the next region. +// Extracted out of runToolFromCSV to keep it under the project's gocyclo +// budget. +func processCSVRegionPurchases(ctx context.Context, awsCfg aws.Config, service common.ServiceType, region string, recs []common.Recommendation, isDryRun bool, cfg Config, runID string) (processedRecs []common.Recommendation, results []common.PurchaseResult, ok bool, auditErr error) { + clientRegion, err := clientRegionFor(service, region) + if err != nil { + AppLogger.Printf(" āŒ Cannot purchase %d %s recommendation(s) from the CSV: %v\n", len(recs), getServiceDisplayName(service), err) + return nil, nil, false, nil + } + regionalCfg := awsCfg.Copy() + regionalCfg.Region = clientRegion + serviceClient := createServiceClient(service, regionalCfg) + + if serviceClient == nil { + AppLogger.Printf(" āš ļø Service client not yet implemented for %s\n", getServiceDisplayName(service)) + AppLogger.Printf(" (Skipping purchase phase for this service)\n") + return nil, nil, false, nil + } + + // Check for duplicate RIs to avoid double purchasing. + adjustedRecs, dedupOK := checkDuplicatesForCSVRegion(ctx, recs, serviceClient, service, region, isDryRun) + if !dedupOK { + return nil, nil, false, nil + } + // Deducting existing commitments shrinks Count, which can push a row + // that cleared the floor in filterAndAdjustRecommendations back under it + // (--min-count 5, a row of 6, and 5 matching recent commitments would + // otherwise be purchased at 1). --min-count is a floor on what gets + // bought, so it is re-applied to whatever the deduction left, not only + // to the pre-deduction counts. + processedRecs = applyMinCountFloor(adjustedRecs, cfg.MinCount) + + results, auditErr = processPurchaseLoop(ctx, processedRecs, region, isDryRun, serviceClient, cfg, runID) + return processedRecs, results, true, auditErr +} + +// checkDuplicatesForCSVRegion runs the duplicate check for a single +// (service, region) pair in CSV mode and reports whether the caller should +// still process that region (ok). The duplicate check is the only guard +// between a re-run and a double purchase, so a failed check must not fall +// back to the un-deduplicated counts on a purchase run: that would buy +// reserved capacity the account already owns. A dry run logs a loud warning +// and continues with the un-deduplicated counts (nothing is bought, so +// reporting fidelity wins); a purchase run refuses to spend and returns +// ok=false, mirroring the fail-closed behavior of checkDuplicates on the +// non-CSV pipeline. +func checkDuplicatesForCSVRegion(ctx context.Context, recs []common.Recommendation, serviceClient provider.ServiceClient, service common.ServiceType, region string, isDryRun bool) (adjustedRecs []common.Recommendation, ok bool) { + adjustedRecs, err := adjustRecsForDuplicates(ctx, recs, serviceClient) + if err == nil { + return adjustedRecs, true + } + if ctx.Err() != nil { + // Interrupted: terminal in dry and real runs; runToolFromCSV stops on ctx.Err(). + return nil, false + } + if !isDryRun { + AppLogger.Printf(" āŒ Refusing to purchase %s/%s: could not check for existing RIs (%v). Skipping this region rather than risking a duplicate purchase.\n", getServiceDisplayName(service), region, err) + return nil, false + } + AppLogger.Printf(" āš ļø Warning: Could not check for existing RIs: %v (dry run; continuing with un-deduplicated counts)\n", err) + return recs, true +} + +// filterAndAdjustRecommendations applies filters, coverage, count override, +// the --min-count floor and the run-wide --max-instances cap to +// recommendations loaded from --input-csv. +// +// It returns an error when the cap cannot be enforced honestly; see +// requireRankingSignal. +func filterAndAdjustRecommendations(ctx context.Context, recs []common.Recommendation, csvModeCoverage float64, cfg Config) ([]common.Recommendation, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + engineData, err := engineVersionFetcher(ctx, cfg, !cfg.IncludeExtendedSupport && hasDatabaseRecs(recs)) + if err != nil { + return nil, err + } + if err := requireFilterInputs(recs, cfg, engineData); err != nil { + return nil, err + } + + // Apply filters (empty currentRegion since we're processing from CSV, not iterating regions). + // Drop tracking is skipped on the CSV path (nil drops). + originalCount := len(recs) + recs = applyFilters(recs, &cfg, engineData.instanceVersions, engineData.versionInfo, "", nil) + if len(recs) < originalCount { + AppLogger.Printf("šŸ” After filters: %d recs (filtered out %d)\n", len(recs), originalCount-len(recs)) + } + + // Apply sizing — target-coverage if set, otherwise coverage. + // Coverage 100% is a no-op (early-returned inside ApplyCoverage), but + // --target-coverage always applies even at coverage 100%, so the + // CSV-path short-circuit is conditional on TargetCoverage == 0. + if cfg.TargetCoverage > 0 || csvModeCoverage < 100 { + beforeSize := len(recs) + recs = applySizing(recs, cfg, csvModeCoverage, nil) + if cfg.TargetCoverage > 0 { + AppLogger.Printf("šŸŽÆ Applying %.1f%% target-coverage: %d recs selected (from %d)\n", cfg.TargetCoverage, len(recs), beforeSize) + } else { + AppLogger.Printf("šŸ“ˆ Applying %.1f%% coverage: %d recs selected (from %d)\n", csvModeCoverage, len(recs), beforeSize) + } + } + + // Apply count override if specified + if cfg.OverrideCount > 0 { + recs = ApplyCountOverride(recs, cfg.OverrideCount) + } + + // Enforce --min-count and the run-wide --max-instances cap. + return scoreAndLimitCSVRecs(recs, cfg) +} + +// requireFilterInputs fails when the data the filters need is missing: RDS +// inventory for the extended-support exclusion, and account names for +// name-based account filters. +func requireFilterInputs(recs []common.Recommendation, cfg Config, engineData engineVersionData) error { + if err := requireRegionInventory(recs, cfg, engineData); err != nil { + return err + } + return requireResolvedAccountNames(recs, &cfg) +} + +// processPurchaseLoop processes purchases for a single region (used by CSV +// mode). runID groups every recommendation processed across the whole CSV +// run into one audit trail, matching how executePurchasePipeline (the main +// pipeline) generates one runID per invocation. Confirmation is not asked +// here: prepareCSVPurchaseRun confirms once for the whole run before this +// loop is reached. +func processPurchaseLoop(ctx context.Context, recs []common.Recommendation, region string, isDryRun bool, serviceClient provider.ServiceClient, cfg Config, runID string) ([]common.PurchaseResult, error) { + results := make([]common.PurchaseResult, 0, len(recs)) + + for j := range recs { + if stopRequested(ctx) { + log.Printf("Shutdown requested; skipping %d remaining recommendation(s) in %s", len(recs)-j, region) + break + } + + if err := probeAuditLog(isDryRun, cfg.AuditLog); err != nil { + return results, err + } + + rec := recs[j] + label := purchaseRegionLabel(rec, region) + AppLogger.Printf(" [%d/%d] Processing: %s %s\n", j+1, len(recs), rec.Service, rec.ResourceType) + AppLogger.Printf(" šŸ’³ Purchasing %d instances\n", rec.Count) + + result, status, attempted := runOnePurchase(ctx, rec, label, region, j+1, isDryRun, serviceClient, cfg) + + // Audit first, then the rate-limit delay: a Ctrl-C during the delay + // must not lose the record of a purchase that already completed. + results = append(results, result) + auditErr := writePurchaseAuditRecord(runID, rec, result, status, isDryRun, cfg.AuditLog) + logPurchaseOutcome(result) + if auditErr != nil { + return results, auditErr + } + + // Rate-limit delay between real purchases; returns at once on Ctrl-C. + if attempted && j < len(recs)-1 { + waitBetweenPurchases(ctx) + } + } + + return results, nil +} + +// runOnePurchase validates and then dry-runs or executes one CSV-path +// recommendation. attempted reports whether a real purchase call was made. +func runOnePurchase(ctx context.Context, rec common.Recommendation, label, region string, index int, isDryRun bool, serviceClient provider.ServiceClient, cfg Config) (result common.PurchaseResult, status string, attempted bool) { + if err := validatePurchasePreconditions(rec, region, cfg.CallerAccountID); err != nil { + return preconditionFailure(rec, err, isDryRun), "error", false + } + if isDryRun { + return createDryRunResult(rec, label, index, cfg), "skipped", false + } + // Detached from Ctrl-C for the same reason as in executePurchasePipeline. + result = executePurchase(context.WithoutCancel(ctx), rec, label, index, serviceClient, cfg) + return result, purchaseAuditStatus(result), true +} + +func logPurchaseOutcome(result common.PurchaseResult) { + if purchaseSucceeded(result) { + AppLogger.Printf(" āœ… Success: %s\n", result.CommitmentID) + return + } + errMsg := "unknown error" + if result.Error != nil { + errMsg = result.Error.Error() + } + AppLogger.Printf(" āŒ Failed: %s\n", errMsg) +} + +// processCSVService runs every region of one service. A non-nil error is +// errAuditTrailLost: the partial results are still returned, and the caller +// stops the service loop as well. +func processCSVService(ctx context.Context, awsCfg aws.Config, service common.ServiceType, regionRecs map[string][]common.Recommendation, isDryRun bool, cfg Config, runID string) (serviceRecs []common.Recommendation, serviceResults []common.PurchaseResult, auditErr error) { + for region, recs := range regionRecs { + if stopRequested(ctx) { + log.Printf("Shutdown requested; skipping remaining regions for %s", getServiceDisplayName(service)) + break + } + + AppLogger.Printf("\n šŸ“ Region: %s (%d recommendations)\n", region, len(recs)) + + processedRecs, regionResults, ok, regionErr := processCSVRegionPurchases(ctx, awsCfg, service, region, recs, isDryRun, cfg, runID) + if !ok { + continue + } + serviceRecs = append(serviceRecs, processedRecs...) + serviceResults = append(serviceResults, regionResults...) + if regionErr != nil { + return serviceRecs, serviceResults, regionErr + } + } + return serviceRecs, serviceResults, nil +} + +// hasDatabaseRecs reports whether any recommendation is one the extended-support +// exclusion applies to (the same *common.DatabaseDetails test the adjuster uses). +func hasDatabaseRecs(recs []common.Recommendation) bool { + for i := range recs { + if d, ok := recs[i].Details.(*common.DatabaseDetails); ok && d != nil { + return true + } + } + return false +} diff --git a/cmd/multi_service_limit.go b/cmd/multi_service_limit.go new file mode 100644 index 000000000..7cd3bcda8 --- /dev/null +++ b/cmd/multi_service_limit.go @@ -0,0 +1,182 @@ +package main + +import ( + "fmt" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" +) + +// rankingRule names the ordering --max-instances consumes, so the cap can say +// which rule decided what survived. The two paths rank on different keys +// because they carry different data, and an operator reading the drop list has +// to know which one applied. Typed rather than a bare string so the call sites +// cannot invent a third wording that does not match any implemented ordering. +type rankingRule string + +const ( + // rankBySavingsPercentage is the recommendation-driven path: scorer.Score + // sorts on SavingsPercentage, an intensive rate, before the cap runs. + rankBySavingsPercentage rankingRule = "highest savings-percentage recommendations first" + // rankBySavingsPerInstance is the --input-csv path: the cap is a budget in + // instances, and a CSV may lack the savings percentage (older reports), so + // sortBySavingsPerInstance derives the rate from the EstimatedSavings and + // Count columns every CSV carries. + rankBySavingsPerInstance rankingRule = "highest savings-per-instance rows first (EstimatedSavings / Count, not savings percentage)" +) + +// capBinds reports whether --max-instances will actually remove instances from +// recs, rather than being unset or already satisfied. +// +// applyGlobalInstanceLimit and requireRankingSignal have to agree on this +// exactly: the second refuses precisely the runs whose outcome the first would +// otherwise decide on an ordering that carries no information. If the two +// conditions drift apart, either a run is refused for a cap that changes +// nothing, or an unrankable run reaches the cap after all. +func capBinds(recs []common.Recommendation, cfg Config) bool { + others := withoutSavingsPlans(recs) + return cfg.MaxInstances > 0 && CalculateTotalInstances(others) > int(cfg.MaxInstances) +} + +// applyGlobalInstanceLimit enforces --max-instances once across the entire run. +// +// The flag is documented as a hard cap on the total number of instances +// purchased across all recommendations, so it has to see every service and +// every region together. Applying it inside the per-region fetch instead caps +// each (service, region) pair independently and multiplies the operator's cap +// by the number of pairs. +// +// passed must already be ordered best-first by rule: ApplyInstanceLimit +// consumes the slice in order and drops the tail, so the ordering decides +// which commitments survive. scorer.Score guarantees the +// rankBySavingsPercentage ordering; scoreAndLimitCSVRecs establishes the +// rankBySavingsPerInstance one. +// +// Truncation can push a recommendation under --min-count, which is a hard +// floor rather than advice, so dropTruncatedBelowMinCount removes any such +// recommendation instead of purchasing it short. +// +// Nothing is truncated silently. Every reduced or dropped recommendation is +// named on stdout, and the drops are counted into the end-of-run summary. +func applyGlobalInstanceLimit(passed []common.Recommendation, cfg Config, rule rankingRule, drops *common.DropSummary) []common.Recommendation { + if !capBinds(passed, cfg) { + return passed + } + + // Savings Plans are exempt from the cap and the post-cap floor: their Count + // is a placeholder 1, not an instance count. The report and the floor see + // only the instance-based rows, whose capped result is a prefix of its input. + others := withoutSavingsPlans(passed) + totalBefore := CalculateTotalInstances(others) + limited := ApplyInstanceLimit(others, cfg.MaxInstances) + limited, belowMin := dropTruncatedBelowMinCount(limited, cfg.MinCount) + reportInstanceLimit(others, limited, len(belowMin), totalBefore, cfg.MaxInstances, rule, drops) + reportMinCountDrops(belowMin, cfg.MinCount, drops) + return mergeKeepingOrder(passed, limited) +} + +// dropTruncatedBelowMinCount removes recommendations that the cap truncated to +// fewer instances than --min-count allows, returning the survivors and the +// removed recommendations at their truncated counts. +// +// --min-count is a hard floor everywhere else in the codebase, never advice: +// the scorer rejects recommendations under it outright +// (scorer.filterReason, "count %d below minimum %d"), the scheduler's +// meetsMinCount drops them, and both `docs/cli/filtering.md` and +// `docs/cli/README.md` describe it as dropping recommendations below the +// number. filtering.md applies it to "the adjusted instance count (after +// coverage scaling)", so the floor is meant to gate the *sized* count, and +// truncation by --max-instances is another form of sizing. +// +// Buying a commitment smaller than the operator's stated minimum can be worse +// than buying nothing, which is the whole reason the floor exists, so a +// truncated recommendation is dropped rather than purchased short. The freed +// budget is deliberately not redistributed: the next recommendation would have +// to fit in an even smaller remainder and would fail the same floor. +// +// Removal is always from the tail. ApplyInstanceLimit reduces at most one +// recommendation (the one where the budget runs out, which is the last it +// keeps), and every earlier one still carries the full count that already +// cleared the scorer's floor. Taking only from the tail keeps the result a +// prefix of the input, which reportInstanceLimit relies on. +func dropTruncatedBelowMinCount(limited []common.Recommendation, minCount int) (kept, removed []common.Recommendation) { + if minCount <= 0 { + return limited, nil + } + kept = limited + for len(kept) > 0 && kept[len(kept)-1].Count < minCount { + removed = append(removed, kept[len(kept)-1]) + kept = kept[:len(kept)-1] + } + return kept, removed +} + +// reportMinCountDrops names each recommendation the --min-count floor rejected +// after --max-instances truncated it. It continues the reportInstanceLimit +// listing, where these already appear as dropped, and explains why they were +// not simply purchased at the reduced count. +func reportMinCountDrops(removed []common.Recommendation, minCount int, drops *common.DropSummary) { + for i := range removed { + rec := removed[i] + AppLogger.Printf(" ↳ %s %s %s: the cap left room for only %d instances, below --min-count %d, so it is dropped rather than purchased short\n", + rec.Service, rec.Region, rec.ResourceType, rec.Count, minCount) + } + drops.Add(common.DropMinCountAfterCap, len(removed)) +} + +// reportInstanceLimit prints what --max-instances removed from the run. +// after must be the prefix of before produced by ApplyInstanceLimit, so +// after[i] and before[i] describe the same recommendation. +// +// rule is the ordering the caller sorted before by, and is printed verbatim: +// the drop list is unreadable without knowing which key decided it, and the +// two paths do not rank on the same key. +// +// belowMinCount is how many of the missing entries were removed by the +// --min-count floor rather than by the budget. They are still listed here as +// dropped (they were), but they are attributed to --min-count-after-cap by +// reportMinCountDrops, so excluding them from this tally keeps each dropped +// recommendation counted exactly once in the end-of-run summary. +func reportInstanceLimit(before, after []common.Recommendation, belowMinCount, totalBefore int, maxInstances int32, rule rankingRule, drops *common.DropSummary) { + AppLogger.Printf("\nšŸ”’ --max-instances=%d caps the whole run: the %d recommendations that passed scoring total %d instances.\n", + maxInstances, len(before), totalBefore) + AppLogger.Printf(" Keeping the %s. The following are reduced or dropped:\n", rule) + + // after is a prefix of before, so membership is positional: every entry past + // len(after) was removed, whatever its Count (a zero-Count entry equals its + // own "kept" count of 0 and must still be counted as removed). + reduced, dropped := 0, 0 + for i := range before { + rec := before[i] + if i >= len(after) { + dropped++ + AppLogger.Printf(" • dropped: %s %s %s (%d instances)\n", rec.Service, rec.Region, rec.ResourceType, rec.Count) + continue + } + if kept := after[i].Count; kept != rec.Count { + reduced++ + AppLogger.Printf(" • reduced: %s %s %s %d → %d instances\n", rec.Service, rec.Region, rec.ResourceType, rec.Count, kept) + } + } + + // The floor only removes entries from the tail of what the cap kept, so + // they are a subset of the removed entries. DropSummary.Add panics on a + // negative count; surface the broken invariant with context instead. + budgetDropped := dropped - belowMinCount + if budgetDropped < 0 { + panic(fmt.Sprintf("reportInstanceLimit: %d recommendations attributed to --min-count-after-cap but only %d were removed (before=%d, after=%d)", + belowMinCount, dropped, len(before), len(after))) + } + drops.Add(common.DropMaxInstances, budgetDropped) + AppLogger.Printf(" Proceeding with %d instances across %d recommendations (%d reduced, %d dropped).\n", + CalculateTotalInstances(after), len(after), reduced, dropped) +} + +// sumPassedRecs returns total instance count and total estimated savings for passed recs. +func sumPassedRecs(recs []common.Recommendation) (total int, totalSavings float64) { + for _rvc := range recs { + r := recs[_rvc] + total += r.Count + totalSavings += r.EstimatedSavings + } + return +} diff --git a/cmd/multi_service_purchase.go b/cmd/multi_service_purchase.go new file mode 100644 index 000000000..f14bf97e9 --- /dev/null +++ b/cmd/multi_service_purchase.go @@ -0,0 +1,155 @@ +package main + +import ( + "context" + "fmt" + "log" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/aws/aws-sdk-go-v2/aws" +) + +// executePurchasePipeline purchases each rec in the passed list (or dry-runs) and writes audit records. +// It stops at the first audit failure on a real run and returns errAuditTrailLost. +func executePurchasePipeline(ctx context.Context, awsCfg aws.Config, recs []common.Recommendation, isDryRun bool, runID string, cfg Config) ([]common.PurchaseResult, error) { + results := make([]common.PurchaseResult, 0, len(recs)) + for i := range recs { + rec := recs[i] + if stopRequested(ctx) { + log.Printf("Shutdown requested — skipping %d remaining recommendations", len(recs)-i) + break + } + if err := probeAuditLog(isDryRun, cfg.AuditLog); err != nil { + return results, err + } + // The purchase runs on a context that survives Ctrl-C: canceling the + // request mid-flight can leave AWS having created the commitment + // while the CLI records a failure, and a re-run then buys it again. + // The per-call SDK timeouts still bound it. + result, status := purchaseSingleRec(context.WithoutCancel(ctx), awsCfg, rec, i+1, isDryRun, cfg) + results = append(results, result) + if err := writePurchaseAuditRecord(runID, rec, result, status, isDryRun, cfg.AuditLog); err != nil { + return results, err + } + if !isDryRun && i < len(recs)-1 { + waitBetweenPurchases(ctx) + } + } + return results, nil +} + +// writePurchaseAuditRecord writes a single purchase's audit record. Shared by +// both purchase entry points -- executePurchasePipeline (the main pipeline) +// and processPurchaseLoop (the --input-csv path) -- so every recommendation +// that reaches a purchase attempt, dry-run or real, is recorded to +// cfg.AuditLog regardless of which one produced it. Before #1609, +// processPurchaseLoop never wrote a record at all, so CSV-mode purchases left +// no audit trail: on a partial failure there was no durable, per-recommendation +// record of which rows succeeded, so an operator could only re-run the whole +// file, which is a double purchase for the rows that already succeeded. +// +// A failed write on a real run returns errAuditTrailLost (#2103): the caller +// must stop buying. The record may be missing or written but not durable +// (sync failed); both count as lost. A dry run only warns. +func writePurchaseAuditRecord(runID string, rec common.Recommendation, result common.PurchaseResult, status string, isDryRun bool, auditLogPath string) error { + auditRec := common.NewAuditRecord(runID, rec, result, status, isDryRun, common.PurchaseSourceCLI) + if err := writeAuditRecordFn(auditRec, auditLogPath); err != nil { + if isDryRun { + log.Printf("Warning: failed to write audit record: %v", err) + return nil + } + return fmt.Errorf("%w: write: %w", errAuditTrailLost, err) + } + return nil +} + +// probeAuditLog checks, before a real purchase, that the audit log can still +// be appended to, so a log that broke after the start-of-run check stops the +// run before money moves. Dry runs skip it. +func probeAuditLog(isDryRun bool, auditLogPath string) error { + if isDryRun { + return nil + } + if err := common.CheckAuditLogWritable(auditLogPath); err != nil { + return fmt.Errorf("%w: probe before purchase: %w", errAuditTrailLost, err) + } + return nil +} + +// newServiceClient is the service-client factory used by purchaseSingleRec. +// It is a seam so tests can drive the real purchase path with a fake provider. +var newServiceClient = createServiceClient + +// purchaseSucceeded reports whether a purchase result is a success: the +// provider set Success AND reported no error. A provider may return +// Success=true with a non-nil Error; that is a failure (issue #1900). +func purchaseSucceeded(result common.PurchaseResult) bool { + return result.Success && result.Error == nil +} + +// purchaseAuditStatus derives the audit status for a completed (non-dry-run) +// purchase attempt. Callers handle the dry-run ("skipped") case separately, +// since that never reaches a PurchaseResult from an actual API call. +func purchaseAuditStatus(result common.PurchaseResult) string { + if purchaseSucceeded(result) { + return "success" + } + return "error" +} + +// purchaseSingleRec executes or dry-runs a single purchase and returns the result + audit status. +func purchaseSingleRec(ctx context.Context, awsCfg aws.Config, rec common.Recommendation, index int, isDryRun bool, cfg Config) (purchaseResult common.PurchaseResult, auditStatus string) { + label := purchaseRegionLabel(rec, rec.Region) + AppLogger.Printf(" [%d] %s %s %s (count=%d)\n", index, rec.Service, label, rec.ResourceType, rec.Count) + if err := validatePurchasePreconditions(rec, rec.Region, cfg.CallerAccountID); err != nil { + AppLogger.Printf(" āŒ cannot purchase: %v\n", err) + return preconditionFailure(rec, err, isDryRun), "error" + } + if isDryRun { + result := createDryRunResult(rec, label, index, cfg) + AppLogger.Printf(" [dry-run] %s\n", result.CommitmentID) + return result, "skipped" + } + + clientRegion, err := clientRegionFor(rec.Service, rec.Region) + if err != nil { + AppLogger.Printf(" āŒ %v\n", err) + return common.PurchaseResult{Recommendation: rec, Success: false, Error: err, Timestamp: time.Now()}, "error" + } + regionalCfg := awsCfg.Copy() + regionalCfg.Region = clientRegion + serviceClient := newServiceClient(rec.Service, regionalCfg) + if serviceClient == nil { + AppLogger.Printf(" āš ļø No service client for %s\n", rec.Service) + return common.PurchaseResult{Success: false}, "error" + } + + result := executePurchase(ctx, rec, label, index, serviceClient, cfg) + status := purchaseAuditStatus(result) + if purchaseSucceeded(result) { + AppLogger.Printf(" āœ… %s\n", result.CommitmentID) + } else { + AppLogger.Printf(" āŒ %v\n", result.Error) + } + return result, status +} + +// buildServiceStats computes per-service statistics from a purchase run. +// Results are assumed to be in the same order as recs (1:1 correspondence). +func buildServiceStats(recs []common.Recommendation, results []common.PurchaseResult) map[common.ServiceType]ServiceProcessingStats { + byService := make(map[common.ServiceType][]common.Recommendation) + resultsByService := make(map[common.ServiceType][]common.PurchaseResult) + for i := range recs { + rec := recs[i] + byService[rec.Service] = append(byService[rec.Service], rec) + if i < len(results) { + resultsByService[rec.Service] = append(resultsByService[rec.Service], results[i]) + } + } + stats := make(map[common.ServiceType]ServiceProcessingStats) + for service, serviceRecs := range byService { + stats[service] = calculateServiceStats(service, serviceRecs, resultsByService[service]) + } + return stats +}