Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions internal/config/recommendation_identity.go
Original file line number Diff line number Diff line change
Expand Up @@ -64,3 +64,16 @@ func RecIdentityKey(rec *RecommendationRecord) (string, error) {
rec.AvailabilityZone, ci.Platform, ci.Tenancy,
}, "\x1f"), nil
}

// SuppressionVariant is the rec's availability zone, platform and tenancy as
// the purchase suppression key carries them (issue #816): the same AZ field and
// the same Details read (ComputeIdentity) as RecIdentityKey, so the write side
// (BuildSuppressions) and the read side (the scheduler's subtraction pass)
// cannot drift from the identity key or from each other.
func (r *RecommendationRecord) SuppressionVariant() (SuppressionVariant, error) {
ci, err := r.ComputeIdentity()
if err != nil {
return SuppressionVariant{}, err
}
return SuppressionVariant{AvailabilityZone: r.AvailabilityZone, Platform: ci.Platform, Tenancy: ci.Tenancy}, nil
}
39 changes: 32 additions & 7 deletions internal/config/store_postgres_suppressions.go
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
package config

// store_postgres_suppressions.go — CRUD + transaction-accepting variants
// for the purchase_suppressions table (migration 000037) plus the
// for the purchase_suppressions table (migrations 000037, 000134) plus the
// WithTx helper used by executePurchase and cancelExecution to commit
// the execution row + suppression rows atomically.

Expand Down Expand Up @@ -103,16 +103,24 @@ func (s *PostgresStore) CreateSuppressionTx(ctx context.Context, tx pgx.Tx, sup
sup.ID = uuid.New().String()
}

