fix(aws): credit replacement RIs bought inside the coverage lookback in the EC2 expiry deduction

CHANGELOG.md

@@ -32,6 +32,17 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
 ### Added
 
+- **`recommendations.AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions`:**
+  the per-pool (EC2 and other non-family) expiry deduction, plus credit for
+  replacements already bought. Coverage is a lookback average, so RIs bought
+  inside `CoverageLookbackDays` are not in it yet; with R the count of active,
+  non-expiring RIs in the same pool that started inside the lookback, the
+  deduction is `max(0, expiring - R)`. It can only buy less, never more. A
+  growth purchase in the pool counts as a replacement, and a replacement of
+  a different size in the same family is not credited.
+  `CoverageLookbackDays <= 0` credits none. Additive; the existing functions
+  are unchanged (#346).
+
 - **`recommendations.Client.GetSPCommitmentRate`:** commitment dollars per
   on-demand dollar (r) of the used part of a new Savings Plan, derived from CE
   purchase recommendations as `(utilization/100) * EstimatedSPCost /

providers/aws/recommendations/expiry.go

@@ -25,61 +25,122 @@ func AdjustExistingCoverageForExpiringCommitments(
 			demand[lookupPoolKey(recs[i])] += avg
 		}
 	}
-	adjusted, _ := adjustExpiringCoverage(recs, commitments, windowDays, demand)
+	adjusted, _ := adjustExpiringCoverage(recs, commitments, PoolExpiryOptions{WindowDays: windowDays}, demand)
 	return adjusted
 }
 
 // AdjustExistingCoverageForExpiringCommitmentsWithCoverage uses authoritative pool demand, never row sums.
 // Eligible rows lacking positive finite demand remain unchanged and are counted in missingDemand.
+// It ignores replacements already bought; see
+// AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions.
 func AdjustExistingCoverageForExpiringCommitmentsWithCoverage(
 	recs []common.Recommendation,
 	commitments []common.Commitment,
 	windowDays int,
 	coverage PoolCoverageMap,
+) (adjusted, missingDemand int) {
+	return AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions(
+		recs, commitments, coverage, PoolExpiryOptions{WindowDays: windowDays})
+}
+
+// PoolExpiryOptions configures the per-pool (EC2 and other non-family) expiry
+// deduction. It mirrors FamilyExpiryOptions, which does the same for RDS.
+//
+// CoverageLookbackDays must be the lookbackDays given to the coverage fetch
+// (GetRICoverageMap/GetEC2RICoverageMap). Coverage is a lookback average, so an
+// RI bought inside that window is only partly in it. Active, non-expiring RIs
+// in the same region and instance type that started inside the window are
+// credited against the expiring count of that pool. A value <= 0 credits none
+// (the coverage fetch itself defaults a non-positive lookback to 30, so pass
+// 30 here in that case).
+//
+// The rule is conservative: it can under-buy, never over-buy. The credit is
+// taken in full although lagged coverage already holds part of the new RI. A
+// growth purchase in the same pool is indistinguishable from a replacement. A
+// replacement of a different size in the same family is not credited.
+type PoolExpiryOptions struct {
+	WindowDays           int
+	CoverageLookbackDays int
+}
+
+// AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions deducts RIs
+// expiring inside opts.WindowDays from the existing coverage of each
+// recommendation's pool, net of replacements bought inside
+// opts.CoverageLookbackDays: per pool the deduction is max(0, expiring -
+// replacements). A pool whose expiring RIs are fully replaced is left
+// untouched and is not counted as missingDemand.
+//
+// The rule is conservative: it can under-buy, never over-buy. A growth
+// purchase in the same pool is indistinguishable from a replacement, and a
+// replacement of a different size in the same family is not credited. See
+// PoolExpiryOptions for the lookback contract.
+func AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions(
+	recs []common.Recommendation,
+	commitments []common.Commitment,
+	coverage PoolCoverageMap,
+	opts PoolExpiryOptions,
 ) (adjusted, missingDemand int) {
 	demand := make(map[string]float64, len(coverage))
 	for key, cov := range coverage {
 		demand[key] = cov.AvgInstancesPerHour
 	}
-	return adjustExpiringCoverage(recs, commitments, windowDays, demand)
+	return adjustExpiringCoverage(recs, commitments, opts, demand)
 }
 
 func adjustExpiringCoverage(
 	recs []common.Recommendation,
 	commitments []common.Commitment,
-	windowDays int,
+	opts PoolExpiryOptions,
 	demand map[string]float64,
 ) (adjusted, missingDemand int) {
-	if windowDays <= 0 || len(commitments) == 0 {
+	if opts.WindowDays <= 0 || len(commitments) == 0 {
 		return 0, 0
 	}
-	cutoff := time.Now().Add(time.Duration(windowDays) * 24 * time.Hour)
-	expiringByPool := expiringCountsByPool(commitments, cutoff)
+	now := time.Now()
+	cutoff := now.Add(time.Duration(opts.WindowDays) * 24 * time.Hour)
+	lagStart := now.Add(-time.Duration(opts.CoverageLookbackDays) * 24 * time.Hour)
+	expiringByPool, replacedByPool := tallyPoolCommitments(commitments, cutoff, lagStart, opts.CoverageLookbackDays > 0)
 	if len(expiringByPool) == 0 {
 		return 0, 0
 	}
-	return applyExpiringAdjustments(recs, expiringByPool, demand)
+	return applyExpiringAdjustments(recs, netExpiringByPool(expiringByPool, replacedByPool), demand)
 }
 
-// expiringCountsByPool aggregates active commitments expiring at-or-before
-// cutoff into a pool-keyed count map. The State filter mirrors
-// DuplicateChecker.filterRecentCommitments — only RIs currently providing
-// coverage are counted (queued / retired RIs aren't covering demand now).
-func expiringCountsByPool(commitments []common.Commitment, cutoff time.Time) map[string]int {
-	out := make(map[string]int)
+// tallyPoolCommitments sums active commitments per pool: those expiring
+// at-or-before cutoff, and (when credit is set) the non-expiring ones that
+// started after lagStart, which lagged coverage does not fully reflect yet.
+// The State filter mirrors DuplicateChecker.filterRecentCommitments, so
+// queued / retired RIs, which are not covering demand now, are ignored.
+// The credit test mirrors tallyFamilyCommitments in family_nu.go:511-512.
+func tallyPoolCommitments(
+	commitments []common.Commitment, cutoff, lagStart time.Time, credit bool,
+) (expiring, replaced map[string]int) {
+	expiring = make(map[string]int)
+	replaced = make(map[string]int)
 	for i := range commitments {
 		c := &commitments[i]
 		if !commitmentIsActive(*c) {
 			continue
 		}
-		if c.EndDate.IsZero() || c.EndDate.After(cutoff) {
-			continue
-		}
 		key := commitmentPoolKey(*c)
 		if key == "" {
 			continue
 		}
-		out[key] += c.Count
+		if !c.EndDate.IsZero() && !c.EndDate.After(cutoff) {
+			expiring[key] += c.Count
+		} else if credit && c.StartDate.After(lagStart) {
+			replaced[key] += c.Count
+		}
+	}
+	return expiring, replaced
+}
+
+// netExpiringByPool returns max(0, expiring - replaced) for every pool with
+// expiring commitments; replacements never push coverage past the lagged value.
+func netExpiringByPool(expiring, replaced map[string]int) map[string]int {
+	out := make(map[string]int, len(expiring))
+	for key, e := range expiring {
+		out[key] = max(0, e-replaced[key])
 	}
 	return out
 }
@@ -90,8 +151,8 @@ func applyExpiringAdjustments(recs []common.Recommendation, expiringByPool map[s
 			continue
 		}
 		key := lookupPoolKey(recs[i])
-		expCount, ok := expiringByPool[key]
-		if !ok || expCount == 0 {
+		expCount := expiringByPool[key]
+		if expCount == 0 {
 			continue
 		}
 		avg := demand[key]

providers/aws/recommendations/expiry_lookback_test.go

@@ -0,0 +1,220 @@
+package recommendations
+
+import (
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/recfilter"
+)
+
+const lookbackDays = 30
+
+func daysFromNow(d int) time.Time { return time.Now().Add(time.Duration(d) * 24 * time.Hour) }
+
+func ec2Commit(count int, state common.CommitmentState, start, end time.Time) common.Commitment {
+	return common.Commitment{
+		Service: common.ServiceEC2, Region: "us-east-1", ResourceType: "m5.large",
+		Count: count, State: state, StartDate: start, EndDate: end,
+	}
+}
+
+func expiringRI(count int) common.Commitment {
+	return ec2Commit(count, "active", daysFromNow(-300), daysFromNow(15))
+}
+
+func replacementRI(count int, startedDaysAgo int) common.Commitment {
+	return ec2Commit(count, "active", daysFromNow(-startedDaysAgo), daysFromNow(300))
+}
+
+// runLookback runs the options API on a fresh expiryPrecisionFixture and
+// returns the recs before and after the adjustment, plus the sized result.
+func runLookback(t *testing.T, commits []common.Commitment, opts PoolExpiryOptions, mutate func(PoolCoverageMap)) (
+	before, after []common.Recommendation, adjusted, missing int, sized []common.Recommendation,
+) {
+	t.Helper()
+	recs, _, coverage := expiryPrecisionFixture()
+	if mutate != nil {
+		mutate(coverage)
+	}
+	ApplyCoverageMapToRecommendations(recs, coverage)
+	before = make([]common.Recommendation, len(recs))
+	copy(before, recs)
+	adjusted, missing = AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions(recs, commits, coverage, opts)
+	return before, recs, adjusted, missing, recfilter.ApplyTargetCoverage(recs, 80, nil, nil)
+}
+
+func defaultOpts() PoolExpiryOptions {
+	return PoolExpiryOptions{WindowDays: 30, CoverageLookbackDays: lookbackDays}
+}
+
+func counts(recs []common.Recommendation) []int {
+	out := make([]int, len(recs))
+	for i := range recs {
+		out[i] = recs[i].Count
+	}
+	return out
+}
+
+func TestPoolExpiryReplacementInsideLookback(t *testing.T) {
+	_, _, _, _, baseline := runLookback(t, nil, defaultOpts(), nil)
+
+	before, after, adjusted, missing, sized := runLookback(t,
+		[]common.Commitment{expiringRI(2), replacementRI(2, 5)}, defaultOpts(), nil)
+	assert.Zero(t, adjusted)
+	assert.Zero(t, missing)
+	for i := range after {
+		assert.Equal(t, before[i].ExistingCoveragePct, after[i].ExistingCoveragePct)
+	}
+	assert.Equal(t, counts(baseline), counts(sized), "no re-buy of what was already bought")
+
+	t.Run("lookback 0 still deducts", func(t *testing.T) {
+		opts := PoolExpiryOptions{WindowDays: 30}
+		before, after, adjusted, _, _ := runLookback(t,
+			[]common.Commitment{expiringRI(2), replacementRI(2, 5)}, opts, nil)
+		assert.Equal(t, 2, adjusted)
+		assert.Less(t, after[0].ExistingCoveragePct, before[0].ExistingCoveragePct)
+	})
+}
+
+func TestPoolExpiryReplacementBeforeLookback(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		ago  int
+		want int
+	}{{"40d ago deducts fully", 40, 2}, {"29d ago is credited", 29, 0}} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, _, adjusted, _, _ := runLookback(t,
+				[]common.Commitment{expiringRI(2), replacementRI(2, tc.ago)}, defaultOpts(), nil)
+			assert.Equal(t, tc.want, adjusted)
+		})
+	}
+}
+
+func TestPoolExpiryPartialAndOverReplacement(t *testing.T) {
+	// demand 30/h, coverage 60%: each expiring RI is 100/30 points.
+	before, after, adjusted, _, _ := runLookback(t,
+		[]common.Commitment{expiringRI(4), replacementRI(1, 5)}, defaultOpts(), nil)
+	assert.Equal(t, 2, adjusted)
+	assert.InDelta(t, before[0].ExistingCoveragePct-3*100.0/30, after[0].ExistingCoveragePct, 1e-9)
+
+	before, after, adjusted, _, _ = runLookback(t,
+		[]common.Commitment{expiringRI(2), replacementRI(5, 5)}, defaultOpts(), nil)
+	assert.Zero(t, adjusted)
+	assert.Equal(t, before[0].ExistingCoveragePct, after[0].ExistingCoveragePct, "never above lagged coverage")
+}
+
+func TestPoolExpiryReplacementStates(t *testing.T) {
+	start, end := daysFromNow(-5), daysFromNow(300)
+	for _, tc := range []struct {
+		name     string
+		c        common.Commitment
+		credited bool
+	}{
+		{"retired", ec2Commit(2, "retired", start, end), false},
+		{"expired", ec2Commit(2, "expired", start, end), false},
+		{"queued", ec2Commit(2, "queued", start, end), false},
+		{"pending-return", ec2Commit(2, "pending-return", start, end), false},
+		{"also expiring", ec2Commit(2, "active", start, daysFromNow(20)), false},
+		{"payment-pending", ec2Commit(2, "payment-pending", start, end), true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, _, adjusted, _, _ := runLookback(t, []common.Commitment{expiringRI(2), tc.c}, defaultOpts(), nil)
+			if tc.credited {
+				assert.Zero(t, adjusted)
+			} else {
+				assert.Equal(t, 2, adjusted)
+			}
+		})
+	}
+}
+
+func TestPoolExpiryReplacementPoolMatching(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		mod      func(*common.Commitment)
+		credited bool
+	}{
+		{"region mismatch", func(c *common.Commitment) { c.Region = "us-west-2" }, false},
+		{"type mismatch", func(c *common.Commitment) { c.ResourceType = "m5.xlarge" }, false},
+		{"service mismatch", func(c *common.Commitment) { c.Service = common.ServiceRDS }, false},
+		{"account ignored", func(c *common.Commitment) { c.Account = "other-linked-account" }, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			repl := replacementRI(2, 5)
+			tc.mod(&repl)
+			_, _, adjusted, _, _ := runLookback(t, []common.Commitment{expiringRI(2), repl}, defaultOpts(), nil)
+			if tc.credited {
+				assert.Zero(t, adjusted)
+			} else {
+				assert.Equal(t, 2, adjusted)
+			}
+		})
+	}
+}
+
+func TestPoolExpiryNoExpiringCommitments(t *testing.T) {
+	before, after, adjusted, missing, _ := runLookback(t,
+		[]common.Commitment{replacementRI(3, 5)}, defaultOpts(), nil)
+	assert.Zero(t, adjusted)
+	assert.Zero(t, missing)
+	assert.Equal(t, before, after)
+}
+
+func TestPoolExpiryMissingDemand(t *testing.T) {
+	noDemand := func(cov PoolCoverageMap) {
+		for k, v := range cov {
+			v.AvgInstancesPerHour = 0
+			cov[k] = v
+		}
+	}
+	_, _, adjusted, missing, _ := runLookback(t,
+		[]common.Commitment{expiringRI(2), replacementRI(2, 5)}, defaultOpts(), noDemand)
+	assert.Zero(t, adjusted)
+	assert.Zero(t, missing, "fully replaced: nothing to deduct")
+
+	_, _, adjusted, missing, _ = runLookback(t,
+		[]common.Commitment{expiringRI(3), replacementRI(1, 5)}, defaultOpts(), noDemand)
+	assert.Zero(t, adjusted)
+	assert.Equal(t, 2, missing)
+}
+
+func TestPoolExpirySkippedRecsKeepExactState(t *testing.T) {
+	recs, _, coverage := expiryPrecisionFixture()
+	ApplyCoverageMapToRecommendations(recs, coverage)
+	seeded := make([]*big.Rat, len(recs))
+	for i := range recs {
+		seeded[i] = big.NewRat(1, 2)
+		recs[i].ExistingCoveragePercentExact = seeded[i]
+	}
+	adjusted, _ := AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions(recs,
+		[]common.Commitment{expiringRI(2), replacementRI(2, 5)}, coverage, defaultOpts())
+	require.Zero(t, adjusted)
+	for i := range recs {
+		assert.Same(t, seeded[i], recs[i].ExistingCoveragePercentExact)
+		assert.Zero(t, big.NewRat(1, 2).Cmp(recs[i].ExistingCoveragePercentExact))
+	}
+}
+
+func TestPoolExpiryZeroOptionsMatchLegacyCoverageAPI(t *testing.T) {
+	commits := []common.Commitment{expiringRI(2), replacementRI(2, 5)}
+	recsA, _, coverage := expiryPrecisionFixture()
+	recsB, _, _ := expiryPrecisionFixture()
+	ApplyCoverageMapToRecommendations(recsA, coverage)
+	ApplyCoverageMapToRecommendations(recsB, coverage)
+	a1, m1 := AdjustExistingCoverageForExpiringCommitmentsWithCoverage(recsA, commits, 30, coverage)
+	a2, m2 := AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions(recsB, commits, coverage,
+		PoolExpiryOptions{WindowDays: 30})
+	assert.Equal(t, a1, a2)
+	assert.Equal(t, m1, m2)
+	assert.Equal(t, recsA, recsB)
+
+	a, m := AdjustExistingCoverageForExpiringCommitmentsWithCoverageOptions(recsB, commits, coverage,
+		PoolExpiryOptions{CoverageLookbackDays: lookbackDays})
+	assert.Zero(t, a)
+	assert.Zero(t, m)
+}