diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 76ec0b712..6577d7cb1 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -496,6 +496,9 @@ func (m *mockConfigStore) UpsertRecommendations(_ context.Context, _ time.Time, func (m *mockConfigStore) ListStoredRecommendations(_ context.Context, _ config.RecommendationFilter) ([]config.RecommendationRecord, error) { return nil, nil } +func (m *mockConfigStore) ListStoredRecommendationsChunk(_ context.Context, _ config.RecommendationFilter, _ *config.RecommendationScanKey, _ int) (config.RecommendationChunk, error) { + return config.RecommendationChunk{Done: true}, nil +} func (m *mockConfigStore) GetRecommendationsFreshness(_ context.Context) (*config.RecommendationsFreshness, error) { return &config.RecommendationsFreshness{}, nil } diff --git a/internal/api/handler_recommendations.go b/internal/api/handler_recommendations.go index 0c2d33e75..2a0fe1570 100644 --- a/internal/api/handler_recommendations.go +++ b/internal/api/handler_recommendations.go @@ -94,6 +94,10 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc if err != nil { return nil, err } + page, err := parseRecommendationsPage(params) + if err != nil { + return nil, err + } // Resolve the account scope once: it decides both whether this read may // kick a background collection (issue #627) and the post-filter below. @@ -102,6 +106,10 @@ func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunc return nil, fmt.Errorf("failed to get allowed accounts: %w", err) } + if page.enabled { + return h.getRecommendationsPage(ctx, scope, filter, page) + } + recommendations, err := h.scheduler.ListRecommendations(recommendationsReadContext(ctx, scope), filter) if err != nil { return nil, fmt.Errorf("failed to get recommendations: %w", err) @@ -142,10 +150,18 @@ func (h *Handler) filterRecommendationsByScope(ctx context.Context, allowedAccou if allowedAccounts.AllowsAll() { return recs, nil } + nameByID, err := h.cloudAccountNamesByID(ctx) + if err != nil { + return nil, err + } + return filterRecommendationsInScope(allowedAccounts, nameByID, recs), nil +} - // Resolve account names so the allowed list can match on either ID or - // display name. Recommendations only carry CloudAccountID, so the name - // lookup needs a one-time fetch of the account list. +// cloudAccountNamesByID resolves account names so an allowed-accounts list +// can match on either ID or display name. Recommendations only carry +// CloudAccountID, so the name lookup needs a one-time fetch of the account +// list. +func (h *Handler) cloudAccountNamesByID(ctx context.Context) (map[string]string, error) { accounts, err := h.config.ListCloudAccounts(ctx, config.CloudAccountFilter{}) if err != nil { return nil, fmt.Errorf("failed to list accounts for filter: %w", err) @@ -155,7 +171,12 @@ func (h *Handler) filterRecommendationsByScope(ctx context.Context, allowedAccou a := accounts[_rvc] nameByID[a.ID] = a.Name } + return nameByID, nil +} +// filterRecommendationsInScope keeps the recs whose account allowedAccounts +// allows, in place and in order. Recs without an account are dropped. +func filterRecommendationsInScope(allowedAccounts auth.AccountScope, nameByID map[string]string, recs []config.RecommendationRecord) []config.RecommendationRecord { filtered := recs[:0] for _rvc := range recs { rec := recs[_rvc] @@ -167,7 +188,7 @@ func (h *Handler) filterRecommendationsByScope(ctx context.Context, allowedAccou filtered = append(filtered, rec) } } - return filtered, nil + return filtered } // filterRecommendationsByEnabledProviders removes recommendations for providers @@ -195,17 +216,22 @@ func (h *Handler) filterRecommendationsByEnabledProviders(ctx context.Context, r } // buildRecommendationsResponse calculates summary statistics and collects -// unique regions from a set of recommendations. +// unique regions from a set of recommendations, returning all of them as rows +// (the legacy unpaginated shape). func buildRecommendationsResponse(recommendations []config.RecommendationRecord) *RecommendationsResponse { - regionSet := make(map[string]struct{}) - var totalSavings, totalUpfront float64 - for i := range recommendations { - rec := &recommendations[i] - totalSavings += rec.Savings - totalUpfront += rec.UpfrontCost - if rec.Region != "" { - regionSet[rec.Region] = struct{}{} - } + return summarizeRecommendations(recommendations, recommendations) +} + +// summarizeRecommendations computes the summary and region list over all (the +// full filtered set) and returns rows as the response body. rows is either all +// itself or one page of it (issue #723), so the totals never shrink to the +// page. +func summarizeRecommendations(all, rows []config.RecommendationRecord) *RecommendationsResponse { + var summary recommendationsSummary + for i := range all { + summary.add(&all[i]) + } + for i := range rows { // Surface the compute size (VCPU / MemoryGB) at the top level so // the frontend Capacity column can render it without parsing the // opaque Details blob. The canonical values live in the typed @@ -214,26 +240,54 @@ func buildRecommendationsResponse(recommendations []config.RecommendationRecord) // pointers when this is a compute rec with a known size. Non-compute // recs, legacy rows with empty Details, and decode failures all // leave the pointers nil, so JSON omits them and the frontend - // renders "—". - stampComputeCapacity(rec) + // renders "—". Only returned rows are decoded. + stampComputeCapacity(&rows[i]) } + return summary.response(rows) +} + +// recommendationsSummary accumulates the full-set totals and region list of +// a list response one rec at a time, so a paginated scan can total every +// filtered row without holding them all (issue #981). +type recommendationsSummary struct { + count int + totalSavings float64 + totalUpfront float64 + regionSet map[string]struct{} +} + +func (s *recommendationsSummary) add(rec *config.RecommendationRecord) { + s.count++ + s.totalSavings += rec.Savings + s.totalUpfront += rec.UpfrontCost + if rec.Region == "" { + return + } + if s.regionSet == nil { + s.regionSet = make(map[string]struct{}) + } + s.regionSet[rec.Region] = struct{}{} +} - regions := make([]string, 0, len(regionSet)) - for r := range regionSet { +// response returns rows as the response body with the accumulated summary +// and regions. +func (s *recommendationsSummary) response(rows []config.RecommendationRecord) *RecommendationsResponse { + regions := make([]string, 0, len(s.regionSet)) + for r := range s.regionSet { regions = append(regions, r) } var avgPayback float64 - if totalSavings > 0 { - avgPayback = totalUpfront / totalSavings + if s.totalSavings > 0 { + avgPayback = s.totalUpfront / s.totalSavings } return &RecommendationsResponse{ - Recommendations: recommendations, + Recommendations: rows, Summary: RecommendationsSummary{ - TotalCount: len(recommendations), - TotalMonthlySavings: totalSavings, - TotalUpfrontCost: totalUpfront, + TotalCount: s.count, + TotalMonthlySavings: s.totalSavings, + TotalUpfrontCost: s.totalUpfront, AvgPaybackMonths: avgPayback, }, Regions: regions, diff --git a/internal/api/handler_recommendations_page.go b/internal/api/handler_recommendations_page.go new file mode 100644 index 000000000..68a23a9ba --- /dev/null +++ b/internal/api/handler_recommendations_page.go @@ -0,0 +1,166 @@ +package api + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "strconv" + "strings" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// Opt-in cursor pagination for GET /api/recommendations (refs issue #723). +// +// The list endpoint used to return the whole filtered cache in one body, +// which passes the 6 MB Lambda response cap once a tenant accumulates a few +// thousand recommendations. A caller that sends `limit` (and, for later +// pages, the `cursor` returned as `next_cursor`) gets one bounded page. +// A paginated request scans the filtered cache in ID-ordered chunks and keeps +// only the page (issue #981, see handler_recommendations_scan.go). A +// page is bounded twice: by row count (limit) and by serialized bytes +// (recommendationsPageByteBudget), because the opaque Details blob makes the +// size of a row data-dependent. +// A request carrying neither parameter keeps the existing unpaginated shape +// unchanged, so the current frontend (which sorts, groups and totals the +// full set client-side) keeps working until it adopts pagination. +// +// In both modes `summary` and `regions` describe the full filtered set, not +// just the returned page, so headline totals stay correct while paging. +const ( + // DefaultRecommendationsPageSize is the page size used when a caller + // sends a cursor without a limit. + DefaultRecommendationsPageSize = 500 + + // MaxRecommendationsPageSize caps the row count of one page. It is an + // upper bound only: recommendationsPageByteBudget can cut a page shorter. + MaxRecommendationsPageSize = 1000 + + // recommendationsPageByteBudget caps the bytes one page adds to the + // Lambda response payload. The Lambda limit (6 MiB, 6291456 bytes) + // applies to the whole invoke response: the JSON envelope with status + // code and headers, plus the body re-encoded as a JSON string, which + // escapes every quote and backslash again. Rows are therefore measured + // as they appear inside that string (see envelopeLen), and the budget is + // checked against uncompressed bytes because the #775 gzip only runs + // when the client sends Accept-Encoding: gzip. 4 MiB leaves about 2 MiB + // for the envelope, headers and next_cursor, which are a few KiB. + recommendationsPageByteBudget = 4 << 20 + + // maxRecommendationsCursorLen bounds the encoded cursor so a hostile + // query string cannot force a large decode. + maxRecommendationsCursorLen = 1024 +) + +// recommendationsPage is the parsed pagination request. enabled is false when +// the caller sent neither limit nor cursor (legacy unpaginated mode). +type recommendationsPage struct { + enabled bool + limit int + afterID string +} + +// parseRecommendationsPage validates the optional limit and cursor query +// parameters. limit must be an integer in 1..MaxRecommendationsPageSize; +// out-of-range values are rejected rather than clamped so a client bug does +// not silently return a different page size than it asked for. Legacy +// unpaginated mode is selected only when neither parameter is present: a +// present but empty value (?limit= or ?cursor=) is a 400, so a client bug +// cannot silently fall back to the unbounded response. +func parseRecommendationsPage(params map[string]string) (recommendationsPage, error) { + rawLimit, hasLimit := params["limit"] + rawCursor, hasCursor := params["cursor"] + if !hasLimit && !hasCursor { + return recommendationsPage{}, nil + } + // From here a present but empty value fails the checks below: "" is not + // an integer, and an empty cursor does not decode to an ID. + rawLimit = strings.TrimSpace(rawLimit) + rawCursor = strings.TrimSpace(rawCursor) + + page := recommendationsPage{enabled: true, limit: DefaultRecommendationsPageSize} + if hasLimit { + n, err := strconv.Atoi(rawLimit) + if err != nil || n < 1 || n > MaxRecommendationsPageSize { + return recommendationsPage{}, NewClientError(400, + "invalid limit: must be an integer between 1 and "+strconv.Itoa(MaxRecommendationsPageSize)) + } + page.limit = n + } + if hasCursor { + afterID, err := decodeRecommendationsCursor(rawCursor) + if err != nil { + return recommendationsPage{}, err + } + page.afterID = afterID + } + return page, nil +} + +// encodeRecommendationsCursor turns the last row ID of a page into an opaque, +// URL-safe cursor. Recommendation IDs can carry characters such as "|", so +// the raw ID is not used directly in the query string. +func encodeRecommendationsCursor(id string) string { + return base64.RawURLEncoding.EncodeToString([]byte(id)) +} + +// decodeRecommendationsCursor reverses encodeRecommendationsCursor and +// rejects malformed or oversized cursors with a 400. +func decodeRecommendationsCursor(cursor string) (string, error) { + if len(cursor) > maxRecommendationsCursorLen { + return "", NewClientError(400, "invalid cursor: too long") + } + raw, err := base64.RawURLEncoding.DecodeString(cursor) + if err != nil || len(raw) == 0 { + return "", NewClientError(400, "invalid cursor") + } + return string(raw), nil +} + +// envelopeLen returns how many bytes v adds to the Lambda invoke response: +// its JSON encoding, re-encoded as the content of the response's body +// string (without the surrounding quotes). +func envelopeLen(v any) (int, error) { + raw, err := json.Marshal(v) + if err != nil { + return 0, err + } + quoted, err := json.Marshal(string(raw)) + if err != nil { + return 0, err + } + return len(quoted) - 2, nil +} + +// selectRecommendationsPage stamps and measures rows from sorted[start:] and +// returns how many fit in one page: at most limit rows, and no more bytes +// than budget once the fixed overhead (summary, regions, keys) and the next +// cursor are counted. It always takes at least one row when any remain, so a +// row that alone exceeds the budget cannot stall the cursor walk; that row +// is returned on its own page with a logged warning (it only fails at Lambda +// if it is over the 6 MiB hard cap by itself, which no page size could fix). +func selectRecommendationsPage(sorted []config.RecommendationRecord, start, limit, overhead, budget int) (int, error) { + used := overhead + n := 0 + for i := start; i < len(sorted) && n < limit; i++ { + stampComputeCapacity(&sorted[i]) + size, err := envelopeLen(&sorted[i]) + if err != nil { + return 0, fmt.Errorf("failed to size recommendation %q: %w", sorted[i].ID, err) + } + // +1 for the separating comma; the cursor that would point at this + // row is reserved so next_cursor always fits. + cost := size + 1 + cursorLen := base64.RawURLEncoding.EncodedLen(len(sorted[i].ID)) + if n > 0 && used+cost+cursorLen > budget { + break + } + if n == 0 && used+cost+cursorLen > budget { + logging.Warnf("recommendation %q alone is %d bytes, over the %d-byte page budget; returning it on its own page", sorted[i].ID, size, budget) + } + used += cost + n++ + } + return n, nil +} diff --git a/internal/api/handler_recommendations_page_test.go b/internal/api/handler_recommendations_page_test.go new file mode 100644 index 000000000..e5f3ebd01 --- /dev/null +++ b/internal/api/handler_recommendations_page_test.go @@ -0,0 +1,424 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strconv" + "strings" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// lambdaResponseCapBytes is the AWS Lambda synchronous response payload cap +// that GET /api/recommendations used to exceed (issue #723). +const lambdaResponseCapBytes = 6 * 1024 * 1024 + +// pageTestRecs returns n recommendations in deliberately non-ID order, each +// padded so a single row is about 1 KB of JSON. +func pageTestRecs(n int) []config.RecommendationRecord { + pad := strings.Repeat("x", 900) + recs := make([]config.RecommendationRecord, n) + for i := range n { + // Reverse order so the handler has to sort to page stably. + idx := n - 1 - i + recs[i] = config.RecommendationRecord{ + ID: fmt.Sprintf("rec-%06d", idx), + Provider: "aws", + Service: "ec2", + Region: "us-east-" + strconv.Itoa(idx%3+1), + ResourceType: "m5.large-" + pad, + Savings: 10, + UpfrontCost: 20, + } + } + return recs +} + +func pageTestHandler(t *testing.T, recs []config.RecommendationRecord) *Handler { + t.Helper() + ms := new(MockScheduler) + // Unpaginated requests read the full list; paginated ones scan it in + // ID order (issue #981). + ms.On("ListRecommendations", mock.Anything, mock.Anything).Return(recs, nil) + ms.On("ScanRecommendations", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(recs, nil) + return &Handler{scheduler: ms, apiKey: "test-key"} +} + +// buildBudgetedPage pages recs the way a paginated request does: the rows +// reach the collector in ID order, as the store scan delivers them, and the +// page is cut with the given byte budget. +func buildBudgetedPage(recs []config.RecommendationRecord, page recommendationsPage, budget int) (*RecommendationsResponse, error) { + sorted := append([]config.RecommendationRecord(nil), recs...) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + collector := newRecommendationsPageCollector(page) + if err := collector.add(sorted); err != nil { + return nil, err + } + return collector.response(budget) +} + +func pageTestReq() *events.LambdaFunctionURLRequest { + return &events.LambdaFunctionURLRequest{Headers: map[string]string{"x-api-key": "test-key"}} +} + +// TestGetRecommendations_PaginationBoundsResponse is the #723 regression: a +// tenant whose full list is well over 6 MB must be able to fetch it in pages +// that each fit under the Lambda cap, and walking the cursor must return +// every row exactly once. +func TestGetRecommendations_PaginationBoundsResponse(t *testing.T) { + const total = 8000 + recs := pageTestRecs(total) + h := pageTestHandler(t, recs) + + full, err := json.Marshal(buildRecommendationsResponse(append([]config.RecommendationRecord(nil), recs...))) + require.NoError(t, err) + require.Greater(t, len(full), lambdaResponseCapBytes, "fixture must exceed the cap to be meaningful") + + seen := make(map[string]int, total) + var order []string + params := map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize)} + for pages := 0; ; pages++ { + require.Less(t, pages, total, "cursor walk did not terminate") + resp, err := h.getRecommendations(context.Background(), pageTestReq(), params) + require.NoError(t, err) + + body, err := json.Marshal(resp) + require.NoError(t, err) + assert.Less(t, len(body), lambdaResponseCapBytes, "page %d exceeds the Lambda response cap", pages) + assert.LessOrEqual(t, len(resp.Recommendations), MaxRecommendationsPageSize) + + // Summary describes the full filtered set on every page. + assert.Equal(t, total, resp.Summary.TotalCount) + assert.InDelta(t, float64(total*10), resp.Summary.TotalMonthlySavings, 1e-6) + assert.Len(t, resp.Regions, 3) + + for i := range resp.Recommendations { + id := resp.Recommendations[i].ID + seen[id]++ + order = append(order, id) + } + if resp.NextCursor == "" { + break + } + params = map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize), "cursor": resp.NextCursor} + } + + assert.Len(t, seen, total, "every row must be returned") + for id, n := range seen { + assert.Equal(t, 1, n, "row %s returned %d times", id, n) + } + assert.IsIncreasing(t, order, "rows must be returned in ID order across pages") +} + +// TestGetRecommendations_UnpaginatedShapeUnchanged pins backward +// compatibility: without limit/cursor the endpoint still returns every row, +// in the order the scheduler gave them, with no next_cursor. +func TestGetRecommendations_UnpaginatedShapeUnchanged(t *testing.T) { + recs := pageTestRecs(5) + // Copy the expected order before the call: the mock hands the handler + // this same backing array, so an in-place sort would also reorder recs. + wantIDs := make([]string, len(recs)) + for i := range recs { + wantIDs[i] = recs[i].ID + } + h := pageTestHandler(t, recs) + + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{}) + require.NoError(t, err) + require.Len(t, resp.Recommendations, 5) + gotIDs := make([]string, len(resp.Recommendations)) + for i := range resp.Recommendations { + gotIDs[i] = resp.Recommendations[i].ID + } + assert.Equal(t, wantIDs, gotIDs, "legacy mode must keep the scheduler's order") + assert.Empty(t, resp.NextCursor) + assert.Equal(t, 5, resp.Summary.TotalCount) + + body, err := json.Marshal(resp) + require.NoError(t, err) + assert.NotContains(t, string(body), "next_cursor") +} + +// TestGetRecommendations_PaginationRejectsDuplicateIDs pins the total-order +// guard: two rows sharing an ID would let the strict "after" comparison skip +// one of them at a page boundary (a,b,b,c with limit=2 used to return only +// three rows), so a paginated request fails instead of walking short. Legacy +// mode does not sort or page and still returns every row. +func TestGetRecommendations_PaginationRejectsDuplicateIDs(t *testing.T) { + recs := pageTestRecs(4) + recs[1].ID = recs[2].ID + h := pageTestHandler(t, recs) + ctx := context.Background() + + _, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "2"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "duplicate recommendation ID") + + legacy, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{}) + require.NoError(t, err) + assert.Len(t, legacy.Recommendations, 4) +} + +func TestGetRecommendations_PaginationPages(t *testing.T) { + recs := pageTestRecs(5) // rec-000000 .. rec-000004, stored in reverse + h := pageTestHandler(t, recs) + ctx := context.Background() + + first, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "2"}) + require.NoError(t, err) + require.Len(t, first.Recommendations, 2) + assert.Equal(t, "rec-000000", first.Recommendations[0].ID) + assert.Equal(t, "rec-000001", first.Recommendations[1].ID) + assert.Equal(t, encodeRecommendationsCursor("rec-000001"), first.NextCursor) + assert.Equal(t, 5, first.Summary.TotalCount) + + last, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{ + "limit": "2", "cursor": encodeRecommendationsCursor("rec-000003"), + }) + require.NoError(t, err) + require.Len(t, last.Recommendations, 1) + assert.Equal(t, "rec-000004", last.Recommendations[0].ID) + assert.Empty(t, last.NextCursor, "last page must not carry a cursor") + + // Exact fit: the page ending on the final row has no next cursor. + exact, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"limit": "5"}) + require.NoError(t, err) + assert.Len(t, exact.Recommendations, 5) + assert.Empty(t, exact.NextCursor) + + // Keyset: a cursor naming a row that has since disappeared resumes at + // the next ID instead of failing or restarting. + gone, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{ + "limit": "2", "cursor": encodeRecommendationsCursor("rec-000001x"), + }) + require.NoError(t, err) + require.Len(t, gone.Recommendations, 2) + assert.Equal(t, "rec-000002", gone.Recommendations[0].ID) + + // Cursor past the end yields an empty page, not an error. + past, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{ + "cursor": encodeRecommendationsCursor("rec-999999"), + }) + require.NoError(t, err) + assert.Empty(t, past.Recommendations) + assert.Empty(t, past.NextCursor) + assert.Equal(t, 5, past.Summary.TotalCount) +} + +func TestGetRecommendations_CursorWithoutLimitUsesDefaultPageSize(t *testing.T) { + recs := pageTestRecs(DefaultRecommendationsPageSize + 3) + h := pageTestHandler(t, recs) + + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{ + "cursor": encodeRecommendationsCursor("rec-"), + }) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, DefaultRecommendationsPageSize) + assert.NotEmpty(t, resp.NextCursor) +} + +func TestGetRecommendations_PaginationRejectsBadParams(t *testing.T) { + cases := map[string]map[string]string{ + "zero limit": {"limit": "0"}, + "negative limit": {"limit": "-1"}, + "limit above max": {"limit": strconv.Itoa(MaxRecommendationsPageSize + 1)}, + "non-numeric limit": {"limit": "abc"}, + "fractional limit": {"limit": "1.5"}, + "non-base64 cursor": {"cursor": "!!!"}, + // Valid base64 (length is a multiple of 4), so only the length cap + // can reject it. + "oversized cursor": {"cursor": strings.Repeat("A", maxRecommendationsCursorLen+4)}, + "bad cursor + limit": {"limit": "10", "cursor": "%%%"}, + // Present but empty must not fall back to the unbounded legacy shape. + "empty limit": {"limit": ""}, + "blank limit": {"limit": " "}, + "empty cursor": {"cursor": ""}, + "blank cursor": {"cursor": " \t"}, + "valid limit + empty cursor": {"limit": "10", "cursor": ""}, + "empty limit + valid cursor": {"limit": "", "cursor": encodeRecommendationsCursor("rec-000001")}, + } + for name, params := range cases { + t.Run(name, func(t *testing.T) { + // No scheduler: validation must reject before any store read. + h := &Handler{apiKey: "test-key"} + _, err := h.getRecommendations(context.Background(), pageTestReq(), params) + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "expected ClientError, got %T: %v", err, err) + assert.Equal(t, 400, ce.code) + }) + } +} + +func TestGetRecommendations_PaginationMaxLimitAccepted(t *testing.T) { + h := pageTestHandler(t, pageTestRecs(3)) + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{ + "limit": strconv.Itoa(MaxRecommendationsPageSize), + }) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, 3) + assert.Empty(t, resp.NextCursor) +} + +// bigDetailsRecs returns n recommendations whose Details blob is detailBytes +// of quote-heavy JSON. Each quote doubles again when the body is re-encoded +// into the Lambda response envelope, so measuring raw JSON would undercount. +func bigDetailsRecs(n, detailBytes int) []config.RecommendationRecord { + recs := pageTestRecs(n) + note := strings.Repeat(`"<&\\`, detailBytes/4) + details, err := json.Marshal(map[string]string{"note": note}) + if err != nil { + panic(err) + } + for i := range recs { + recs[i].ResourceType = "m5.large" + recs[i].Details = details + } + return recs +} + +// lambdaPayloadLen is the size of the Lambda invoke response that would carry +// resp, matching how handler.go and the Lambda runtime encode it. +func lambdaPayloadLen(t *testing.T, resp *RecommendationsResponse) (bodyInEnvelope, total int) { + t.Helper() + body, err := json.Marshal(resp) + require.NoError(t, err) + quoted, err := json.Marshal(string(body)) + require.NoError(t, err) + envelope, err := json.Marshal(events.LambdaFunctionURLResponse{ + StatusCode: 200, + Headers: map[string]string{"Content-Type": "application/json"}, + Body: string(body), + }) + require.NoError(t, err) + return len(quoted) - 2, len(envelope) +} + +// TestGetRecommendations_PageCutByByteBudget: with large Details, limit rows +// would be far over the Lambda cap, so pages must be cut by serialized bytes +// while still visiting every row exactly once, in order, with full totals. +func TestGetRecommendations_PageCutByByteBudget(t *testing.T) { + const total = 1500 + recs := bigDetailsRecs(total, 1500) // ~10 KB per row in the envelope + h := pageTestHandler(t, recs) + + _, fullSize := lambdaPayloadLen(t, buildRecommendationsResponse(append([]config.RecommendationRecord(nil), recs...))) + require.Greater(t, fullSize, 2*lambdaResponseCapBytes, "fixture must be far over the cap") + + seen := make(map[string]int, total) + var order []string + cutByBytes := false + params := map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize)} + for pages := 0; ; pages++ { + require.Less(t, pages, total, "cursor walk did not terminate") + resp, err := h.getRecommendations(context.Background(), pageTestReq(), params) + require.NoError(t, err) + + inEnvelope, payload := lambdaPayloadLen(t, resp) + assert.LessOrEqual(t, inEnvelope, recommendationsPageByteBudget, "page %d over the byte budget", pages) + assert.Less(t, payload, lambdaResponseCapBytes, "page %d over the Lambda cap", pages) + require.NotEmpty(t, resp.Recommendations, "a page must never be empty mid-walk") + if resp.NextCursor != "" && len(resp.Recommendations) < MaxRecommendationsPageSize { + cutByBytes = true + } + + assert.Equal(t, total, resp.Summary.TotalCount) + assert.InDelta(t, float64(total*10), resp.Summary.TotalMonthlySavings, 1e-6) + assert.InDelta(t, float64(total*20), resp.Summary.TotalUpfrontCost, 1e-6) + + for i := range resp.Recommendations { + seen[resp.Recommendations[i].ID]++ + order = append(order, resp.Recommendations[i].ID) + } + if resp.NextCursor == "" { + break + } + params = map[string]string{"limit": strconv.Itoa(MaxRecommendationsPageSize), "cursor": resp.NextCursor} + } + + assert.True(t, cutByBytes, "expected at least one page cut by the byte budget before limit") + assert.Len(t, seen, total) + for id, n := range seen { + assert.Equal(t, 1, n, "row %s returned %d times", id, n) + } + assert.IsIncreasing(t, order) +} + +// TestBuildBudgetedPage_OversizedRowAlone: a row that alone exceeds the +// budget is returned on its own page so the walk advances, and the rows +// around it still pack normally. +func TestBuildBudgetedPage_OversizedRowAlone(t *testing.T) { + recs := pageTestRecs(6) // ~1 KB rows: rec-000000 .. rec-000005 + huge := bigDetailsRecs(1, 8000)[0] + recs[2].Details = huge.Details // rec-000003 (pageTestRecs is reversed) + const budget = 4000 // fits a few small rows, never the big one + + var pages [][]string + page := recommendationsPage{enabled: true, limit: MaxRecommendationsPageSize} + for range len(recs) + 1 { + resp, err := buildBudgetedPage(append([]config.RecommendationRecord(nil), recs...), page, budget) + require.NoError(t, err) + require.NotEmpty(t, resp.Recommendations) + ids := make([]string, 0, len(resp.Recommendations)) + for i := range resp.Recommendations { + ids = append(ids, resp.Recommendations[i].ID) + } + pages = append(pages, ids) + if len(ids) > 1 { + inEnvelope, _ := lambdaPayloadLen(t, resp) + assert.LessOrEqual(t, inEnvelope, budget, "multi-row page over budget: %v", ids) + } + if resp.NextCursor == "" { + break + } + cursor, err := decodeRecommendationsCursor(resp.NextCursor) + require.NoError(t, err) + page.afterID = cursor + } + + assert.Contains(t, pages, []string{"rec-000003"}, "oversized row must be alone on its page") + all := make([]string, 0, len(recs)) + for _, p := range pages { + all = append(all, p...) + } + assert.Equal(t, []string{"rec-000000", "rec-000001", "rec-000002", "rec-000003", "rec-000004", "rec-000005"}, all) + assert.Greater(t, len(pages[0]), 1, "small rows before the big one must still share a page") +} + +// TestBuildBudgetedPage_TightBudgetSweep sweeps the budget one byte at a time +// so the accounting is checked at its exact edges: every multi-row page must +// fit (counting the summary, regions, keys and next_cursor), and the page +// must be maximal, i.e. one more row would not have fit. +func TestBuildBudgetedPage_TightBudgetSweep(t *testing.T) { + recs := pageTestRecs(6) + page := recommendationsPage{enabled: true, limit: MaxRecommendationsPageSize} + for budget := 1500; budget <= 7000; budget++ { + resp, err := buildBudgetedPage(append([]config.RecommendationRecord(nil), recs...), page, budget) + require.NoError(t, err) + n := len(resp.Recommendations) + require.GreaterOrEqual(t, n, 1) + if n > 1 { + inEnvelope, _ := lambdaPayloadLen(t, resp) + require.LessOrEqual(t, inEnvelope, budget, "budget %d: %d rows do not fit", budget, n) + } + if n < len(recs) { + // The same page plus the next row, with that row's cursor. + bigger, err := buildBudgetedPage(append([]config.RecommendationRecord(nil), recs...), + recommendationsPage{enabled: true, limit: n + 1}, 1<<30) + require.NoError(t, err) + bigger.NextCursor = encodeRecommendationsCursor(bigger.Recommendations[n].ID) + inEnvelope, _ := lambdaPayloadLen(t, bigger) + // The paginator overcounts by at most 2 bytes (one comma and the + // one-byte cursor placeholder in the overhead), so anything it + // rejected must have been within 2 bytes of overflowing. + require.Greater(t, inEnvelope+2, budget, "budget %d: page of %d rows is not maximal", budget, n) + } + } +} diff --git a/internal/api/handler_recommendations_scan.go b/internal/api/handler_recommendations_scan.go new file mode 100644 index 000000000..4ef8aae48 --- /dev/null +++ b/internal/api/handler_recommendations_scan.go @@ -0,0 +1,142 @@ +package api + +import ( + "context" + "fmt" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// recommendationsScanChunkSize is how many cache rows a paginated +// GET /api/recommendations reads per store query. Peak memory is about one +// chunk plus one page, whatever the tenant's size. +const recommendationsScanChunkSize = 500 + +// getRecommendationsPage serves a paginated GET /api/recommendations (issue +// #981). Instead of loading and sorting the whole filtered set, it scans the +// cache in ID-ordered chunks through the scheduler (which applies +// suppressions and account overrides), drops rows for disabled providers and +// accounts outside the caller's scope, and hands the survivors to a collector +// that keeps only the requested page while totaling every row for the +// summary. +// +// The summary still needs every filtered row, so each request still reads the +// whole filtered set from the store; what no longer grows with the tenant is +// the memory held and the in-process sort. Pushing the summary into SQL would +// need suppressions, account overrides, the provider and scope filters and +// the savings-percentage floor in SQL too; until then a SQL aggregate would +// count rows the response hides. +func (h *Handler) getRecommendationsPage(ctx context.Context, scope auth.AccountScope, filter config.RecommendationFilter, page recommendationsPage) (*RecommendationsResponse, error) { + keep, err := h.recommendationsRowFilter(ctx, scope) + if err != nil { + return nil, err + } + collector := newRecommendationsPageCollector(page) + err = h.scheduler.ScanRecommendations(recommendationsReadContext(ctx, scope), filter, recommendationsScanChunkSize, + func(chunk []config.RecommendationRecord) error { + return collector.add(keep(chunk)) + }) + if err != nil { + return nil, fmt.Errorf("failed to get recommendations: %w", err) + } + return collector.response(recommendationsPageByteBudget) +} + +// recommendationsRowFilter returns the handler-side row filters of +// getRecommendations (enabled providers, issue #1409, then the caller's +// account scope) as one function, with their inputs read once up front +// rather than once per chunk. +func (h *Handler) recommendationsRowFilter(ctx context.Context, scope auth.AccountScope) (func([]config.RecommendationRecord) []config.RecommendationRecord, error) { + var enabledProviders []string + if h.config != nil { + globalCfg, err := h.config.GetGlobalConfig(ctx) + if err != nil { + return nil, fmt.Errorf("failed to read global config for enabled-providers filter: %w", err) + } + enabledProviders = globalCfg.EnabledProviders + } + var nameByID map[string]string + if !scope.AllowsAll() { + var err error + if nameByID, err = h.cloudAccountNamesByID(ctx); err != nil { + return nil, err + } + } + return func(recs []config.RecommendationRecord) []config.RecommendationRecord { + recs = config.FilterRecommendationsByEnabledProviders(recs, enabledProviders) + if scope.AllowsAll() { + return recs + } + return filterRecommendationsInScope(scope, nameByID, recs) + }, nil +} + +// recommendationsPageCollector builds one page from a scan's surviving rows, +// which arrive in strictly increasing ID order. It totals every row into the +// summary but keeps at most limit+1 rows after the cursor: the page, plus one +// row that proves a next page exists. +type recommendationsPageCollector struct { + page recommendationsPage + summary recommendationsSummary + seenAny bool + lastID string + rows []config.RecommendationRecord +} + +func newRecommendationsPageCollector(page recommendationsPage) *recommendationsPageCollector { + return &recommendationsPageCollector{page: page, rows: []config.RecommendationRecord{}} +} + +// add consumes the next rows of the scan. Paging needs ID order to be a total +// order: two rows sharing an ID would let the strict "after" comparison skip +// the second one at a page boundary, so a duplicate fails the request rather +// than silently walking short. The ID is the application recommendation ID +// built from the natural-key columns (recommendationID in +// internal/scheduler), not the cache table's UUID primary key. A row out of +// order means the scan broke its contract and would make the cursor skip +// rows, so it fails too. +func (c *recommendationsPageCollector) add(recs []config.RecommendationRecord) error { + for i := range recs { + rec := &recs[i] + if c.seenAny && rec.ID <= c.lastID { + if rec.ID == c.lastID { + return fmt.Errorf("duplicate recommendation ID %q: cannot page the recommendations list", rec.ID) + } + return fmt.Errorf("recommendations scan out of ID order: %q after %q", rec.ID, c.lastID) + } + c.seenAny, c.lastID = true, rec.ID + c.summary.add(rec) + // Resuming after the last ID the client saw (rather than at an + // offset) means a row added or removed between two requests does not + // shift later pages, even if the cursor row itself is gone. + if len(c.rows) <= c.page.limit && (c.page.afterID == "" || rec.ID > c.page.afterID) { + c.rows = append(c.rows, *rec) + } + } + return nil +} + +// response cuts the collected rows to one page by row count and serialized +// bytes (see selectRecommendationsPage) and sets next_cursor when any row is +// left after it. +func (c *recommendationsPageCollector) response(budget int) (*RecommendationsResponse, error) { + resp := c.summary.response(c.rows[:0]) + // Overhead: everything except the rows and the cursor value, measured + // with an empty next_cursor key present. + resp.NextCursor = " " + overhead, err := envelopeLen(resp) + if err != nil { + return nil, fmt.Errorf("failed to size recommendations page: %w", err) + } + n, err := selectRecommendationsPage(c.rows, 0, c.page.limit, overhead, budget) + if err != nil { + return nil, err + } + resp.Recommendations = c.rows[:n] + resp.NextCursor = "" + if n < len(c.rows) { + resp.NextCursor = encodeRecommendationsCursor(c.rows[n-1].ID) + } + return resp, nil +} diff --git a/internal/api/handler_recommendations_scan_integration_test.go b/internal/api/handler_recommendations_scan_integration_test.go new file mode 100644 index 000000000..a66c2a2a7 --- /dev/null +++ b/internal/api/handler_recommendations_scan_integration_test.go @@ -0,0 +1,90 @@ +//go:build integration + +package api + +import ( + "context" + "fmt" + "sort" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" +) + +// TestGetRecommendations_PagedScanMatchesLegacyOnPostgres runs the paginated +// GET /api/recommendations end to end on a real migrated database (issue +// #981): handler, scheduler scan and the keyset chunk query. The cache holds +// more rows than one scan chunk, with IDs whose byte order differs from a +// linguistic order, and the request carries the in-process +// min_savings_pct floor. Walking the cursor must return exactly the legacy +// unpaginated rows in Go string order, with the legacy summary on every page. +func TestGetRecommendations_PagedScanMatchesLegacyOnPostgres(t *testing.T) { + ctx := t.Context() + pg, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) }) + require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", "")) + + store := config.NewPostgresStore(pg.DB) + prefixes := []string{"aws", "AWS", "aws-x", "aws x", "Aws", "élan", "elan", "_z", "aws/x"} + const total = 2*recommendationsScanChunkSize + 37 + recs := make([]config.RecommendationRecord, total) + for i := range recs { + pct := float64(10 + i%50) // rows with pct < 20 fall under the floor below + recs[i] = config.RecommendationRecord{ + ID: fmt.Sprintf("%s|%04d", prefixes[i%len(prefixes)], total-i), + Provider: "aws", + Service: "ec2", + Region: "us-east-" + strconv.Itoa(i%3+1), + ResourceType: fmt.Sprintf("m5.r%04d", i), + Count: 1, + Term: 1, + Payment: "no-upfront", + Savings: float64(i % 7), + UpfrontCost: float64(i % 5), + SavingsPercentage: &pct, + } + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + sched := scheduler.NewScheduler(scheduler.SchedulerConfig{ConfigStore: store, IsLambda: true}) + h := &Handler{config: store, scheduler: sched, apiKey: "test-key"} + + legacy, err := h.getRecommendations(ctx, pageTestReq(), map[string]string{"min_savings_pct": "20"}) + require.NoError(t, err) + want := make([]string, len(legacy.Recommendations)) + for i := range legacy.Recommendations { + want[i] = legacy.Recommendations[i].ID + } + sort.Strings(want) + require.Greater(t, len(want), recommendationsScanChunkSize, "the walk must span several scan chunks") + require.Less(t, len(want), total, "the pct floor must drop some rows") + + const limit = 97 + var got []string + params := map[string]string{"min_savings_pct": "20", "limit": strconv.Itoa(limit)} + for page := 0; ; page++ { + require.LessOrEqual(t, page, len(want)/limit+1, "cursor walk did not terminate") + resp, err := h.getRecommendations(ctx, pageTestReq(), params) + require.NoError(t, err) + assert.Equal(t, legacy.Summary, resp.Summary, "page %d summary", page) + assert.ElementsMatch(t, legacy.Regions, resp.Regions) + for i := range resp.Recommendations { + got = append(got, resp.Recommendations[i].ID) + } + if resp.NextCursor == "" { + break + } + require.Len(t, resp.Recommendations, limit, "only the last page may be short") + params = map[string]string{"min_savings_pct": "20", "limit": strconv.Itoa(limit), "cursor": resp.NextCursor} + } + assert.Equal(t, want, got) +} diff --git a/internal/api/handler_recommendations_scan_test.go b/internal/api/handler_recommendations_scan_test.go new file mode 100644 index 000000000..7a43669e1 --- /dev/null +++ b/internal/api/handler_recommendations_scan_test.go @@ -0,0 +1,254 @@ +package api + +import ( + "context" + "sort" + "strconv" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// chunkScanScheduler serves ScanRecommendations from recs like the store +// scan does (ID order, chunk rows per call) but with a chunk size the test +// picks, so page and chunk boundaries can be made to cross. Every other +// scheduler method goes to the embedded mock, which has no expectations: a +// paginated request that falls back to ListRecommendations panics. +type chunkScanScheduler struct { + *MockScheduler + recs []config.RecommendationRecord + chunk int + ctxs []context.Context +} + +func (s *chunkScanScheduler) ScanRecommendations(ctx context.Context, _ config.RecommendationFilter, _ int, fn func([]config.RecommendationRecord) error) error { + s.ctxs = append(s.ctxs, ctx) + sorted := append([]config.RecommendationRecord(nil), s.recs...) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + for start := 0; start < len(sorted); start += s.chunk { + end := min(start+s.chunk, len(sorted)) + // A fresh slice per chunk, as the store returns. + if err := fn(append([]config.RecommendationRecord(nil), sorted[start:end]...)); err != nil { + return err + } + } + return nil +} + +// scanFilterRecs mixes IDs that a linguistic collation orders differently +// from bytes with rows the handler-side filters drop: an Azure row (only AWS +// is enabled in recsReadStore), a row in an account outside the scoped +// user's scope and a row with no account. +func scanFilterRecs() []config.RecommendationRecord { + in, out := scopedInAccount, "22222222-2222-4222-8222-222222222222" + mk := func(id, provider string, account *string, region string, savings float64) config.RecommendationRecord { + return config.RecommendationRecord{ + ID: id, Provider: provider, Service: "ec2", Region: region, ResourceType: "m5.large", + CloudAccountID: account, Savings: savings, UpfrontCost: 2 * savings, + } + } + return []config.RecommendationRecord{ + mk("aws|b", "aws", &in, "us-east-1", 1), + mk("AWS|a", "aws", &in, "us-east-2", 2), + mk("aws-a", "aws", &in, "eu-west-1", 3), + mk("aws a", "aws", &in, "us-east-1", 4), + mk("azure|x", "azure", &in, "westeurope", 5), + mk("aws/a", "aws", &out, "ap-south-1", 6), + mk("aws|a", "aws", &in, "us-east-1", 7), + mk("Zeta", "aws", &in, "us-west-2", 8), + mk("ambient", "aws", nil, "us-east-1", 9), + mk("élan", "aws", &in, "us-east-1", 10), + mk("elan", "aws", &in, "sa-east-1", 11), + } +} + +func scanTestHandler(recs []config.RecommendationRecord, chunk int, scoped bool) (*Handler, *chunkScanScheduler) { + sched := &chunkScanScheduler{MockScheduler: new(MockScheduler), recs: recs, chunk: chunk} + accounts := []string(nil) + if scoped { + accounts = []string{scopedInAccount} + } + return &Handler{config: recsReadStore(), scheduler: sched, auth: scopedAuth(accounts)}, sched +} + +// legacyResponse is the unpaginated response for recs, read through +// ListRecommendations on a separate handler with its own copy (the legacy +// scope filter compacts its input in place). +func legacyResponse(t *testing.T, recs []config.RecommendationRecord, scoped bool) *RecommendationsResponse { + t.Helper() + ms := new(MockScheduler) + ms.On("ListRecommendations", mock.Anything, mock.Anything). + Return(append([]config.RecommendationRecord(nil), recs...), nil) + accounts := []string(nil) + if scoped { + accounts = []string{scopedInAccount} + } + h := &Handler{config: recsReadStore(), scheduler: ms, auth: scopedAuth(accounts)} + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{}) + require.NoError(t, err) + return resp +} + +// TestGetRecommendations_PaginatedRequestDoesNotLoadFullList is the #981 +// regression: a paginated request must page from the chunked scan and never +// read the whole filtered set through ListRecommendations. +func TestGetRecommendations_PaginatedRequestDoesNotLoadFullList(t *testing.T) { + recs := pageTestRecs(5) + ms := new(MockScheduler) + ms.On("ScanRecommendations", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(recs, nil) + h := &Handler{scheduler: ms, apiKey: "test-key"} + + resp, err := h.getRecommendations(context.Background(), pageTestReq(), map[string]string{"limit": "2"}) + require.NoError(t, err) + require.Len(t, resp.Recommendations, 2) + assert.Equal(t, 5, resp.Summary.TotalCount) + ms.AssertNotCalled(t, "ListRecommendations", mock.Anything, mock.Anything) + ms.AssertCalled(t, "ScanRecommendations", mock.Anything, mock.Anything, recommendationsScanChunkSize, mock.Anything) +} + +// TestGetRecommendations_PagedWalkMatchesLegacyOrder pins the paginated +// output to what the in-memory paging returned: the legacy (unpaginated) rows +// after every filter, sorted by Go string order, cut into consecutive pages +// of limit rows, with next_cursor on every page but the last and the +// full-set summary on every page. It runs for an unrestricted and a scoped +// reader, across chunk sizes that put chunk boundaries before, on and after +// page boundaries, including the exact-fit last page. +func TestGetRecommendations_PagedWalkMatchesLegacyOrder(t *testing.T) { + recs := scanFilterRecs() + for _, scoped := range []bool{false, true} { + legacy := legacyResponse(t, recs, scoped) + want := make([]string, len(legacy.Recommendations)) + for i := range legacy.Recommendations { + want[i] = legacy.Recommendations[i].ID + } + sort.Strings(want) + require.NotEmpty(t, want) + if scoped { + require.Len(t, want, 8, "fixture: Azure, out-of-scope and account-less rows dropped") + } + n := len(want) + for _, chunk := range []int{1, 2, 3, n, n + 1} { + for _, limit := range []int{1, 2, 3, n - 1, n, n + 1} { + t.Run("scoped="+strconv.FormatBool(scoped)+"/chunk="+strconv.Itoa(chunk)+"/limit="+strconv.Itoa(limit), func(t *testing.T) { + h, _ := scanTestHandler(recs, chunk, scoped) + var got []string + params := map[string]string{"limit": strconv.Itoa(limit)} + for page := 0; ; page++ { + require.LessOrEqual(t, page, n, "cursor walk did not terminate") + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), params) + require.NoError(t, err) + + start := page * limit + end := min(start+limit, n) + ids := make([]string, len(resp.Recommendations)) + for i := range resp.Recommendations { + ids[i] = resp.Recommendations[i].ID + } + require.Equal(t, want[start:end], ids, "page %d", page) + assert.Equal(t, legacy.Summary, resp.Summary, "summary must cover the full filtered set") + assert.ElementsMatch(t, legacy.Regions, resp.Regions) + got = append(got, ids...) + if end == n { + assert.Empty(t, resp.NextCursor, "the last page carries no cursor") + break + } + require.Equal(t, encodeRecommendationsCursor(want[end-1]), resp.NextCursor) + params = map[string]string{"limit": strconv.Itoa(limit), "cursor": resp.NextCursor} + } + assert.Equal(t, want, got) + }) + } + } + } +} + +// TestGetRecommendations_PagedDuplicateIDs pins the tie handling to the +// in-memory paging: two surviving rows sharing an ID fail every paginated +// request, wherever the chunk and page boundaries fall and even when the +// duplicate lies beyond the requested page, while a duplicate whose twin a +// filter removes is not a tie at all. +func TestGetRecommendations_PagedDuplicateIDs(t *testing.T) { + recs := scanFilterRecs() + dup := recs[0] // "aws|b", in scope + dup.Region = "us-west-1" + recs = append(recs, dup) + + for _, chunk := range []int{1, 2, 3, len(recs)} { + for _, limit := range []string{"1", "2", "100"} { + h, _ := scanTestHandler(recs, chunk, false) + _, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": limit}) + require.Error(t, err, "chunk=%d limit=%s", chunk, limit) + assert.Contains(t, err.Error(), `duplicate recommendation ID "aws|b"`) + } + } + + // The twin is in an account outside the scoped reader's scope, so the + // reader sees one "aws|b" and pages normally. + outside := "22222222-2222-4222-8222-222222222222" + recs[len(recs)-1].CloudAccountID = &outside + h, _ := scanTestHandler(recs, 1, true) + resp, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": "100"}) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, 8) +} + +// TestGetRecommendations_PagedReadBackgroundCollectionScope keeps issue #627 +// on the paginated path: only an unrestricted reader's scan may kick a +// background collection. +func TestGetRecommendations_PagedReadBackgroundCollectionScope(t *testing.T) { + for _, scoped := range []bool{false, true} { + h, sched := scanTestHandler(scanFilterRecs(), 4, scoped) + _, err := h.getRecommendations(context.Background(), scopedRequest(""), map[string]string{"limit": "3"}) + require.NoError(t, err) + require.Len(t, sched.ctxs, 1, "one scan per request") + assert.Equal(t, !scoped, scheduler.BackgroundCollectionAllowed(sched.ctxs[0]), "scoped=%v", scoped) + } +} + +// TestRecommendationsPageCollector_HoldsOnePage checks the memory bound: after +// a long scan the collector holds the page plus one look-ahead row, while +// the summary counts every row. +func TestRecommendationsPageCollector_HoldsOnePage(t *testing.T) { + recs := pageTestRecs(2000) + sort.Slice(recs, func(i, j int) bool { return recs[i].ID < recs[j].ID }) + c := newRecommendationsPageCollector(recommendationsPage{enabled: true, limit: 10, afterID: recs[99].ID}) + for start := 0; start < len(recs); start += 37 { + require.NoError(t, c.add(recs[start:min(start+37, len(recs))])) + } + assert.Len(t, c.rows, 11, "page plus one look-ahead row") + assert.Equal(t, recs[100].ID, c.rows[0].ID, "the page starts strictly after the cursor") + + resp, err := c.response(recommendationsPageByteBudget) + require.NoError(t, err) + assert.Len(t, resp.Recommendations, 10) + assert.Equal(t, 2000, resp.Summary.TotalCount) + assert.Equal(t, encodeRecommendationsCursor(recs[109].ID), resp.NextCursor) +} + +// TestRecommendationsPageCollector_RejectsOutOfOrderScan fails loudly when the +// scan breaks its ordering contract, which would make the cursor skip rows. +func TestRecommendationsPageCollector_RejectsOutOfOrderScan(t *testing.T) { + c := newRecommendationsPageCollector(recommendationsPage{enabled: true, limit: 10}) + require.NoError(t, c.add([]config.RecommendationRecord{{ID: "b"}})) + err := c.add([]config.RecommendationRecord{{ID: "a"}}) + require.ErrorContains(t, err, "out of ID order") +} + +// TestGetRecommendations_PagedEmptyCache returns an empty page as a JSON +// array, with zero totals and no cursor. +func TestGetRecommendations_PagedEmptyCache(t *testing.T) { + h, _ := scanTestHandler(nil, 3, false) + resp, err := h.getRecommendations(context.Background(), &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer scoped-token"}, + }, map[string]string{"limit": "5"}) + require.NoError(t, err) + require.NotNil(t, resp.Recommendations) + assert.Empty(t, resp.Recommendations) + assert.Empty(t, resp.NextCursor) + assert.Equal(t, 0, resp.Summary.TotalCount) +} diff --git a/internal/api/mocks_test.go b/internal/api/mocks_test.go index 2851eb04c..2f9c05d57 100644 --- a/internal/api/mocks_test.go +++ b/internal/api/mocks_test.go @@ -3,6 +3,7 @@ package api import ( "context" "fmt" + "sort" "sync" "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" @@ -138,6 +139,26 @@ func (m *MockScheduler) ListRecommendations(ctx context.Context, filter config.R return args.Get(0).([]config.RecommendationRecord), args.Error(1) } +// ScanRecommendations replays the []config.RecommendationRecord registered +// for it the way the store-backed scan does: sorted by ID (a stable sort, so +// duplicate IDs keep their relative order), chunkSize rows per call to fn, +// stopping at fn's first error. The second return value is the scan's own +// error. +func (m *MockScheduler) ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error { + args := m.Called(ctx, filter, chunkSize, fn) + if recs, ok := args.Get(0).([]config.RecommendationRecord); ok { + sorted := append([]config.RecommendationRecord(nil), recs...) + sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID }) + for start := 0; start < len(sorted); start += chunkSize { + end := min(start+chunkSize, len(sorted)) + if err := fn(sorted[start:end]); err != nil { + return err + } + } + } + return args.Error(1) +} + func (m *MockScheduler) GetRecommendationByID(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) { args := m.Called(ctx, id) var rec *config.RecommendationRecord diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index ba01d7f22..d5a0b4bd4 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -294,6 +294,30 @@ paths: schema: type: string - $ref: '#/components/parameters/AccountID' + - name: limit + in: query + description: >- + Opt-in maximum page size in rows (1-1000). When neither limit nor + cursor is present, every matching recommendation is returned in + one response (legacy behavior); a present but empty value is a + 400. When either is sent, rows are returned in ascending id order, + one page at a time, and a page can hold fewer than limit rows + (follow next_cursor, not the row count) because each page is also + capped at 4 MiB of serialized JSON. A single row larger than that + is returned alone. summary and regions still describe the full + filtered set. + schema: + type: integer + minimum: 1 + maximum: 1000 + - name: cursor + in: query + description: >- + Opaque next_cursor value from the previous page. Implies + pagination; the page size defaults to 500 when limit is omitted. + schema: + type: string + maxLength: 1024 responses: '200': description: Recommendations list @@ -301,6 +325,17 @@ paths: application/json: schema: $ref: '#/components/schemas/RecommendationsResponse' + '400': + description: > + Invalid query parameter: a limit that is empty, not an integer or + outside 1-1000; a cursor that is empty, longer than 1024 + characters or not a cursor this endpoint issued; or an invalid + provider, service, region, account_ids, min_savings_usd or + min_savings_pct filter. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' '401': $ref: '#/components/responses/Unauthorized' '403': @@ -3516,11 +3551,33 @@ components: type: array items: $ref: '#/components/schemas/RecommendationRecord' - total_savings: - type: number - format: double - count: - type: integer + regions: + type: array + items: + type: string + summary: + type: object + description: Totals over the full filtered set, not just the returned page. + properties: + total_count: + type: integer + total_monthly_savings: + type: number + format: double + total_upfront_cost: + type: number + format: double + avg_payback_months: + type: number + format: double + next_cursor: + type: string + description: >- + Present only on a paginated request that has more rows; pass it as + the cursor query parameter to fetch the next page. Its absence is + the only end-of-list signal: a page with fewer rows than limit is + not necessarily the last one, because each page is also capped by + serialized size. RecommendationRecord: type: object diff --git a/internal/api/types.go b/internal/api/types.go index 9ff2a91f1..ff69ba36a 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -170,6 +170,10 @@ type PurchaseManagerInterface interface { type SchedulerInterface interface { CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) ListRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) + // ScanRecommendations walks the same rows as ListRecommendations in + // byte-wise ID order, chunkSize store rows at a time, calling fn with + // each chunk's survivors (issue #981). + ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error // GetRecommendationByID fetches a single rec by its application-level id, // bypassing account-override filtering so deep-linked URLs to override- // hidden recs resolve. hiddenBy is non-nil when the rec would be dropped by @@ -463,6 +467,9 @@ type RecommendationsResponse struct { Recommendations []config.RecommendationRecord `json:"recommendations"` Regions []string `json:"regions"` Summary RecommendationsSummary `json:"summary"` + // NextCursor is set only on a paginated request (limit or cursor sent) + // that has more rows; pass it back as ?cursor= for the next page. + NextCursor string `json:"next_cursor,omitempty"` } // UsagePoint is a single sample in the per-recommendation usage time diff --git a/internal/config/export_recommendations_scan_test.go b/internal/config/export_recommendations_scan_test.go new file mode 100644 index 000000000..cf1fb3e76 --- /dev/null +++ b/internal/config/export_recommendations_scan_test.go @@ -0,0 +1,6 @@ +package config + +// BuildRecommendationChunkQuery exposes the keyset chunk query to the external +// integration tests, which EXPLAIN it to prove recommendations_scan_key_idx +// (migration 000142) can serve it. +var BuildRecommendationChunkQuery = buildRecommendationChunkQuery diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 9a078709a..e448e0bea 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -556,6 +556,10 @@ type StoreInterface interface { ReplaceRecommendations(ctx context.Context, collectedAt time.Time, recs []RecommendationRecord) error UpsertRecommendations(ctx context.Context, collectedAt time.Time, recs []RecommendationRecord, successfulCollects []SuccessfulCollect) error ListStoredRecommendations(ctx context.Context, filter RecommendationFilter) ([]RecommendationRecord, error) + // ListStoredRecommendationsChunk is ListStoredRecommendations for a + // keyset walk: at most limit rows after the position after, in + // byte-wise ID order then primary-key order (issue #981). + ListStoredRecommendationsChunk(ctx context.Context, filter RecommendationFilter, after *RecommendationScanKey, limit int) (RecommendationChunk, error) GetRecommendationsFreshness(ctx context.Context) (*RecommendationsFreshness, error) SetRecommendationsCollectionError(ctx context.Context, errMsg string) error // SetRecommendationsCollectionPartial records (or, with nil, clears) the diff --git a/internal/config/store_postgres_recommendations.go b/internal/config/store_postgres_recommendations.go index d95877c47..87513e902 100644 --- a/internal/config/store_postgres_recommendations.go +++ b/internal/config/store_postgres_recommendations.go @@ -410,19 +410,13 @@ func (s *PostgresStore) ListStoredRecommendations(ctx context.Context, filter Re if err := rows.Scan(&payload); err != nil { return nil, fmt.Errorf("failed to scan payload: %w", err) } - var rec RecommendationRecord - if err := json.Unmarshal(payload, &rec); err != nil { - return nil, fmt.Errorf("failed to unmarshal payload: %w", err) + rec, keep, err := decodeStoredRecommendation(payload, filter) + if err != nil { + return nil, err } - // Apply the in-process percentage floor. Recs where the percentage - // cannot be computed (missing on-demand baseline) pass through -- we - // only drop a rec when pct is computable AND below the threshold. - if filter.MinSavingsPct > 0 { - if pct, ok := recEffectiveSavingsPct(&rec); ok && pct < filter.MinSavingsPct { - continue - } + if keep { + out = append(out, rec) } - out = append(out, rec) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("rows iteration: %w", err) @@ -430,6 +424,23 @@ func (s *PostgresStore) ListStoredRecommendations(ctx context.Context, filter Re return out, nil } +// decodeStoredRecommendation unmarshals one stored payload and applies the +// in-process MinSavingsPct floor. keep is false when the floor drops the rec. +// Recs where the percentage cannot be computed (missing on-demand baseline) +// pass through: a rec is only dropped when pct is computable AND below the +// threshold. +func decodeStoredRecommendation(payload []byte, filter RecommendationFilter) (rec RecommendationRecord, keep bool, err error) { + if err := json.Unmarshal(payload, &rec); err != nil { + return RecommendationRecord{}, false, fmt.Errorf("failed to unmarshal payload: %w", err) + } + if filter.MinSavingsPct > 0 { + if pct, ok := recEffectiveSavingsPct(&rec); ok && pct < filter.MinSavingsPct { + return rec, false, nil + } + } + return rec, true, nil +} + // GetRecommendationsFreshness returns the singleton freshness row. The // table is seeded with id=1 by the migration so a row always exists; // LastCollectedAt, LastCollectionError, LastCollectionStartedAt and diff --git a/internal/config/store_postgres_recommendations_scan.go b/internal/config/store_postgres_recommendations_scan.go new file mode 100644 index 000000000..a6bdf67f5 --- /dev/null +++ b/internal/config/store_postgres_recommendations_scan.go @@ -0,0 +1,92 @@ +package config + +import ( + "context" + "errors" + "fmt" +) + +// recommendationScanKeyExpr is the application recommendation ID in the +// byte-wise order the keyset scan walks (issue #981). COLLATE "C" compares +// bytes, the same order as Go's string comparison, so paging through SQL +// returns rows in the order the API used to produce by sorting in Go. +// COALESCE maps a missing or JSON-null id to "", which is what decoding the +// payload yields. It must stay identical to the expression indexed by +// migration 000142 (recommendations_scan_key_idx), or the planner falls back +// to a full scan and sort for every chunk. +const recommendationScanKeyExpr = `(COALESCE(payload->>'id', '') COLLATE "C")` + +// ListStoredRecommendationsChunk reads at most limit rows matching filter, +// ordered by recommendation ID (byte-wise) and then by the table's UUID +// primary key, starting strictly after the position after (nil means from +// the beginning). The primary key gives rows that share an ID a total order, +// so resuming from Last never skips or repeats a row, even at a run of +// duplicate IDs. +// +// The SQL-pushed conditions are the same as ListStoredRecommendations; the +// MinSavingsPct floor is still applied in-process, so Recs can be shorter +// than the rows read. Last is the position of the last row read (before that +// floor), and Done reports that fewer than limit rows were read, meaning +// nothing matches after Last. Each call is a single statement, so a chunk is +// one consistent snapshot; consecutive chunks are not. +func (s *PostgresStore) ListStoredRecommendationsChunk(ctx context.Context, filter RecommendationFilter, after *RecommendationScanKey, limit int) (RecommendationChunk, error) { + if limit < 1 { + return RecommendationChunk{}, errors.New("recommendations chunk limit must be positive") + } + if err := s.checkAzurePricingMigration(ctx, filter.RequireAzurePricingMigration); err != nil { + return RecommendationChunk{}, err + } + query, args := buildRecommendationChunkQuery(filter, after, limit) + rows, err := s.db.Query(ctx, query, args...) + if err != nil { + return RecommendationChunk{}, fmt.Errorf("failed to query recommendations chunk: %w", err) + } + defer rows.Close() + + var chunk RecommendationChunk + read := 0 + for rows.Next() { + var key RecommendationScanKey + var payload []byte + if err := rows.Scan(&key.RowID, &key.ID, &payload); err != nil { + return RecommendationChunk{}, fmt.Errorf("failed to scan recommendations chunk row: %w", err) + } + read++ + chunk.Last = &key + rec, keep, err := decodeStoredRecommendation(payload, filter) + if err != nil { + return RecommendationChunk{}, err + } + if keep { + chunk.Recs = append(chunk.Recs, rec) + } + } + if err := rows.Err(); err != nil { + return RecommendationChunk{}, fmt.Errorf("recommendations chunk rows iteration: %w", err) + } + chunk.Done = read < limit + return chunk, nil +} + +// buildRecommendationChunkQuery appends the keyset bound, ORDER BY and LIMIT +// to the ListStoredRecommendations filter. The row comparison against +// (ID, primary key) is what recommendations_scan_key_idx can serve directly. +func buildRecommendationChunkQuery(filter RecommendationFilter, after *RecommendationScanKey, limit int) (query string, args []any) { + where, args := buildRecommendationFilter(filter) + if after != nil { + cond := fmt.Sprintf("(%s, id) > ($%d, $%d::uuid)", recommendationScanKeyExpr, len(args)+1, len(args)+2) + if where == "" { + where = " WHERE " + cond + } else { + where += " AND " + cond + } + args = append(args, after.ID, after.RowID) + } + args = append(args, limit) + // The output columns are aliased and the primary key in ORDER BY is + // table-qualified: a bare "id" there would resolve to the id::text output + // column, a different sort key the index cannot serve. + query = fmt.Sprintf(`SELECT recommendations.id::text AS row_id, %s AS rec_id, payload FROM recommendations%s ORDER BY %s, recommendations.id LIMIT $%d`, + recommendationScanKeyExpr, where, recommendationScanKeyExpr, len(args)) + return query, args +} diff --git a/internal/config/store_postgres_recommendations_scan_test.go b/internal/config/store_postgres_recommendations_scan_test.go new file mode 100644 index 000000000..89d16539d --- /dev/null +++ b/internal/config/store_postgres_recommendations_scan_test.go @@ -0,0 +1,263 @@ +//go:build integration +// +build integration + +package config_test + +import ( + "context" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// scanRec is an AWS rec with the given application ID. resourceType must be +// unique per row: it is the part of the natural key these tests vary. +func scanRec(id, resourceType string) config.RecommendationRecord { + return awsRec(id, "ec2", "us-east-1", resourceType, 10) +} + +// walkChunks reads every chunk for filter, chunkSize rows at a time, and +// returns the IDs of the recs in scan order plus the number of reads. +func walkChunks(ctx context.Context, t *testing.T, store *config.PostgresStore, filter config.RecommendationFilter, chunkSize int) (ids []string, reads int) { + t.Helper() + var after *config.RecommendationScanKey + for { + reads++ + require.Less(t, reads, 10000, "chunk walk did not terminate") + chunk, err := store.ListStoredRecommendationsChunk(ctx, filter, after, chunkSize) + require.NoError(t, err) + for i := range chunk.Recs { + ids = append(ids, chunk.Recs[i].ID) + } + if chunk.Done { + return ids, reads + } + require.NotNil(t, chunk.Last, "a chunk that is not done must say where to resume") + after = chunk.Last + } +} + +// TestListStoredRecommendationsChunk_OrderMatchesGoStringOrder pins the scan +// order to Go's byte-wise string order (issue #981), which is the order the +// paginated API sorted rows into before the keyset moved into SQL. The IDs mix +// case, "|", "-", "/", spaces, non-ASCII and the empty ID, where a linguistic +// collation (en_US and friends) orders differently from bytes. +func TestListStoredRecommendationsChunk_OrderMatchesGoStringOrder(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + ids := []string{ + "aws|b", "AWS|a", "aws-a", "aws a", "aws/a", "aws|a|Linux/UNIX|default", + "aws|a", "Zeta", "alpha", "élan", "elan", "_x", "", "aws|a|Windows|default", + } + recs := make([]config.RecommendationRecord, len(ids)) + for i, id := range ids { + recs[i] = scanRec(id, fmt.Sprintf("t%02d", i)) + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + want := append([]string(nil), ids...) + sort.Strings(want) + + for _, chunkSize := range []int{1, 2, 3, len(ids), len(ids) + 1} { + t.Run(fmt.Sprintf("chunk=%d", chunkSize), func(t *testing.T) { + got, reads := walkChunks(ctx, t, store, config.RecommendationFilter{}, chunkSize) + assert.Equal(t, want, got, "scan order must equal Go's sort.Strings order") + // The walk ends on the first short chunk: exactly enough reads to + // cover every row plus, when the count divides evenly, one empty + // read to learn there is nothing after the last row. + assert.Equal(t, len(ids)/chunkSize+1, reads) + }) + } +} + +// TestListStoredRecommendationsChunk_DuplicateIDsNeverSkipped pins the +// tie-break: rows that share an application ID are ordered by primary key, and +// a chunk boundary that falls inside such a run neither skips nor repeats a +// row, so a caller further up sees every duplicate and can reject it. +func TestListStoredRecommendationsChunk_DuplicateIDsNeverSkipped(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + recs := []config.RecommendationRecord{ + scanRec("a", "t1"), scanRec("b", "t2"), scanRec("b", "t3"), scanRec("b", "t4"), scanRec("c", "t5"), + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + for chunkSize := 1; chunkSize <= len(recs)+1; chunkSize++ { + got, _ := walkChunks(ctx, t, store, config.RecommendationFilter{}, chunkSize) + assert.Equal(t, []string{"a", "b", "b", "b", "c"}, got, "chunk size %d", chunkSize) + } +} + +// TestListStoredRecommendationsChunk_FiltersAndPctFloor checks the SQL filters +// still apply, and that rows dropped by the in-process MinSavingsPct floor +// still advance the position, so a chunk whose rows are all below the floor is +// empty but not the end of the walk. +func TestListStoredRecommendationsChunk_FiltersAndPctFloor(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + pct := func(id, resourceType string, p float64) config.RecommendationRecord { + rec := scanRec(id, resourceType) + rec.SavingsPercentage = &p + return rec + } + other := scanRec("r00", "t0") + other.Region = "eu-west-1" + recs := []config.RecommendationRecord{ + other, + pct("r01", "t1", 5), pct("r02", "t2", 5), // a whole chunk below the floor + pct("r03", "t3", 50), pct("r04", "t4", 5), pct("r05", "t5", 60), + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), recs, nil)) + + filter := config.RecommendationFilter{Region: "us-east-1", MinSavingsPct: 30} + first, err := store.ListStoredRecommendationsChunk(ctx, filter, nil, 2) + require.NoError(t, err) + assert.Empty(t, first.Recs, "both rows of the first chunk are below the floor") + assert.False(t, first.Done, "a full chunk is not the end even when the floor emptied it") + require.NotNil(t, first.Last) + assert.Equal(t, "r02", first.Last.ID) + assert.NotEmpty(t, first.Last.RowID) + + got, _ := walkChunks(ctx, t, store, filter, 2) + assert.Equal(t, []string{"r03", "r05"}, got) + + _, err = store.ListStoredRecommendationsChunk(ctx, filter, nil, 0) + require.Error(t, err, "a non-positive limit is a caller bug, not an empty result") +} + +// TestListStoredRecommendationsChunk_EmptyTable returns a done, empty chunk +// with no position. +func TestListStoredRecommendationsChunk_EmptyTable(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + chunk, err := store.ListStoredRecommendationsChunk(ctx, config.RecommendationFilter{}, nil, 10) + require.NoError(t, err) + assert.Empty(t, chunk.Recs) + assert.Nil(t, chunk.Last) + assert.True(t, chunk.Done) +} + +// TestListStoredRecommendationsChunk_UsesScanKeyIndex proves the chunk query +// matches the expression migration 000142 indexes: with sequential and bitmap +// scans disabled the plan reads recommendations_scan_key_idx in order and has no +// Sort node. A drift between recommendationScanKeyExpr and the index (for +// example dropping COLLATE "C") brings the full scan plus sort back for every +// chunk. +func TestListStoredRecommendationsChunk_UsesScanKeyIndex(t *testing.T) { + ctx := context.Background() + db := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()).DB.Pool() + + for name, after := range map[string]*config.RecommendationScanKey{ + "first chunk": nil, + "resume": {ID: "aws|x", RowID: "00000000-0000-0000-0000-000000000001"}, + } { + t.Run(name, func(t *testing.T) { + query, args := config.BuildRecommendationChunkQuery(config.RecommendationFilter{}, after, 500) + plan := explainWithoutSeqScan(ctx, t, db, query, args) + assert.Contains(t, plan, "recommendations_scan_key_idx", plan) + assert.NotContains(t, plan, "Sort", plan) + }) + } +} + +func explainWithoutSeqScan(ctx context.Context, t *testing.T, db interface { + Begin(context.Context) (pgx.Tx, error) +}, query string, args []any) string { + t.Helper() + tx, err := db.Begin(ctx) + require.NoError(t, err) + defer func() { _ = tx.Rollback(ctx) }() + // An empty test table makes any plan cheap, so steer the planner away + // from the scans that do not return rows in index order; the question + // is whether the ordered index scan can serve the query at all. + _, err = tx.Exec(ctx, `SET LOCAL enable_seqscan = off; SET LOCAL enable_bitmapscan = off`) + require.NoError(t, err) + rows, err := tx.Query(ctx, "EXPLAIN "+query, args...) + require.NoError(t, err) + defer rows.Close() + var lines []string + for rows.Next() { + var line string + require.NoError(t, rows.Scan(&line)) + lines = append(lines, line) + } + require.NoError(t, rows.Err()) + return strings.Join(lines, "\n") +} + +// TestListStoredRecommendationsChunk_ConcurrentUpsertsDuringWalk runs chunk +// walks while collections keep upserting the same rows (new payload values, +// same natural keys) and inserting and evicting other rows. Every row that +// exists for the whole walk must come back exactly once and in order: an +// upsert keeps the primary key, so the (ID, primary key) position stays +// valid between chunks. +func TestListStoredRecommendationsChunk_ConcurrentUpsertsDuringWalk(t *testing.T) { + ctx := context.Background() + store := setupRecommendationsStore(ctx, t) + + const stable = 300 + base := make([]config.RecommendationRecord, stable) + wantStable := make([]string, stable) + for i := range base { + base[i] = scanRec(fmt.Sprintf("s%04d", i), fmt.Sprintf("s%04d", i)) + wantStable[i] = base[i].ID + } + require.NoError(t, store.UpsertRecommendations(ctx, time.Now().UTC(), base, nil)) + + stop := make(chan struct{}) + var wg sync.WaitGroup + writerErr := make(chan error, 1) + wg.Add(1) + go func() { + defer wg.Done() + for round := 0; ; round++ { + select { + case <-stop: + return + default: + } + recs := append([]config.RecommendationRecord(nil), base...) + for i := range recs { + recs[i].Savings = float64(round) + } + // A churning row that sorts between stable ones. + recs = append(recs, scanRec(fmt.Sprintf("s%04d-churn-%d", round%stable, round), fmt.Sprintf("c%06d", round))) + if err := store.UpsertRecommendations(ctx, time.Now().UTC(), recs, []config.SuccessfulCollect{{Provider: "aws"}}); err != nil { + writerErr <- err + return + } + } + }() + + for walk := 0; walk < 20; walk++ { + got, _ := walkChunks(ctx, t, store, config.RecommendationFilter{}, 7) + assert.True(t, sort.StringsAreSorted(got), "walk %d out of order", walk) + var gotStable []string + for _, id := range got { + if !strings.Contains(id, "-churn-") { + gotStable = append(gotStable, id) + } + } + require.Equal(t, wantStable, gotStable, "walk %d must return every stable row exactly once", walk) + } + close(stop) + wg.Wait() + select { + case err := <-writerErr: + require.NoError(t, err) + default: + } +} diff --git a/internal/config/types.go b/internal/config/types.go index 4b43d6a0f..0f66f07e9 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -789,6 +789,26 @@ type RecommendationFilter struct { ID string // "" = all ids; non-empty = exact match on the id column } +// RecommendationScanKey is a stored recommendation's position in the keyset +// scan order of ListStoredRecommendationsChunk (issue #981): the application +// recommendation ID compared byte-wise, then the cache table's UUID primary +// key (RowID), which keeps rows that share an ID in a total order. +type RecommendationScanKey struct { + ID string + RowID string +} + +// RecommendationChunk is one keyset read of the recommendations cache. +// Recs holds the rows that passed the filter, in scan order. Last is the +// position of the last row read (nil when no row was read) and is where the +// next chunk resumes. Done is true when fewer rows than the limit were read, +// so nothing matches after Last. +type RecommendationChunk struct { + Recs []RecommendationRecord + Last *RecommendationScanKey + Done bool +} + // PurchasePlanFilter parameterises ListPurchasePlans. Zero-value means "no // filter" (all plans are returned). Non-empty AccountIDs restricts the result // to plans that reference at least one of the given account IDs via the diff --git a/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql new file mode 100644 index 000000000..77faacddf --- /dev/null +++ b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS recommendations_scan_key_idx; diff --git a/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql new file mode 100644 index 000000000..551d0b428 --- /dev/null +++ b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx.up.sql @@ -0,0 +1,21 @@ +-- Index the keyset order GET /api/recommendations pages in (issue #981). +-- +-- A paginated request walks the recommendations cache in chunks ordered by the +-- application recommendation ID (payload->>'id'), with the table's UUID +-- primary key as the tie-breaker, and resumes each chunk after the last row it +-- read. Without an index on that order every chunk is a full scan plus a +-- top-N sort, so a walk of N rows in chunks of C costs O(N^2 / C). +-- +-- COLLATE "C" compares bytes, which is the order Go's string comparison uses, +-- so the SQL order matches the order the API has always paged in. COALESCE +-- maps a missing or null id to '' (what the Go decoder produces), so no row +-- falls outside the order. The expression must stay identical to +-- recommendationScanKeyExpr in internal/config/store_postgres_recommendations.go +-- or the planner cannot use the index. +-- +-- The table is a regenerable cache rewritten every collection cycle, so a +-- plain CREATE INDEX (which blocks writes while it builds) is acceptable; the +-- migration runner wraps every migration in a transaction, which rules out +-- CREATE INDEX CONCURRENTLY anyway. +CREATE INDEX IF NOT EXISTS recommendations_scan_key_idx + ON recommendations ((COALESCE(payload->>'id', '') COLLATE "C"), id); diff --git a/internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go new file mode 100644 index 000000000..1e7bf95fb --- /dev/null +++ b/internal/database/postgres/migrations/000142_recommendations_scan_key_idx_test.go @@ -0,0 +1,52 @@ +//go:build integration + +package migrations_test + +import ( + "context" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/require" +) + +// TestMigration_RecommendationsScanKeyIndex checks that 000142 creates the +// keyset index the paginated recommendations scan relies on (issue #981), in +// byte-wise ("C") order with the primary key as tie-breaker, and that the down +// migration removes it. +func TestMigration_RecommendationsScanKeyIndex(t *testing.T) { + ctx := context.Background() + container := testhelpers.RequirePostgresContainer(ctx, t) + t.Cleanup(func() { require.NoError(t, container.Cleanup(context.Background())) }) + pool, path := container.DB.Pool(), getMigrationsPath() + + const version = 142 + prev := previousMigrationVersion(t, version) + + indexDef := func() string { + t.Helper() + var def string + require.NoError(t, pool.QueryRow(ctx, ` + SELECT COALESCE((SELECT indexdef FROM pg_indexes + WHERE tablename = 'recommendations' + AND indexname = 'recommendations_scan_key_idx'), '')`).Scan(&def)) + return def + } + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev)) + require.Empty(t, indexDef(), "before 000142 the index must not exist") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version)) + def := indexDef() + require.Contains(t, def, `COALESCE((payload ->> 'id'::text), ''::text) COLLATE "C", id)`, + "the index must be on the byte-wise id, then the primary key") + require.NotContains(t, def, "UNIQUE", "rows may share an application id; the primary key breaks ties") + require.NotContains(t, def, "WHERE", "a partial index cannot serve an unfiltered scan") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, prev)) + require.Empty(t, indexDef(), "down must drop the index") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version)) + require.NotEmpty(t, indexDef(), "up must re-apply cleanly after down") +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index e946ed5f8..97ede6250 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1645,6 +1645,19 @@ func (m *MockConfigStore) ListStoredRecommendations(ctx context.Context, filter return v, args.Error(1) } +func (m *MockConfigStore) ListStoredRecommendationsChunk(ctx context.Context, filter config.RecommendationFilter, after *config.RecommendationScanKey, limit int) (config.RecommendationChunk, error) { + m.record("ListStoredRecommendationsChunk", ctx, filter, after, limit) + if !isExpected(&m.Mock, "ListStoredRecommendationsChunk") { + return config.RecommendationChunk{Done: true}, nil + } + args := m.Called(ctx, filter, after, limit) + v, ok := args.Get(0).(config.RecommendationChunk) + if !ok { + panic(fmt.Sprintf("mock: expected config.RecommendationChunk, got %T", args.Get(0))) + } + return v, args.Error(1) +} + func (m *MockConfigStore) GetRecommendationsFreshness(ctx context.Context) (*config.RecommendationsFreshness, error) { m.record("GetRecommendationsFreshness", ctx) if !isExpected(&m.Mock, "GetRecommendationsFreshness") { diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index c7514c959..42b6ee480 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -1197,14 +1197,9 @@ func recommendationID(rec config.RecommendationRecord) (string, error) { func (s *Scheduler) ListRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) { logging.Info("Reading recommendations from cache...") - freshness, err := s.config.GetRecommendationsFreshness(ctx) + freshness, err := s.startCacheRead(ctx) if err != nil { - return nil, fmt.Errorf("failed to read recommendations freshness: %w", err) - } - - mayCollect := BackgroundCollectionAllowed(ctx) - if mayCollect && freshness.LastCollectedAt == nil { - s.kickColdStartCollection(ctx) + return nil, err } recs, err := s.config.ListStoredRecommendations(ctx, filter) @@ -1229,19 +1224,40 @@ func (s *Scheduler) ListRecommendations(ctx context.Context, filter config.Recom logging.Errorf("failed to apply account overrides; returning un-filtered recs: %v", err) } - // Background refresh is only attempted on non-Lambda runtimes; skip - // the DB config read entirely on Lambda so GetGlobalConfig is not called - // on the hot read path (Lambda has no persistent goroutines). - if mayCollect && !s.isLambda { - ttl, disabled := s.resolveEffectiveCacheTTL(ctx) - if disabled { - return recs, nil - } - s.maybeKickBackgroundRefresh(freshness, ttl) - } + s.finishCacheRead(ctx, freshness) return recs, nil } +// startCacheRead is steps 1 and 2 of ListRecommendations (and of +// ScanRecommendations): read freshness and, for a caller allowed to trigger +// collection, kick the cold-start collection. +func (s *Scheduler) startCacheRead(ctx context.Context) (*config.RecommendationsFreshness, error) { + freshness, err := s.config.GetRecommendationsFreshness(ctx) + if err != nil { + return nil, fmt.Errorf("failed to read recommendations freshness: %w", err) + } + if BackgroundCollectionAllowed(ctx) && freshness.LastCollectedAt == nil { + s.kickColdStartCollection(ctx) + } + return freshness, nil +} + +// finishCacheRead is step 4 of ListRecommendations (and of +// ScanRecommendations): the stale-while-revalidate kick. Background refresh is +// only attempted on non-Lambda runtimes; the DB config read is skipped +// entirely on Lambda so GetGlobalConfig is not called on the hot read path +// (Lambda has no persistent goroutines). +func (s *Scheduler) finishCacheRead(ctx context.Context, freshness *config.RecommendationsFreshness) { + if !BackgroundCollectionAllowed(ctx) || s.isLambda { + return + } + ttl, disabled := s.resolveEffectiveCacheTTL(ctx) + if disabled { + return + } + s.maybeKickBackgroundRefresh(freshness, ttl) +} + // ListRecommendationsStrict is ListRecommendations for the money path: the // same stored rows, suppressions and account overrides, but a suppression or // override read failure is returned instead of logged. Over-showing is fine diff --git a/internal/scheduler/scheduler_scan.go b/internal/scheduler/scheduler_scan.go new file mode 100644 index 000000000..cef7057de --- /dev/null +++ b/internal/scheduler/scheduler_scan.go @@ -0,0 +1,147 @@ +package scheduler + +import ( + "context" + "errors" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// ScanRecommendations is ListRecommendations for a caller that pages through +// the cache without holding all of it (issue #981). It walks the cached +// recommendations in chunks of at most chunkSize rows, in byte-wise ID order +// (the order of Go's string comparison), applies the same suppressions and +// account overrides as ListRecommendations to each chunk, and calls fn with +// each chunk's surviving rows in that order. fn may keep what it needs; the +// scan itself holds one chunk at a time. An error from fn stops the scan and +// is returned. +// +// It has the same cold-start and stale-while-revalidate side effects as +// ListRecommendations, and the same over-show trade-off: a suppression or +// override read failure is logged and the affected rows pass through. +// +// Each chunk is one consistent read; the walk as a whole is not one snapshot. +// A row upserted in place keeps its position, so it is visited once. A row +// inserted behind the scan position during the walk is missed by this walk, +// and a row evicted ahead of it is not visited. +func (s *Scheduler) ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error { + if chunkSize < 1 { + return errors.New("recommendations scan chunk size must be positive") + } + logging.Info("Scanning recommendations from cache...") + + freshness, err := s.startCacheRead(ctx) + if err != nil { + return err + } + + suppressions := s.loadSuppressionIndex(ctx) + overrides := newOverrideScanFilter(s.config) + var after *config.RecommendationScanKey + for { + chunk, err := s.config.ListStoredRecommendationsChunk(ctx, filter, after, chunkSize) + if err != nil { + return err + } + recs := chunk.Recs + if len(suppressions) > 0 { + recs = applySuppressionIndex(recs, suppressions) + } + recs = overrides.apply(ctx, recs) + if len(recs) > 0 { + if err := fn(recs); err != nil { + return err + } + } + if chunk.Done || chunk.Last == nil { + break + } + after = chunk.Last + } + + s.finishCacheRead(ctx, freshness) + return nil +} + +// loadSuppressionIndex reads the active suppressions once for a whole scan and +// groups them by 6-tuple, as applySuppressions does. Which rows of a group +// apply to a rec depends only on that rec's variant (issue #816), so applying +// the index chunk by chunk suppresses exactly what ListRecommendations does. +// A read failure is logged and yields no index, so the scan over-shows like +// ListRecommendations does. +func (s *Scheduler) loadSuppressionIndex(ctx context.Context) map[suppressionKey][]config.PurchaseSuppression { + sups, err := s.config.ListActiveSuppressions(ctx) + if err != nil { + logging.Errorf("failed to apply suppressions; returning un-suppressed recs: %v", err) + return nil + } + if len(sups) == 0 { + return nil + } + return aggregateSuppressions(sups) +} + +// overrideScanFilter applies account overrides chunk by chunk, resolving each +// (account, provider, service) triple once per scan instead of once per chunk. +type overrideScanFilter struct { + store config.AccountConfigReader + resolved map[string]*config.ServiceConfig + seen map[string]struct{} +} + +func newOverrideScanFilter(store config.AccountConfigReader) *overrideScanFilter { + return &overrideScanFilter{ + store: store, + resolved: make(map[string]*config.ServiceConfig), + seen: make(map[string]struct{}), + } +} + +// apply resolves the triples recs introduces and filters recs with every +// configuration resolved so far. On a resolution error the chunk passes +// through unfiltered (logged), and its new triples stay unresolved so a later +// chunk retries them. +func (o *overrideScanFilter) apply(ctx context.Context, recs []config.RecommendationRecord) []config.RecommendationRecord { + if len(recs) == 0 { + return recs + } + var unseen []config.RecommendationRecord + var unseenKeys []string + for i := range recs { + key := recAccountConfigKey(&recs[i]) + if _, ok := o.seen[key]; ok { + continue + } + o.seen[key] = struct{}{} + unseen = append(unseen, recs[i]) + unseenKeys = append(unseenKeys, key) + } + if len(unseen) > 0 { + resolved, err := config.ResolveAccountConfigsForRecs(ctx, o.store, unseen) + if err != nil { + for _, key := range unseenKeys { + delete(o.seen, key) + } + logging.Errorf("failed to apply account overrides; returning un-filtered recs: %v", err) + return recs + } + for key, cfg := range resolved { + o.resolved[key] = cfg + } + } + if len(o.resolved) == 0 { + return recs + } + return filterRecsByResolvedConfigs(recs, o.resolved) +} + +// recAccountConfigKey is the AccountConfigKey filterRecsByResolvedConfigs +// looks a rec up by. +func recAccountConfigKey(rec *config.RecommendationRecord) string { + accountID := "" + if rec.CloudAccountID != nil { + accountID = *rec.CloudAccountID + } + return config.AccountConfigKey(accountID, rec.Provider, rec.Service) +} diff --git a/internal/scheduler/scheduler_scan_test.go b/internal/scheduler/scheduler_scan_test.go new file mode 100644 index 000000000..304ba86b9 --- /dev/null +++ b/internal/scheduler/scheduler_scan_test.go @@ -0,0 +1,298 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "sort" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// scanStore serves ListStoredRecommendationsChunk from recs the way the +// Postgres store does: ordered by ID (byte-wise) then by row position, +// resuming strictly after the given key. It also counts reads so tests can +// check that override lookups are not repeated per chunk. +type scanStore struct { + mockOverrideStore + sups []config.PurchaseSuppression + supsErr error + chunkErr error + chunkReads int + overrideLookup map[string]int +} + +func (m *scanStore) ListActiveSuppressions(_ context.Context) ([]config.PurchaseSuppression, error) { + return m.sups, m.supsErr +} + +func (m *scanStore) GetAccountServiceOverride(ctx context.Context, accountID, provider, service string) (*config.AccountServiceOverride, error) { + if m.overrideLookup == nil { + m.overrideLookup = map[string]int{} + } + m.overrideLookup[accountID+"|"+provider+"|"+service]++ + return m.mockOverrideStore.GetAccountServiceOverride(ctx, accountID, provider, service) +} + +func (m *scanStore) ListStoredRecommendationsChunk(_ context.Context, _ config.RecommendationFilter, after *config.RecommendationScanKey, limit int) (config.RecommendationChunk, error) { + m.chunkReads++ + if m.chunkErr != nil { + return config.RecommendationChunk{}, m.chunkErr + } + type row struct { + key config.RecommendationScanKey + rec config.RecommendationRecord + } + rows := make([]row, len(m.recs)) + for i := range m.recs { + rows[i] = row{config.RecommendationScanKey{ID: m.recs[i].ID, RowID: fmt.Sprintf("%08d", i)}, m.recs[i]} + } + less := func(a, b config.RecommendationScanKey) bool { + if a.ID != b.ID { + return a.ID < b.ID + } + return a.RowID < b.RowID + } + sort.Slice(rows, func(i, j int) bool { return less(rows[i].key, rows[j].key) }) + var chunk config.RecommendationChunk + for i := range rows { + if after != nil && !less(*after, rows[i].key) { + continue + } + if len(chunk.Recs) == limit { + return chunk, nil + } + chunk.Recs = append(chunk.Recs, rows[i].rec) + k := rows[i].key + chunk.Last = &k + } + // Like the store: only a short read proves there is nothing after Last. + chunk.Done = len(chunk.Recs) < limit + return chunk, nil +} + +// scanFixture is a cache where suppressions and account overrides each drop +// some rows and change others, spread over several chunks. +func scanFixture() *scanStore { + var recs []config.RecommendationRecord + for i := 9; i >= 0; i-- { // reverse order: the scan must not rely on input order + for _, acct := range []string{"acct-A", "acct-B", "acct-C"} { + rec := rdsRec(acct, "us-east-1", "mysql") + rec.ID = fmt.Sprintf("%s|%02d", acct, i) + rec.ResourceType = fmt.Sprintf("db.t3.r%02d", i) + rec.Count = 2 + recs = append(recs, rec) + } + } + future := time.Now().Add(time.Hour) + return &scanStore{ + mockOverrideStore: mockOverrideStore{ + recs: recs, + globals: map[string]*config.ServiceConfig{ + "aws|rds": {Provider: "aws", Service: "rds", Enabled: true}, + }, + overrides: map[string]*config.AccountServiceOverride{ + "acct-B|aws|rds": {Enabled: boolPtr(false)}, + }, + }, + sups: []config.PurchaseSuppression{ + // Drops acct-C|03 (2 - 2 = 0) and annotates acct-A|05 (2 - 1 = 1). + {ExecutionID: "e1", AccountID: "acct-C", Provider: "aws", Service: "rds", Region: "us-east-1", + ResourceType: "db.t3.r03", Engine: "mysql", SuppressedCount: 2, ExpiresAt: future}, + {ExecutionID: "e2", AccountID: "acct-A", Provider: "aws", Service: "rds", Region: "us-east-1", + ResourceType: "db.t3.r05", Engine: "mysql", SuppressedCount: 1, ExpiresAt: future}, + }, + } +} + +func collectScan(t *testing.T, s *Scheduler, chunkSize int) ([]config.RecommendationRecord, int) { + t.Helper() + var got []config.RecommendationRecord + calls := 0 + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, chunkSize, func(chunk []config.RecommendationRecord) error { + calls++ + require.NotEmpty(t, chunk, "fn must not be called with an empty chunk") + got = append(got, chunk...) + return nil + }) + require.NoError(t, err) + return got, calls +} + +// TestScanRecommendations_MatchesListRecommendations pins the scan to the +// legacy read: for every chunk size it yields exactly the rows +// ListRecommendations returns, with the same suppression counts and override +// drops, in ID order. +func TestScanRecommendations_MatchesListRecommendations(t *testing.T) { + store := scanFixture() + s := &Scheduler{config: store} + + want, err := s.ListRecommendations(context.Background(), config.RecommendationFilter{}) + require.NoError(t, err) + sort.Slice(want, func(i, j int) bool { return want[i].ID < want[j].ID }) + require.Len(t, want, 19, "fixture: 30 rows, acct-B's 10 overridden away, one suppressed away") + + for _, chunkSize := range []int{1, 2, 3, 7, 30, 31} { + t.Run(fmt.Sprintf("chunk=%d", chunkSize), func(t *testing.T) { + store.overrideLookup = nil + got, _ := collectScan(t, s, chunkSize) + assert.Equal(t, want, got) + for key, n := range store.overrideLookup { + assert.Equal(t, 1, n, "override for %s looked up %d times in one scan", key, n) + } + }) + } +} + +// variantScanFixture is an EC2 cache exercising the variant-aware suppression +// rules of issue #816: a purchase of one variant (AZ, platform, tenancy) must +// reduce only that variant, a purchase covering a variant fully hides it, a +// legacy row with no variant is a wildcard over its 6-tuple, and a rec whose +// Details cannot be read matches every row of its 6-tuple. +func variantScanFixture(t *testing.T) *scanStore { + t.Helper() + bought := ec2VariantRec("ec2|bought", "us-east-1a", "Linux/UNIX", "default") + covered := ec2VariantRec("ec2|covered", "us-east-1c", "Linux/UNIX", "default") + unreadable := ec2VariantRec("ec2|unreadable", "us-east-1b", "Windows", "default") + unreadable.Details = []byte(`{"platform":42}`) + unreadable.Count = 10 + legacyA := ec2VariantRec("ec2|legacy-a", "us-east-1a", "Linux/UNIX", "default") + legacyA.ResourceType = "m5.xlarge" + legacyB := ec2VariantRec("ec2|legacy-b", "us-east-1b", "Windows", "dedicated") + legacyB.ResourceType = "m5.xlarge" + recs := []config.RecommendationRecord{ + legacyB, + ec2VariantRec("ec2|sib-windows", "us-east-1a", "Windows", "default"), + unreadable, + ec2VariantRec("ec2|sib-other-az", "us-east-1b", "Linux/UNIX", "default"), + covered, + ec2VariantRec("ec2|sib-dedicated", "us-east-1a", "Linux/UNIX", "dedicated"), + bought, + ec2VariantRec("ec2|sib-regional", "", "Linux/UNIX", "default"), + legacyA, + } + + boughtPurchase := bought + boughtPurchase.Count = 3 + coveredPurchase := covered // the full count of 5 + sups := config.BuildSuppressions([]config.RecommendationRecord{boughtPurchase, coveredPurchase}, "e-variant", + &config.GlobalConfig{GracePeriodDays: map[string]int{"aws": 7}}, time.Now()) + require.Len(t, sups, 2) + for i := range sups { + require.NotNil(t, sups[i].Variant, "BuildSuppressions must write variant rows") + } + sups = append(sups, config.PurchaseSuppression{ExecutionID: "e-legacy", AccountID: "acct-1", Provider: "aws", + Service: "ec2", Region: "us-east-1", ResourceType: "m5.xlarge", SuppressedCount: 1, + ExpiresAt: time.Now().Add(time.Hour)}) + + return &scanStore{ + mockOverrideStore: mockOverrideStore{ + recs: recs, + globals: map[string]*config.ServiceConfig{ + "aws|ec2": {Provider: "aws", Service: "ec2", Enabled: true}, + }, + }, + sups: sups, + } +} + +// TestScanRecommendations_VariantSuppressionsMatchListRecommendations pins +// the scan to ListRecommendations under issue #816's variant-aware +// suppressions, at every chunk size, and checks the expected per-variant +// result so the equality is not vacuous. +func TestScanRecommendations_VariantSuppressionsMatchListRecommendations(t *testing.T) { + store := variantScanFixture(t) + s := &Scheduler{config: store} + + want, err := s.ListRecommendations(context.Background(), config.RecommendationFilter{}) + require.NoError(t, err) + sort.Slice(want, func(i, j int) bool { return want[i].ID < want[j].ID }) + + for chunkSize := 1; chunkSize <= len(store.recs)+1; chunkSize++ { + t.Run(fmt.Sprintf("chunk=%d", chunkSize), func(t *testing.T) { + got, _ := collectScan(t, s, chunkSize) + assert.Equal(t, want, got) + + byID := map[string]config.RecommendationRecord{} + for _, r := range got { + byID[r.ID] = r + } + require.Len(t, byID, 8, "only the fully covered variant is hidden") + assert.NotContains(t, byID, "ec2|covered") + assert.Equal(t, 2, byID["ec2|bought"].Count, "5 - 3 on the purchased variant") + assert.Equal(t, 3, byID["ec2|bought"].SuppressedCount) + for _, id := range []string{"ec2|sib-windows", "ec2|sib-other-az", "ec2|sib-dedicated", "ec2|sib-regional"} { + assert.Equal(t, 5, byID[id].Count, "%s: a sibling variant keeps its full count", id) + assert.Zero(t, byID[id].SuppressedCount, id) + assert.Nil(t, byID[id].PrimarySuppressionExecutionID, id) + } + assert.Equal(t, 2, byID["ec2|unreadable"].Count, "10 - 3 - 5: unreadable Details match every row") + for _, id := range []string{"ec2|legacy-a", "ec2|legacy-b"} { + assert.Equal(t, 4, byID[id].Count, "%s: the legacy row is a wildcard over its 6-tuple", id) + require.NotNil(t, byID[id].PrimarySuppressionExecutionID, id) + assert.Equal(t, "e-legacy", *byID[id].PrimarySuppressionExecutionID, id) + } + }) + } +} + +// TestScanRecommendations_SkipsEmptyChunks checks that a chunk whose rows are +// all filtered away is not passed to fn, and that the walk still continues +// past it. +func TestScanRecommendations_SkipsEmptyChunks(t *testing.T) { + store := scanFixture() + s := &Scheduler{config: store} + got, calls := collectScan(t, s, 1) + assert.Len(t, got, 19) + assert.Equal(t, 19, calls, "one call per surviving row with chunk size 1") + assert.Equal(t, 31, store.chunkReads, "30 one-row chunks and the empty final read") +} + +// TestScanRecommendations_StopsOnCallbackError propagates fn's error and +// reads no further chunks. +func TestScanRecommendations_StopsOnCallbackError(t *testing.T) { + store := scanFixture() + s := &Scheduler{config: store} + boom := errors.New("boom") + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, 5, func([]config.RecommendationRecord) error { + return boom + }) + require.ErrorIs(t, err, boom) + assert.Equal(t, 1, store.chunkReads) +} + +// TestScanRecommendations_StoreErrorFails returns a chunk read error instead +// of a short walk. +func TestScanRecommendations_StoreErrorFails(t *testing.T) { + store := scanFixture() + store.chunkErr = errors.New("db down") + s := &Scheduler{config: store} + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, 5, func([]config.RecommendationRecord) error { + t.Fatal("fn must not run when the read fails") + return nil + }) + require.ErrorContains(t, err, "db down") +} + +// TestScanRecommendations_ReadFailuresOverShow mirrors ListRecommendations' +// trade-off: a suppression or override read failure is logged and the rows +// pass through rather than failing the page. +func TestScanRecommendations_ReadFailuresOverShow(t *testing.T) { + store := scanFixture() + store.supsErr = errors.New("sups down") + store.getOverrideErr = errors.New("overrides down") + s := &Scheduler{config: store} + got, _ := collectScan(t, s, 4) + assert.Len(t, got, 30, "nothing suppressed or overridden away") +} + +// TestScanRecommendations_RejectsBadChunkSize guards the loop bound. +func TestScanRecommendations_RejectsBadChunkSize(t *testing.T) { + s := &Scheduler{config: scanFixture()} + err := s.ScanRecommendations(context.Background(), config.RecommendationFilter{}, 0, func([]config.RecommendationRecord) error { return nil }) + require.Error(t, err) +} diff --git a/internal/server/interfaces.go b/internal/server/interfaces.go index ebdbb78f6..2a072b0f4 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -13,6 +13,9 @@ import ( type SchedulerInterface interface { CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) ListRecommendations(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) + // ScanRecommendations walks the same rows as ListRecommendations in + // ID-ordered chunks for the paginated list endpoint (issue #981). + ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error // GetRecommendationByID fetches a single rec by application-level id, // bypassing account-override filtering. hiddenBy is non-nil when the rec // exists but would be dropped by the override filter. Returns nil, nil, diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 845618a1d..b7d462888 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -347,6 +347,9 @@ func (m *mockConfigStoreForHealth) UpsertRecommendations(_ context.Context, _ ti func (m *mockConfigStoreForHealth) ListStoredRecommendations(_ context.Context, _ config.RecommendationFilter) ([]config.RecommendationRecord, error) { return nil, nil } +func (m *mockConfigStoreForHealth) ListStoredRecommendationsChunk(_ context.Context, _ config.RecommendationFilter, _ *config.RecommendationScanKey, _ int) (config.RecommendationChunk, error) { + return config.RecommendationChunk{Done: true}, nil +} func (m *mockConfigStoreForHealth) GetRecommendationsFreshness(_ context.Context) (*config.RecommendationsFreshness, error) { return &config.RecommendationsFreshness{}, nil } diff --git a/internal/testutil/mocks.go b/internal/testutil/mocks.go index bf6559c14..646a8155b 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -14,6 +14,7 @@ type MockScheduler struct { CollectRecommendationsFunc func(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) ListRecommendationsFunc func(ctx context.Context, filter config.RecommendationFilter) ([]config.RecommendationRecord, error) GetRecommendationByIDFunc func(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) + ScanRecommendationsFunc func(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error } func (m *MockScheduler) CollectRecommendations(ctx context.Context, ownerToken string) (*scheduler.CollectResult, error) { @@ -30,6 +31,14 @@ func (m *MockScheduler) ListRecommendations(ctx context.Context, filter config.R return []config.RecommendationRecord{}, nil } +// ScanRecommendations defaults to an empty cache: fn is never called. +func (m *MockScheduler) ScanRecommendations(ctx context.Context, filter config.RecommendationFilter, chunkSize int, fn func([]config.RecommendationRecord) error) error { + if m.ScanRecommendationsFunc != nil { + return m.ScanRecommendationsFunc(ctx, filter, chunkSize, fn) + } + return nil +} + func (m *MockScheduler) GetRecommendationByID(ctx context.Context, id string) (*config.RecommendationRecord, []string, error) { if m.GetRecommendationByIDFunc != nil { return m.GetRecommendationByIDFunc(ctx, id)