From 5796a164f1cc6d02fa3e94211f3c6f8953f39edc Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 20:18:17 +0100 Subject: [PATCH 1/3] fix(aws): batch daily coverage sparkline fetch per region AttachDailyUsageHistory issued one GetReservationCoverage request per distinct (service, region, instance type), so CE requests scaled with SKU count. Group recommendations by (service, region). Buckets with more distinct types than lookback days now issue one region-wide request per day grouped by INSTANCE_TYPE (CE rejects DAILY granularity combined with GroupBy, so a single grouped request cannot return a daily series) and fan each type's series out to its recommendations. Smaller buckets keep the existing per-type DAILY request, so a bucket never costs more than min(types, lookback days) requests plus continuation pages. Per-type series, missing-day zeros and nil-for-no-data semantics match the per-type path; a failed grouped fetch leaves the bucket's usage history nil instead of reporting zero coverage. Closes #85 Co-Authored-By: Claude Opus 5.5 --- .../aws/recommendations/usage_history.go | 210 ++++++++++-- .../usage_history_batch_test.go | 322 ++++++++++++++++++ 2 files changed, 509 insertions(+), 23 deletions(-) create mode 100644 providers/aws/recommendations/usage_history_batch_test.go diff --git a/providers/aws/recommendations/usage_history.go b/providers/aws/recommendations/usage_history.go index 8f7f721b..d9c21167 100644 --- a/providers/aws/recommendations/usage_history.go +++ b/providers/aws/recommendations/usage_history.go @@ -3,6 +3,7 @@ package recommendations import ( "context" "fmt" + "strings" "time" "github.com/aws/aws-sdk-go-v2/aws" @@ -122,61 +123,224 @@ func applyPeriodsToDayMap(periods []types.CoverageByTime, dayPct map[string]floa return anyData } -// tupleKey is the map key for a unique (serviceFilter, region, resourceType) triple. -type tupleKey struct{ service, region, resourceType string } +// regionKey identifies one (CE service filter, region) pair. Every +// recommendation sharing a regionKey can be served by the same region-wide +// GetReservationCoverage query grouped by INSTANCE_TYPE. +type regionKey struct{ service, region string } -// groupRecsByTuple builds an ordered unique list of tuples and an index map -// from each tuple to the recommendation indices that share it. -func groupRecsByTuple(recs []common.Recommendation) (order []tupleKey, recsByTuple map[tupleKey][]int) { - order = make([]tupleKey, 0) - seen := make(map[tupleKey]struct{}) - recsByTuple = make(map[tupleKey][]int) +// regionBucket holds the distinct resource types requested for one regionKey +// (in first-seen order) and the recommendation indices for each type. +type regionBucket struct { + key regionKey + types []string + recsByType map[string][]int +} + +// groupRecsByRegion builds an ordered list of (service, region) buckets, each +// listing its distinct resource types and the recommendation indices that +// share each (service, region, resourceType) tuple. +func groupRecsByRegion(recs []common.Recommendation) []*regionBucket { + buckets := make([]*regionBucket, 0) + byKey := make(map[regionKey]*regionBucket) for i := range recs { r := &recs[i] sf := getServiceStringForCostExplorer(r.Service) if sf == "" || r.Region == "" || r.ResourceType == "" { continue } - k := tupleKey{sf, r.Region, r.ResourceType} - if _, ok := seen[k]; !ok { - seen[k] = struct{}{} - order = append(order, k) + k := regionKey{sf, r.Region} + b, ok := byKey[k] + if !ok { + b = ®ionBucket{key: k, recsByType: make(map[string][]int)} + byKey[k] = b + buckets = append(buckets, b) + } + if _, seen := b.recsByType[r.ResourceType]; !seen { + b.types = append(b.types, r.ResourceType) } - recsByTuple[k] = append(recsByTuple[k], i) + b.recsByType[r.ResourceType] = append(b.recsByType[r.ResourceType], i) } - return order, recsByTuple + return buckets } // AttachDailyUsageHistory enriches each recommendation in recs with a -// UsageHistory slice sourced from GetDailyUsagePcts. It batches by unique -// (serviceFilter, region, resourceType) so a 20-rec list for the same SKU -// fires only one CE call. Errors from individual tuples are logged and -// skipped so a single CE failure doesn't drop the whole collection. +// UsageHistory slice of daily RI-coverage percentages. Recs are batched by +// (serviceFilter, region) so the number of Cost Explorer requests scales with +// regions, not with distinct SKUs: +// +// - A bucket with at most usageHistoryLookbackDays distinct resource types +// issues one DAILY query per type (GetDailyUsagePcts). +// - A larger bucket issues one region-wide query per day grouped by +// INSTANCE_TYPE (getDailyUsagePctsByType) and fans the per-type series +// out to every rec. CE rejects DAILY granularity combined with GroupBy, +// so a single grouped request cannot return a daily series; one grouped +// request per day is the smallest batch CE can express. +// +// Either way a bucket costs at most min(types, usageHistoryLookbackDays) +// requests (plus extra pages). Errors are logged and skipped so a single CE +// failure doesn't drop the whole collection; a failed fetch leaves +// UsageHistory nil rather than reporting zero coverage. // // SavingsPlans have no per-instance-type coverage breakdown in CE, so recs // whose Service resolves to the empty coverage-filter string are silently // skipped. func (c *Client) AttachDailyUsageHistory(ctx context.Context, recs []common.Recommendation) { - order, recsByTuple := groupRecsByTuple(recs) - for _, k := range order { + for _, b := range groupRecsByRegion(recs) { if ctx.Err() != nil { return } - pcts, err := c.GetDailyUsagePcts(ctx, k.service, k.resourceType, k.region) + if len(b.types) > usageHistoryLookbackDays { + c.attachBucketGrouped(ctx, recs, b) + continue + } + c.attachBucketPerType(ctx, recs, b) + } +} + +// attachBucketPerType fetches one DAILY coverage series per resource type. +func (c *Client) attachBucketPerType(ctx context.Context, recs []common.Recommendation, b *regionBucket) { + for _, rt := range b.types { + if ctx.Err() != nil { + return + } + pcts, err := c.GetDailyUsagePcts(ctx, b.key.service, rt, b.key.region) if err != nil { - logging.Warnf("usage_history: failed to fetch daily coverage for %s/%s/%s: %v", k.service, k.resourceType, k.region, err) + logging.Warnf("usage_history: failed to fetch daily coverage for %s/%s/%s: %v", b.key.service, rt, b.key.region, err) continue } // pcts==nil means no CE data for this tuple; leave UsageHistory nil. if pcts == nil { continue } - for _, idx := range recsByTuple[k] { + for _, idx := range b.recsByType[rt] { recs[idx].UsageHistory = pcts } } } +// attachBucketGrouped fetches the whole bucket with per-day grouped queries. +// Any failure leaves every rec in the bucket without UsageHistory, since a +// partial window would otherwise show the failed days as 0% coverage. +func (c *Client) attachBucketGrouped(ctx context.Context, recs []common.Recommendation, b *regionBucket) { + series, err := c.getDailyUsagePctsByType(ctx, b.key.service, b.key.region, b.types) + if err != nil { + logging.Warnf("usage_history: failed to fetch daily coverage for %s/%s (%d instance types): %v", b.key.service, b.key.region, len(b.types), err) + return + } + for _, rt := range b.types { + pcts := series[rt] + if pcts == nil { + continue + } + for _, idx := range b.recsByType[rt] { + recs[idx].UsageHistory = pcts + } + } +} + +// getDailyUsagePctsByType returns, for each of resourceTypes, the same +// series GetDailyUsagePcts would return for that type: usageHistoryLookbackDays +// points ordered oldest-to-newest, with 0 for days CE omits, and no entry +// (nil) for a type CE has no data for at all. It issues one region-wide +// GetReservationCoverage request per day (plus pages) grouped by +// INSTANCE_TYPE, because CE does not accept DAILY granularity with GroupBy. +func (c *Client) getDailyUsagePctsByType(ctx context.Context, serviceFilter, region string, resourceTypes []string) (map[string][]float64, error) { + if serviceFilter == "" || region == "" || len(resourceTypes) == 0 { + return nil, nil + } + end := time.Now().UTC() + start := end.AddDate(0, 0, -usageHistoryLookbackDays) + + // CE reports instanceType attributes lower-cased by extractGroupAttributes; + // index the requested types the same way. + acc := groupedDayAccumulator{dayPct: make(map[string]map[string]float64), byLower: make(map[string][]string), withData: make(map[string]bool)} + for _, rt := range resourceTypes { + acc.dayPct[rt], _ = newDayWindow(start) + lower := strings.ToLower(rt) + acc.byLower[lower] = append(acc.byLower[lower], rt) + } + for i := 0; i < usageHistoryLookbackDays; i++ { + if err := c.fetchGroupedDayCoverage(ctx, serviceFilter, region, start.AddDate(0, 0, i), &acc); err != nil { + return nil, err + } + } + out := make(map[string][]float64, len(acc.withData)) + for rt := range acc.withData { + out[rt] = orderedDaySlice(start, acc.dayPct[rt]) + } + return out, nil +} + +// groupedDayAccumulator collects per-type daily percentages from grouped +// coverage responses. +type groupedDayAccumulator struct { + dayPct map[string]map[string]float64 // resourceType -> day -> pct + byLower map[string][]string // lower-cased type -> requested types + withData map[string]bool // types with at least one data point +} + +// fetchGroupedDayCoverage pages through one region-wide, single-day +// GetReservationCoverage request grouped by INSTANCE_TYPE. +func (c *Client) fetchGroupedDayCoverage(ctx context.Context, serviceFilter, region string, day time.Time, acc *groupedDayAccumulator) error { + input := &costexplorer.GetReservationCoverageInput{ + TimePeriod: &types.DateInterval{ + Start: aws.String(day.Format("2006-01-02")), + End: aws.String(day.AddDate(0, 0, 1).Format("2006-01-02")), + }, + GroupBy: []types.GroupDefinition{ + {Type: types.GroupDefinitionTypeDimension, Key: aws.String(string(types.DimensionInstanceType))}, + }, + Filter: serviceRegionFilter(serviceFilter, region), + Metrics: []string{"Hour"}, + } + var token *string + for { + input.NextPageToken = token + result, err := c.fetchCoveragePage(ctx, input) + if err != nil { + return fmt.Errorf("failed to get daily coverage for %s/%s on %s: %w", serviceFilter, region, day.Format("2006-01-02"), err) + } + acc.addPeriods(result.CoveragesByTime) + if result.NextPageToken == nil || *result.NextPageToken == "" { + return nil + } + token = result.NextPageToken + } +} + +// addPeriods applies each group's coverage percentage to the requested +// types it matches, skipping periods without a start date and groups without +// a CoverageHoursPercentage, exactly as applyPeriodsToDayMap does for Totals. +func (a *groupedDayAccumulator) addPeriods(periods []types.CoverageByTime) { + for _, period := range periods { + if period.TimePeriod == nil || period.TimePeriod.Start == nil { + continue + } + day := aws.ToString(period.TimePeriod.Start) + for _, group := range period.Groups { + pct, ok := groupCoveragePct(group) + if !ok { + continue + } + instType, _ := extractGroupAttributes(group.Attributes) + for _, rt := range a.byLower[instType] { + a.dayPct[rt][day] = pct + a.withData[rt] = true + } + } + } +} + +// groupCoveragePct returns the group's parsed CoverageHoursPercentage, or +// false when CE omitted it. +func groupCoveragePct(group types.ReservationCoverageGroup) (float64, bool) { + if group.Coverage == nil || group.Coverage.CoverageHours == nil || + group.Coverage.CoverageHours.CoverageHoursPercentage == nil { + return 0, false + } + return parseFloat(aws.ToString(group.Coverage.CoverageHours.CoverageHoursPercentage)), true +} + func dailyUsageFilter(serviceFilter, resourceType, region string) *types.Expression { filter := serviceRegionFilter(serviceFilter, region) filter.And = append(filter.And, types.Expression{Dimensions: &types.DimensionValues{ diff --git a/providers/aws/recommendations/usage_history_batch_test.go b/providers/aws/recommendations/usage_history_batch_test.go new file mode 100644 index 00000000..716aba02 --- /dev/null +++ b/providers/aws/recommendations/usage_history_batch_test.go @@ -0,0 +1,322 @@ +package recommendations + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sort" + "strings" + "sync" + "testing" + "time" + + "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" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" +) + +// coverageFixture is one in-memory CE coverage table served in both request +// shapes: per-type DAILY queries (filtered by INSTANCE_TYPE, read from Total) +// and per-day grouped queries (GroupBy INSTANCE_TYPE, read from Groups). A +// nil *string value models CE returning the row without a percentage. +type coverageFixture struct { + t *testing.T + pageSize int // grouped groups per page; 0 = single page + data map[string]map[string]map[string]*string // region -> type -> day -> pct + failDay string // grouped request for this day returns an error + + mu sync.Mutex + calls map[string]int // region -> requests + grouped int +} + +func (f *coverageFixture) serve(w http.ResponseWriter, r *http.Request) { + var input costexplorer.GetReservationCoverageInput + if !assert.NoError(f.t, json.NewDecoder(r.Body).Decode(&input)) { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + region := filterValue(input.Filter, types.DimensionRegion) + f.mu.Lock() + if f.calls == nil { + f.calls = make(map[string]int) + } + f.calls[region]++ + if len(input.GroupBy) > 0 { + f.grouped++ + } + f.mu.Unlock() + w.Header().Set("Content-Type", "application/x-amz-json-1.1") + var out map[string]any + if len(input.GroupBy) > 0 { + out = f.groupedResponse(w, &input, region) + } else { + out = f.perTypeResponse(&input, region) + } + if out != nil { + assert.NoError(f.t, json.NewEncoder(w).Encode(out)) + } +} + +func (f *coverageFixture) perTypeResponse(input *costexplorer.GetReservationCoverageInput, region string) map[string]any { + assert.Equal(f.t, types.GranularityDaily, input.Granularity) + rt := filterValue(input.Filter, types.DimensionInstanceType) + start, err := time.Parse("2006-01-02", aws.ToString(input.TimePeriod.Start)) + require.NoError(f.t, err) + periods := []types.CoverageByTime{} + for i := 0; i < usageHistoryLookbackDays; i++ { + day := start.AddDate(0, 0, i).Format("2006-01-02") + pct, ok := f.data[region][rt][day] + if !ok { + continue + } + p := types.CoverageByTime{TimePeriod: &types.DateInterval{Start: aws.String(day)}} + if pct != nil { + p.Total = &types.Coverage{CoverageHours: &types.CoverageHours{CoverageHoursPercentage: pct}} + } + periods = append(periods, p) + } + return map[string]any{"CoveragesByTime": periods} +} + +func (f *coverageFixture) groupedResponse(w http.ResponseWriter, input *costexplorer.GetReservationCoverageInput, region string) map[string]any { + day := aws.ToString(input.TimePeriod.Start) + start, err := time.Parse("2006-01-02", day) + require.NoError(f.t, err) + assert.Empty(f.t, input.Granularity, "CE rejects Granularity with GroupBy") + assert.Equal(f.t, start.AddDate(0, 0, 1).Format("2006-01-02"), aws.ToString(input.TimePeriod.End)) + assert.Equal(f.t, []types.GroupDefinition{{Type: types.GroupDefinitionTypeDimension, Key: aws.String("INSTANCE_TYPE")}}, input.GroupBy) + assert.Equal(f.t, serviceRegionFilter(ec2CoverageService, region), input.Filter) + assert.Equal(f.t, []string{"Hour"}, input.Metrics) + if day == f.failDay { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"__type":"DataUnavailableException","message":"no coverage available"}`)) + return nil + } + groups := f.dayGroups(region, day) + offset := 0 + if tok := aws.ToString(input.NextPageToken); tok != "" { + _, err := fmt.Sscanf(tok, "page-%d", &offset) + require.NoError(f.t, err) + } + end := len(groups) + var next *string + if f.pageSize > 0 && offset+f.pageSize < end { + end = offset + f.pageSize + next = aws.String(fmt.Sprintf("page-%d", end)) + } + period := types.CoverageByTime{TimePeriod: &types.DateInterval{Start: aws.String(day)}, Groups: groups[offset:end]} + return map[string]any{"CoveragesByTime": []types.CoverageByTime{period}, "NextPageToken": next} +} + +// dayGroups returns the fixture's groups for one day in a stable order, plus +// an unrequested instance type that the client must ignore. +func (f *coverageFixture) dayGroups(region, day string) []types.ReservationCoverageGroup { + names := make([]string, 0, len(f.data[region])) + for rt := range f.data[region] { + names = append(names, rt) + } + sort.Strings(names) + groups := []types.ReservationCoverageGroup{{ + Attributes: map[string]string{"instanceType": "z9.unrequested"}, + Coverage: &types.Coverage{CoverageHours: &types.CoverageHours{CoverageHoursPercentage: aws.String("1")}}, + }} + for _, rt := range names { + pct, ok := f.data[region][rt][day] + if !ok { + continue + } + g := types.ReservationCoverageGroup{Attributes: map[string]string{"instanceType": rt, "region": region}} + if pct != nil { + g.Coverage = &types.Coverage{CoverageHours: &types.CoverageHours{CoverageHoursPercentage: pct}} + } + groups = append(groups, g) + } + return groups +} + +// counts returns the requests seen for region and the grouped request total. +func (f *coverageFixture) counts(region string) (perRegion, grouped int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls[region], f.grouped +} + +func filterValue(expr *types.Expression, key types.Dimension) string { + if expr == nil { + return "" + } + for _, e := range expr.And { + if e.Dimensions != nil && e.Dimensions.Key == key && len(e.Dimensions.Values) == 1 { + return e.Dimensions.Values[0] + } + } + return "" +} + +// usageWindowDays returns the lookback window's day strings, matching the +// client's own window computation. +func usageWindowDays() []string { + start := time.Now().UTC().AddDate(0, 0, -usageHistoryLookbackDays) + days := make([]string, usageHistoryLookbackDays) + for i := range days { + days[i] = start.AddDate(0, 0, i).Format("2006-01-02") + } + return days +} + +// manyTypeFixture builds n instance types for one region. Type i has data on +// days where (i+d)%3 != 0, a row without a percentage on day 1 for type 0, the +// second-to-last type has only rows without a percentage, and the last type +// has no data at all. +func manyTypeFixture(n int) (map[string]map[string]*string, []string) { + days := usageWindowDays() + data := make(map[string]map[string]*string, n) + names := make([]string, n) + for i := 0; i < n; i++ { + rt := fmt.Sprintf("m%d.large", i+1) + names[i] = rt + data[rt] = map[string]*string{} + if i == n-1 { + continue // missing from every response + } + for d, day := range days { + if (i+d)%3 != 0 { + data[rt][day] = aws.String(fmt.Sprintf("%d.5", 10*i+d)) + } + } + } + data[names[0]][days[1]] = nil + for _, day := range days { + data[names[n-2]][day] = nil + } + return data, names +} + +func newFixtureClient(t *testing.T, f *coverageFixture) *Client { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(f.serve)) + t.Cleanup(server.Close) + sdk := costexplorer.New(costexplorer.Options{ + Region: "us-east-1", BaseEndpoint: aws.String(server.URL), + Credentials: aws.AnonymousCredentials{}, HTTPClient: server.Client(), + }) + return NewClientWithAPI(sdk, "us-east-1") +} + +func recsFor(region string, names []string) []common.Recommendation { + recs := make([]common.Recommendation, 0, len(names)+1) + for _, rt := range names { + recs = append(recs, common.Recommendation{Service: common.ServiceEC2, Region: region, ResourceType: rt}) + } + // A second rec sharing the first SKU must receive the same series. + return append(recs, common.Recommendation{Service: common.ServiceEC2, Region: region, ResourceType: names[0]}) +} + +// legacySeries computes each type's series through the per-type path +// (GetDailyUsagePcts), i.e. what callers received before batching. +func legacySeries(t *testing.T, data map[string]map[string]map[string]*string, region string, names []string) map[string][]float64 { + t.Helper() + legacy := newFixtureClient(t, &coverageFixture{t: t, data: data}) + out := make(map[string][]float64, len(names)) + for _, rt := range names { + pcts, err := legacy.GetDailyUsagePcts(context.Background(), ec2CoverageService, rt, region) + require.NoError(t, err) + out[rt] = pcts + } + return out +} + +func TestAttachDailyUsageHistory_BatchesManyTypesPerRegion(t *testing.T) { + t.Parallel() + const region = "us-east-1" + regionData, names := manyTypeFixture(12) + data := map[string]map[string]map[string]*string{region: regionData} + want := legacySeries(t, data, region, names) + + for _, tc := range []struct { + name string + pageSize int + wantCalls int + }{ + // Before batching: one request per instance type (12). + {name: "single page", wantCalls: usageHistoryLookbackDays}, + // Each day has 1 unrequested + up to 11 typed groups; pages of 4. + {name: "paginated", pageSize: 4}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + f := &coverageFixture{t: t, data: data, pageSize: tc.pageSize} + client := newFixtureClient(t, f) + recs := recsFor(region, names) + client.AttachDailyUsageHistory(context.Background(), recs) + + wantCalls := tc.wantCalls + if tc.pageSize > 0 { + for _, day := range usageWindowDays() { + n := len(f.dayGroups(region, day)) + wantCalls += (n + tc.pageSize - 1) / tc.pageSize + } + } + calls, grouped := f.counts(region) + assert.Equal(t, wantCalls, calls, "CE requests for %d instance types", len(names)) + assert.Equal(t, wantCalls, grouped, "every request should be a grouped request") + for _, r := range recs { + assert.Equal(t, want[r.ResourceType], r.UsageHistory, "series for %s must match the per-type path", r.ResourceType) + } + assert.Nil(t, recs[len(names)-1].UsageHistory, "type missing from every response stays nil") + assert.Nil(t, recs[len(names)-2].UsageHistory, "type with only percentage-less rows stays nil") + require.Len(t, recs[0].UsageHistory, usageHistoryLookbackDays) + assert.Equal(t, 0.0, recs[0].UsageHistory[1], "row without a percentage keeps the zero slot") + }) + } +} + +func TestAttachDailyUsageHistory_SmallBucketStaysPerType(t *testing.T) { + t.Parallel() + const region = "eu-west-1" + regionData, names := manyTypeFixture(3) + data := map[string]map[string]map[string]*string{region: regionData} + want := legacySeries(t, data, region, names) + f := &coverageFixture{t: t, data: data} + recs := recsFor(region, names) + newFixtureClient(t, f).AttachDailyUsageHistory(context.Background(), recs) + + calls, grouped := f.counts(region) + assert.Equal(t, len(names), calls, "a small bucket keeps one request per type") + assert.Zero(t, grouped) + for _, r := range recs { + assert.Equal(t, want[r.ResourceType], r.UsageHistory) + } +} + +func TestAttachDailyUsageHistory_GroupedErrorLeavesBucketNil(t *testing.T) { + t.Parallel() + big, bigNames := manyTypeFixture(9) + small, smallNames := manyTypeFixture(4) + data := map[string]map[string]map[string]*string{"us-east-1": big, "us-west-2": small} + f := &coverageFixture{t: t, data: data, failDay: usageWindowDays()[3]} + recs := append(recsFor("us-east-1", bigNames), recsFor("us-west-2", smallNames)...) + newFixtureClient(t, f).AttachDailyUsageHistory(context.Background(), recs) + + for _, r := range recs { + if r.Region == "us-east-1" { + assert.Nil(t, r.UsageHistory, "%s: a CE error must not be reported as coverage", r.ResourceType) + } + } + firstSmall := len(bigNames) + 1 // recsFor appends one duplicate rec per region + assert.Equal(t, "us-west-2", recs[firstSmall].Region) + assert.NotNil(t, recs[firstSmall].UsageHistory, "other region buckets are unaffected") + calls, _ := f.counts("us-east-1") + assert.Equal(t, 4, calls, "fetch stops at the failing day") + + _, err := newFixtureClient(t, f).getDailyUsagePctsByType(context.Background(), ec2CoverageService, "us-east-1", bigNames) + require.Error(t, err) + assert.True(t, strings.Contains(err.Error(), "DataUnavailableException"), err.Error()) +} From 98b490e2c4bab44f33c94951394cfc2b12386cf1 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Sat, 17 Oct 2026 00:42:53 +0100 Subject: [PATCH 2/3] fix(aws): stop grouped coverage paging on a repeated page token If Cost Explorer returned the same NextPageToken twice, the grouped per-day loop re-requested the same page until ctx was canceled. Return an error instead (CodeRabbit finding on #464). Co-Authored-By: Claude Opus 5.5 --- .../aws/recommendations/usage_history.go | 5 +++++ .../usage_history_batch_test.go | 20 +++++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/providers/aws/recommendations/usage_history.go b/providers/aws/recommendations/usage_history.go index d9c21167..4b154eda 100644 --- a/providers/aws/recommendations/usage_history.go +++ b/providers/aws/recommendations/usage_history.go @@ -304,6 +304,11 @@ func (c *Client) fetchGroupedDayCoverage(ctx context.Context, serviceFilter, reg if result.NextPageToken == nil || *result.NextPageToken == "" { return nil } + // A token that doesn't advance would repeat the same rate-limited + // request until ctx is canceled. + if token != nil && *token == *result.NextPageToken { + return fmt.Errorf("failed to get daily coverage for %s/%s on %s: repeated page token", serviceFilter, region, day.Format("2006-01-02")) + } token = result.NextPageToken } } diff --git a/providers/aws/recommendations/usage_history_batch_test.go b/providers/aws/recommendations/usage_history_batch_test.go index 716aba02..f936c318 100644 --- a/providers/aws/recommendations/usage_history_batch_test.go +++ b/providers/aws/recommendations/usage_history_batch_test.go @@ -30,6 +30,7 @@ type coverageFixture struct { pageSize int // grouped groups per page; 0 = single page data map[string]map[string]map[string]*string // region -> type -> day -> pct failDay string // grouped request for this day returns an error + stuckTok string // when set, every grouped page returns this NextPageToken mu sync.Mutex calls map[string]int // region -> requests @@ -111,6 +112,9 @@ func (f *coverageFixture) groupedResponse(w http.ResponseWriter, input *costexpl end = offset + f.pageSize next = aws.String(fmt.Sprintf("page-%d", end)) } + if f.stuckTok != "" { + next = aws.String(f.stuckTok) + } period := types.CoverageByTime{TimePeriod: &types.DateInterval{Start: aws.String(day)}, Groups: groups[offset:end]} return map[string]any{"CoveragesByTime": []types.CoverageByTime{period}, "NextPageToken": next} } @@ -320,3 +324,19 @@ func TestAttachDailyUsageHistory_GroupedErrorLeavesBucketNil(t *testing.T) { require.Error(t, err) assert.True(t, strings.Contains(err.Error(), "DataUnavailableException"), err.Error()) } + +func TestFetchGroupedDayCoverage_RepeatedPageTokenFails(t *testing.T) { + t.Parallel() + data, _ := manyTypeFixture(9) + f := &coverageFixture{t: t, data: map[string]map[string]map[string]*string{"us-east-1": data}, stuckTok: "page-0"} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + day, err := time.Parse("2006-01-02", usageWindowDays()[0]) + require.NoError(t, err) + + err = newFixtureClient(t, f).fetchGroupedDayCoverage(ctx, ec2CoverageService, "us-east-1", day, &groupedDayAccumulator{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "repeated page token") + _, grouped := f.counts("us-east-1") + assert.Equal(t, 2, grouped, "stops at the first repeated token") +} From d7630caf529963ac51c2aac42fc1b42805dcbc1b Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Sat, 17 Oct 2026 00:54:02 +0100 Subject: [PATCH 3/3] fix(aws): cap grouped daily coverage paging through paginateCE fetchGroupedDayCoverage was the only Cost Explorer token loop in the package left uncapped after #462/#463. Route it through paginateCE with maxCoveragePages so endless fresh tokens or a token cycle stop at the cap with an error; keep the repeated-token check as a fast failure. Co-Authored-By: Claude Opus 5.5 --- .../aws/recommendations/usage_history.go | 20 +++--- .../usage_history_batch_test.go | 71 ++++++++++++++++--- 2 files changed, 69 insertions(+), 22 deletions(-) diff --git a/providers/aws/recommendations/usage_history.go b/providers/aws/recommendations/usage_history.go index 52b13045..f362ec76 100644 --- a/providers/aws/recommendations/usage_history.go +++ b/providers/aws/recommendations/usage_history.go @@ -292,24 +292,20 @@ func (c *Client) fetchGroupedDayCoverage(ctx context.Context, serviceFilter, reg Filter: serviceRegionFilter(serviceFilter, region), Metrics: []string{"Hour"}, } - var token *string - for { + return paginateCE(ctx, maxCoveragePages, "grouped daily coverage", func(token *string) (*string, error) { input.NextPageToken = token result, err := c.fetchCoveragePage(ctx, input) if err != nil { - return fmt.Errorf("failed to get daily coverage for %s/%s on %s: %w", serviceFilter, region, day.Format("2006-01-02"), err) + return nil, fmt.Errorf("failed to get daily coverage for %s/%s on %s: %w", serviceFilter, region, day.Format("2006-01-02"), err) } acc.addPeriods(result.CoveragesByTime) - if result.NextPageToken == nil || *result.NextPageToken == "" { - return nil + // Fail fast on a token that doesn't advance; paginateCE's cap catches + // longer cycles and endless fresh tokens. + if next := aws.ToString(result.NextPageToken); next != "" && next == aws.ToString(token) { + return nil, fmt.Errorf("failed to get daily coverage for %s/%s on %s: repeated page token", serviceFilter, region, day.Format("2006-01-02")) } - // A token that doesn't advance would repeat the same rate-limited - // request until ctx is canceled. - if token != nil && *token == *result.NextPageToken { - return fmt.Errorf("failed to get daily coverage for %s/%s on %s: repeated page token", serviceFilter, region, day.Format("2006-01-02")) - } - token = result.NextPageToken - } + return result.NextPageToken, nil + }) } // addPeriods applies each group's coverage percentage to the requested diff --git a/providers/aws/recommendations/usage_history_batch_test.go b/providers/aws/recommendations/usage_history_batch_test.go index f936c318..ace2f03e 100644 --- a/providers/aws/recommendations/usage_history_batch_test.go +++ b/providers/aws/recommendations/usage_history_batch_test.go @@ -30,7 +30,7 @@ type coverageFixture struct { pageSize int // grouped groups per page; 0 = single page data map[string]map[string]map[string]*string // region -> type -> day -> pct failDay string // grouped request for this day returns an error - stuckTok string // when set, every grouped page returns this NextPageToken + nextTok func(sent string) string // when set, grouped pages carry no groups and this NextPageToken mu sync.Mutex calls map[string]int // region -> requests @@ -100,6 +100,10 @@ func (f *coverageFixture) groupedResponse(w http.ResponseWriter, input *costexpl _, _ = w.Write([]byte(`{"__type":"DataUnavailableException","message":"no coverage available"}`)) return nil } + if f.nextTok != nil { + period := types.CoverageByTime{TimePeriod: &types.DateInterval{Start: aws.String(day)}} + return map[string]any{"CoveragesByTime": []types.CoverageByTime{period}, "NextPageToken": f.nextTok(aws.ToString(input.NextPageToken))} + } groups := f.dayGroups(region, day) offset := 0 if tok := aws.ToString(input.NextPageToken); tok != "" { @@ -112,9 +116,6 @@ func (f *coverageFixture) groupedResponse(w http.ResponseWriter, input *costexpl end = offset + f.pageSize next = aws.String(fmt.Sprintf("page-%d", end)) } - if f.stuckTok != "" { - next = aws.String(f.stuckTok) - } period := types.CoverageByTime{TimePeriod: &types.DateInterval{Start: aws.String(day)}, Groups: groups[offset:end]} return map[string]any{"CoveragesByTime": []types.CoverageByTime{period}, "NextPageToken": next} } @@ -325,18 +326,68 @@ func TestAttachDailyUsageHistory_GroupedErrorLeavesBucketNil(t *testing.T) { assert.True(t, strings.Contains(err.Error(), "DataUnavailableException"), err.Error()) } -func TestFetchGroupedDayCoverage_RepeatedPageTokenFails(t *testing.T) { - t.Parallel() +// fetchOneGroupedDay runs one grouped day against tokens served by nextTok +// and returns the grouped request count and the error. +func fetchOneGroupedDay(t *testing.T, nextTok func(string) string) (int, error) { + t.Helper() data, _ := manyTypeFixture(9) - f := &coverageFixture{t: t, data: map[string]map[string]map[string]*string{"us-east-1": data}, stuckTok: "page-0"} - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + f := &coverageFixture{t: t, data: map[string]map[string]map[string]*string{"us-east-1": data}, nextTok: nextTok} + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() day, err := time.Parse("2006-01-02", usageWindowDays()[0]) require.NoError(t, err) - err = newFixtureClient(t, f).fetchGroupedDayCoverage(ctx, ec2CoverageService, "us-east-1", day, &groupedDayAccumulator{}) + _, grouped := f.counts("us-east-1") + return grouped, err +} + +func TestFetchGroupedDayCoverage_RepeatedPageTokenFails(t *testing.T) { + t.Parallel() + grouped, err := fetchOneGroupedDay(t, func(string) string { return "same" }) require.Error(t, err) assert.Contains(t, err.Error(), "repeated page token") - _, grouped := f.counts("us-east-1") assert.Equal(t, 2, grouped, "stops at the first repeated token") } + +func TestFetchGroupedDayCoverage_EndlessFreshTokensHitCap(t *testing.T) { + t.Parallel() + grouped, err := fetchOneGroupedDay(t, func(sent string) string { + n := 0 + if sent != "" { + _, err := fmt.Sscanf(sent, "page-%d", &n) + require.NoError(t, err) + } + return fmt.Sprintf("page-%d", n+1) + }) + require.Error(t, err) + assert.Contains(t, err.Error(), fmt.Sprintf("pagination cap reached after %d pages", maxCoveragePages)) + assert.Equal(t, maxCoveragePages, grouped, "requests stop at the cap") +} + +func TestFetchGroupedDayCoverage_TokenCycleHitsCap(t *testing.T) { + t.Parallel() + grouped, err := fetchOneGroupedDay(t, func(sent string) string { + if sent == "A" { + return "B" + } + return "A" + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "pagination cap reached") + assert.Equal(t, maxCoveragePages, grouped, "an A, B, A, B cycle stops at the cap") +} + +func TestAttachDailyUsageHistory_EndlessGroupedTokensLeaveBucketNil(t *testing.T) { + t.Parallel() + data, names := manyTypeFixture(9) + f := &coverageFixture{t: t, data: map[string]map[string]map[string]*string{"us-east-1": data}, nextTok: func(sent string) string { return sent + "x" }} + recs := recsFor("us-east-1", names) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + newFixtureClient(t, f).AttachDailyUsageHistory(ctx, recs) + for _, r := range recs { + assert.Nil(t, r.UsageHistory, "%s: a capped fetch must not be reported as coverage", r.ResourceType) + } + _, grouped := f.counts("us-east-1") + assert.Equal(t, maxCoveragePages, grouped, "fetch stops at the cap on the first day") +}