diff --git a/CHANGELOG.md b/CHANGELOG.md index 620b2109..5f448ea0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -133,6 +133,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/). ### Changed +- **`aws/recommendations`:** the four Cost Explorer coverage and utilization page fetchers share one generic `fetchWithRetry` helper (rate limiter plus concurrency cap); behaviour and error messages are unchanged (#120). - **`aws/rds`:** the RDS purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged (#10). - **`aws/memorydb`:** the MemoryDB purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged; only the lookup-failure error wording changes to "before MemoryDB reservation purchase" (#10). - **`aws/opensearch`:** the OpenSearch purchase path uses the shared `purchaseguard` idempotency guard and AlreadyExists recovery, like ElastiCache. Behavior for valid purchases is unchanged (#10). diff --git a/providers/aws/recommendations/coverage.go b/providers/aws/recommendations/coverage.go index 53ad2b41..0aa206fc 100644 --- a/providers/aws/recommendations/coverage.go +++ b/providers/aws/recommendations/coverage.go @@ -12,7 +12,6 @@ import ( "github.com/aws/aws-sdk-go-v2/service/costexplorer/types" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" - "github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency" ) // coverageServiceFilters lists the Cost Explorer service-dimension values @@ -384,23 +383,9 @@ func serviceRegionFilter(service, region string) *types.Expression { // held only across the outbound SDK call, matching fetchRIPageWithRetry, // so the coverage-enrichment fan-out stays inside the global IO cap. func (c *Client) fetchCoveragePage(ctx context.Context, input *costexplorer.GetReservationCoverageInput) (*costexplorer.GetReservationCoverageOutput, error) { - rateLimiter := c.rateLimiter.newOperation() - for { - if waitErr := rateLimiter.Wait(ctx); waitErr != nil { - return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr) - } - if acqErr := concurrency.Acquire(ctx); acqErr != nil { - return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr) - } - result, err := c.costExplorerClient.GetReservationCoverage(ctx, input) - concurrency.Release(ctx) - if !rateLimiter.ShouldRetry(err) { - if err != nil { - return nil, fmt.Errorf("failed to get reservation coverage: %w", err) - } - return result, nil - } - } + return fetchWithRetry(ctx, c.rateLimiter, "failed to get reservation coverage", func() (*costexplorer.GetReservationCoverageOutput, error) { + return c.costExplorerClient.GetReservationCoverage(ctx, input) + }) } // poolCoverageFromGroup pulls the (pct, avg) pair from a CE response diff --git a/providers/aws/recommendations/fetch_retry.go b/providers/aws/recommendations/fetch_retry.go new file mode 100644 index 00000000..30eb9def --- /dev/null +++ b/providers/aws/recommendations/fetch_retry.go @@ -0,0 +1,33 @@ +package recommendations + +import ( + "context" + "fmt" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency" +) + +// fetchWithRetry runs one Cost Explorer call under the shared rate limiter +// and the global concurrency cap, retrying while the limiter classifies the +// error as retryable. The semaphore is held only across the SDK call, so +// back-off waits never occupy a slot. A terminal SDK error is wrapped with +// errPrefix; limiter and semaphore failures carry their own messages. +func fetchWithRetry[T any](ctx context.Context, limiter *RateLimiter, errPrefix string, call func() (*T, error)) (*T, error) { + op := limiter.newOperation() + for { + if waitErr := op.Wait(ctx); waitErr != nil { + return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr) + } + if acqErr := concurrency.Acquire(ctx); acqErr != nil { + return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr) + } + result, err := call() + concurrency.Release(ctx) + if !op.ShouldRetry(err) { + if err != nil { + return nil, fmt.Errorf("%s: %w", errPrefix, err) + } + return result, nil + } + } +} diff --git a/providers/aws/recommendations/fetch_retry_test.go b/providers/aws/recommendations/fetch_retry_test.go new file mode 100644 index 00000000..629718a1 --- /dev/null +++ b/providers/aws/recommendations/fetch_retry_test.go @@ -0,0 +1,66 @@ +package recommendations + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func fastLimiter(maxRetries int) *RateLimiter { + return NewRateLimiterWithOptions(time.Millisecond, time.Millisecond, maxRetries) +} + +func TestFetchWithRetry_RetriesThrottleThenSucceeds(t *testing.T) { + calls := 0 + got, err := fetchWithRetry(context.Background(), fastLimiter(3), "boom", func() (*string, error) { + calls++ + if calls < 3 { + return nil, newThrottleError() + } + v := "ok" + return &v, nil + }) + require.NoError(t, err) + assert.Equal(t, "ok", *got) + assert.Equal(t, 3, calls) +} + +func TestFetchWithRetry_WrapsTerminalError(t *testing.T) { + sentinel := errors.New("access denied") + calls := 0 + got, err := fetchWithRetry(context.Background(), fastLimiter(3), "failed to get x", func() (*string, error) { + calls++ + return nil, sentinel + }) + require.Error(t, err) + assert.Nil(t, got) + assert.ErrorIs(t, err, sentinel) + assert.Contains(t, err.Error(), "failed to get x") + assert.Equal(t, 1, calls, "non-throttle errors must not be retried") +} + +func TestFetchWithRetry_StopsAfterMaxRetries(t *testing.T) { + calls := 0 + _, err := fetchWithRetry(context.Background(), fastLimiter(2), "p", func() (*string, error) { + calls++ + return nil, newThrottleError() + }) + require.Error(t, err) + assert.Equal(t, 3, calls) +} + +func TestFetchWithRetry_CanceledContextFailsBeforeCalling(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + calls := 0 + _, err := fetchWithRetry(ctx, fastLimiter(1), "p", func() (*string, error) { + calls++ + return nil, nil + }) + require.ErrorIs(t, err, context.Canceled) + assert.Zero(t, calls) +} diff --git a/providers/aws/recommendations/sp_coverage.go b/providers/aws/recommendations/sp_coverage.go index 84cf9649..98096138 100644 --- a/providers/aws/recommendations/sp_coverage.go +++ b/providers/aws/recommendations/sp_coverage.go @@ -11,8 +11,6 @@ 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/LeanerCloud/cloud-commitments-go/pkg/concurrency" ) // SPCoverageSummary is the Savings Plans coverage summary (optionally scoped @@ -377,23 +375,9 @@ func (c *Client) GetSPCoverageSummary(ctx context.Context, region string, lookba // fetchSPCoveragePage calls GetSavingsPlansCoverage with rate-limit retry. // Mirrors fetchCoveragePage in coverage.go so both paths back off consistently. func (c *Client) fetchSPCoveragePage(ctx context.Context, input *costexplorer.GetSavingsPlansCoverageInput) (*costexplorer.GetSavingsPlansCoverageOutput, error) { - rateLimiter := c.rateLimiter.newOperation() - for { - if waitErr := rateLimiter.Wait(ctx); waitErr != nil { - return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr) - } - if acqErr := concurrency.Acquire(ctx); acqErr != nil { - return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr) - } - result, err := c.costExplorerClient.GetSavingsPlansCoverage(ctx, input) - concurrency.Release(ctx) - if !rateLimiter.ShouldRetry(err) { - if err != nil { - return nil, fmt.Errorf("failed to get SP coverage: %w", err) - } - return result, nil - } - } + return fetchWithRetry(ctx, c.rateLimiter, "failed to get SP coverage", func() (*costexplorer.GetSavingsPlansCoverageOutput, error) { + return c.costExplorerClient.GetSavingsPlansCoverage(ctx, input) + }) } // GetSPUtilization fetches Savings Plans utilization for one plan type over @@ -470,23 +454,9 @@ func (c *Client) GetSPUtilization(ctx context.Context, planType types.SupportedS // fetchSPUtilizationPage calls GetSavingsPlansUtilization with rate-limit retry. // Mirrors fetchUtilizationPage in utilization.go so both paths back off consistently. func (c *Client) fetchSPUtilizationPage(ctx context.Context, input *costexplorer.GetSavingsPlansUtilizationInput) (*costexplorer.GetSavingsPlansUtilizationOutput, error) { - rateLimiter := c.rateLimiter.newOperation() - for { - if waitErr := rateLimiter.Wait(ctx); waitErr != nil { - return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr) - } - if acqErr := concurrency.Acquire(ctx); acqErr != nil { - return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr) - } - result, err := c.costExplorerClient.GetSavingsPlansUtilization(ctx, input) - concurrency.Release(ctx) - if !rateLimiter.ShouldRetry(err) { - if err != nil { - return nil, fmt.Errorf("failed to get SP utilization: %w", err) - } - return result, nil - } - } + return fetchWithRetry(ctx, c.rateLimiter, "failed to get SP utilization", func() (*costexplorer.GetSavingsPlansUtilizationOutput, error) { + return c.costExplorerClient.GetSavingsPlansUtilization(ctx, input) + }) } // buildSPUtilizationSummary converts a CE GetSavingsPlansUtilization response diff --git a/providers/aws/recommendations/utilization.go b/providers/aws/recommendations/utilization.go index aa4ac1a6..af35800d 100644 --- a/providers/aws/recommendations/utilization.go +++ b/providers/aws/recommendations/utilization.go @@ -10,7 +10,6 @@ import ( "github.com/aws/aws-sdk-go-v2/service/costexplorer/types" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" - "github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency" ) // RIUtilization is an alias for common.RIUtilization so existing callers that @@ -126,24 +125,9 @@ func buildUtilizations(agg map[string]*riAccumulator) []RIUtilization { // across the outbound SDK call, matching fetchRIPageWithRetry, so callers // running under a semaphore-carrying context stay inside the global IO cap. func (c *Client) fetchUtilizationPage(ctx context.Context, input *costexplorer.GetReservationUtilizationInput) (*costexplorer.GetReservationUtilizationOutput, error) { - rateLimiter := c.rateLimiter.newOperation() - for { - if waitErr := rateLimiter.Wait(ctx); waitErr != nil { - return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr) - } - - if acqErr := concurrency.Acquire(ctx); acqErr != nil { - return nil, fmt.Errorf("concurrency acquire failed: %w", acqErr) - } - result, err := c.costExplorerClient.GetReservationUtilization(ctx, input) - concurrency.Release(ctx) - if !rateLimiter.ShouldRetry(err) { - if err != nil { - return nil, fmt.Errorf("failed to get reservation utilization: %w", err) - } - return result, nil - } - } + return fetchWithRetry(ctx, c.rateLimiter, "failed to get reservation utilization", func() (*costexplorer.GetReservationUtilizationOutput, error) { + return c.costExplorerClient.GetReservationUtilization(ctx, input) + }) } // aggregateUtilizationDetail accumulates hours from a single utilization detail