// A nil Variant writes NULL into all three variant columns (a legacy,
// 6-tuple-wide row); otherwise all three are written, empty strings included
// (migration 000134 keeps them all-NULL or all-set).
var az, platform, tenancy *string
if sup.Variant != nil {
az, platform, tenancy = &sup.Variant.AvailabilityZone, &sup.Variant.Platform, &sup.Variant.Tenancy
}
const q = `
INSERT INTO purchase_suppressions (
id, execution_id, account_id, provider, service, region,
resource_type, engine, suppressed_count, expires_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
resource_type, engine, suppressed_count, expires_at,
availability_zone, platform, tenancy
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
`
if _, err := tx.Exec(ctx, q,
sup.ID, sup.ExecutionID, sup.AccountID, sup.Provider, sup.Service,
sup.Region, sup.ResourceType, sup.Engine, sup.SuppressedCount,
sup.ExpiresAt,
sup.ExpiresAt, az, platform, tenancy,
); err != nil {
return fmt.Errorf("failed to insert purchase suppression: %w", err)
}
Expand Down Expand Up @@ -144,12 +152,15 @@ func (s *PostgresStore) DeleteSuppressionsByExecutionTx(ctx context.Context, tx

// ListActiveSuppressions returns every suppression row whose expiry
// hasn't passed. The scheduler groups these by the 6-tuple key
// (account, provider, service, region, resource_type, engine) to
// subtract cumulative suppressed counts from its rec-list output.
// (account, provider, service, region, resource_type, engine) and
// matches each row's variant (availability zone, platform, tenancy;
// issue #816) to subtract cumulative suppressed counts from its rec-list
// output. A row with NULL variant columns comes back with a nil Variant.
func (s *PostgresStore) ListActiveSuppressions(ctx context.Context) ([]PurchaseSuppression, error) {
const q = `
SELECT id, execution_id, account_id, provider, service, region,
resource_type, engine, suppressed_count, expires_at, created_at
resource_type, engine, suppressed_count, expires_at, created_at,
availability_zone, platform, tenancy
FROM purchase_suppressions
WHERE expires_at > NOW()
`
Expand All @@ -162,17 +173,31 @@ func (s *PostgresStore) ListActiveSuppressions(ctx context.Context) ([]PurchaseS
var out []PurchaseSuppression
for rows.Next() {
var sup PurchaseSuppression
var az, platform, tenancy *string
if err := rows.Scan(
&sup.ID, &sup.ExecutionID, &sup.AccountID, &sup.Provider,
&sup.Service, &sup.Region, &sup.ResourceType, &sup.Engine,
&sup.SuppressedCount, &sup.ExpiresAt, &sup.CreatedAt,
&az, &platform, &tenancy,
); err != nil {
return nil, fmt.Errorf("failed to scan suppression row: %w", err)
}
sup.Variant = suppressionVariantFromColumns(az, platform, tenancy)
out = append(out, sup)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("failed to iterate suppression rows: %w", err)
}
return out, nil
}

// suppressionVariantFromColumns rebuilds a row's variant from its nullable
// columns. The 000134 CHECK keeps them all-NULL or all-set; a row with any
// NULL is read as having no variant (a 6-tuple-wide wildcard), which
// over-suppresses rather than over-shows.
func suppressionVariantFromColumns(az, platform, tenancy *string) *SuppressionVariant {
if az == nil || platform == nil || tenancy == nil {
return nil
}
return &SuppressionVariant{AvailabilityZone: *az, Platform: *platform, Tenancy: *tenancy}
}
36 changes: 33 additions & 3 deletions internal/config/store_postgres_suppressions_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ func TestPGXMock_CreateSuppression_RoundTrip(t *testing.T) {
mock.ExpectBegin()
mock.ExpectExec(`INSERT INTO purchase_suppressions`).
WithArgs(pgxmock.AnyArg(), "exec-1", "acct-1", "aws", "ec2", "us-east-1",
"t4g.nano", "", 3, pgxmock.AnyArg()).
"t4g.nano", "", 3, pgxmock.AnyArg(), (*string)(nil), (*string)(nil), (*string)(nil)).
WillReturnResult(pgxmock.NewResult("INSERT", 1))
mock.ExpectCommit()

Expand All @@ -33,6 +33,32 @@ func TestPGXMock_CreateSuppression_RoundTrip(t *testing.T) {
assert.NoError(t, mock.ExpectationsWereMet())
}

// TestPGXMock_CreateSuppression_WritesVariant pins issue #816: a row with a
// variant writes all three variant columns, empty strings included (a
// regional, known-empty AZ is not NULL).
func TestPGXMock_CreateSuppression_WritesVariant(t *testing.T) {
mock := newMock(t)
store := storeWith(mock)
ctx := context.Background()

az, platform, tenancy := "", "Linux/UNIX", "default"
mock.ExpectBegin()
mock.ExpectExec(`INSERT INTO purchase_suppressions`).
WithArgs(pgxmock.AnyArg(), "exec-1", "acct-1", "aws", "ec2", "us-east-1",
"t4g.nano", "", 3, pgxmock.AnyArg(), &az, &platform, &tenancy).
WillReturnResult(pgxmock.NewResult("INSERT", 1))
mock.ExpectCommit()

err := store.CreateSuppression(ctx, &PurchaseSuppression{
ExecutionID: "exec-1", AccountID: "acct-1", Provider: "aws",
Service: "ec2", Region: "us-east-1", ResourceType: "t4g.nano",
SuppressedCount: 3, ExpiresAt: time.Now().Add(time.Hour),
Variant: &SuppressionVariant{Platform: "Linux/UNIX", Tenancy: "default"},
})
require.NoError(t, err)
assert.NoError(t, mock.ExpectationsWereMet())
}

func TestPGXMock_CreateSuppression_Validation(t *testing.T) {
mock := newMock(t)
store := storeWith(mock)
Expand Down Expand Up @@ -94,12 +120,14 @@ func TestPGXMock_ListActiveSuppressions(t *testing.T) {
cols := []string{
"id", "execution_id", "account_id", "provider", "service", "region",
"resource_type", "engine", "suppressed_count", "expires_at", "created_at",
"availability_zone", "platform", "tenancy",
}
az, platform, tenancy := "us-east-1a", "Linux/UNIX", "default"
rows := pgxmock.NewRows(cols).
AddRow("sup-1", "exec-1", "acct-1", "aws", "ec2", "us-east-1",
"t4g.nano", "", 3, future, now).
"t4g.nano", "", 3, future, now, nil, nil, nil).
AddRow("sup-2", "exec-2", "acct-1", "aws", "ec2", "us-east-1",
"t4g.nano", "", 2, future, now)
"t4g.nano", "", 2, future, now, &az, &platform, &tenancy)
mock.ExpectQuery(`SELECT .* FROM purchase_suppressions WHERE expires_at > NOW\(\)`).
WillReturnRows(rows)

Expand All @@ -108,8 +136,10 @@ func TestPGXMock_ListActiveSuppressions(t *testing.T) {
assert.Len(t, sups, 2)
assert.Equal(t, "sup-1", sups[0].ID)
assert.Equal(t, 3, sups[0].SuppressedCount)
assert.Nil(t, sups[0].Variant, "NULL variant columns read back as a legacy, 6-tuple-wide row")
assert.Equal(t, "sup-2", sups[1].ID)
assert.Equal(t, 2, sups[1].SuppressedCount)
assert.Equal(t, &SuppressionVariant{AvailabilityZone: "us-east-1a", Platform: "Linux/UNIX", Tenancy: "default"}, sups[1].Variant)
assert.NoError(t, mock.ExpectationsWereMet())
}

Expand Down
102 changes: 69 additions & 33 deletions internal/config/suppressions_build.go
Original file line number Diff line number Diff line change
@@ -1,65 +1,101 @@
package config

import "time"
import (
"time"

"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
)

// suppressionBucket is the key BuildSuppressions aggregates on: the 6-tuple
// (account, provider, service, region, resource_type, engine) plus the rec's
// variant (availability zone, platform, tenancy; issue #816). unknownVariant
// marks a rec whose Details could not be read.
type suppressionBucket struct {
accountID, provider, service, region, resourceType, engine string
variant SuppressionVariant
unknownVariant bool
}

// suppressionBucketOf returns rec's bucket. A rec whose Details cannot be read
// gets no variant: its row is then a wildcard over the 6-tuple, the pre-#816
// behavior, which over-suppresses (under-buys) rather than leaving the
// purchased capacity on show. The purchase paths reject such recs before
// they get here (RecIdentityKey reads the same Details), so this is a
// fallback, not a path in normal use.
func suppressionBucketOf(rec *RecommendationRecord) suppressionBucket {
accountID := ""
if rec.CloudAccountID != nil {
accountID = *rec.CloudAccountID
}
b := suppressionBucket{
accountID: accountID,
provider: rec.Provider,
service: rec.Service,
region: rec.Region,
resourceType: rec.ResourceType,
engine: rec.Engine,
}
v, err := rec.SuppressionVariant()
if err != nil {
logging.Warnf("suppression for %s/%s/%s: %v; suppressing the whole 6-tuple", rec.Provider, rec.Region, rec.ResourceType, err)
b.unknownVariant = true
return b
}
b.variant = v
return b
}

// BuildSuppressions converts the recs on an executePurchase request into
// purchase_suppressions rows, one per unique 6-tuple (account, provider,
// service, region, resource_type, engine). Counts for duplicate tuples
// within the same request are summed (defensive — the UI should collapse
// them client-side, but the backend doesn't trust that). Providers with
// a grace period of 0 (feature disabled) contribute no rows.
// service, region, resource_type, engine) and variant (availability zone,
// platform, tenancy; issue #816), so a purchase of one zonal EC2 variant does
// not hide its sibling AZ, regional, platform or tenancy recs. Counts for
// duplicate keys within the same request are summed (defensive — the UI
// should collapse them client-side, but the backend doesn't trust that).
// Providers with a grace period of 0 (feature disabled) contribute no rows.
func BuildSuppressions(recs []RecommendationRecord, executionID string, cfg *GlobalConfig, now time.Time) []PurchaseSuppression {
// Aggregate count per 6-tuple before writing suppression rows so
// the UNIQUE(execution_id, ...6-tuple) constraint can't fire from
// a request that repeated the same tuple.
type key struct {
accountID, provider, service, region, resourceType, engine string
}
agg := map[key]int{}
order := []key{}
for _rvc := range recs {
rec := recs[_rvc]
if rec.Count <= 0 {
// Aggregate count per key before writing suppression rows so the
// UNIQUE(execution_id, ...key) constraint can't fire from a request
// that repeated the same key.
agg := map[suppressionBucket]int{}
order := []suppressionBucket{}
for i := range recs {
if recs[i].Count <= 0 {
continue
}
accountID := ""
if rec.CloudAccountID != nil {
accountID = *rec.CloudAccountID
}
k := key{
accountID: accountID,
provider: rec.Provider,
service: rec.Service,
region: rec.Region,
resourceType: rec.ResourceType,
engine: rec.Engine,
}
k := suppressionBucketOf(&recs[i])
if _, seen := agg[k]; !seen {
order = append(order, k)
}
agg[k] += rec.Count
agg[k] += recs[i].Count
}

out := make([]PurchaseSuppression, 0, len(order))
for _, k := range order {
for i := range order {
k := &order[i]
graceDays := DefaultGracePeriodDays
if cfg != nil {
graceDays = cfg.GracePeriodFor(k.provider)
}
if graceDays <= 0 {
continue // feature disabled for this provider
}
out = append(out, PurchaseSuppression{
sup := PurchaseSuppression{
ExecutionID: executionID,
AccountID: k.accountID,
Provider: k.provider,
Service: k.service,
Region: k.region,
ResourceType: k.resourceType,
Engine: k.engine,
SuppressedCount: agg[k],
SuppressedCount: agg[*k],
ExpiresAt: now.Add(time.Duration(graceDays) * 24 * time.Hour),
})
}
if !k.unknownVariant {
v := k.variant
sup.Variant = &v
}
out = append(out, sup)
}
return out
}
84 changes: 84 additions & 0 deletions internal/config/suppressions_build_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -68,3 +68,87 @@ func TestBuildSuppressions_SkipsZeroCountRecs(t *testing.T) {

assert.Empty(t, sups, "zero-count recs don't produce suppression rows")
}

// TestBuildSuppressions_SeparatesVariants pins issue #816: recs on one 6-tuple
// that differ only by availability zone, platform or tenancy get one row each,
// carrying the rec's variant; recs with the same variant are still summed.
func TestBuildSuppressions_SeparatesVariants(t *testing.T) {
ec2 := func(az, platform, tenancy string, count int) RecommendationRecord {
return RecommendationRecord{
Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large",
Count: count, CloudAccountID: strPtr("acct-1"), AvailabilityZone: az,
Details: []byte(`{"platform":"` + platform + `","tenancy":"` + tenancy + `"}`),
}
}
sups := BuildSuppressions([]RecommendationRecord{
ec2("us-east-1a", "Linux/UNIX", "default", 1),
ec2("us-east-1a", "Linux/UNIX", "default", 2),
ec2("us-east-1b", "Linux/UNIX", "default", 3),
ec2("", "Linux/UNIX", "default", 4),
ec2("us-east-1a", "Windows", "default", 5),
ec2("us-east-1a", "Linux/UNIX", "dedicated", 6),
}, "exec-1", &GlobalConfig{}, time.Now())

require.Len(t, sups, 5)
want := []struct {
variant SuppressionVariant
count int
}{
{SuppressionVariant{"us-east-1a", "Linux/UNIX", "default"}, 3},
{SuppressionVariant{"us-east-1b", "Linux/UNIX", "default"}, 3},
{SuppressionVariant{"", "Linux/UNIX", "default"}, 4},
{SuppressionVariant{"us-east-1a", "Windows", "default"}, 5},
{SuppressionVariant{"us-east-1a", "Linux/UNIX", "dedicated"}, 6},
}
for i, w := range want {
require.NotNil(t, sups[i].Variant, "row %d", i)
assert.Equal(t, w.variant, *sups[i].Variant, "row %d", i)
assert.Equal(t, w.count, sups[i].SuppressedCount, "row %d", i)
}
}

// TestBuildSuppressions_NonComputeVariantIsKnownEmpty: a rec without platform
// or tenancy (non-EC2, no Details) gets an empty but known variant, not a
// legacy wildcard.
func TestBuildSuppressions_NonComputeVariantIsKnownEmpty(t *testing.T) {
sups := BuildSuppressions([]RecommendationRecord{
{Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.t4g.micro", Engine: "postgres", Count: 1},
}, "exec-1", &GlobalConfig{}, time.Now())
require.Len(t, sups, 1)
assert.Equal(t, &SuppressionVariant{}, sups[0].Variant)
}

// TestBuildSuppressions_UnreadableDetailsFallBackToWildcard: a rec whose
// Details cannot be read gets no variant, so its row suppresses the whole
// 6-tuple (pre-#816 behavior) instead of nothing.
func TestBuildSuppressions_UnreadableDetailsFallBackToWildcard(t *testing.T) {
sups := BuildSuppressions([]RecommendationRecord{
{Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large", Count: 2,
AvailabilityZone: "us-east-1a", Details: []byte(`{"platform":42}`)},
{Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large", Count: 1,
AvailabilityZone: "us-east-1a", Details: []byte(`{"platform":"Linux/UNIX","tenancy":"default"}`)},
}, "exec-1", &GlobalConfig{}, time.Now())
require.Len(t, sups, 2)
assert.Nil(t, sups[0].Variant)
assert.Equal(t, 2, sups[0].SuppressedCount)
require.NotNil(t, sups[1].Variant)
assert.Equal(t, 1, sups[1].SuppressedCount)
}

// TestPurchaseSuppression_AppliesTo pins the match rule shared by the
// scheduler: exact variant match, legacy rows and unknown rec variants are
// wildcards.
func TestPurchaseSuppression_AppliesTo(t *testing.T) {
v := SuppressionVariant{AvailabilityZone: "us-east-1a", Platform: "Linux/UNIX", Tenancy: "default"}
scoped := PurchaseSuppression{Variant: &v}
legacy := PurchaseSuppression{}

assert.True(t, scoped.AppliesTo(&SuppressionVariant{"us-east-1a", "Linux/UNIX", "default"}))
assert.False(t, scoped.AppliesTo(&SuppressionVariant{"us-east-1b", "Linux/UNIX", "default"}))
assert.False(t, scoped.AppliesTo(&SuppressionVariant{"", "Linux/UNIX", "default"}))
assert.False(t, scoped.AppliesTo(&SuppressionVariant{"us-east-1a", "Windows", "default"}))
assert.False(t, scoped.AppliesTo(&SuppressionVariant{"us-east-1a", "Linux/UNIX", "dedicated"}))
assert.True(t, scoped.AppliesTo(nil), "an unreadable rec variant matches every row")
assert.True(t, legacy.AppliesTo(&SuppressionVariant{"us-east-1b", "Windows", "dedicated"}), "a legacy row is a wildcard")
assert.True(t, legacy.AppliesTo(nil))
}
Loading
Loading