From e3861eaebee90da9bcb75f09fa849343476cf550 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Wed, 10 Jun 2026 17:45:57 -0700 Subject: [PATCH] fix(aws/recs): cap coverage/utilization CE calls with shared semaphore fetchCoveragePage and fetchUtilizationPage called the Cost Explorer SDK without acquiring the shared concurrency semaphore, so the scheduler's coverage-enrichment fan-out (AttachDailyUsageHistory per service, region, resourceType tuple) escaped the CUDLY_MAX_PARALLELISM cap that protects Lambda memory and CE rate limits. Worst case added roughly 6 x min(N,20) uncapped concurrent CE calls on top of the capped 20, and made the GetAllRecommendations docstring invariant ("the Acquire/Release boundary lives around the individual CE SDK call") false for a third of the CE call sites. Wrap both SDK calls in concurrency.Acquire/Release, holding the slot only across the outbound call (released during rate-limit backoffs), matching fetchRIPageWithRetry and fetchSPPageWithRetry. Add regression tests that saturate a capacity-1 shared semaphore and assert the CE mock is never reached (both fail on the pre-fix code), plus a slot-release guard against Acquire/Release pairing leaks. Closes #1144 --- .../recommendations/concurrency_cap_test.go | 101 ++++++++++++++++++ providers/aws/recommendations/coverage.go | 9 +- providers/aws/recommendations/utilization.go | 8 ++ 3 files changed, 117 insertions(+), 1 deletion(-) create mode 100644 providers/aws/recommendations/concurrency_cap_test.go diff --git a/providers/aws/recommendations/concurrency_cap_test.go b/providers/aws/recommendations/concurrency_cap_test.go new file mode 100644 index 000000000..e9f58e74b --- /dev/null +++ b/providers/aws/recommendations/concurrency_cap_test.go @@ -0,0 +1,101 @@ +package recommendations + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/costexplorer" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "golang.org/x/sync/semaphore" + + "github.com/LeanerCloud/CUDly/pkg/concurrency" +) + +// semCountingCE counts coverage/utilization SDK calls so the tests below can +// assert whether a CE call escaped the shared concurrency semaphore. +type semCountingCE struct { + mockCostExplorerAPI + coverageCalls atomic.Int32 + utilizationCalls atomic.Int32 +} + +func (m *semCountingCE) GetReservationCoverage(ctx context.Context, params *costexplorer.GetReservationCoverageInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationCoverageOutput, error) { + m.coverageCalls.Add(1) + return &costexplorer.GetReservationCoverageOutput{}, nil +} + +func (m *semCountingCE) GetReservationUtilization(ctx context.Context, params *costexplorer.GetReservationUtilizationInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationUtilizationOutput, error) { + m.utilizationCalls.Add(1) + return &costexplorer.GetReservationUtilizationOutput{}, nil +} + +// fullSemaphoreCtx returns a cancelled context carrying a capacity-1 shared +// semaphore whose only slot is already held, replicating the scheduler's +// collection path when the CUDLY_MAX_PARALLELISM cap is saturated. A correct +// leaf call must block on Acquire (and surface ctx cancellation) instead of +// issuing the CE request; the pre-fix code escaped the cap and called CE +// immediately (review finding PERF-02). +func fullSemaphoreCtx(t *testing.T) context.Context { + t.Helper() + sem := semaphore.NewWeighted(1) + require.NoError(t, sem.Acquire(context.Background(), 1)) + t.Cleanup(func() { sem.Release(1) }) + + ctx, cancel := context.WithCancel(concurrency.WithSharedSemaphore(context.Background(), sem)) + cancel() + return ctx +} + +// TestFetchCoveragePage_RespectsSharedSemaphore is the PERF-02 regression +// test for the coverage path (scheduler's AttachDailyUsageHistory fan-out): +// with the cap saturated, fetchCoveragePage must not reach the CE SDK. +// Pre-fix this test fails: the call escaped the semaphore, hit the mock, +// and returned success. +func TestFetchCoveragePage_RespectsSharedSemaphore(t *testing.T) { + mock := &semCountingCE{} + client := NewClientWithAPI(mock, "us-east-1") + + _, err := client.fetchCoveragePage(fullSemaphoreCtx(t), &costexplorer.GetReservationCoverageInput{}) + + require.Error(t, err, "fetch must fail when the cap is saturated and ctx is cancelled") + assert.ErrorIs(t, err, context.Canceled) + assert.Equal(t, int32(0), mock.coverageCalls.Load(), + "GetReservationCoverage escaped the shared concurrency semaphore") +} + +// TestFetchUtilizationPage_RespectsSharedSemaphore is the PERF-02 regression +// test for the utilization path: same invariant as the coverage test. +func TestFetchUtilizationPage_RespectsSharedSemaphore(t *testing.T) { + mock := &semCountingCE{} + client := NewClientWithAPI(mock, "us-east-1") + + _, err := client.fetchUtilizationPage(fullSemaphoreCtx(t), &costexplorer.GetReservationUtilizationInput{}) + + require.Error(t, err, "fetch must fail when the cap is saturated and ctx is cancelled") + assert.ErrorIs(t, err, context.Canceled) + assert.Equal(t, int32(0), mock.utilizationCalls.Load(), + "GetReservationUtilization escaped the shared concurrency semaphore") +} + +// TestFetchPages_ReleaseSemaphoreSlot verifies the Acquire/Release pairing: +// after a successful fetch under a free capacity-1 semaphore, the slot must +// be available again (no leak that would starve later leaf calls). +func TestFetchPages_ReleaseSemaphoreSlot(t *testing.T) { + mock := &semCountingCE{} + client := NewClientWithAPI(mock, "us-east-1") + + sem := semaphore.NewWeighted(1) + ctx := concurrency.WithSharedSemaphore(context.Background(), sem) + + _, err := client.fetchCoveragePage(ctx, &costexplorer.GetReservationCoverageInput{}) + require.NoError(t, err) + _, err = client.fetchUtilizationPage(ctx, &costexplorer.GetReservationUtilizationInput{}) + require.NoError(t, err) + + assert.Equal(t, int32(1), mock.coverageCalls.Load()) + assert.Equal(t, int32(1), mock.utilizationCalls.Load()) + require.True(t, sem.TryAcquire(1), "semaphore slot leaked: fetch did not release after the SDK call") + sem.Release(1) +} diff --git a/providers/aws/recommendations/coverage.go b/providers/aws/recommendations/coverage.go index d4b4b97ad..6c7ac438f 100644 --- a/providers/aws/recommendations/coverage.go +++ b/providers/aws/recommendations/coverage.go @@ -11,6 +11,7 @@ import ( "github.com/aws/aws-sdk-go-v2/service/costexplorer/types" "github.com/LeanerCloud/CUDly/pkg/common" + "github.com/LeanerCloud/CUDly/pkg/concurrency" ) // coverageServiceFilters lists the Cost Explorer service-dimension values @@ -319,14 +320,20 @@ func serviceRegionFilter(service, region string) *types.Expression { // fetchCoveragePage calls the Cost Explorer API with rate-limit retry. // Mirrors fetchUtilizationPage so the two paths fail and back off the -// same way. +// same way. The shared concurrency semaphore (CUDLY_MAX_PARALLELISM) is +// 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) { c.rateLimiter.Reset() for { if waitErr := c.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 !c.rateLimiter.ShouldRetry(err) { if err != nil { return nil, fmt.Errorf("failed to get reservation coverage: %w", err) diff --git a/providers/aws/recommendations/utilization.go b/providers/aws/recommendations/utilization.go index 7c79d7511..c17f6a4db 100644 --- a/providers/aws/recommendations/utilization.go +++ b/providers/aws/recommendations/utilization.go @@ -12,6 +12,7 @@ import ( "github.com/aws/aws-sdk-go-v2/service/costexplorer/types" "github.com/LeanerCloud/CUDly/pkg/common" + "github.com/LeanerCloud/CUDly/pkg/concurrency" ) // RIUtilization is an alias for common.RIUtilization so existing callers that @@ -97,6 +98,9 @@ func buildUtilizations(agg map[string]*riAccumulator) []RIUtilization { } // fetchUtilizationPage calls the Cost Explorer API with rate-limit retry. +// The shared concurrency semaphore (CUDLY_MAX_PARALLELISM) is held only +// 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) { c.rateLimiter.Reset() for { @@ -104,7 +108,11 @@ func (c *Client) fetchUtilizationPage(ctx context.Context, input *costexplorer.G 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 !c.rateLimiter.ShouldRetry(err) { if err != nil { return nil, fmt.Errorf("failed to get reservation utilization: %w", err)