refactor(aws): move the capped Cost Explorer loops onto the shared pagination helper

providers/aws/recommendations/ce_paginate_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/costexplorer"
+	"github.com/aws/aws-sdk-go-v2/service/costexplorer/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -101,8 +102,9 @@ const endlessCeiling = 500
 // utilization call.
 type endlessCE struct {
 	mockCostExplorerAPI
-	covCalls  atomic.Int32
-	utilCalls atomic.Int32
+	covCalls                          atomic.Int32
+	utilCalls                         atomic.Int32
+	spCovCalls, costCalls, spRecCalls atomic.Int32
 }
 
 func (m *endlessCE) GetReservationCoverage(_ context.Context, _ *costexplorer.GetReservationCoverageInput, _ ...func(*costexplorer.Options)) (*costexplorer.GetReservationCoverageOutput, error) {
@@ -163,3 +165,115 @@ func TestGetDailyUsagePcts_PreCanceledContext(t *testing.T) {
 	assert.ErrorIs(t, err, context.Canceled)
 	assert.Zero(t, mock.covCalls.Load())
 }
+
+func (m *endlessCE) GetSavingsPlansCoverage(_ context.Context, _ *costexplorer.GetSavingsPlansCoverageInput, _ ...func(*costexplorer.Options)) (*costexplorer.GetSavingsPlansCoverageOutput, error) {
+	n := m.spCovCalls.Add(1)
+	if n > endlessCeiling {
+		return nil, errors.New("mock: runaway pagination")
+	}
+	return &costexplorer.GetSavingsPlansCoverageOutput{NextToken: aws.String(fmt.Sprintf("t%d", n))}, nil
+}
+
+func (m *endlessCE) GetCostAndUsage(_ context.Context, _ *costexplorer.GetCostAndUsageInput, _ ...func(*costexplorer.Options)) (*costexplorer.GetCostAndUsageOutput, error) {
+	n := m.costCalls.Add(1)
+	if n > endlessCeiling {
+		return nil, errors.New("mock: runaway pagination")
+	}
+	return &costexplorer.GetCostAndUsageOutput{NextPageToken: aws.String(fmt.Sprintf("t%d", n))}, nil
+}
+
+func (m *endlessCE) GetSavingsPlansPurchaseRecommendation(_ context.Context, _ *costexplorer.GetSavingsPlansPurchaseRecommendationInput, _ ...func(*costexplorer.Options)) (*costexplorer.GetSavingsPlansPurchaseRecommendationOutput, error) {
+	n := m.spRecCalls.Add(1)
+	if n > endlessCeiling {
+		return nil, errors.New("mock: runaway pagination")
+	}
+	return &costexplorer.GetSavingsPlansPurchaseRecommendationOutput{
+		SavingsPlansPurchaseRecommendation: &types.SavingsPlansPurchaseRecommendation{},
+		NextPageToken:                      aws.String(fmt.Sprintf("t%d", n)),
+	}, nil
+}
+
+func TestGetSPCoveredSeries_CapsEndlessTokens(t *testing.T) {
+	assert.Equal(t, 20, maxSPCoveragePages)
+	mock := &endlessCE{}
+	_, err := NewClientWithAPI(mock, "us-east-1").GetSPCoveredSeries(context.Background(), "us-east-1", 7)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "sp covered series: pagination cap reached after 20 pages")
+	assert.Equal(t, int32(20), mock.spCovCalls.Load())
+}
+
+func TestGetOnDemandSeries_CapsEndlessTokens(t *testing.T) {
+	assert.Equal(t, 20, maxOnDemandSeriesPages)
+	mock := &endlessCE{}
+	_, err := NewClientWithAPI(mock, "us-east-1").GetOnDemandSeries(context.Background(), "us-east-1", 7)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "GetOnDemandSeries: pagination cap reached after 20 pages")
+	assert.Equal(t, int32(20), mock.costCalls.Load())
+}
+
+func TestGetSPCommitmentRate_CapsEndlessTokens(t *testing.T) {
+	assert.Equal(t, 20, maxRecommendationPages)
+	mock := &endlessCE{}
+	_, err := NewClientWithAPI(mock, "us-east-1").GetSPCommitmentRate(context.Background(), rateParams(types.SupportedSavingsPlansTypeComputeSp))
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "sp commitment rate: pagination cap reached after 20 pages")
+	assert.Equal(t, int32(20), mock.spRecCalls.Load())
+}
+
+func TestGetSPCommitmentRate_PreCanceledContext(t *testing.T) {
+	mock := &endlessCE{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	_, err := NewClientWithAPI(mock, "us-east-1").GetSPCommitmentRate(ctx, rateParams(types.SupportedSavingsPlansTypeComputeSp))
+	assert.Equal(t, context.Canceled, err)
+	assert.Zero(t, mock.spRecCalls.Load())
+}
+
+// scriptedCE serves SP recommendation pages from a script and fails
+// GetCostAndUsage with costErr.
+type scriptedCE struct {
+	mockCostExplorerAPI
+	spPages []*costexplorer.GetSavingsPlansPurchaseRecommendationOutput
+	spErr   error
+	costErr error
+	spCalls int
+}
+
+func (m *scriptedCE) GetSavingsPlansPurchaseRecommendation(_ context.Context, _ *costexplorer.GetSavingsPlansPurchaseRecommendationInput, _ ...func(*costexplorer.Options)) (*costexplorer.GetSavingsPlansPurchaseRecommendationOutput, error) {
+	if m.spErr != nil {
+		return nil, m.spErr
+	}
+	out := m.spPages[m.spCalls]
+	m.spCalls++
+	return out, nil
+}
+
+func (m *scriptedCE) GetCostAndUsage(_ context.Context, _ *costexplorer.GetCostAndUsageInput, _ ...func(*costexplorer.Options)) (*costexplorer.GetCostAndUsageOutput, error) {
+	return nil, m.costErr
+}
+
+func TestGetSPCommitmentRate_AbsentRecommendationReportsPageIndex(t *testing.T) {
+	mock := &scriptedCE{spPages: []*costexplorer.GetSavingsPlansPurchaseRecommendationOutput{
+		{SavingsPlansPurchaseRecommendation: &types.SavingsPlansPurchaseRecommendation{}, NextPageToken: aws.String("t1")},
+		{},
+	}}
+	_, err := NewClientWithAPI(mock, "us-east-1").GetSPCommitmentRate(context.Background(), rateParams(types.SupportedSavingsPlansTypeComputeSp))
+	require.ErrorIs(t, err, ErrNoSPCommitmentRate)
+	assert.Contains(t, err.Error(), "page 1")
+}
+
+func TestGetSPCommitmentRate_FetchErrorKeepsPrefix(t *testing.T) {
+	boom := errors.New("boom")
+	mock := &scriptedCE{spErr: boom}
+	_, err := NewClientWithAPI(mock, "us-east-1").GetSPCommitmentRate(context.Background(), rateParams(types.SupportedSavingsPlansTypeComputeSp))
+	require.ErrorIs(t, err, boom)
+	assert.Contains(t, err.Error(), "sp commitment rate: ")
+}
+
+func TestGetOnDemandSeries_FetchErrorKeepsPrefix(t *testing.T) {
+	boom := errors.New("boom")
+	mock := &scriptedCE{costErr: boom}
+	_, err := NewClientWithAPI(mock, "us-east-1").GetOnDemandSeries(context.Background(), "us-east-1", 7)
+	require.ErrorIs(t, err, boom)
+	assert.Contains(t, err.Error(), "GetOnDemandSeries: ")
+}

providers/aws/recommendations/client.go

@@ -187,33 +187,18 @@ func (c *Client) fetchRIAllPages(
 	service common.ServiceType,
 ) ([]types.ReservationPurchaseRecommendation, error) {
 	var allRecs []types.ReservationPurchaseRecommendation
-	var nextPageToken *string
-
-	for pageIdx := 0; ; pageIdx++ {
-		if err := ctx.Err(); err != nil {
-			return nil, err
-		}
-		if pageIdx >= maxRecommendationPages {
-			return nil, fmt.Errorf(
-				"pagination cap reached after %d pages for RI %s (issue #692)",
-				maxRecommendationPages, service,
-			)
-		}
-		input.NextPageToken = nextPageToken
-
+	err := paginateCE(ctx, maxRecommendationPages, "RI "+string(service), func(token *string) (*string, error) {
+		input.NextPageToken = token
 		result, err := c.fetchRIPageWithRetry(ctx, input)
 		if err != nil {
 			return nil, err
 		}
-
 		allRecs = append(allRecs, result.Recommendations...)
-
-		if result.NextPageToken == nil || aws.ToString(result.NextPageToken) == "" {
-			break
-		}
-		nextPageToken = result.NextPageToken
+		return result.NextPageToken, nil
+	})
+	if err != nil {
+		return nil, err
 	}
-
 	return allRecs, nil
 }
 

providers/aws/recommendations/ondemand_series.go

@@ -135,23 +135,12 @@ func (c *Client) getOnDemandSeries(ctx context.Context, accountID, region string
 
 	byDate := make(map[string]float64)
 
-	var nextToken *string
-	page := 0
-	for {
-		if err := ctx.Err(); err != nil {
-			return nil, fmt.Errorf("GetOnDemandSeries: context canceled during pagination: %w", err)
-		}
-		page++
-		if page > maxOnDemandSeriesPages {
-			return nil, fmt.Errorf("GetOnDemandSeries: exceeded %d page cap; possible CE token loop", maxOnDemandSeriesPages)
-		}
-		input.NextPageToken = nextToken
-
+	err := paginateCE(ctx, maxOnDemandSeriesPages, "GetOnDemandSeries", func(token *string) (*string, error) {
+		input.NextPageToken = token
 		out, err := c.fetchOnDemandPage(ctx, input)
 		if err != nil {
 			return nil, fmt.Errorf("GetOnDemandSeries: %w", err)
 		}
-
 		if accountID != "" {
 			err = accumulateScopedDailyResults(byDate, out, start, end)
 		} else {
@@ -160,13 +149,10 @@ func (c *Client) getOnDemandSeries(ctx context.Context, accountID, region string
 		if err != nil {
 			return nil, fmt.Errorf("GetOnDemandSeries: %w", err)
 		}
-
-		// aws.ToString returns "" for nil, so this single check covers both nil
-		// and empty-string tokens.
-		if aws.ToString(out.NextPageToken) == "" {
-			break
-		}
-		nextToken = out.NextPageToken
+		return out.NextPageToken, nil
+	})
+	if err != nil {
+		return nil, err
 	}
 
 	return buildDailySeries(byDate, region, lookbackDays)

providers/aws/recommendations/sp_commitment_rate.go

@@ -81,35 +81,35 @@ func (c *Client) GetSPCommitmentRate(ctx context.Context, p SPCommitmentRatePara
 // sumSPRateDetails pages through the recommendation and sums the used SP spend
 // and covered on-demand spend of every matching detail.
 func (c *Client) sumSPRateDetails(ctx context.Context, input *costexplorer.GetSavingsPlansPurchaseRecommendationInput, p *SPCommitmentRateParams) (spCost, covered float64, matched int, err error) {
-	for page := 0; ; page++ {
-		if page >= maxRecommendationPages {
-			return 0, 0, 0, fmt.Errorf("sp commitment rate: pagination cap reached after %d pages", maxRecommendationPages)
-		}
+	page := 0
+	err = paginateCE(ctx, maxRecommendationPages, "sp commitment rate", func(token *string) (*string, error) {
+		input.NextPageToken = token
 		out, fetchErr := c.fetchSPPageWithRetry(ctx, input)
 		if fetchErr != nil {
-			return 0, 0, 0, fmt.Errorf("sp commitment rate: %w", fetchErr)
+			return nil, fmt.Errorf("sp commitment rate: %w", fetchErr)
 		}
 		if out == nil || out.SavingsPlansPurchaseRecommendation == nil {
-			return 0, 0, 0, fmt.Errorf("%w: recommendation absent on page %d", ErrNoSPCommitmentRate, page)
+			return nil, fmt.Errorf("%w: recommendation absent on page %d", ErrNoSPCommitmentRate, page)
 		}
+		page++
 		rec := out.SavingsPlansPurchaseRecommendation
 		for i := range rec.SavingsPlansPurchaseRecommendationDetails {
 			d := &rec.SavingsPlansPurchaseRecommendationDetails[i]
 			if !spRateDetailMatches(d, p) {
 				continue
 			}
-			used, cov, err := spRateTerms(d)
-			if err != nil {
-				return 0, 0, 0, err
+			used, cov, termErr := spRateTerms(d)
+			if termErr != nil {
+				return nil, termErr
 			}
 			spCost += used
 			covered += cov
 			matched++
 		}
-		input.NextPageToken = out.NextPageToken
-		if aws.ToString(input.NextPageToken) == "" {
-			break
-		}
+		return out.NextPageToken, nil
+	})
+	if err != nil {
+		return 0, 0, 0, err
 	}
 	return spCost, covered, matched, nil
 }

providers/aws/recommendations/sp_coverage.go

@@ -351,30 +351,21 @@ func (c *Client) GetSPCoverageSummary(ctx context.Context, region string, lookba
 	}
 
 	var acc spCoverageAccumulator
-	var token *string
-	for pageIdx := 0; ; pageIdx++ {
-		if err := ctx.Err(); err != nil {
-			return SPCoverageSummary{}, fmt.Errorf("sp coverage: pagination canceled: %w", err)
-		}
-		if pageIdx >= maxSPCoveragePages {
-			return SPCoverageSummary{}, fmt.Errorf(
-				"sp coverage: pagination cap reached after %d pages (mirrors the issue #692 guard)",
-				maxSPCoveragePages)
-		}
+	err := paginateCE(ctx, maxSPCoveragePages, "sp coverage", func(token *string) (*string, error) {
 		input.NextToken = token
 		result, err := c.fetchSPCoveragePage(ctx, input)
 		if err != nil {
-			return SPCoverageSummary{}, err
+			return nil, err
 		}
 		for _, cov := range result.SavingsPlansCoverages {
 			if addErr := acc.add(cov); addErr != nil {
-				return SPCoverageSummary{}, fmt.Errorf("sp coverage: %w", addErr)
+				return nil, fmt.Errorf("sp coverage: %w", addErr)
 			}
 		}
-		if result.NextToken == nil || *result.NextToken == "" {
-			break
-		}
-		token = result.NextToken
+		return result.NextToken, nil
+	})
+	if err != nil {
+		return SPCoverageSummary{}, err
 	}
 
 	windowHours := float64(lookbackDays * 24)

providers/aws/recommendations/sp_coverage_series.go

@@ -90,14 +90,7 @@ func (c *Client) GetSPCoveredSeries(ctx context.Context, region string, lookback
 // USD/hr keyed by "YYYY-MM-DD".
 func (c *Client) fetchSPCoveredByDate(ctx context.Context, input *costexplorer.GetSavingsPlansCoverageInput) (map[string]float64, error) {
 	byDate := make(map[string]float64)
-	var token *string
-	for pageIdx := 0; ; pageIdx++ {
-		if err := ctx.Err(); err != nil {
-			return nil, fmt.Errorf("sp covered series: pagination canceled: %w", err)
-		}
-		if pageIdx >= maxSPCoveragePages {
-			return nil, fmt.Errorf("sp covered series: pagination cap reached after %d pages", maxSPCoveragePages)
-		}
+	err := paginateCE(ctx, maxSPCoveragePages, "sp covered series", func(token *string) (*string, error) {
 		input.NextToken = token
 		result, err := c.fetchSPCoveragePage(ctx, input)
 		if err != nil {
@@ -106,11 +99,12 @@ func (c *Client) fetchSPCoveredByDate(ctx context.Context, input *costexplorer.G
 		if err := accumulateSPCoveredDays(byDate, result.SavingsPlansCoverages); err != nil {
 			return nil, fmt.Errorf("sp covered series: %w", err)
 		}
-		if result.NextToken == nil || *result.NextToken == "" {
-			return byDate, nil
-		}
-		token = result.NextToken
+		return result.NextToken, nil
+	})
+	if err != nil {
+		return nil, err
 	}
+	return byDate, nil
 }
 
 // sortedDailyCosts orders a date-keyed map oldest-to-newest (lexicographic