diff --git a/providers/azure/recommendations_multi_subscription.go b/providers/azure/recommendations_multi_subscription.go index af0b1f55..31d4f215 100644 --- a/providers/azure/recommendations_multi_subscription.go +++ b/providers/azure/recommendations_multi_subscription.go @@ -12,6 +12,7 @@ import ( "golang.org/x/sync/errgroup" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/concurrency" "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" azrecs "github.com/LeanerCloud/cloud-commitments-go/providers/azure/internal/recommendations" @@ -213,10 +214,13 @@ func NewMultiSubscriptionRecommendationsClient(cred azcore.TokenCredential, acco // each per-subscription goroutine captures its own error and returns nil to // the group, so one subscription failing (e.g. the principal lost Reader // access mid-run, or a subscription-specific throttle) never cancels -// sibling subscriptions. The semaphore that bounds aggregate concurrent ARM -// calls is acquired inside each per-subscription client's own -// GetRecommendations (around the outbound API calls, not around this -// fan-out), so no additional semaphore is needed at this layer. +// sibling subscriptions. The number of subscriptions in flight is capped with +// g.SetLimit(concurrency.MaxParallelismFromEnv()) so a tenant with many +// subscriptions cannot fan out unbounded. This is an errgroup limit, not a +// second semaphore: the shared semaphore that bounds aggregate concurrent ARM +// calls is acquired only inside each per-subscription client's leaf calls +// (never held while waiting on this fan-out), so the two bounds cannot +// deadlock against each other. // // After g.Wait(), ctx.Err() is checked explicitly: g.Wait() only reports // errors returned to the group, and every goroutine here returns nil, so a @@ -253,6 +257,7 @@ func (m *MultiSubscriptionRecommendationsClient) GetRecommendations(ctx context. errs := make([]error, len(targets)) g, gctx := errgroup.WithContext(ctx) + g.SetLimit(concurrency.MaxParallelismFromEnv()) for i, sub := range targets { i, sub := i, sub g.Go(func() error { diff --git a/providers/azure/recommendations_multi_subscription_test.go b/providers/azure/recommendations_multi_subscription_test.go index f6dd9bf4..15acc5b0 100644 --- a/providers/azure/recommendations_multi_subscription_test.go +++ b/providers/azure/recommendations_multi_subscription_test.go @@ -6,6 +6,7 @@ import ( "fmt" "sync/atomic" "testing" + "time" "github.com/Azure/azure-sdk-for-go/sdk/azcore" "github.com/stretchr/testify/assert" @@ -517,3 +518,62 @@ func TestNewMultiSubscriptionRecommendationsClient_DeduplicatesAccounts(t *testi assert.Len(t, partial.Failed, 1) }) } + +// peakRecommendationsClient records the maximum number of concurrently +// in-flight GetRecommendations calls across all instances sharing the same +// counters. +type peakRecommendationsClient struct { + inFlight *atomic.Int64 + peak *atomic.Int64 +} + +func (p *peakRecommendationsClient) GetRecommendations(ctx context.Context, _ *common.RecommendationParams) ([]common.Recommendation, error) { + cur := p.inFlight.Add(1) + defer p.inFlight.Add(-1) + for { + old := p.peak.Load() + if cur <= old || p.peak.CompareAndSwap(old, cur) { + break + } + } + select { + case <-time.After(5 * time.Millisecond): + case <-ctx.Done(): + return nil, ctx.Err() + } + return nil, nil +} + +func (p *peakRecommendationsClient) GetRecommendationsForService(ctx context.Context, service common.ServiceType) ([]common.Recommendation, error) { + return p.GetRecommendations(ctx, &common.RecommendationParams{Service: service}) +} + +func (p *peakRecommendationsClient) GetAllRecommendations(ctx context.Context) ([]common.Recommendation, error) { + return p.GetRecommendations(ctx, &common.RecommendationParams{}) +} + +// The fan-out must not exceed CUDLY_MAX_PARALLELISM concurrent subscriptions, +// otherwise a tenant with many subscriptions issues unbounded ARM calls. +func TestMultiSubscriptionRecommendationsClient_GetRecommendations_BoundsConcurrency(t *testing.T) { + const limit, subs = 3, 24 + t.Setenv("CUDLY_MAX_PARALLELISM", fmt.Sprint(limit)) + + var inFlight, peak atomic.Int64 + orig := newSubscriptionRecommendationsClientFn + t.Cleanup(func() { newSubscriptionRecommendationsClientFn = orig }) + newSubscriptionRecommendationsClientFn = func(azcore.TokenCredential, string) (provider.RecommendationsClient, error) { + return &peakRecommendationsClient{inFlight: &inFlight, peak: &peak}, nil + } + + accounts := make([]common.Account, subs) + for i := range accounts { + accounts[i] = common.Account{Provider: common.ProviderAzure, ID: fmt.Sprintf("sub-%d", i)} + } + client, err := NewMultiSubscriptionRecommendationsClient(&mockAzureTokenCredential{}, accounts) + require.NoError(t, err) + + _, err = client.GetAllRecommendations(context.Background()) + require.NoError(t, err) + assert.LessOrEqual(t, peak.Load(), int64(limit)) + assert.Greater(t, peak.Load(), int64(1), "fan-out should still run subscriptions concurrently") +}