Skip to content

Commit 650221d

Browse files
committed
fix(aws/recommendations): per-call RateLimiter to fix -race (closes #271)
RateLimiter was a single *RateLimiter field on Client, shared across the 6 concurrent goroutines that GetAllRecommendations fans out via errgroup. Reset/ShouldRetry/GetRetryCount all mutate retryCount without a lock, triggering data races under go test -race. Per feedback_rate_limiter_per_call.md: each goroutine needs its own independent retry budget (not shared throughput), so per-call instantiation is correct over adding a mutex. Replace rateLimiter *RateLimiter with newRateLimiter func() *RateLimiter. Each fetch*WithRetry / fetch*Page function calls rl := c.newRateLimiter() at entry. Tests inject a factory returning a faster limiter for speed. Also fix mockCostExplorerAPI: callCount and riCalls were mutated by concurrent goroutines without synchronisation. Add sync.Mutex guard.
1 parent cbdc4be commit 650221d

5 files changed

Lines changed: 44 additions & 27 deletions

File tree

‎providers/aws/recommendations/client.go‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,10 @@ type CostExplorerAPI interface {
3434
type Client struct {
3535
costExplorerClient CostExplorerAPI
3636
region string
37-
rateLimiter *RateLimiter
37+
38+
// newRateLimiter is called once per API call (not shared across goroutines).
39+
// Tests can replace it with a factory returning a faster limiter.
40+
newRateLimiter func() *RateLimiter
3841

3942
// ec2API is the EC2 client used to build the DescribeInstanceTypes paginator.
4043
// Populated by NewClient from aws.Config; nil when created via NewClientWithAPI.
@@ -62,7 +65,7 @@ func NewClient(cfg aws.Config) *Client {
6265
return &Client{
6366
costExplorerClient: costexplorer.NewFromConfig(ceConfig),
6467
region: cfg.Region,
65-
rateLimiter: NewRateLimiter(),
68+
newRateLimiter: NewRateLimiter,
6669
ec2API: ec2Client,
6770
// Factory wraps the EC2 client so the paginator is created lazily
6871
// on the first EC2 recommendation parse (not at construction time).
@@ -77,7 +80,7 @@ func NewClientWithAPI(api CostExplorerAPI, region string) *Client {
7780
return &Client{
7881
costExplorerClient: api,
7982
region: region,
80-
rateLimiter: NewRateLimiter(),
83+
newRateLimiter: NewRateLimiter,
8184
// ec2API left nil: instanceTypeLookup falls back to VCPU=0/MemoryGB=0
8285
// unless the caller sets instanceTypePagerFactory.
8386
}
@@ -177,12 +180,12 @@ func (c *Client) fetchRIPageWithRetry(
177180
ctx context.Context,
178181
input *costexplorer.GetReservationPurchaseRecommendationInput,
179182
) (*costexplorer.GetReservationPurchaseRecommendationOutput, error) {
180-
c.rateLimiter.Reset()
183+
rl := c.newRateLimiter()
181184
var result *costexplorer.GetReservationPurchaseRecommendationOutput
182185
var err error
183186

184187
for {
185-
if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
188+
if waitErr := rl.Wait(ctx); waitErr != nil {
186189
return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
187190
}
188191

@@ -191,13 +194,13 @@ func (c *Client) fetchRIPageWithRetry(
191194
}
192195
result, err = c.costExplorerClient.GetReservationPurchaseRecommendation(ctx, input)
193196
concurrency.Release(ctx)
194-
if !c.rateLimiter.ShouldRetry(err) {
197+
if !rl.ShouldRetry(err) {
195198
break
196199
}
197200
}
198201

199202
if err != nil {
200-
return nil, fmt.Errorf("failed to get RI recommendations after %d retries: %w", c.rateLimiter.GetRetryCount(), err)
203+
return nil, fmt.Errorf("failed to get RI recommendations after %d retries: %w", rl.GetRetryCount(), err)
201204
}
202205

203206
return result, nil

‎providers/aws/recommendations/client_test.go‎

Lines changed: 25 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package recommendations
33
import (
44
"context"
55
"fmt"
6+
"sync"
67
"testing"
78
"time"
89

@@ -17,6 +18,7 @@ import (
1718

1819
// Mock CostExplorerAPI for testing
1920
type mockCostExplorerAPI struct {
21+
mu sync.Mutex
2022
riRecommendations *costexplorer.GetReservationPurchaseRecommendationOutput
2123
spRecommendations *costexplorer.GetSavingsPlansPurchaseRecommendationOutput
2224
riError error
@@ -28,20 +30,28 @@ type mockCostExplorerAPI struct {
2830
}
2931

3032
func (m *mockCostExplorerAPI) GetReservationPurchaseRecommendation(ctx context.Context, params *costexplorer.GetReservationPurchaseRecommendationInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationPurchaseRecommendationOutput, error) {
33+
m.mu.Lock()
3134
m.callCount++
3235
m.riCalls = append(m.riCalls, params)
33-
if m.riError != nil {
34-
return nil, m.riError
36+
riErr := m.riError
37+
riRecs := m.riRecommendations
38+
m.mu.Unlock()
39+
if riErr != nil {
40+
return nil, riErr
3541
}
36-
return m.riRecommendations, nil
42+
return riRecs, nil
3743
}
3844

3945
func (m *mockCostExplorerAPI) GetSavingsPlansPurchaseRecommendation(ctx context.Context, params *costexplorer.GetSavingsPlansPurchaseRecommendationInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetSavingsPlansPurchaseRecommendationOutput, error) {
46+
m.mu.Lock()
4047
m.callCount++
41-
if m.spError != nil {
42-
return nil, m.spError
48+
spErr := m.spError
49+
spRecs := m.spRecommendations
50+
m.mu.Unlock()
51+
if spErr != nil {
52+
return nil, spErr
4353
}
44-
return m.spRecommendations, nil
54+
return spRecs, nil
4555
}
4656

4757
func (m *mockCostExplorerAPI) GetReservationUtilization(ctx context.Context, params *costexplorer.GetReservationUtilizationInput, optFns ...func(*costexplorer.Options)) (*costexplorer.GetReservationUtilizationOutput, error) {
@@ -61,7 +71,7 @@ func TestNewClient(t *testing.T) {
6171

6272
assert.NotNil(t, client)
6373
assert.NotNil(t, client.costExplorerClient)
64-
assert.NotNil(t, client.rateLimiter)
74+
assert.NotNil(t, client.newRateLimiter)
6575
assert.Equal(t, "us-west-2", client.region)
6676
}
6777

@@ -74,7 +84,7 @@ func TestNewClientWithAPI(t *testing.T) {
7484
assert.NotNil(t, client)
7585
assert.Equal(t, mockAPI, client.costExplorerClient)
7686
assert.Equal(t, region, client.region)
77-
assert.NotNil(t, client.rateLimiter)
87+
assert.NotNil(t, client.newRateLimiter)
7888
}
7989

8090
func TestGetRecommendations_EC2_Success(t *testing.T) {
@@ -262,9 +272,11 @@ func TestGetRecommendations_Error(t *testing.T) {
262272
riError: newThrottleError(),
263273
}
264274

265-
// Use custom rate limiter to speed up test
275+
// Use custom rate limiter factory to speed up test
266276
client := NewClientWithAPI(mockAPI, "us-east-1")
267-
client.rateLimiter = NewRateLimiterWithOptions(1*time.Millisecond, 10*time.Millisecond, 2)
277+
client.newRateLimiter = func() *RateLimiter {
278+
return NewRateLimiterWithOptions(1*time.Millisecond, 10*time.Millisecond, 2)
279+
}
268280

269281
params := common.RecommendationParams{
270282
Service: common.ServiceEC2,
@@ -506,7 +518,9 @@ func TestGetRecommendations_ContextCancellation(t *testing.T) {
506518
}
507519

508520
client := NewClientWithAPI(mockAPI, "us-east-1")
509-
client.rateLimiter = NewRateLimiterWithOptions(100*time.Millisecond, 1*time.Second, 5)
521+
client.newRateLimiter = func() *RateLimiter {
522+
return NewRateLimiterWithOptions(100*time.Millisecond, 1*time.Second, 5)
523+
}
510524

511525
ctx, cancel := context.WithCancel(context.Background())
512526
cancel() // Cancel immediately

‎providers/aws/recommendations/coverage.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -321,13 +321,13 @@ func serviceRegionFilter(service, region string) *types.Expression {
321321
// Mirrors fetchUtilizationPage so the two paths fail and back off the
322322
// same way.
323323
func (c *Client) fetchCoveragePage(ctx context.Context, input *costexplorer.GetReservationCoverageInput) (*costexplorer.GetReservationCoverageOutput, error) {
324-
c.rateLimiter.Reset()
324+
rl := c.newRateLimiter()
325325
for {
326-
if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
326+
if waitErr := rl.Wait(ctx); waitErr != nil {
327327
return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
328328
}
329329
result, err := c.costExplorerClient.GetReservationCoverage(ctx, input)
330-
if !c.rateLimiter.ShouldRetry(err) {
330+
if !rl.ShouldRetry(err) {
331331
if err != nil {
332332
return nil, fmt.Errorf("failed to get reservation coverage: %w", err)
333333
}

‎providers/aws/recommendations/parser_sp.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -120,12 +120,12 @@ func (c *Client) fetchSPPageWithRetry(
120120
ctx context.Context,
121121
input *costexplorer.GetSavingsPlansPurchaseRecommendationInput,
122122
) (*costexplorer.GetSavingsPlansPurchaseRecommendationOutput, error) {
123-
c.rateLimiter.Reset()
123+
rl := c.newRateLimiter()
124124
var result *costexplorer.GetSavingsPlansPurchaseRecommendationOutput
125125
var err error
126126

127127
for {
128-
if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
128+
if waitErr := rl.Wait(ctx); waitErr != nil {
129129
return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
130130
}
131131

@@ -134,7 +134,7 @@ func (c *Client) fetchSPPageWithRetry(
134134
}
135135
result, err = c.costExplorerClient.GetSavingsPlansPurchaseRecommendation(ctx, input)
136136
concurrency.Release(ctx)
137-
if !c.rateLimiter.ShouldRetry(err) {
137+
if !rl.ShouldRetry(err) {
138138
break
139139
}
140140
}

‎providers/aws/recommendations/utilization.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -98,14 +98,14 @@ func buildUtilizations(agg map[string]*riAccumulator) []RIUtilization {
9898

9999
// fetchUtilizationPage calls the Cost Explorer API with rate-limit retry.
100100
func (c *Client) fetchUtilizationPage(ctx context.Context, input *costexplorer.GetReservationUtilizationInput) (*costexplorer.GetReservationUtilizationOutput, error) {
101-
c.rateLimiter.Reset()
101+
rl := c.newRateLimiter()
102102
for {
103-
if waitErr := c.rateLimiter.Wait(ctx); waitErr != nil {
103+
if waitErr := rl.Wait(ctx); waitErr != nil {
104104
return nil, fmt.Errorf("rate limiter wait failed: %w", waitErr)
105105
}
106106

107107
result, err := c.costExplorerClient.GetReservationUtilization(ctx, input)
108-
if !c.rateLimiter.ShouldRetry(err) {
108+
if !rl.ShouldRetry(err) {
109109
if err != nil {
110110
return nil, fmt.Errorf("failed to get reservation utilization: %w", err)
111111
}

0 commit comments

Comments
 (0)