diff --git a/internal/config/store_postgres_recommendations.go b/internal/config/store_postgres_recommendations.go index 1baaaf0e..f3d8b93b 100644 --- a/internal/config/store_postgres_recommendations.go +++ b/internal/config/store_postgres_recommendations.go @@ -63,7 +63,7 @@ func (s *PostgresStore) ReplaceRecommendations(ctx context.Context, collectedAt // UpsertRecommendations is the incremental write path: it upserts each row // by natural key (cloud_account_id, provider, service, region, -// resource_type, engine, term, payment_option, availability_zone) and then evicts stale rows for the +// resource_type, engine, term, payment_option, availability_zone, platform, tenancy) and then evicts stale rows for the // set of (provider, account) pairs that successfully collected in this run. // Pairs whose collection failed are NOT in successfulCollects and their // stale rows stay — callers see older data with a banner rather than a @@ -72,8 +72,9 @@ func (s *PostgresStore) ReplaceRecommendations(ctx context.Context, collectedAt // Migration 000032 broadened the natural key to include term + payment_option, // so per-rec ON CONFLICT no longer collides on SQLSTATE 21000. Migration // 000109 added availability_zone, a GENERATED column derived from payload that -// is never written here; the upsert arbiter is the 9-column -// recommendations_natural_key_az_idx (issue #792). +// is never written here. Migration 000111 added the platform and tenancy +// generated columns the same way; the upsert arbiter is the 11-column +// recommendations_natural_key_az_tp_idx (issues #792, #806). // // The eviction predicate uses (provider, account_key) IN (unnest($2, $3)) // where account_key matches the generated column on the table — nil @@ -227,7 +228,7 @@ func insertRecommendationsBatch(ctx context.Context, tx pgx.Tx, collectedAt time if onConflict { query += ` - ON CONFLICT (account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone) + ON CONFLICT (account_key, provider, service, region, resource_type, engine, term, payment_option, availability_zone, platform, tenancy) DO UPDATE SET payload = EXCLUDED.payload, upfront_cost = EXCLUDED.upfront_cost, diff --git a/internal/config/store_postgres_recommendations_test.go b/internal/config/store_postgres_recommendations_test.go index 9c4198a6..23367043 100644 --- a/internal/config/store_postgres_recommendations_test.go +++ b/internal/config/store_postgres_recommendations_test.go @@ -581,13 +581,14 @@ func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) { t.Run("row written by pre-PR2b code is found and updated", func(t *testing.T) { require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil)) - // What the old code stored: AZ only inside details, 8-column semantics. + // What the old code stored: AZ, platform and tenancy only inside details (the + // collector always sets all three on a zonal EC2 rec), 8-column semantics. _, err := pool.Exec(ctx, ` INSERT INTO recommendations (collected_at, provider, service, region, resource_type, engine, payload, upfront_cost, monthly_savings, term, payment_option) VALUES ($1, 'aws', 'ec2', 'us-east-1', 'm5.large', '', - '{"details":{"availability_zone":"us-east-1a"}}'::jsonb, 100, 5, 12, 'no-upfront')`, t0) + '{"details":{"platform":"Linux/UNIX","tenancy":"default","availability_zone":"us-east-1a"}}'::jsonb, 100, 5, 12, 'no-upfront')`, t0) require.NoError(t, err) require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{zonalAWSRec("us-east-1a", 40)}, collect)) var n int @@ -600,6 +601,82 @@ func TestPostgresStore_UpsertRecommendations_AZArbiter(t *testing.T) { }) } +// Issue #806, PR-C: the upsert arbiter is the 11-column index that adds the +// generated platform and tenancy columns. A re-upsert on the same platform and +// tenancy updates in place, regional non-EC2 rows (platform and tenancy ”) are +// unchanged, and a row written by pre-PR-C code is found and updated. A batch +// of rows differing only by platform or tenancy is covered with the 9-column +// index drop (migration 000112). +func TestPostgresStore_UpsertRecommendations_PlatformTenancyArbiter(t *testing.T) { + ctx := context.Background() + container, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + defer container.Cleanup(ctx) + pool := container.DB.Pool() + require.NoError(t, migrations.RunMigrations(ctx, pool, getMigrationsPath(), "", "")) + store := config.NewPostgresStore(container.DB) + t0 := time.Now().UTC().Truncate(time.Second) + collect := []config.SuccessfulCollect{{Provider: "aws"}} + + ec2 := func(platform, tenancy string, savings float64) config.RecommendationRecord { + rec := awsRec("p-"+platform+tenancy, "ec2", "us-east-1", "m5.large", savings) + rec.Details = json.RawMessage(`{"instance_type":"m5.large","platform":"` + platform + `","tenancy":"` + tenancy + `"}`) + return rec + } + columns := func(t *testing.T) (platform, tenancy string, n int) { + t.Helper() + require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM recommendations`).Scan(&n)) + require.NoError(t, pool.QueryRow(ctx, `SELECT platform, tenancy FROM recommendations LIMIT 1`).Scan(&platform, &tenancy)) + return platform, tenancy, n + } + + t.Run("same platform and tenancy re-upsert updates in place", func(t *testing.T) { + require.NoError(t, store.UpsertRecommendations(ctx, t0, []config.RecommendationRecord{ec2("Windows", "dedicated", 10)}, collect)) + require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{ec2("Windows", "dedicated", 25)}, collect)) + got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + require.Len(t, got, 1) + assert.InDelta(t, 25.0, got[0].Savings, 0.001) + platform, tenancy, n := columns(t) + assert.Equal(t, 1, n) + assert.Equal(t, "Windows", platform, "the generated column is derived from the payload") + assert.Equal(t, "dedicated", tenancy) + }) + + t.Run("regional non-EC2 row has empty platform and tenancy and updates in place", func(t *testing.T) { + require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil)) + rds := awsRec("r1", "rds", "us-east-1", "db.m5.large", 10) + require.NoError(t, store.UpsertRecommendations(ctx, t0, []config.RecommendationRecord{rds}, collect)) + rds.Savings = 30 + require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{rds}, collect)) + got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + require.Len(t, got, 1) + assert.InDelta(t, 30.0, got[0].Savings, 0.001) + platform, tenancy, n := columns(t) + assert.Equal(t, 1, n) + assert.Empty(t, platform) + assert.Empty(t, tenancy) + }) + + t.Run("row written by pre-PR-C code is found and updated", func(t *testing.T) { + require.NoError(t, store.ReplaceRecommendations(ctx, t0, nil)) + // What the old code stored: platform and tenancy only inside details. + _, err := pool.Exec(ctx, ` + INSERT INTO recommendations + (collected_at, provider, service, region, resource_type, engine, payload, + upfront_cost, monthly_savings, term, payment_option) + VALUES ($1, 'aws', 'ec2', 'us-east-1', 'm5.large', '', + '{"details":{"platform":"Linux/UNIX","tenancy":"default"}}'::jsonb, 100, 5, 12, 'no-upfront')`, t0) + require.NoError(t, err) + require.NoError(t, store.UpsertRecommendations(ctx, t0.Add(time.Hour), []config.RecommendationRecord{ec2("Linux/UNIX", "default", 40)}, collect)) + got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{}) + require.NoError(t, err) + require.Len(t, got, 1, "the legacy row must be the conflict target, not duplicated") + assert.InDelta(t, 40.0, got[0].Savings, 0.001) + }) +} + // Issue #792, PR2c: once migration 000110 drops the 8-column index, two zonal // EC2 recommendations that differ only by availability zone upsert in ONE batch // (the shape persistCollection writes) without "ON CONFLICT DO UPDATE command diff --git a/internal/purchase/azure_pricing_integration_test.go b/internal/purchase/azure_pricing_integration_test.go index 85d98d65..93dae37a 100644 --- a/internal/purchase/azure_pricing_integration_test.go +++ b/internal/purchase/azure_pricing_integration_test.go @@ -102,8 +102,9 @@ func TestApproveAzurePricingRequiresMigrationAndRecollection(t *testing.T) { require.NoError(t, migrations.MigrateToVersion(ctx, pool, rampMigrationsPath(), 106)) t.Run("106 invalidated old prices", func(t *testing.T) { approve(t, ErrStaleAzurePricing, 0) }) rec.UpfrontCost = 700 - // The store's upsert needs the availability_zone column from migration 000109. - require.NoError(t, migrations.MigrateToVersion(ctx, pool, rampMigrationsPath(), 109)) + // The store's upsert needs the availability_zone, platform and tenancy columns and the + // 11-column index from migration 000111. + require.NoError(t, migrations.MigrateToVersion(ctx, pool, rampMigrationsPath(), 111)) require.NoError(t, pricing.UpsertRecommendations(ctx, time.Now().Add(time.Minute), []config.RecommendationRecord{rec}, collects)) t.Run("recollected prices purchase once", func(t *testing.T) { approve(t, nil, 1) }) }