From a859738fa088209dfe18265754ac3c08fd3d2a47 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Mon, 12 Oct 2026 14:44:45 +0200 Subject: [PATCH 1/2] fix(cmd): stop the run when an audit record cannot be written A failed audit write on a real run now stops further purchases and exits 1 (errAuditTrailLost), on the main pipeline and --input-csv. The audit log is probed before each real purchase. Dry runs still only warn. Closes #2103 Co-Authored-By: Claude Sonnet 5.5 --- CHANGELOG.md | 9 ++ cmd/audit_stop_2103_test.go | 163 ++++++++++++++++++++ cmd/cancellation_test.go | 4 +- cmd/interrupt.go | 25 ++++ cmd/multi_service.go | 214 ++++++++++++++++++--------- cmd/multi_service_helpers.go | 4 +- cmd/multi_service_test.go | 14 +- cmd/purchase_safeguards_2121_test.go | 4 +- cmd/savings_plan_purchase_test.go | 8 +- 9 files changed, 359 insertions(+), 86 deletions(-) create mode 100644 cmd/audit_stop_2103_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index de0ade427..4e130edfc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/). ### Fixed +- A `--purchase` run now stops, and exits 1, when an audit record cannot be + written durably (disk full, log removed or made unwritable mid-run) instead + of warning and buying more with no audit trail. The audit log is also probed + before each real purchase, so a log that broke after the start-of-run check + stops the run before buying. The report and summary for what already ran are + still written, and the message says how many recommendations were attempted + and to verify in AWS before re-running. Applies to the main pipeline and + `--input-csv`; dry runs still only warn (#2103). + - Ctrl-C now cancels the whole invocation instead of only setting a flag that was polled between purchases. The engine-version lifecycle queries, region discovery, recommendation fetch, coverage fetch and duplicate checks treat a diff --git a/cmd/audit_stop_2103_test.go b/cmd/audit_stop_2103_test.go new file mode 100644 index 000000000..e43c5428f --- /dev/null +++ b/cmd/audit_stop_2103_test.go @@ -0,0 +1,163 @@ +package main + +import ( + "context" + "errors" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// spHook answers the Savings Plan calls of a successful purchase. +func spHook() *hookTransport { + h := &hookTransport{} + h.handle = func(r *http.Request) (int, string) { + switch { + case strings.HasSuffix(r.URL.Path, "/CreateSavingsPlan"): + return http.StatusOK, `{"savingsPlanId":"sp-111"}` + case strings.HasSuffix(r.URL.Path, "/DescribeSavingsPlans"): + return http.StatusOK, `{"savingsPlans":[]}` + } + return http.StatusBadRequest, `{"__type":"ValidationException","message":"unexpected call"}` + } + return h +} + +// failAuditWrites makes every audit write fail, as a full disk would. +func failAuditWrites(t *testing.T) { + t.Helper() + orig := writeAuditRecordFn + writeAuditRecordFn = func(common.AuditRecord, string) error { + return errors.New("no space left on device") + } + t.Cleanup(func() { writeAuditRecordFn = orig }) +} + +func auditCfg(t *testing.T) Config { + t.Helper() + dir := t.TempDir() + return Config{AuditLog: filepath.Join(dir, "audit.jsonl"), CSVOutput: filepath.Join(dir, "report.csv")} +} + +func TestExecuteAndReport_AuditWriteFailureStopsFurtherPurchases(t *testing.T) { + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + out := captureAppLog(t) + cfg := auditCfg(t) + failAuditWrites(t) + h := spHook() + recs := []common.Recommendation{offeringSP("o-1", 1), offeringSP("o-2", 2), offeringSP("o-3", 3)} + + err := executeAndReport(context.Background(), hookedAWSConfig(h), recs, false, "run-1", cfg, nil) + + require.ErrorIs(t, err, errAuditTrailLost) + assert.Equal(t, 1, h.count("/CreateSavingsPlan"), "no purchase may follow an unaudited one") + assert.Positive(t, h.count("/DescribeSavingsPlans"), "the duplicate check ran before the purchase") + assert.Contains(t, out.String(), "missing or not durable") + assert.Contains(t, out.String(), "1 of 3") + assert.Contains(t, out.String(), "sp-111") + report, rerr := os.ReadFile(cfg.CSVOutput) + require.NoError(t, rerr) + assert.Contains(t, string(report), "sp-111", "the unaudited purchase still reaches the report") +} + +func TestExecuteAndReport_AuditProbeFailureStopsBeforeBuying(t *testing.T) { + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + captureAppLog(t) + cfg := auditCfg(t) + // The seam writes record 1, then the log path becomes a directory, which + // fails the open with EISDIR even when the tests run as root. + orig := writeAuditRecordFn + writeAuditRecordFn = func(rec common.AuditRecord, path string) error { + if err := orig(rec, path); err != nil { + return err + } + if err := os.Rename(path, path+".1"); err != nil { + return err + } + return os.Mkdir(path, 0o755) + } + t.Cleanup(func() { writeAuditRecordFn = orig }) + h := spHook() + recs := []common.Recommendation{offeringSP("o-1", 1), offeringSP("o-2", 2)} + + err := executeAndReport(context.Background(), hookedAWSConfig(h), recs, false, "run-1", cfg, nil) + + require.ErrorIs(t, err, errAuditTrailLost) + assert.Contains(t, err.Error(), "probe") + assert.Equal(t, 1, h.count("/CreateSavingsPlan"), "purchase 2 must never be sent") + rows := auditRows(t, cfg.AuditLog+".1") + require.Len(t, rows, 1, "record 1 exists") +} + +func TestProcessPurchaseLoop_AuditWriteFailureStopsFurtherPurchases(t *testing.T) { + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + captureAppLog(t) + cfg := auditCfg(t) + failAuditWrites(t) + recs := []common.Recommendation{validEC2Rec(), validEC2Rec(), validEC2Rec()} + mockClient := &MockServiceClient{} + mockClient.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything). + Return(common.PurchaseResult{Success: true, CommitmentID: "ri-1"}, nil) + + results, err := processPurchaseLoop(context.Background(), recs, "us-east-1", false, mockClient, cfg, "run-1") + + require.ErrorIs(t, err, errAuditTrailLost) + require.Len(t, results, 1, "the unaudited result is still returned") + mockClient.AssertNumberOfCalls(t, "PurchaseCommitment", 1) +} + +func TestRunCSVPurchases_AuditWriteFailureStopsAcrossRegionsAndServices(t *testing.T) { + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + captureAppLog(t) + cfg := auditCfg(t) + failAuditWrites(t) + h := spHook() + ec2sp := offeringSP("o-3", 3) + ec2sp.Service = common.ServiceSavingsPlansEC2Instance + ec2sp.Details = &common.SavingsPlanDetails{PlanType: "EC2Instance", HourlyCommitment: 3, InstanceFamily: "m5", Region: "eu-west-1", OfferingID: "o-3"} + ec2sp2 := ec2sp + ec2sp2.Details = &common.SavingsPlanDetails{PlanType: "EC2Instance", HourlyCommitment: 4, InstanceFamily: "c5", Region: "eu-west-1", OfferingID: "o-4"} + recs := []common.Recommendation{offeringSP("o-1", 1), offeringSP("o-2", 2), ec2sp, ec2sp2} + + err := runCSVPurchases(context.Background(), hookedAWSConfig(h), recs, false, cfg, "run-1") + + require.ErrorIs(t, err, errAuditTrailLost) + assert.Equal(t, 1, h.count("/CreateSavingsPlan"), "whatever the map order, nothing is bought after the failure") +} + +func TestProcessCSVService_AuditWriteFailureStopsRegionLoop(t *testing.T) { + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + captureAppLog(t) + cfg := auditCfg(t) + failAuditWrites(t) + h := spHook() + byRegion := map[string][]common.Recommendation{ + "a": {offeringSP("o-1", 1)}, + "b": {offeringSP("o-2", 2)}, + } + + _, results, err := processCSVService(context.Background(), hookedAWSConfig(h), common.ServiceSavingsPlansCompute, byRegion, false, cfg, "run-1") + + require.ErrorIs(t, err, errAuditTrailLost) + assert.Len(t, results, 1) + assert.Equal(t, 1, h.count("/CreateSavingsPlan")) +} + +func TestAuditWriteFailureOnDryRunDoesNotStop(t *testing.T) { + captureAppLog(t) + cfg := auditCfg(t) + failAuditWrites(t) + recs := []common.Recommendation{validEC2Rec(), validEC2Rec(), validEC2Rec()} + + results, err := processPurchaseLoop(context.Background(), recs, "us-east-1", true, &MockServiceClient{}, cfg, "run-1") + + require.NoError(t, err) + assert.Len(t, results, 3) +} diff --git a/cmd/cancellation_test.go b/cmd/cancellation_test.go index e57825f19..edb67e668 100644 --- a/cmd/cancellation_test.go +++ b/cmd/cancellation_test.go @@ -197,7 +197,7 @@ func TestProcessPurchaseLoop_AuditRecordWrittenBeforeDelay(t *testing.T) { } t.Cleanup(func() { waitBetweenPurchases = orig }) - results := processPurchaseLoop(ctx, recs, "us-east-1", false, mockClient, cfg, "run-1") + results, _ := processPurchaseLoop(ctx, recs, "us-east-1", false, mockClient, cfg, "run-1") require.Len(t, rowsAtWait, 1, "the purchase that just completed must already be audited when the delay starts") require.Len(t, results, 1, "the second recommendation is not attempted after the interrupt") @@ -297,7 +297,7 @@ func TestProcessPurchaseLoop_CancelDuringPurchaseDoesNotCancelTheInFlightCall(t }). Return(common.PurchaseResult{Success: true, CommitmentID: "ri-1"}, nil) - results := processPurchaseLoop(ctx, recs, "us-east-1", false, mockClient, cfg, "run-1") + results, _ := processPurchaseLoop(ctx, recs, "us-east-1", false, mockClient, cfg, "run-1") require.NoError(t, ctxErrAtPurchase) require.Len(t, results, 1) diff --git a/cmd/interrupt.go b/cmd/interrupt.go index 136ee2194..7787b3348 100644 --- a/cmd/interrupt.go +++ b/cmd/interrupt.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "os" "time" @@ -46,6 +47,30 @@ func printInterruptedSummary(attempted, total int, results []common.PurchaseResu } } +// errAuditTrailLost marks a real run whose audit record could not be written +// durably (or whose log failed the pre-purchase probe). The run stops and +// exits non-zero: continuing would buy with no reconcilable trail (#2103). +var errAuditTrailLost = errors.New("audit trail lost") + +// writeAuditRecordFn is a seam so tests can fail a write after a purchase. +var writeAuditRecordFn = common.WriteAuditRecord + +// auditStopMessage tells the operator what ran before the audit trail was +// lost. The last result may be missing from, or not durable in, the log. +func auditStopMessage(attempted, total int, results []common.PurchaseResult, cause error) string { + last := "no purchase was attempted" + if n := len(results); n > 0 { + r := results[n-1] + last = fmt.Sprintf("last attempted: %s %s", r.Recommendation.Service, r.Recommendation.ResourceType) + if r.Success { + last += " (commitment " + r.CommitmentID + ")" + } else { + last += " (errored: " + fmt.Sprint(r.Error) + ")" + } + } + return fmt.Sprintf("\n❌ Audit record missing or not durable: stopping. %d of %d recommendation(s) attempted, the rest were not purchased; %s. Verify in AWS before re-running, a re-run may buy again. Cause: %v\n", attempted, total, last, cause) +} + func unverifiedPurchaseLine(r common.PurchaseResult) string { return fmt.Sprintf("purchase of %s %s errored (%v): verify in AWS before re-running", r.Recommendation.Service, r.Recommendation.ResourceType, r.Error) } diff --git a/cmd/multi_service.go b/cmd/multi_service.go index 5b6ed11e7..0dcd30931 100644 --- a/cmd/multi_service.go +++ b/cmd/multi_service.go @@ -202,7 +202,16 @@ func runToolMultiService(ctx context.Context, cfg Config) { } // Phases 3-4: confirm, purchase, and produce summary outputs. - runPurchaseAndReport(ctx, awsCfg, scoredResult, isDryRun, cfg, drops) + fatalOnError(runPurchaseAndReport(ctx, awsCfg, scoredResult, isDryRun, cfg, drops)) +} + +// fatalOnError exits 1 on a non-nil error. The report and summary were already +// written, so this is the last step of the run (untested glue, like +// runCSVPathOrFatal). +func fatalOnError(err error) { + if err != nil { + log.Fatalf("%v", err) + } } // mustFetchEngineVersionData is fetchEngineVersionData for the main pipeline: @@ -227,27 +236,31 @@ func exitIfInterrupted(ctx context.Context, phase string) { // runPurchaseAndReport handles the confirm, execute, and report phases of // the multi-service pipeline. It is a separate function to keep // runToolMultiService within the cyclomatic-complexity limit. -func runPurchaseAndReport(ctx context.Context, awsCfg aws.Config, scoredResult scorer.ScoredResult, isDryRun bool, cfg Config, drops *common.DropSummary) { +func runPurchaseAndReport(ctx context.Context, awsCfg aws.Config, scoredResult scorer.ScoredResult, isDryRun bool, cfg Config, drops *common.DropSummary) error { runID := uuid.New().String() if !confirmPurchaseRun(scoredResult.Passed, isDryRun) { printDropSummary(drops) AppLogger.Printf("\n❌ Purchase canceled.\n") - return + return nil } - executeAndReport(ctx, awsCfg, scoredResult.Passed, isDryRun, runID, cfg, drops) + return executeAndReport(ctx, awsCfg, scoredResult.Passed, isDryRun, runID, cfg, drops) } // executeAndReport runs the purchase pipeline and produces the report and // summary for whatever ran, interrupted or not. Split from // runPurchaseAndReport so the post-confirmation half is testable without a // terminal. -func executeAndReport(ctx context.Context, awsCfg aws.Config, passed []common.Recommendation, isDryRun bool, runID string, cfg Config, drops *common.DropSummary) { - allResults := executePurchasePipeline(ctx, awsCfg, passed, isDryRun, runID, cfg) +func executeAndReport(ctx context.Context, awsCfg aws.Config, passed []common.Recommendation, isDryRun bool, runID string, cfg Config, drops *common.DropSummary) error { + allResults, auditErr := executePurchasePipeline(ctx, awsCfg, passed, isDryRun, runID, cfg) writeReportAndSummary(passed, allResults, isDryRun, cfg, drops) if stopRequested(ctx) { printInterruptedSummary(len(allResults), len(passed), allResults, isDryRun) } + if auditErr != nil { + AppLogger.Print(auditStopMessage(len(allResults), len(passed), allResults, auditErr)) + } + return auditErr } // confirmPurchaseRun asks for confirmation once against the full @@ -485,7 +498,8 @@ func sumPassedRecs(recs []common.Recommendation) (total int, totalSavings float6 } // executePurchasePipeline purchases each rec in the passed list (or dry-runs) and writes audit records. -func executePurchasePipeline(ctx context.Context, awsCfg aws.Config, recs []common.Recommendation, isDryRun bool, runID string, cfg Config) []common.PurchaseResult { +// 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] @@ -493,18 +507,23 @@ func executePurchasePipeline(ctx context.Context, awsCfg aws.Config, recs []comm 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) - writePurchaseAuditRecord(runID, rec, result, status, isDryRun, cfg.AuditLog) + 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 + return results, nil } // writePurchaseAuditRecord writes a single purchase's audit record. Shared by @@ -516,11 +535,33 @@ func executePurchasePipeline(ctx context.Context, awsCfg aws.Config, recs []comm // 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. -func writePurchaseAuditRecord(runID string, rec common.Recommendation, result common.PurchaseResult, status string, isDryRun bool, auditLogPath string) { +// +// 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 := common.WriteAuditRecord(auditRec, auditLogPath); err != nil { - log.Printf("Warning: failed to write audit record: %v", err) + 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 } // purchaseAuditStatus derives the audit status for a completed (non-dry-run) @@ -680,6 +721,13 @@ func runToolFromCSV(ctx context.Context, cfg Config) error { return nil } + 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. +func runCSVPurchases(ctx context.Context, awsCfg aws.Config, recs []common.Recommendation, isDryRun bool, cfg Config, runID string) error { // Create account alias cache for lookup accountCache := NewAccountAliasCache(awsCfg) @@ -691,11 +739,12 @@ func runToolFromCSV(ctx context.Context, cfg Config) error { // Process purchases allResults := make([]common.PurchaseResult, 0) - serviceResults := 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) { @@ -703,31 +752,13 @@ func runToolFromCSV(ctx context.Context, cfg Config) error { break } - // Reset service results for each service - serviceResults = serviceResults[:0] - AppLogger.Printf("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n") AppLogger.Printf("🎯 Processing %s\n", getServiceDisplayName(service)) AppLogger.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n") - serviceRecs := make([]common.Recommendation, 0) - 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 := processCSVRegionPurchases(ctx, awsCfg, service, region, recs, isDryRun, cfg, runID) - if !ok { - continue - } - - serviceRecs = append(serviceRecs, processedRecs...) - allAdjustedRecs = append(allAdjustedRecs, processedRecs...) - serviceResults = append(serviceResults, regionResults...) - } + 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...) @@ -736,6 +767,9 @@ func runToolFromCSV(ctx context.Context, cfg Config) error { stats := calculateServiceStats(service, serviceRecs, serviceResults) serviceStats[service] = stats printServiceSummary(service, stats) + if auditErr != nil { + break + } } // Generate CSV filename and write report @@ -754,7 +788,10 @@ func runToolFromCSV(ctx context.Context, cfg Config) error { if stopRequested(ctx) { printInterruptedSummary(len(allResults), len(recs), allResults, isDryRun) } - return nil + if auditErr != nil { + AppLogger.Print(auditStopMessage(len(allResults), len(recs), allResults, auditErr)) + } + return auditErr } // processCSVRegionPurchases handles a single (service, region) pair within @@ -765,11 +802,11 @@ func runToolFromCSV(ctx context.Context, cfg Config) error { // 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) { +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 + return nil, nil, false, nil } regionalCfg := awsCfg.Copy() regionalCfg.Region = clientRegion @@ -778,13 +815,13 @@ func processCSVRegionPurchases(ctx context.Context, awsCfg aws.Config, service c 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 + 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 + return nil, nil, false, nil } // Deducting existing commitments shrinks Count, which can push a row // that cleared the floor in filterAndAdjustRecommendations back under it @@ -794,8 +831,8 @@ func processCSVRegionPurchases(ctx context.Context, awsCfg aws.Config, service c // to the pre-deduction counts. processedRecs = applyMinCountFloor(adjustedRecs, cfg.MinCount) - results = processPurchaseLoop(ctx, processedRecs, region, isDryRun, serviceClient, cfg, runID) - return processedRecs, results, true + results, auditErr = processPurchaseLoop(ctx, processedRecs, region, isDryRun, serviceClient, cfg, runID) + return processedRecs, results, true, auditErr } // checkDuplicatesForCSVRegion runs the duplicate check for a single @@ -914,6 +951,10 @@ func processService(ctx context.Context, awsCfg aws.Config, recClient provider.R ) serviceRecs = append(serviceRecs, regionResult.recommendations...) serviceResults = append(serviceResults, regionResult.results...) + if regionResult.auditErr != nil { + log.Printf("%s", auditStopMessage(len(serviceResults), len(serviceRecs), serviceResults, regionResult.auditErr)) + break + } } return serviceRecs, serviceResults @@ -925,7 +966,7 @@ func processService(ctx context.Context, awsCfg aws.Config, recClient provider.R // 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 { +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 { @@ -934,42 +975,24 @@ func processPurchaseLoop(ctx context.Context, recs []common.Recommendation, regi 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) - var result common.PurchaseResult - var status string - attempted := false - if err := validatePurchasePreconditions(rec, region); err != nil { - result = preconditionFailure(rec, err, isDryRun) - status = "error" - } else if isDryRun { - result = createDryRunResult(rec, label, j+1, cfg) - status = "skipped" - } else { - // Execute actual purchase. - attempted = true - // Detached from Ctrl-C for the same reason as in - // executePurchasePipeline. - result = executePurchase(context.WithoutCancel(ctx), rec, label, j+1, serviceClient, cfg) - status = purchaseAuditStatus(result) - } + 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. - writePurchaseAuditRecord(runID, rec, result, status, isDryRun, cfg.AuditLog) results = append(results, result) - - if result.Success { - AppLogger.Printf(" ✅ Success: %s\n", result.CommitmentID) - } else { - errMsg := "unknown error" - if result.Error != nil { - errMsg = result.Error.Error() - } - AppLogger.Printf(" ❌ Failed: %s\n", errMsg) + 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. @@ -978,5 +1001,56 @@ func processPurchaseLoop(ctx context.Context, recs []common.Recommendation, regi } } - return results + 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); 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 result.Success { + 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 } diff --git a/cmd/multi_service_helpers.go b/cmd/multi_service_helpers.go index a4d170744..8c39e009e 100644 --- a/cmd/multi_service_helpers.go +++ b/cmd/multi_service_helpers.go @@ -359,6 +359,7 @@ func queryMajorVersions(ctx context.Context, cfg Config) (map[string]MajorEngine type regionRecommendations struct { recommendations []common.Recommendation results []common.PurchaseResult + auditErr error // errAuditTrailLost: stop buying } // processRegionRecommendations fetches and processes recommendations for a single region. @@ -448,8 +449,9 @@ func processRegionRecommendations( // it processes still ends up in cfg.AuditLog with a durable, groupable // record; it is simply not grouped with a sibling region's run. runID := uuid.New().String() - regionResults := processPurchaseLoop(ctx, adjustedRecs, region, isDryRun, serviceClient, cfg, runID) + regionResults, auditErr := processPurchaseLoop(ctx, adjustedRecs, region, isDryRun, serviceClient, cfg, runID) result.results = regionResults + result.auditErr = auditErr return result } diff --git a/cmd/multi_service_test.go b/cmd/multi_service_test.go index 2d54e0873..ed5cef8aa 100644 --- a/cmd/multi_service_test.go +++ b/cmd/multi_service_test.go @@ -1282,7 +1282,7 @@ func TestProcessPurchaseLoopPurchaseFailure(t *testing.T) { t.Setenv("DISABLE_PURCHASE_DELAY", "true") - results := processPurchaseLoop(ctx, recs, "ap-south-1", false, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, recs, "ap-south-1", false, mockClient, toolCfg, "test-run") assert.Len(t, results, 1) assert.False(t, results[0].Success) @@ -1301,7 +1301,7 @@ func TestProcessPurchaseLoopEmptyRecommendations(t *testing.T) { mockClient := &MockServiceClient{} - results := processPurchaseLoop(ctx, []common.Recommendation{}, "us-east-1", false, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, []common.Recommendation{}, "us-east-1", false, mockClient, toolCfg, "test-run") assert.Empty(t, results) mockClient.AssertNotCalled(t, "PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything) @@ -1330,7 +1330,7 @@ func TestProcessServicePurchasesUserCancellation(t *testing.T) { t.Setenv("DISABLE_PURCHASE_DELAY", "true") - results := processPurchaseLoop(ctx, recs, "us-west-1", false, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, recs, "us-west-1", false, mockClient, toolCfg, "test-run") assert.Len(t, results, 1) assert.True(t, results[0].Success) @@ -1356,7 +1356,7 @@ func TestProcessServicePurchasesDryRunMultiple(t *testing.T) { mockClient := &MockServiceClient{} // Dry run should not call PurchaseCommitment - results := processPurchaseLoop(ctx, recs, "ap-northeast-1", true, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, recs, "ap-northeast-1", true, mockClient, toolCfg, "test-run") assert.Len(t, results, 3) for i, result := range results { @@ -1432,7 +1432,7 @@ func TestProcessPurchaseLoopDryRun(t *testing.T) { // Logger output disabled for testing - results := processPurchaseLoop(ctx, recs, "us-east-1", true, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, recs, "us-east-1", true, mockClient, toolCfg, "test-run") assert.Len(t, results, 2) for _, result := range results { @@ -1480,7 +1480,7 @@ func TestProcessPurchaseLoopActualPurchase(t *testing.T) { // Disable purchase delay for testing t.Setenv("DISABLE_PURCHASE_DELAY", "true") - results := processPurchaseLoop(ctx, recs, "eu-west-1", false, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, recs, "eu-west-1", false, mockClient, toolCfg, "test-run") assert.Len(t, results, 2) for i, result := range results { @@ -1525,7 +1525,7 @@ func TestProcessPurchaseLoopWithConfirmation(t *testing.T) { // Disable purchase delay for testing t.Setenv("DISABLE_PURCHASE_DELAY", "true") - results := processPurchaseLoop(ctx, recs, "us-west-2", false, mockClient, toolCfg, "test-run") + results, _ := processPurchaseLoop(ctx, recs, "us-west-2", false, mockClient, toolCfg, "test-run") assert.Len(t, results, 1) assert.True(t, results[0].Success) diff --git a/cmd/purchase_safeguards_2121_test.go b/cmd/purchase_safeguards_2121_test.go index c44f76eab..d676b75ab 100644 --- a/cmd/purchase_safeguards_2121_test.go +++ b/cmd/purchase_safeguards_2121_test.go @@ -243,14 +243,14 @@ func TestExecutePurchasePipeline_InterruptsFailClosed_2121(t *testing.T) { auditPath := filepath.Join(t.TempDir(), "audit.jsonl") shutdownRequested.Store(true) defer shutdownRequested.Store(false) - results := executePurchasePipeline(ctx, aws.Config{}, recs, true, "run-2121-int", Config{AuditLog: auditPath}) + results, _ := executePurchasePipeline(ctx, aws.Config{}, recs, true, "run-2121-int", Config{AuditLog: auditPath}) assert.Empty(t, results, "a requested shutdown must stop the pipeline before the first purchase") _, err := os.Stat(auditPath) assert.True(t, os.IsNotExist(err), "no audit records without purchase attempts") }) t.Run("uninterrupted dry run audits every rec as skipped", func(t *testing.T) { auditPath := filepath.Join(t.TempDir(), "audit.jsonl") - results := executePurchasePipeline(ctx, aws.Config{}, recs, true, "run-2121-dry", Config{AuditLog: auditPath}) + results, _ := executePurchasePipeline(ctx, aws.Config{}, recs, true, "run-2121-dry", Config{AuditLog: auditPath}) require.Len(t, results, len(recs)) for _, result := range results { assert.True(t, result.Success) diff --git a/cmd/savings_plan_purchase_test.go b/cmd/savings_plan_purchase_test.go index 1f98c3515..e8b3ce291 100644 --- a/cmd/savings_plan_purchase_test.go +++ b/cmd/savings_plan_purchase_test.go @@ -189,7 +189,7 @@ func TestProcessCSVRegionPurchases_SavingsPlanRowWithEmptyRegionAndProvider(t *t Details: &common.SavingsPlanDetails{PlanType: "Compute", HourlyCommitment: 1.5}, } - _, results, ok := processCSVRegionPurchases(context.Background(), awsCfg, rec.Service, "", []common.Recommendation{rec}, false, cfg, "run-1") + _, results, ok, _ := processCSVRegionPurchases(context.Background(), awsCfg, rec.Service, "", []common.Recommendation{rec}, false, cfg, "run-1") require.True(t, ok) require.Len(t, results, 1) @@ -308,13 +308,13 @@ func TestProcessPurchaseLoop_DryRunReportsPreconditionFailureOnCSVShapedRows(t * csvEC2.Details = &common.ComputeDetails{InstanceType: "m5.large", Platform: "Linux/UNIX"} // loader leaves Tenancy/Scope blank okSP := parserShapedSP() - results := processPurchaseLoop(context.Background(), []common.Recommendation{noDetailsSP, okSP}, "", true, nil, cfg, "run-1") + results, _ := processPurchaseLoop(context.Background(), []common.Recommendation{noDetailsSP, okSP}, "", true, nil, cfg, "run-1") require.Len(t, results, 2) assert.False(t, results[0].Success) assert.True(t, results[1].Success) assert.Contains(t, results[1].CommitmentID, "global") - results = processPurchaseLoop(context.Background(), []common.Recommendation{csvEC2}, "us-east-1", true, nil, cfg, "run-2") + results, _ = processPurchaseLoop(context.Background(), []common.Recommendation{csvEC2}, "us-east-1", true, nil, cfg, "run-2") require.Len(t, results, 1) assert.False(t, results[0].Success) } @@ -393,7 +393,7 @@ func TestProcessCSVRegionPurchases_SavingsPlanRowFromWrittenCSV(t *testing.T) { require.Empty(t, recs[0].Region) require.Empty(t, recs[0].Provider) - _, results, ok := processCSVRegionPurchases(context.Background(), awsCfg, recs[0].Service, recs[0].Region, recs, false, cfg, "run-1") + _, results, ok, _ := processCSVRegionPurchases(context.Background(), awsCfg, recs[0].Service, recs[0].Region, recs, false, cfg, "run-1") require.True(t, ok) require.Len(t, results, 1) From a76c85e32476de729dac0e4b71978c431168a26d Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Mon, 12 Oct 2026 16:20:36 +0200 Subject: [PATCH 2/2] test(cmd): pin CSV audit-stop probe, partial-service report and commitment id Co-Authored-By: Claude Sonnet 5.5 --- cmd/audit_stop_2103_test.go | 65 ++++++++++++++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) diff --git a/cmd/audit_stop_2103_test.go b/cmd/audit_stop_2103_test.go index e43c5428f..bab5c0ca0 100644 --- a/cmd/audit_stop_2103_test.go +++ b/cmd/audit_stop_2103_test.go @@ -61,7 +61,7 @@ func TestExecuteAndReport_AuditWriteFailureStopsFurtherPurchases(t *testing.T) { assert.Positive(t, h.count("/DescribeSavingsPlans"), "the duplicate check ran before the purchase") assert.Contains(t, out.String(), "missing or not durable") assert.Contains(t, out.String(), "1 of 3") - assert.Contains(t, out.String(), "sp-111") + assert.Contains(t, out.String(), "(commitment sp-111)") report, rerr := os.ReadFile(cfg.CSVOutput) require.NoError(t, rerr) assert.Contains(t, string(report), "sp-111", "the unaudited purchase still reaches the report") @@ -161,3 +161,66 @@ func TestAuditWriteFailureOnDryRunDoesNotStop(t *testing.T) { require.NoError(t, err) assert.Len(t, results, 3) } + +// lockAuditAfterFirstPurchase makes the first CreateSavingsPlan call revoke +// write access to the audit file and its directory, the real-shape failure. +func lockAuditAfterFirstPurchase(t *testing.T, h *hookTransport, auditPath string) { + t.Helper() + inner := h.handle + h.handle = func(r *http.Request) (int, string) { + if strings.HasSuffix(r.URL.Path, "/CreateSavingsPlan") { + _ = os.Chmod(auditPath, 0o400) + _ = os.Chmod(filepath.Dir(auditPath), 0o500) + t.Cleanup(func() { _ = os.Chmod(filepath.Dir(auditPath), 0o700) }) + } + return inner(r) + } +} + +func TestRunCSVPurchases_AuditStopKeepsUnauditedPurchaseInReport(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("read-only permissions do not bind root") + } + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + out := captureAppLog(t) + cfg := auditCfg(t) + cfg.CSVOutput = filepath.Join(t.TempDir(), "report.csv") + h := spHook() + lockAuditAfterFirstPurchase(t, h, cfg.AuditLog) + recs := []common.Recommendation{offeringSP("o-1", 1), offeringSP("o-2", 2), offeringSP("o-3", 3)} + + err := runCSVPurchases(context.Background(), hookedAWSConfig(h), recs, false, cfg, "run-1") + + require.ErrorIs(t, err, errAuditTrailLost) + assert.Equal(t, 1, h.count("/CreateSavingsPlan")) + report, rerr := os.ReadFile(cfg.CSVOutput) + require.NoError(t, rerr) + assert.Contains(t, string(report), "sp-111", "the unaudited purchase must stay in the CSV report") + assert.Contains(t, out.String(), "1 of 3") + assert.Contains(t, out.String(), "(commitment sp-111)") +} + +func TestRunCSVPurchases_AuditProbeFailureStopsBeforeBuying(t *testing.T) { + t.Setenv("DISABLE_PURCHASE_DELAY", "true") + captureAppLog(t) + cfg := auditCfg(t) + orig := writeAuditRecordFn + writeAuditRecordFn = func(rec common.AuditRecord, path string) error { + if err := orig(rec, path); err != nil { + return err + } + if err := os.Rename(path, path+".1"); err != nil { + return err + } + return os.Mkdir(path, 0o755) + } + t.Cleanup(func() { writeAuditRecordFn = orig }) + h := spHook() + recs := []common.Recommendation{offeringSP("o-1", 1), offeringSP("o-2", 2)} + + err := runCSVPurchases(context.Background(), hookedAWSConfig(h), recs, false, cfg, "run-1") + + require.ErrorIs(t, err, errAuditTrailLost) + assert.Contains(t, err.Error(), "probe") + assert.Equal(t, 1, h.count("/CreateSavingsPlan"), "purchase 2 must never be sent") +}