diff --git a/go.mod b/go.mod index 0b2c557b1..52c53767e 100644 --- a/go.mod +++ b/go.mod @@ -84,7 +84,7 @@ require ( github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/reservations/armreservations v1.1.0 github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azkeys v1.4.0 github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azsecrets v1.4.0 - github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261018003623-541e05e5fd29 + github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261020134525-e237996e45b4 github.com/LeanerCloud/cloud-commitments-go/providers/aws v0.0.0-20261019192600-aeb8a6f7ff51 github.com/LeanerCloud/cloud-commitments-go/providers/azure v0.0.0-20261017134002-3def3f898617 github.com/LeanerCloud/cloud-commitments-go/providers/gcp v0.0.0-20261017134002-3def3f898617 diff --git a/go.sum b/go.sum index 3a32f2465..9541de833 100644 --- a/go.sum +++ b/go.sum @@ -92,8 +92,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.54.0/go.mod h1:vB2GH9GAYYJTO3mEn8oYwzEdhlayZIdQz6zdzgUIRvA= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0 h1:s0WlVbf9qpvkh1c/uDAPElam0WrL7fHRIidgZJ7UqZI= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0/go.mod h1:Mf6O40IAyB9zR/1J8nGDDPirZQQPbYJni8Yisy7NTMc= -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261018003623-541e05e5fd29 h1:QJkoZBHkwjvMGplDb1TcuBy3HdYPxlZ+AyR/7QEU8Hc= -github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261018003623-541e05e5fd29/go.mod h1:SCDEgHG6evjPh0RsoXof+J3aMmGXNFMrwbJIle7rMpQ= +github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261020134525-e237996e45b4 h1:tE+KD86HKsxS0js0KfaEo8XURVpi2ly7PMVwcfzzJWk= +github.com/LeanerCloud/cloud-commitments-go/pkg v0.0.0-20261020134525-e237996e45b4/go.mod h1:/yCOnBtEJRRydFrDm6g1UcHuJwuB35AqYTBdaHDG8wo= github.com/LeanerCloud/cloud-commitments-go/providers/aws v0.0.0-20261019192600-aeb8a6f7ff51 h1:fr8mS6aF7/y69cJ5UlF9MEvP15mSlH/4yKg25Cvk/YM= github.com/LeanerCloud/cloud-commitments-go/providers/aws v0.0.0-20261019192600-aeb8a6f7ff51/go.mod h1:p6TrYRa4+lvcdsPCAJ+sa0tI0sRfGQ/0YT1Af3fZk3w= github.com/LeanerCloud/cloud-commitments-go/providers/azure v0.0.0-20261017134002-3def3f898617 h1:K9wBX9ac1QwhkjW7BvDt0gl2Kt9fGdDytJyndoPRcBU= diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 1df14783b..1e44b3c41 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -42,9 +42,12 @@ import ( // Cross-family alternatives no longer flow through here — they're // sourced from the cached AWS Cost Explorer purchase recommendations // in Postgres via purchaseRecLookupFromStore (see exchange_lookup.go), -// so the EC2 client only needs to enumerate convertible RIs. +// so the EC2 client only needs to enumerate convertible RIs and the +// region's offered instance types (DescribeInstanceTypeOfferings), which +// limit reshape targets to sizes the region actually sells. type reshapeEC2Client interface { ListConvertibleReservedInstances(ctx context.Context) ([]ec2svc.ConvertibleRI, error) + GetValidResourceTypes(ctx context.Context) ([]string, error) } // reshapeRecsClient is the narrow slice of the recommendations @@ -1754,7 +1757,9 @@ func (h *Handler) getReshapeRecommendations(ctx context.Context, req *events.Lam } currencyCode := firstNonEmptyCurrency(instances) lookup := purchaseRecLookupFromStore(h.config, cloudAccountID) - recs := exchange.AnalyzeReshapingWithRecs(ctx, riInfos, utilInfos, p.threshold, p.region, currencyCode, lookup) + offered := OfferedInstanceTypes(ctx, p.region, ec2Client.GetValidResourceTypes) + recs := exchange.AnalyzeReshapingWithRecs(ctx, riInfos, utilInfos, p.threshold, p.region, currencyCode, lookup, + exchange.WithOfferedInstanceTypes(offered)) resp := &ReshapeRecommendationsResponse{Recommendations: recs, ExcludedRIs: excluded} h.attachReshapeStaleness(ctx, resp) diff --git a/internal/api/handler_ri_exchange_integration_test.go b/internal/api/handler_ri_exchange_integration_test.go index d9df292ae..38512c876 100644 --- a/internal/api/handler_ri_exchange_integration_test.go +++ b/internal/api/handler_ri_exchange_integration_test.go @@ -34,6 +34,11 @@ func (f *fakeReshapeEC2) ListConvertibleReservedInstances(_ context.Context) ([] return f.instances, nil } +// GetValidResourceTypes returns no catalog, so reshape keeps the legacy ladder. +func (f *fakeReshapeEC2) GetValidResourceTypes(_ context.Context) ([]string, error) { + return nil, nil +} + // fakeReshapeRecs is a stub for reshapeRecsClient. Counts calls so the // test can assert cache-hit behavior on the second request. type fakeReshapeRecs struct { diff --git a/internal/api/handler_ri_exchange_test.go b/internal/api/handler_ri_exchange_test.go index 81e4288f1..9e383a890 100644 --- a/internal/api/handler_ri_exchange_test.go +++ b/internal/api/handler_ri_exchange_test.go @@ -905,14 +905,25 @@ func TestRejectRIExchange_EmptyStoredToken(t *testing.T) { // called and the region-normalization assertion would never fire // (silent test trap). Mirrors the integration-test fake but lives // here so this unit test runs without the //go:build integration tag. +// +// offered / offeredErr drive GetValidResourceTypes (the region's +// instance type catalog); offeredCalls counts the catalog lookups. type fakeReshapeEC2Stub struct { - instances []ec2svc.ConvertibleRI + instances []ec2svc.ConvertibleRI + offered []string + offeredErr error + offeredCalls int } func (f *fakeReshapeEC2Stub) ListConvertibleReservedInstances(_ context.Context) ([]ec2svc.ConvertibleRI, error) { return f.instances, nil } +func (f *fakeReshapeEC2Stub) GetValidResourceTypes(_ context.Context) ([]string, error) { + f.offeredCalls++ + return f.offered, f.offeredErr +} + // fakeReshapeRecsStub returns a configurable RI utilization slice so // the reshape pipeline runs without a real Cost Explorer dependency. // AnalyzeReshapingWithRecs only emits a recommendation (and only then diff --git a/internal/api/handler_ri_utilization_scope_test.go b/internal/api/handler_ri_utilization_scope_test.go index b0a858c9d..61b5ec4c3 100644 --- a/internal/api/handler_ri_utilization_scope_test.go +++ b/internal/api/handler_ri_utilization_scope_test.go @@ -35,6 +35,10 @@ func (f *countingUtilEC2) ListConvertibleReservedInstances(_ context.Context) ([ return f.instances, f.err } +func (f *countingUtilEC2) GetValidResourceTypes(_ context.Context) ([]string, error) { + return nil, nil +} + type countingUtilRecs struct { rows []recommendations.RIUtilization calls int diff --git a/internal/api/reshape_offered_types.go b/internal/api/reshape_offered_types.go new file mode 100644 index 000000000..6ca74826f --- /dev/null +++ b/internal/api/reshape_offered_types.go @@ -0,0 +1,29 @@ +package api + +import ( + "context" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" +) + +// OfferedInstanceTypes returns the EC2 instance types offered in region, for +// exchange.WithOfferedInstanceTypes and RunAutoExchangeParams.OfferedInstanceTypes. +// fetch is normally ec2svc.Client.GetValidResourceTypes +// (DescribeInstanceTypeOfferings). +// +// On a fetch error or an empty catalog it logs a warning naming the region and +// returns nil, so reshape analysis proceeds on the legacy size ladder instead of +// failing the request. Nothing is cached, so the next call queries the catalog +// again rather than reusing the fallback. +func OfferedInstanceTypes(ctx context.Context, region string, fetch func(context.Context) ([]string, error)) []string { + types, err := fetch(ctx) + if err != nil { + logging.Warnf("reshape: EC2 instance type offerings for region %q failed, proceeding without the catalog: %v", region, err) + return nil + } + if len(types) == 0 { + logging.Warnf("reshape: EC2 instance type offerings for region %q came back empty, proceeding without the catalog", region) + return nil + } + return types +} diff --git a/internal/api/reshape_offered_types_test.go b/internal/api/reshape_offered_types_test.go new file mode 100644 index 000000000..fdd2f7bf2 --- /dev/null +++ b/internal/api/reshape_offered_types_test.go @@ -0,0 +1,179 @@ +package api + +import ( + "bytes" + "context" + "errors" + "sync" + "testing" + + "github.com/aws/aws-lambda-go/events" + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-go/providers/aws/recommendations" + ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// lockedBuffer is an io.Writer safe for the logger's concurrent writes. +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// captureWarnings routes the default pkg/logging logger into a buffer at warn +// level for the rest of the test. Tests using it must not call t.Parallel. +func captureWarnings(t *testing.T) *lockedBuffer { + t.Helper() + out := &lockedBuffer{} + prevOut := logging.SetOutput(out) + prevLevel := logging.GetLevel() + logging.SetLevelValue(logging.LevelWarn) + t.Cleanup(func() { + logging.SetOutput(prevOut) + logging.SetLevelValue(prevLevel) + }) + return out +} + +const reshapeCatalogRegion = "eu-west-3" + +// newCatalogReshapeHandler builds a reshape handler over one convertible +// 1x m5.4xlarge RI (32 normalized units) at 50% utilization (16 used). On +// the legacy ladder that reshapes to 1x m5.2xlarge; with a catalog lacking +// m5.2xlarge it must become 2x m5.xlarge. +func newCatalogReshapeHandler(ec2 *fakeReshapeEC2Stub) *Handler { + ec2.instances = []ec2svc.ConvertibleRI{ + {ReservedInstanceID: "ri-1", InstanceType: "m5.4xlarge", InstanceCount: 1, CurrencyCode: "USD", Duration: 31536000}, + } + mockStore := &MockConfigStore{} + mockStore.On("ListStoredRecommendations", mock.Anything, mock.Anything). + Return([]config.RecommendationRecord(nil), nil) + h := &Handler{ + config: mockStore, + auth: &mockAuthForExchange{}, + reshapeEC2Factory: func(_ aws.Config) reshapeEC2Client { return ec2 }, + reshapeRecsFactory: func(_ aws.Config) reshapeRecsClient { + return &fakeReshapeRecsStub{utilization: []recommendations.RIUtilization{ + {ReservedInstanceID: "ri-1", UtilizationPercent: 50.0}, + }} + }, + reshapeAccountResolver: func(_ context.Context) (string, error) { return "", nil }, + } + h.awsCfgOnce.Do(func() { h.awsCfg = aws.Config{Region: reshapeCatalogRegion} }) + return h +} + +func reshapeOnce(t *testing.T, h *Handler) *ReshapeRecommendationsResponse { + t.Helper() + req := &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"authorization": "Bearer test-token"}, + QueryStringParameters: map[string]string{"lookback_days": "30", "threshold": "95.0"}, + } + out, err := h.getReshapeRecommendations(context.Background(), req) + require.NoError(t, err) + resp, ok := out.(*ReshapeRecommendationsResponse) + require.True(t, ok) + require.Len(t, resp.Recommendations, 1) + return resp +} + +func TestGetReshapeRecommendations_CatalogLimitsTargetToOfferedSizes(t *testing.T) { + ec2 := &fakeReshapeEC2Stub{offered: []string{"m5.4xlarge", "m5.xlarge", "c5.large"}} + rec := reshapeOnce(t, newCatalogReshapeHandler(ec2)).Recommendations[0] + + assert.Equal(t, 1, ec2.offeredCalls, "the reshape handler must query the region's instance type catalog") + assert.Equal(t, "m5.xlarge", rec.TargetInstanceType, + "m5.2xlarge is not offered, so the target must step down to an offered size") + assert.Equal(t, int32(2), rec.TargetCount) +} + +func TestGetReshapeRecommendations_CatalogMissingFamilySurfacesReason(t *testing.T) { + ec2 := &fakeReshapeEC2Stub{offered: []string{"c5.large", "c5.xlarge"}} + rec := reshapeOnce(t, newCatalogReshapeHandler(ec2)).Recommendations[0] + + assert.Empty(t, rec.TargetInstanceType, "an unoffered family must not get a target") + assert.Zero(t, rec.TargetCount) + assert.Contains(t, rec.Reason, "instance family not offered in region (m5)") +} + +// A catalog error or an empty catalog falls back to the legacy ladder with a +// warning naming the region, and the fallback is not cached: the next request +// queries the catalog again. +func TestGetReshapeRecommendations_CatalogUnavailableFallsBackWithWarning(t *testing.T) { + cases := []struct { + name string + ec2 *fakeReshapeEC2Stub + warning string + }{ + { + name: "catalog error", + ec2: &fakeReshapeEC2Stub{offeredErr: errors.New("UnauthorizedOperation: ec2:DescribeInstanceTypeOfferings")}, + warning: "failed, proceeding without the catalog: UnauthorizedOperation", + }, + { + name: "empty catalog", + ec2: &fakeReshapeEC2Stub{offered: []string{}}, + warning: "came back empty, proceeding without the catalog", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + logs := captureWarnings(t) + h := newCatalogReshapeHandler(tc.ec2) + + for i := 1; i <= 2; i++ { + rec := reshapeOnce(t, h).Recommendations[0] + assert.Equal(t, "m5.2xlarge", rec.TargetInstanceType, "request %d must use the legacy ladder", i) + assert.Equal(t, int32(1), rec.TargetCount, "request %d", i) + assert.Equal(t, i, tc.ec2.offeredCalls, "request %d must query the catalog again, not reuse a cached fallback", i) + } + + out := logs.String() + assert.Contains(t, out, `region "`+reshapeCatalogRegion+`"`) + assert.Contains(t, out, tc.warning) + }) + } +} + +func TestOfferedInstanceTypes(t *testing.T) { + ctx := context.Background() + t.Run("returns the catalog unchanged", func(t *testing.T) { + logs := captureWarnings(t) + got := OfferedInstanceTypes(ctx, "us-east-1", func(context.Context) ([]string, error) { + return []string{"m5.large", "c5.xlarge"}, nil + }) + assert.Equal(t, []string{"m5.large", "c5.xlarge"}, got) + assert.Empty(t, logs.String(), "a usable catalog must not log a warning") + }) + t.Run("error returns nil and warns", func(t *testing.T) { + logs := captureWarnings(t) + got := OfferedInstanceTypes(ctx, "us-east-1", func(context.Context) ([]string, error) { + return []string{"m5.large"}, errors.New("throttled") + }) + assert.Nil(t, got, "a partial result next to an error must not be used as the catalog") + assert.Contains(t, logs.String(), `region "us-east-1" failed, proceeding without the catalog: throttled`) + }) + t.Run("empty returns nil and warns", func(t *testing.T) { + logs := captureWarnings(t) + got := OfferedInstanceTypes(ctx, "ap-south-2", func(context.Context) ([]string, error) { return nil, nil }) + assert.Nil(t, got) + assert.Contains(t, logs.String(), `region "ap-south-2" came back empty`) + }) +} diff --git a/internal/server/handler_ri_exchange.go b/internal/server/handler_ri_exchange.go index 171af4a8b..ed298048e 100644 --- a/internal/server/handler_ri_exchange.go +++ b/internal/server/handler_ri_exchange.go @@ -17,6 +17,7 @@ import ( awsprovider "github.com/LeanerCloud/cloud-commitments-go/providers/aws" "github.com/LeanerCloud/cloud-commitments-go/providers/aws/recommendations" ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/api" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" "github.com/LeanerCloud/cloud-commitments-platform/internal/email" ) @@ -27,10 +28,44 @@ import ( type riExchangeClients struct { listConvertibleRIs func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) getRIUtilization func(ctx context.Context, lookbackDays int) ([]recommendations.RIUtilization, error) - exchangeClient exchange.ExchangeClientInterface - lookupOffering exchange.LookupOfferingFunc - accountID string - region string + // offeredInstanceTypes lists the region's offered EC2 instance types + // (DescribeInstanceTypeOfferings), which limit the reshape targets. + offeredInstanceTypes func(ctx context.Context) ([]string, error) + exchangeClient exchange.ExchangeClientInterface + lookupOffering exchange.LookupOfferingFunc + accountID string + region string +} + +// riExchangeEC2 is the slice of the EC2 client the scheduled and ladder RI +// exchange runs need. Satisfied by *ec2svc.Client. +type riExchangeEC2 interface { + convertibleOfferingFinder + ListConvertibleReservedInstances(ctx context.Context) ([]ec2svc.ConvertibleRI, error) + GetValidResourceTypes(ctx context.Context) ([]string, error) +} + +// riUtilizationSource is the slice of the recommendations client the RI +// exchange runs need. Satisfied by *awsprovider.RecommendationsClientAdapter. +type riUtilizationSource interface { + GetRIUtilization(ctx context.Context, lookbackDays int, region string) ([]recommendations.RIUtilization, error) +} + +// newRIExchangeClients wires the RI exchange dependencies for one region. Both +// the standalone scheduled reshape and the ladder exchange runner use it, so +// they read RIs, utilization and the instance type catalog the same way. +func newRIExchangeClients(ec2Client riExchangeEC2, recs riUtilizationSource, exchangeClient exchange.ExchangeClientInterface, region, accountID string) riExchangeClients { + return riExchangeClients{ + listConvertibleRIs: ec2Client.ListConvertibleReservedInstances, + getRIUtilization: func(ctx context.Context, lookbackDays int) ([]recommendations.RIUtilization, error) { + return recs.GetRIUtilization(ctx, lookbackDays, region) + }, + offeredInstanceTypes: ec2Client.GetValidResourceTypes, + exchangeClient: exchangeClient, + lookupOffering: newConvertibleOfferingLookup(ec2Client), + accountID: accountID, + region: region, + } } // handleRIExchangeReshape runs the automated RI exchange analysis and execution. @@ -54,20 +89,13 @@ func (app *Application) handleRIExchangeReshape(ctx context.Context) (*exchange. return nil, fmt.Errorf("failed to load AWS config: %w", err) } - ec2Client := awsprovider.NewEC2ClientDirect(awsCfg) - clients := riExchangeClients{ - listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { - return ec2Client.ListConvertibleReservedInstances(ctx) - }, - getRIUtilization: func(ctx context.Context, lookbackDays int) ([]recommendations.RIUtilization, error) { - recsClient := awsprovider.NewRecommendationsClientDirect(awsCfg) - return recsClient.GetRIUtilization(ctx, lookbackDays, awsCfg.Region) - }, - exchangeClient: exchange.NewExchangeClient(awsCfg), - lookupOffering: newConvertibleOfferingLookup(ec2Client), - accountID: resolveAccountID(ctx, awsCfg), - region: awsCfg.Region, - } + clients := newRIExchangeClients( + awsprovider.NewEC2ClientDirect(awsCfg), + awsprovider.NewRecommendationsClientDirect(awsCfg), + exchange.NewExchangeClient(awsCfg), + awsCfg.Region, + resolveAccountID(ctx, awsCfg), + ) return app.executeRIExchangeReshape(ctx, cfg, clients) } @@ -143,11 +171,12 @@ func (app *Application) executeRIExchangeReshape(ctx context.Context, cfg *confi store := newConfigExchangeStoreAdapter(app.Config) result, err := exchange.RunAutoExchange(ctx, exchange.RunAutoExchangeParams{ - Store: store, - ExchangeClient: clients.exchangeClient, - LookupOffering: clients.lookupOffering, - RIs: riInfos, - Utilization: utilInfos, + Store: store, + ExchangeClient: clients.exchangeClient, + LookupOffering: clients.lookupOffering, + RIs: riInfos, + Utilization: utilInfos, + OfferedInstanceTypes: api.OfferedInstanceTypes(ctx, clients.region, clients.offeredInstanceTypes), Config: exchange.RIExchangeConfig{ Mode: cfg.RIExchangeMode, UtilizationThreshold: cfg.RIExchangeUtilizationThreshold, diff --git a/internal/server/handler_ri_exchange_reservation_integration_test.go b/internal/server/handler_ri_exchange_reservation_integration_test.go index 69a50d244..8a5ae31f5 100644 --- a/internal/server/handler_ri_exchange_reservation_integration_test.go +++ b/internal/server/handler_ri_exchange_reservation_integration_test.go @@ -69,9 +69,10 @@ func TestRIExchangeAutoWorkers_ConcreteStoreAndFixtureClient(t *testing.T) { standaloneCfg := &config.GlobalConfig{RIExchangeEnabled: true, RIExchangeMode: "auto", RIExchangeUtilizationThreshold: 95, RIExchangeMaxPerExchangeUSD: 1000, RIExchangeMaxDailyUSD: baseline + 1000, RIExchangeLookbackDays: 30} standalone, err := app.executeRIExchangeReshape(ctx, standaloneCfg, riExchangeClients{ - listConvertibleRIs: func(context.Context) ([]ec2svc.ConvertibleRI, error) { return ri(standaloneRIID), nil }, - getRIUtilization: func(context.Context, int) ([]recommendations.RIUtilization, error) { return util(standaloneRIID), nil }, - exchangeClient: quoteClient(standaloneRIID, &standaloneExecutions), + offeredInstanceTypes: noOfferedCatalog, + listConvertibleRIs: func(context.Context) ([]ec2svc.ConvertibleRI, error) { return ri(standaloneRIID), nil }, + getRIUtilization: func(context.Context, int) ([]recommendations.RIUtilization, error) { return util(standaloneRIID), nil }, + exchangeClient: quoteClient(standaloneRIID, &standaloneExecutions), lookupOffering: func(context.Context, exchange.LookupOfferingParams) (string, error) { return "offering-test", nil }, diff --git a/internal/server/handler_ri_exchange_test.go b/internal/server/handler_ri_exchange_test.go index d887971d0..f8ca06fba 100644 --- a/internal/server/handler_ri_exchange_test.go +++ b/internal/server/handler_ri_exchange_test.go @@ -498,6 +498,7 @@ func TestExecuteRIExchangeReshape_NoConvertibleRIs(t *testing.T) { } clients := riExchangeClients{ + offeredInstanceTypes: noOfferedCatalog, listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { return nil, nil // no RIs }, @@ -531,6 +532,7 @@ func TestExecuteRIExchangeReshape_ListRIsFailure(t *testing.T) { } clients := riExchangeClients{ + offeredInstanceTypes: noOfferedCatalog, listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { return nil, errors.New("EC2 API throttled") }, @@ -556,6 +558,7 @@ func TestExecuteRIExchangeReshape_UtilizationFailure(t *testing.T) { } clients := riExchangeClients{ + offeredInstanceTypes: noOfferedCatalog, listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { return []ec2svc.ConvertibleRI{{ReservedInstanceID: "ri-1"}}, nil }, @@ -603,6 +606,7 @@ func TestExecuteRIExchangeReshape_ManualMode(t *testing.T) { } clients := riExchangeClients{ + offeredInstanceTypes: noOfferedCatalog, listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { return []ec2svc.ConvertibleRI{ { @@ -710,6 +714,7 @@ func TestExecuteRIExchangeReshape_AutoMode(t *testing.T) { } clients := riExchangeClients{ + offeredInstanceTypes: noOfferedCatalog, listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { return []ec2svc.ConvertibleRI{ { @@ -818,6 +823,7 @@ func TestExecuteRIExchangeReshape_DailyCapHitMidRun(t *testing.T) { } clients := riExchangeClients{ + offeredInstanceTypes: noOfferedCatalog, listConvertibleRIs: func(ctx context.Context) ([]ec2svc.ConvertibleRI, error) { return []ec2svc.ConvertibleRI{ { diff --git a/internal/server/ladder_write.go b/internal/server/ladder_write.go index afce36e10..ceed9f8f6 100644 --- a/internal/server/ladder_write.go +++ b/internal/server/ladder_write.go @@ -11,6 +11,7 @@ import ( pkgladder "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" awsprovider "github.com/LeanerCloud/cloud-commitments-go/providers/aws" awsladder "github.com/LeanerCloud/cloud-commitments-go/providers/aws/ladder" + "github.com/LeanerCloud/cloud-commitments-platform/internal/api" ) // exchangeRunnerAdapter bridges internal/server wiring (exchange store, EC2 exchange @@ -44,14 +45,24 @@ func (a *exchangeRunnerAdapter) RunAutoExchange(ctx context.Context, cfg exchang return nil, fmt.Errorf("exchangeRunnerAdapter: load AWS config: %w", err) } - ec2Client := awsprovider.NewEC2ClientDirect(awsCfg) - recsClient := awsprovider.NewRecommendationsClientDirect(awsCfg) + clients := newRIExchangeClients( + awsprovider.NewEC2ClientDirect(awsCfg), + awsprovider.NewRecommendationsClientDirect(awsCfg), + exchange.NewExchangeClient(awsCfg), + a.region, + a.accountID, + ) + return a.runWithClients(ctx, cfg, ladderRunID, dryRun, clients) +} - instances, err := ec2Client.ListConvertibleReservedInstances(ctx) +// runWithClients is RunAutoExchange after client construction, split out so +// tests can drive it with fakes. +func (a *exchangeRunnerAdapter) runWithClients(ctx context.Context, cfg exchange.RIExchangeConfig, ladderRunID *string, dryRun bool, clients riExchangeClients) (*exchange.AutoExchangeResult, error) { + instances, err := clients.listConvertibleRIs(ctx) if err != nil { return nil, fmt.Errorf("exchangeRunnerAdapter: list convertible RIs: %w", err) } - utilData, err := recsClient.GetRIUtilization(ctx, cfg.LookbackDays, a.region) + utilData, err := clients.getRIUtilization(ctx, cfg.LookbackDays) if err != nil { return nil, fmt.Errorf("exchangeRunnerAdapter: get RI utilization: %w", err) } @@ -60,18 +71,19 @@ func (a *exchangeRunnerAdapter) RunAutoExchange(ctx context.Context, cfg exchang store := newConfigExchangeStoreAdapter(a.app.Config) return exchange.RunAutoExchange(ctx, exchange.RunAutoExchangeParams{ - Store: store, - ExchangeClient: exchange.NewExchangeClient(awsCfg), - LookupOffering: newConvertibleOfferingLookup(ec2Client), - RIs: riInfos, - Utilization: utilInfos, - Config: cfg, - AccountID: a.accountID, - Region: a.region, - DashboardURL: a.app.appConfig.DashboardURL, - RIMetadata: riMetadata, - LadderRunID: ladderRunID, - DryRun: dryRun, + Store: store, + ExchangeClient: clients.exchangeClient, + LookupOffering: clients.lookupOffering, + RIs: riInfos, + Utilization: utilInfos, + OfferedInstanceTypes: api.OfferedInstanceTypes(ctx, clients.region, clients.offeredInstanceTypes), + Config: cfg, + AccountID: clients.accountID, + Region: clients.region, + DashboardURL: a.app.appConfig.DashboardURL, + RIMetadata: riMetadata, + LadderRunID: ladderRunID, + DryRun: dryRun, }) } diff --git a/internal/server/ri_exchange_offered_types_test.go b/internal/server/ri_exchange_offered_types_test.go new file mode 100644 index 000000000..aa5c89065 --- /dev/null +++ b/internal/server/ri_exchange_offered_types_test.go @@ -0,0 +1,257 @@ +package server + +import ( + "bytes" + "context" + "errors" + "math/big" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/exchange" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-go/providers/aws/recommendations" + ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// noOfferedCatalog is an offeredInstanceTypes source with no catalog, so the +// reshape keeps the legacy size ladder (after a warning). +func noOfferedCatalog(context.Context) ([]string, error) { return nil, nil } + +const ( + catalogTestRegion = "eu-west-3" + catalogTestRIID = "ri-catalog" +) + +// catalogEC2 serves one convertible 1x m5.4xlarge RI (32 normalized units) +// and a configurable instance type catalog, and records offering lookups. +type catalogEC2 struct { + offered []string + offeredErr error + offeredCalls int + lookups []string +} + +func (f *catalogEC2) ListConvertibleReservedInstances(context.Context) ([]ec2svc.ConvertibleRI, error) { + return []ec2svc.ConvertibleRI{{ + ReservedInstanceID: catalogTestRIID, + InstanceType: "m5.4xlarge", + InstanceCount: 1, + ProductDescription: "Linux/UNIX", + InstanceTenancy: "default", + Scope: "Region", + Duration: 31536000, + CurrencyCode: "USD", + }}, nil +} + +func (f *catalogEC2) GetValidResourceTypes(context.Context) ([]string, error) { + f.offeredCalls++ + return f.offered, f.offeredErr +} + +func (f *catalogEC2) FindConvertibleOffering(_ context.Context, p ec2svc.FindConvertibleOfferingParams) (string, error) { + f.lookups = append(f.lookups, p.InstanceType) + return "offering-1", nil +} + +// catalogRecs reports the RI at 50% utilization (16 of 32 units used): the +// legacy ladder reshapes it to 1x m5.2xlarge. +type catalogRecs struct{ regions []string } + +func (r *catalogRecs) GetRIUtilization(_ context.Context, _ int, region string) ([]recommendations.RIUtilization, error) { + r.regions = append(r.regions, region) + return []recommendations.RIUtilization{{ReservedInstanceID: catalogTestRIID, UtilizationPercent: 50}}, nil +} + +func validQuoteClient() *mockExchangeClient { + return &mockExchangeClient{ + getQuoteFunc: func(context.Context, exchange.ExchangeQuoteRequest) (*exchange.ExchangeQuoteSummary, error) { + return &exchange.ExchangeQuoteSummary{ + IsValidExchange: true, + PaymentDueUSD: new(big.Rat).SetFloat64(5), + PaymentDueUSDStr: "5.00", + CurrencyCode: "USD", + }, nil + }, + } +} + +type lockedLog struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (l *lockedLog) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.Write(p) +} + +func (l *lockedLog) String() string { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.String() +} + +// captureWarnLogs routes pkg/logging into a buffer at warn level for the rest +// of the test. Tests using it must not call t.Parallel. +func captureWarnLogs(t *testing.T) *lockedLog { + t.Helper() + out := &lockedLog{} + prevOut := logging.SetOutput(out) + prevLevel := logging.GetLevel() + logging.SetLevelValue(logging.LevelWarn) + t.Cleanup(func() { + logging.SetOutput(prevOut) + logging.SetLevelValue(prevLevel) + }) + return out +} + +func TestNewRIExchangeClients_RoutesToClients(t *testing.T) { + ctx := context.Background() + ec2 := &catalogEC2{offered: []string{"m5.large"}} + recs := &catalogRecs{} + exch := validQuoteClient() + c := newRIExchangeClients(ec2, recs, exch, catalogTestRegion, "123456789012") + + offered, err := c.offeredInstanceTypes(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"m5.large"}, offered) + assert.Equal(t, 1, ec2.offeredCalls) + + ris, err := c.listConvertibleRIs(ctx) + require.NoError(t, err) + require.Len(t, ris, 1) + + _, err = c.getRIUtilization(ctx, 30) + require.NoError(t, err) + assert.Equal(t, []string{catalogTestRegion}, recs.regions) + + _, err = c.lookupOffering(ctx, exchange.LookupOfferingParams{InstanceType: "m5.xlarge"}) + require.NoError(t, err) + assert.Equal(t, []string{"m5.xlarge"}, ec2.lookups) + + assert.Same(t, exch, c.exchangeClient) + assert.Equal(t, catalogTestRegion, c.region) + assert.Equal(t, "123456789012", c.accountID) +} + +// reshapeEntryPoints are the two scheduled paths that run RunAutoExchange: +// the standalone RI-exchange task and the ladder's exchange runner. +var reshapeEntryPoints = []struct { + name string + run func(t *testing.T, clients riExchangeClients) *exchange.AutoExchangeResult +}{ + { + name: "standalone", + run: func(t *testing.T, clients riExchangeClients) *exchange.AutoExchangeResult { + app := &Application{Config: &mockConfigStoreForExchange{}, Email: &mockEmailSender{}} + cfg := &config.GlobalConfig{ + RIExchangeEnabled: true, + RIExchangeMode: "manual", + RIExchangeUtilizationThreshold: 95, + RIExchangeMaxPerExchangeUSD: 100, + RIExchangeMaxDailyUSD: 500, + RIExchangeLookbackDays: 30, + } + result, err := app.executeRIExchangeReshape(context.Background(), cfg, clients) + require.NoError(t, err) + return result + }, + }, + { + name: "ladder", + run: func(t *testing.T, clients riExchangeClients) *exchange.AutoExchangeResult { + a := &exchangeRunnerAdapter{app: &Application{Config: &mockConfigStoreForExchange{}}} + cfg := exchange.RIExchangeConfig{ + Mode: "manual", + UtilizationThreshold: 95, + MaxPaymentPerExchangeUSD: 100, + MaxPaymentDailyUSD: 500, + LookbackDays: 30, + } + ladderRunID := "ladder-run-1" + result, err := a.runWithClients(context.Background(), cfg, &ladderRunID, true, clients) + require.NoError(t, err) + return result + }, + }, +} + +func TestRIExchangeReshape_OfferedCatalog(t *testing.T) { + cases := []struct { + name string + ec2 func() *catalogEC2 + wantLookups []string + wantTarget string + wantCount int32 + wantSkip string + wantWarning string + }{ + { + name: "catalog limits the target to offered sizes", + ec2: func() *catalogEC2 { return &catalogEC2{offered: []string{"m5.4xlarge", "m5.xlarge", "c5.large"}} }, + wantLookups: []string{"m5.xlarge"}, + wantTarget: "m5.xlarge", + wantCount: 2, + }, + { + name: "catalog without the family skips with a reason", + ec2: func() *catalogEC2 { return &catalogEC2{offered: []string{"c5.large", "c5.xlarge"}} }, + wantSkip: "instance family not offered in region (m5)", + }, + { + name: "catalog error falls back to the legacy ladder", + ec2: func() *catalogEC2 { return &catalogEC2{offeredErr: errors.New("UnauthorizedOperation")} }, + wantLookups: []string{"m5.2xlarge"}, + wantTarget: "m5.2xlarge", + wantCount: 1, + wantWarning: "failed, proceeding without the catalog: UnauthorizedOperation", + }, + { + name: "empty catalog falls back to the legacy ladder", + ec2: func() *catalogEC2 { return &catalogEC2{offered: []string{}} }, + wantLookups: []string{"m5.2xlarge"}, + wantTarget: "m5.2xlarge", + wantCount: 1, + wantWarning: "came back empty, proceeding without the catalog", + }, + } + for _, ep := range reshapeEntryPoints { + for _, tc := range cases { + t.Run(ep.name+"/"+tc.name, func(t *testing.T) { + logs := captureWarnLogs(t) + ec2 := tc.ec2() + clients := newRIExchangeClients(ec2, &catalogRecs{}, validQuoteClient(), catalogTestRegion, "123456789012") + + result := ep.run(t, clients) + + assert.Equal(t, 1, ec2.offeredCalls, "the run must query the region's instance type catalog once") + assert.Equal(t, tc.wantLookups, ec2.lookups, "offering lookups") + outcomes := append(append([]exchange.ExchangeOutcome{}, result.Pending...), result.Completed...) + if tc.wantSkip != "" { + assert.Empty(t, outcomes) + require.Len(t, result.Skipped, 1) + assert.Contains(t, result.Skipped[0].Reason, tc.wantSkip) + } else { + assert.Empty(t, result.Skipped) + require.Len(t, outcomes, 1) + assert.Equal(t, tc.wantTarget, outcomes[0].TargetInstanceType) + assert.Equal(t, tc.wantCount, outcomes[0].TargetCount) + } + if tc.wantWarning != "" { + assert.Contains(t, logs.String(), `region "`+catalogTestRegion+`"`) + assert.Contains(t, logs.String(), tc.wantWarning) + } else { + assert.NotContains(t, logs.String(), "proceeding without the catalog") + } + }) + } + } +} diff --git a/internal/server/ri_exchange_zone_test.go b/internal/server/ri_exchange_zone_test.go index 1515fd590..b6c3b7a0e 100644 --- a/internal/server/ri_exchange_zone_test.go +++ b/internal/server/ri_exchange_zone_test.go @@ -89,7 +89,8 @@ func runZoneReshape(t *testing.T, ri ec2types.ReservedInstances) (*exchange.Auto RIExchangeLookbackDays: 30, } clients := riExchangeClients{ - listConvertibleRIs: c.ListConvertibleReservedInstances, + offeredInstanceTypes: noOfferedCatalog, + listConvertibleRIs: c.ListConvertibleReservedInstances, getRIUtilization: func(context.Context, int) ([]recommendations.RIUtilization, error) { return []recommendations.RIUtilization{{ReservedInstanceID: zoneTestRIID, UtilizationPercent: 40.0}}, nil },