diff --git a/test/pkg/environment/common/karpenter_metrics_poller.go b/test/pkg/environment/common/karpenter_metrics_poller.go index 8c268bb43d..8d4ebc25a7 100644 --- a/test/pkg/environment/common/karpenter_metrics_poller.go +++ b/test/pkg/environment/common/karpenter_metrics_poller.go @@ -17,7 +17,6 @@ limitations under the License. package common import ( - "bytes" "context" "errors" "fmt" @@ -27,8 +26,6 @@ import ( "github.com/montanaflynn/stats" . "github.com/onsi/ginkgo/v2" dto "github.com/prometheus/client_model/go" - "github.com/prometheus/common/expfmt" - "github.com/prometheus/common/model" ) type ResourceSample struct { @@ -211,20 +208,12 @@ func (mp *KarpenterMetricsPoller) recordSample(state *pollerState, now time.Time // scrapeMetrics uses the API server pod proxy to fetch /metrics from the Karpenter pod. func (mp *KarpenterMetricsPoller) scrapeMetrics(ctx context.Context, podName string) (memBytes float64, cpuSeconds float64, err error) { - data, err := mp.env.KubeClient.CoreV1().Pods("kube-system").ProxyGet("http", podName, "8080", "/metrics", nil).DoRaw(ctx) + families, err := scrapeKarpenterMetricFamilies(ctx, mp.env, podName) if err != nil { - return 0, 0, fmt.Errorf("proxy GET /metrics: %w", err) + return 0, 0, err } - - parser := expfmt.NewTextParser(model.UTF8Validation) - families, err := parser.TextToMetricFamilies(bytes.NewReader(data)) - if err != nil { - return 0, 0, fmt.Errorf("parsing metrics: %w", err) - } - memBytes = getGaugeValue(families, "process_resident_memory_bytes") cpuSeconds = getCounterValue(families, "process_cpu_seconds_total") - if memBytes == 0 && cpuSeconds == 0 { return 0, 0, &metricsNotFoundError{foundMem: false, foundCPU: false} } diff --git a/test/pkg/environment/common/latency_harness.go b/test/pkg/environment/common/latency_harness.go new file mode 100644 index 0000000000..1776e0daba --- /dev/null +++ b/test/pkg/environment/common/latency_harness.go @@ -0,0 +1,382 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "bytes" + "context" + "fmt" + "sort" + "strings" + + . "github.com/onsi/ginkgo/v2" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" +) + +// HistogramStats is the derived percentile summary of one labeled histogram +// series over the observations added between LatencyHarness.Start and +// LatencyHarness.Stop. +type HistogramStats struct { + MetricName string `json:"metric_name"` + Labels map[string]string `json:"labels,omitempty"` + Count uint64 `json:"count"` + Sum float64 `json:"sum"` + Mean float64 `json:"mean"` + P50 float64 `json:"p50"` + P90 float64 `json:"p90"` + P95 float64 `json:"p95"` + P99 float64 `json:"p99"` + Max float64 `json:"max"` + BucketTruncationRate float64 `json:"bucket_truncation_rate"` +} + +// TargetHistograms is the Karpenter histogram set the harness scrapes. +var TargetHistograms = []string{ + "karpenter_pods_scheduling_decision_duration_seconds", + "karpenter_pods_bound_duration_seconds", + "karpenter_pods_provisioning_bound_duration_seconds", + "karpenter_pods_provisioning_startup_duration_seconds", + "karpenter_scheduler_scheduling_duration_seconds", + "karpenter_voluntary_disruption_decision_evaluation_duration_seconds", + "karpenter_cloudprovider_duration_seconds", + "karpenter_nodeclaims_instance_termination_duration_seconds", + "karpenter_nodeclaims_termination_duration_seconds", + "karpenter_consolidation_score", +} + +// TargetCounters is the Karpenter counter set the harness reports as deltas +// between Start and Stop, keyed the same way as LatencyStats. +var TargetCounters = []string{ + "karpenter_voluntary_disruption_consolidation_timeouts_total", + "karpenter_consolidation_moves_total", + "karpenter_nodeclaims_created_total", + "karpenter_nodes_created_total", +} + +// LatencyResult is what LatencyHarness.Stop returns. Keys of LatencyStats and +// Counters are series fingerprints (see seriesKey). Process-level memory and +// CPU are covered by KarpenterMetricsPoller; run both harnesses in tandem if +// resource-usage stats are needed. +type LatencyResult struct { + LatencyStats map[string]HistogramStats + Counters map[string]uint64 +} + +// LatencyHarness captures a start-of-phase snapshot of Karpenter's /metrics +// endpoint and produces per-histogram percentile summaries by bucket-count +// delta at Stop. It reuses the pod-proxy scrape pattern from +// KarpenterMetricsPoller. +type LatencyHarness struct { + env *Environment + podName string + start map[string]*dto.MetricFamily +} + +// StartLatencyHarness discovers the active Karpenter pod, scrapes /metrics +// once, and stores a compacted snapshot (target series only) for later delta +// reduction. Symmetric with StartKarpenterMetricsPoller. +func StartLatencyHarness(env *Environment) (*LatencyHarness, error) { + pod, err := env.FindActiveKarpenterPod(env.Context) + if err != nil || pod == nil { + return nil, fmt.Errorf("finding karpenter pod: %w", err) + } + h := &LatencyHarness{env: env, podName: pod.Name} + families, err := scrapeKarpenterMetricFamilies(env.Context, env, pod.Name) + if err != nil { + return nil, fmt.Errorf("initial scrape: %w", err) + } + h.start = compactFamilies(families) + GinkgoWriter.Printf("LatencyHarness: started, scraping pod kube-system/%s\n", pod.Name) + return h, nil +} + +// Stop scrapes the end snapshot and reduces the histogram / counter deltas +// into a LatencyResult. On the first scrape failure the harness refreshes the +// active-pod name once (matches KarpenterMetricsPoller's leader-election +// handling) and retries; a second failure returns the error. +func (h *LatencyHarness) Stop() (*LatencyResult, error) { + ctx := h.env.Context + end, err := scrapeKarpenterMetricFamilies(ctx, h.env, h.podName) + if err != nil { + if pod, findErr := h.env.FindActiveKarpenterPod(ctx); findErr == nil && pod != nil && pod.Name != h.podName { + GinkgoWriter.Printf("LatencyHarness: active pod changed from %s to %s, retrying scrape\n", h.podName, pod.Name) + h.podName = pod.Name + end, err = scrapeKarpenterMetricFamilies(ctx, h.env, pod.Name) + } + if err != nil { + return nil, fmt.Errorf("end scrape: %w", err) + } + } + res := &LatencyResult{ + LatencyStats: map[string]HistogramStats{}, + Counters: map[string]uint64{}, + } + for _, name := range TargetHistograms { + for key, stats := range deltaHistogram(name, h.start[name], end[name]) { + res.LatencyStats[key] = stats + } + } + for _, name := range TargetCounters { + for key, delta := range deltaCounter(name, h.start[name], end[name]) { + res.Counters[key] = delta + } + } + GinkgoWriter.Printf("LatencyHarness: stopped, %d histogram series, %d counter series\n", + len(res.LatencyStats), len(res.Counters)) + return res, nil +} + +// scrapeKarpenterMetricFamilies fetches and parses /metrics from a Karpenter +// pod via the API-server pod proxy. Shared between LatencyHarness and +// KarpenterMetricsPoller. +func scrapeKarpenterMetricFamilies(ctx context.Context, env *Environment, podName string) (map[string]*dto.MetricFamily, error) { + data, err := env.KubeClient.CoreV1().Pods("kube-system").ProxyGet("http", podName, "8080", "/metrics", nil).DoRaw(ctx) + if err != nil { + return nil, fmt.Errorf("proxy GET /metrics: %w", err) + } + parser := expfmt.NewTextParser(model.UTF8Validation) + families, err := parser.TextToMetricFamilies(bytes.NewReader(data)) + if err != nil { + return nil, fmt.Errorf("parsing metrics: %w", err) + } + return families, nil +} + +// compactFamilies retains only the metric families the harness reduces at +// Stop, plus process gauges the poller reads. The full Karpenter /metrics +// response contains hundreds of families; retaining only the target set +// keeps memory bounded across long test phases. +func compactFamilies(families map[string]*dto.MetricFamily) map[string]*dto.MetricFamily { + keep := make(map[string]*dto.MetricFamily, len(TargetHistograms)+len(TargetCounters)) + for _, n := range TargetHistograms { + if f, ok := families[n]; ok { + keep[n] = f + } + } + for _, n := range TargetCounters { + if f, ok := families[n]; ok { + keep[n] = f + } + } + return keep +} + +// seriesKey returns the canonical fingerprint for a labeled sample: +// "metric_name" or "metric_name{k=v,k2=v2,...}" with keys sorted lexically. +func seriesKey(name string, labels []*dto.LabelPair) string { + if len(labels) == 0 { + return name + } + pairs := make([]string, 0, len(labels)) + for _, l := range labels { + pairs = append(pairs, l.GetName()+"="+l.GetValue()) + } + sort.Strings(pairs) + return name + "{" + strings.Join(pairs, ",") + "}" +} + +// labelMap returns the labels of a Metric as a plain map for HistogramStats. +func labelMap(labels []*dto.LabelPair) map[string]string { + if len(labels) == 0 { + return nil + } + out := make(map[string]string, len(labels)) + for _, l := range labels { + out[l.GetName()] = l.GetValue() + } + return out +} + +// deltaHistogram computes per-series stats from the count delta between two +// snapshots of the same MetricFamily. Nil start or end families are treated +// as empty. Series present only at end are emitted with their end histogram +// as the whole delta. +func deltaHistogram(name string, start, end *dto.MetricFamily) map[string]HistogramStats { + out := map[string]HistogramStats{} + if end == nil { + return out + } + startBySeries := indexBySeries(name, start) + for _, m := range end.GetMetric() { + if m.GetHistogram() == nil { + continue + } + key := seriesKey(name, m.GetLabel()) + s := reduceHistogramDelta(m.GetHistogram(), startBySeries[key].GetHistogram()) + s.MetricName = name + s.Labels = labelMap(m.GetLabel()) + out[key] = s + } + return out +} + +// deltaCounter computes counter-value deltas between two snapshots. Nil start +// yields the raw end value; a counter reset (end < start) yields end (the new +// baseline is treated as fresh observation). +func deltaCounter(name string, start, end *dto.MetricFamily) map[string]uint64 { + out := map[string]uint64{} + if end == nil { + return out + } + startBySeries := indexBySeries(name, start) + for _, m := range end.GetMetric() { + if m.GetCounter() == nil { + continue + } + key := seriesKey(name, m.GetLabel()) + endV := m.GetCounter().GetValue() + startV := 0.0 + if prev, ok := startBySeries[key]; ok && prev.GetCounter() != nil { + startV = prev.GetCounter().GetValue() + } + delta := endV - startV + if delta < 0 { + // Counter reset (pod restart); take end as-is. endV is a Prometheus + // counter value and cannot be negative. + delta = endV + } + out[key] = uint64(delta) + } + return out +} + +// indexBySeries returns metrics from mf keyed by seriesKey. +func indexBySeries(name string, mf *dto.MetricFamily) map[string]*dto.Metric { + out := map[string]*dto.Metric{} + if mf == nil { + return out + } + for _, m := range mf.GetMetric() { + out[seriesKey(name, m.GetLabel())] = m + } + return out +} + +// reduceHistogramDelta subtracts the start histogram from end (bucket-wise +// and on sample_count / sample_sum) and derives percentile stats over the +// resulting bucket distribution. Both histograms MUST share the same bucket +// layout; deltas for missing start-buckets treat startCumulative as 0. +func reduceHistogramDelta(end *dto.Histogram, startHistogram *dto.Histogram) HistogramStats { + if end == nil { + return HistogramStats{} + } + endCount := end.GetSampleCount() + endSum := end.GetSampleSum() + startCount, startSum, startCumBy := resolveDeltaBaseline(startHistogram, endCount) + deltaCount := endCount - startCount + if deltaCount == 0 { + return HistogramStats{Count: 0, Sum: endSum - startSum} + } + endBuckets := end.GetBucket() + deltaCum := make([]uint64, len(endBuckets)) + for i, b := range endBuckets { + endCum := b.GetCumulativeCount() + startCum := startCumBy[b.GetUpperBound()] + if endCum < startCum { + startCum = 0 + } + deltaCum[i] = endCum - startCum + } + // The +Inf bucket count is deltaCount (per Prometheus contract). Truncation + // rate is what escaped the finite tail. + trunc := 0.0 + if len(deltaCum) > 0 && deltaCount > deltaCum[len(deltaCum)-1] { + trunc = float64(deltaCount-deltaCum[len(deltaCum)-1]) / float64(deltaCount) + } + deltaSum := endSum - startSum + if deltaSum < 0 { + deltaSum = endSum + } + return HistogramStats{ + Count: deltaCount, + Sum: deltaSum, + Mean: deltaSum / float64(deltaCount), + P50: interpolatePercentile(endBuckets, deltaCum, deltaCount, 0.50), + P90: interpolatePercentile(endBuckets, deltaCum, deltaCount, 0.90), + P95: interpolatePercentile(endBuckets, deltaCum, deltaCount, 0.95), + P99: interpolatePercentile(endBuckets, deltaCum, deltaCount, 0.99), + Max: inferMaxBound(endBuckets, deltaCum), + BucketTruncationRate: trunc, + } +} + +// resolveDeltaBaseline returns the baseline sample count, sum, and cumulative +// bucket counts (keyed by upper bound) that the delta reduction subtracts +// from end. A nil startHistogram or a counter-reset (endCount < startCount, +// typically from a pod restart) yields a zero baseline so end is treated as +// the whole delta. +func resolveDeltaBaseline(startHistogram *dto.Histogram, endCount uint64) (uint64, float64, map[float64]uint64) { + if startHistogram == nil { + return 0, 0, nil + } + startCount := startHistogram.GetSampleCount() + if endCount < startCount { + return 0, 0, nil + } + buckets := startHistogram.GetBucket() + cumBy := make(map[float64]uint64, len(buckets)) + for _, b := range buckets { + cumBy[b.GetUpperBound()] = b.GetCumulativeCount() + } + return startCount, startHistogram.GetSampleSum(), cumBy +} + +// inferMaxBound returns the tightest finite bucket upper bound that saw a +// non-zero delta. When every delta observation fell beyond the last finite +// bucket (the +Inf bucket), the last finite bound is returned; callers pair +// this with BucketTruncationRate to detect that condition. +func inferMaxBound(endBuckets []*dto.Bucket, deltaCum []uint64) float64 { + if len(endBuckets) == 0 { + return 0 + } + for i := len(deltaCum) - 1; i >= 0; i-- { + if deltaCum[i] > 0 { + return endBuckets[i].GetUpperBound() + } + } + return endBuckets[len(endBuckets)-1].GetUpperBound() +} + +// interpolatePercentile returns the linearly-interpolated percentile from a +// cumulative delta distribution. Follows Prometheus' histogram_quantile +// convention: uniform-within-bucket, linear from the previous upper bound to +// the current upper bound. Percentiles landing beyond the last finite bucket +// return the last finite bound (a coarse under-estimate under truncation). +func interpolatePercentile(buckets []*dto.Bucket, cum []uint64, total uint64, q float64) float64 { + if total == 0 || len(buckets) == 0 { + return 0 + } + target := q * float64(total) + prevCum := uint64(0) + prevUpper := 0.0 + for i, b := range buckets { + c := cum[i] + if float64(c) >= target { + upper := b.GetUpperBound() + bucketDelta := c - prevCum + if bucketDelta == 0 { + return upper + } + return prevUpper + (upper-prevUpper)*(target-float64(prevCum))/float64(bucketDelta) + } + prevCum = c + prevUpper = b.GetUpperBound() + } + return prevUpper +} diff --git a/test/pkg/environment/common/latency_harness_test.go b/test/pkg/environment/common/latency_harness_test.go new file mode 100644 index 0000000000..cae0bfa964 --- /dev/null +++ b/test/pkg/environment/common/latency_harness_test.go @@ -0,0 +1,297 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "math" + "testing" + + dto "github.com/prometheus/client_model/go" +) + +// mkBucket returns a *dto.Bucket with the given upper bound and cumulative count. +func mkBucket(upper float64, cum uint64) *dto.Bucket { + return &dto.Bucket{UpperBound: &upper, CumulativeCount: &cum} +} + +// mkHistogram returns a *dto.Histogram with the given cumulative buckets and +// total sample_count / sample_sum. The buckets slice MUST be sorted by +// upper bound ascending; cum is cumulative (Prometheus convention). +func mkHistogram(count uint64, sum float64, buckets []*dto.Bucket) *dto.Histogram { + return &dto.Histogram{SampleCount: &count, SampleSum: &sum, Bucket: buckets} +} + +// mkMetric wraps a histogram into a labeled dto.Metric. +func mkMetric(h *dto.Histogram, labels map[string]string) *dto.Metric { + m := &dto.Metric{Histogram: h} + for k, v := range labels { + name, val := k, v + m.Label = append(m.Label, &dto.LabelPair{Name: &name, Value: &val}) + } + return m +} + +// mkFamily wraps a set of Metric into a MetricFamily of the given type. +func mkFamily(name string, mtype dto.MetricType, metrics ...*dto.Metric) *dto.MetricFamily { + n, t := name, mtype + return &dto.MetricFamily{Name: &n, Type: &t, Metric: metrics} +} + +// mkCounterMetric wraps a counter value into a labeled dto.Metric. +func mkCounterMetric(v float64, labels map[string]string) *dto.Metric { + m := &dto.Metric{Counter: &dto.Counter{Value: &v}} + for k, val := range labels { + name, value := k, val + m.Label = append(m.Label, &dto.LabelPair{Name: &name, Value: &value}) + } + return m +} + +// Test 1. Uniform bucket layout, single-series, easy percentiles. +// 100 observations delta split evenly across buckets [0.1, 0.5, 1.0, 2.0]. +// P50 = 0.5 (median lands at bucket 2 upper edge), P90 = 1.6 (interpolated). +func TestReduceHistogramDelta_UniformDistribution(t *testing.T) { + end := mkHistogram(100, 30.0, []*dto.Bucket{ + mkBucket(0.1, 25), + mkBucket(0.5, 50), + mkBucket(1.0, 75), + mkBucket(2.0, 100), + }) + stats := reduceHistogramDelta(end, nil) + if stats.Count != 100 { + t.Errorf("Count: got %d, want 100", stats.Count) + } + if math.Abs(stats.Sum-30.0) > 1e-9 { + t.Errorf("Sum: got %v, want 30.0", stats.Sum) + } + if math.Abs(stats.Mean-0.3) > 1e-9 { + t.Errorf("Mean: got %v, want 0.3", stats.Mean) + } + if math.Abs(stats.P50-0.5) > 1e-9 { + t.Errorf("P50: got %v, want 0.5", stats.P50) + } + // P90 target = 90. prev bucket cum=75, cur=100, prev upper=1.0, cur upper=2.0. + // P90 = 1.0 + (2.0 - 1.0) * (90 - 75) / (100 - 75) = 1.0 + 0.6 = 1.6. + if math.Abs(stats.P90-1.6) > 1e-9 { + t.Errorf("P90: got %v, want 1.6", stats.P90) + } + if stats.BucketTruncationRate != 0 { + t.Errorf("BucketTruncationRate: got %v, want 0", stats.BucketTruncationRate) + } + if math.Abs(stats.Max-2.0) > 1e-9 { + t.Errorf("Max: got %v, want 2.0", stats.Max) + } +} + +// Test 2. Delta reduction subtracts start-of-phase observations correctly. +// Start snapshot has 50 total; end has 150. Delta is 100 with the tail newly +// filled; P50 should reflect only the new observations, not the start pool. +func TestReduceHistogramDelta_SubtractsStartSnapshot(t *testing.T) { + start := mkHistogram(50, 5.0, []*dto.Bucket{ + mkBucket(0.1, 50), + mkBucket(0.5, 50), + mkBucket(1.0, 50), + mkBucket(2.0, 50), + }) + end := mkHistogram(150, 55.0, []*dto.Bucket{ + mkBucket(0.1, 50), // no new observations in this bucket + mkBucket(0.5, 75), + mkBucket(1.0, 100), + mkBucket(2.0, 150), + }) + stats := reduceHistogramDelta(end, start) + if stats.Count != 100 { + t.Errorf("Count: got %d, want 100", stats.Count) + } + if math.Abs(stats.Sum-50.0) > 1e-9 { + t.Errorf("Sum: got %v, want 50.0", stats.Sum) + } + // Delta cumulative buckets: 0, 25, 50, 100. + // P50 target = 50. cum=50 at upper=1.0. Return 1.0 exactly. + if math.Abs(stats.P50-1.0) > 1e-9 { + t.Errorf("P50: got %v, want 1.0", stats.P50) + } + // P90 target = 90. prev cum=50 (upper=1.0), cur cum=100 (upper=2.0). + // P90 = 1.0 + (2.0-1.0) * (90-50)/(100-50) = 1.0 + 0.8 = 1.8. + if math.Abs(stats.P90-1.8) > 1e-9 { + t.Errorf("P90: got %v, want 1.8", stats.P90) + } +} + +// Test 3. Truncation-rate reports observations that fell into +Inf. +// End buckets total 90 within the finite tail while sample_count is 100; +// 10 observations exceeded the top bucket. Truncation rate = 0.10. +func TestReduceHistogramDelta_BucketTruncation(t *testing.T) { + end := mkHistogram(100, 500.0, []*dto.Bucket{ + mkBucket(1.0, 40), + mkBucket(5.0, 70), + mkBucket(10.0, 90), + }) + stats := reduceHistogramDelta(end, nil) + if stats.Count != 100 { + t.Errorf("Count: got %d, want 100", stats.Count) + } + if math.Abs(stats.BucketTruncationRate-0.10) > 1e-9 { + t.Errorf("BucketTruncationRate: got %v, want 0.10", stats.BucketTruncationRate) + } + // P95 target = 95. prev cum=90 (upper=10.0), no next finite bucket -> +Inf. + // Falls back to last finite upper bound. + if math.Abs(stats.P95-10.0) > 1e-9 { + t.Errorf("P95 under truncation: got %v, want 10.0", stats.P95) + } +} + +// Test 4. Zero-observation phase yields zero-valued stats. +func TestReduceHistogramDelta_NoNewObservations(t *testing.T) { + same := mkHistogram(50, 5.0, []*dto.Bucket{ + mkBucket(0.1, 25), + mkBucket(1.0, 50), + }) + stats := reduceHistogramDelta(same, same) + if stats.Count != 0 { + t.Errorf("Count: got %d, want 0", stats.Count) + } + if stats.P50 != 0 || stats.P90 != 0 || stats.P95 != 0 || stats.P99 != 0 { + t.Errorf("percentiles under zero-count: want all zero, got P50=%v P90=%v P95=%v P99=%v", + stats.P50, stats.P90, stats.P95, stats.P99) + } +} + +// Test 5. Counter-reset (pod restart) between snapshots. end_count < start_count +// should fall back to end as fresh observations. +func TestReduceHistogramDelta_CounterReset(t *testing.T) { + start := mkHistogram(200, 50.0, []*dto.Bucket{ + mkBucket(1.0, 200), + mkBucket(5.0, 200), + }) + // Pod restarted; new counter is smaller than the pre-restart baseline. + end := mkHistogram(30, 3.0, []*dto.Bucket{ + mkBucket(1.0, 20), + mkBucket(5.0, 30), + }) + stats := reduceHistogramDelta(end, start) + if stats.Count != 30 { + t.Errorf("Count under reset: got %d, want 30", stats.Count) + } + if math.Abs(stats.Sum-3.0) > 1e-9 { + t.Errorf("Sum under reset: got %v, want 3.0", stats.Sum) + } +} + +// Test 6. Multi-series histogram: same metric name, different label sets. +// deltaHistogram should emit one HistogramStats per (name, label-fingerprint). +func TestDeltaHistogram_MultiSeries(t *testing.T) { + name := "karpenter_voluntary_disruption_decision_evaluation_duration_seconds" + single := mkMetric(mkHistogram(10, 1.0, []*dto.Bucket{ + mkBucket(0.1, 10), + }), map[string]string{"consolidation_type": "single", "reason": "underutilized"}) + multi := mkMetric(mkHistogram(5, 2.5, []*dto.Bucket{ + mkBucket(0.1, 2), + mkBucket(1.0, 5), + }), map[string]string{"consolidation_type": "multi", "reason": "underutilized"}) + end := mkFamily(name, dto.MetricType_HISTOGRAM, single, multi) + out := deltaHistogram(name, nil, end) + if len(out) != 2 { + t.Fatalf("series count: got %d, want 2 (%v)", len(out), out) + } + singleKey := name + "{consolidation_type=single,reason=underutilized}" + multiKey := name + "{consolidation_type=multi,reason=underutilized}" + if _, ok := out[singleKey]; !ok { + t.Errorf("missing series key %q; got %v", singleKey, out) + } + if _, ok := out[multiKey]; !ok { + t.Errorf("missing series key %q; got %v", multiKey, out) + } + if out[singleKey].Count != 10 { + t.Errorf("single count: got %d, want 10", out[singleKey].Count) + } + if out[multiKey].Count != 5 { + t.Errorf("multi count: got %d, want 5", out[multiKey].Count) + } + if lbl := out[singleKey].Labels["consolidation_type"]; lbl != "single" { + t.Errorf("single labels.consolidation_type: got %q, want %q", lbl, "single") + } +} + +// Test 7. seriesKey is deterministic under label reordering. +func TestSeriesKey_StableSort(t *testing.T) { + name := "karpenter_consolidation_score" + a, av := "decision", "approved" + b, bv := "nodepool", "pool-a" + c, cv := "policy", "Balanced" + forward := []*dto.LabelPair{{Name: &a, Value: &av}, {Name: &b, Value: &bv}, {Name: &c, Value: &cv}} + reverse := []*dto.LabelPair{{Name: &c, Value: &cv}, {Name: &b, Value: &bv}, {Name: &a, Value: &av}} + if seriesKey(name, forward) != seriesKey(name, reverse) { + t.Errorf("seriesKey not stable under reorder: %q vs %q", seriesKey(name, forward), seriesKey(name, reverse)) + } + want := name + "{decision=approved,nodepool=pool-a,policy=Balanced}" + if got := seriesKey(name, forward); got != want { + t.Errorf("seriesKey format: got %q, want %q", got, want) + } +} + +// Test 8. Counter delta subtracts start value; reset falls back to end. +func TestDeltaCounter_DeltaAndReset(t *testing.T) { + name := "karpenter_voluntary_disruption_consolidation_timeouts_total" + lbl := map[string]string{"consolidation_type": "single"} + start := mkFamily(name, dto.MetricType_COUNTER, mkCounterMetric(3, lbl)) + end := mkFamily(name, dto.MetricType_COUNTER, mkCounterMetric(8, lbl)) + out := deltaCounter(name, start, end) + key := name + "{consolidation_type=single}" + if out[key] != 5 { + t.Errorf("counter delta: got %d, want 5", out[key]) + } + // Reset case: end < start -> take end as the delta. + resetV := 2.0 + end.Metric[0].Counter.Value = &resetV + out = deltaCounter(name, start, end) + if out[key] != 2 { + t.Errorf("counter reset: got %d, want 2", out[key]) + } +} + +// Test 9. deltaHistogram tolerates a missing metric family from either side. +func TestDeltaHistogram_MissingMetric(t *testing.T) { + out := deltaHistogram("karpenter_missing_metric", nil, nil) + if len(out) != 0 { + t.Errorf("missing metric: got %d series, want 0", len(out)) + } +} + +// Test 10. compactFamilies keeps only the target metric families. +func TestCompactFamilies(t *testing.T) { + families := map[string]*dto.MetricFamily{ + "karpenter_pods_scheduling_decision_duration_seconds": mkFamily("karpenter_pods_scheduling_decision_duration_seconds", dto.MetricType_HISTOGRAM), + "karpenter_voluntary_disruption_consolidation_timeouts_total": mkFamily("karpenter_voluntary_disruption_consolidation_timeouts_total", dto.MetricType_COUNTER), + "go_gc_duration_seconds": mkFamily("go_gc_duration_seconds", dto.MetricType_SUMMARY), + "process_open_fds": mkFamily("process_open_fds", dto.MetricType_GAUGE), + "workqueue_adds_total": mkFamily("workqueue_adds_total", dto.MetricType_COUNTER), + } + out := compactFamilies(families) + if _, ok := out["karpenter_pods_scheduling_decision_duration_seconds"]; !ok { + t.Errorf("compact dropped a target histogram") + } + if _, ok := out["karpenter_voluntary_disruption_consolidation_timeouts_total"]; !ok { + t.Errorf("compact dropped a target counter") + } + if _, ok := out["go_gc_duration_seconds"]; ok { + t.Errorf("compact retained a non-target family") + } + if len(out) != 2 { + t.Errorf("compact size: got %d, want 2", len(out)) + } +} diff --git a/test/pkg/environment/common/latency_sidecar.go b/test/pkg/environment/common/latency_sidecar.go new file mode 100644 index 0000000000..49123b7ad2 --- /dev/null +++ b/test/pkg/environment/common/latency_sidecar.go @@ -0,0 +1,64 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package common + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// LatencySidecar is the JSON schema written alongside a PerformanceReport +// when the LatencyHarness observed histogram or counter deltas. Offline +// analysis tools unmarshal into this type to pair a PerformanceReport with +// its latency companion; keep it stable across performance-suite tests so +// prior artifacts continue to load. +type LatencySidecar struct { + TestName string `json:"test_name"` + ConsolidationPolicy string `json:"consolidation_policy"` + Timestamp time.Time `json:"timestamp"` + LatencyStats map[string]HistogramStats `json:"latency_stats,omitempty"` + Counters map[string]uint64 `json:"counters,omitempty"` +} + +// WriteLatencySidecar writes sc to /_latency.json. Returns +// nil (no-op) when dir is empty, matching the report.go artifact posture so +// suites can run without OUTPUT_DIR configured. filePrefix is sanitized via +// filepath.Base + filepath.Clean and the resolved path is checked to stay +// under dir before writing. +func WriteLatencySidecar(dir, filePrefix string, sc LatencySidecar) error { + if dir == "" { + return nil + } + data, err := json.MarshalIndent(sc, "", " ") + if err != nil { + return fmt.Errorf("marshal latency sidecar: %w", err) + } + safeDir := filepath.Clean(dir) + safePrefix := filepath.Base(filepath.Clean(filePrefix)) + path := filepath.Join(safeDir, fmt.Sprintf("%s_latency.json", safePrefix)) + if rel, relErr := filepath.Rel(safeDir, path); relErr != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + return fmt.Errorf("latency sidecar path escapes %q", safeDir) + } + if err := os.WriteFile(path, data, 0600); err != nil { + return fmt.Errorf("write latency sidecar %s: %w", path, err) + } + return nil +} diff --git a/test/suites/performance/balanced_baseline_marginal_test.go b/test/suites/performance/balanced_baseline_marginal_test.go new file mode 100644 index 0000000000..987dffa023 --- /dev/null +++ b/test/suites/performance/balanced_baseline_marginal_test.go @@ -0,0 +1,215 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package performance + +import ( + "fmt" + "os" + "strconv" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + + v1 "sigs.k8s.io/karpenter/pkg/apis/v1" + "sigs.k8s.io/karpenter/pkg/test" + "sigs.k8s.io/karpenter/test/pkg/debug" + "sigs.k8s.io/karpenter/test/pkg/environment/common" +) + +// buildMarginalDeployments returns two deployments with the basic_test.go +// resource profile (500 pods each: small 900m/3100Mi, large 3500m/28Gi). +// When deletionCost > 0 the small deployment carries a pod-deletion-cost +// annotation, raising each pod's EvictionCost by cost/2^27 (clamped at 10) +// and driving the pool's total_disruption_cost up. That pushes the +// consolidation score BELOW the 1/k=0.5 Balanced threshold on scale-in. +// When deletionCost is zero the same fixture stays ABOVE threshold. The +// single toggle isolates score-side behavior from the workload shape. +func buildMarginalDeployments(deletionCost int) (*appsv1.Deployment, *appsv1.Deployment) { + var smallExtras []test.DeploymentOptionModifier + if deletionCost > 0 { + smallExtras = append(smallExtras, test.WithAnnotations(map[string]string{ + corev1.PodDeletionCost: strconv.Itoa(deletionCost), + })) + } + smallOpts := test.CreateDeploymentOptions("marginal-small-app", 500, "900m", "3100Mi", smallExtras...) + largeOpts := test.CreateDeploymentOptions("marginal-large-app", 500, "3500m", "28Gi") + return test.Deployment(smallOpts), test.Deployment(largeOpts) +} + +// runBaselineMarginalPhases runs the shared two-phase fixture (1000-pod +// scale-out then 700-pod consolidation) with LatencyHarness capture and +// sidecar-JSON emission for both phases. The scale-in phase drives both +// deployments to 350 replicas at once, matching basic_test.go. +// The caller pins the ConsolidationPolicy before invocation; this function +// reads it back from the NodePool so the sidecar records the policy that +// actually ran. Returns the consolidation phase's latency result so callers +// can run spec-specific soft checks on score-histogram deltas. +func runBaselineMarginalPhases(filePrefixBase string, smallDeployment, largeDeployment *appsv1.Deployment) *common.LatencyResult { + scaleOutPrefix := filePrefixBase + "_scale_out" + consolidationPrefix := filePrefixBase + "_consolidation" + policy := nodePool.Spec.Disruption.ConsolidationPolicy + + env.ExpectCreated(nodePool, nodeClass, smallDeployment, largeDeployment) + + scaleOutHarness, err := common.StartLatencyHarness(env) + Expect(err).ToNot(HaveOccurred()) + + scaleOutReport, err := ReportScaleOutWithOutput(env, + filePrefixBase+" Scale Out", 1000, 15*time.Minute, scaleOutPrefix) + Expect(err).ToNot(HaveOccurred()) + Expect(scaleOutReport.TotalPods).To(Equal(1000)) + initialNodes := scaleOutReport.TotalNodes + + scaleOutLatency, err := scaleOutHarness.Stop() + Expect(err).ToNot(HaveOccurred()) + GinkgoWriter.Printf("LatencyHarness [%s, %s]: %d histogram series, %d counter series\n", + scaleOutReport.TestName, policy, len(scaleOutLatency.LatencyStats), len(scaleOutLatency.Counters)) + if err := common.WriteLatencySidecar(os.Getenv("OUTPUT_DIR"), scaleOutPrefix, common.LatencySidecar{ + TestName: scaleOutReport.TestName, + ConsolidationPolicy: string(policy), + Timestamp: time.Now(), + LatencyStats: scaleOutLatency.LatencyStats, + Counters: scaleOutLatency.Counters, + }); err != nil { + GinkgoWriter.Printf("LatencyHarness: %v\n", err) + } + + By("Scaling both deployments down and capturing consolidation latency") + smallDeployment.Spec.Replicas = new(int32(350)) + largeDeployment.Spec.Replicas = new(int32(350)) + env.ExpectUpdated(smallDeployment, largeDeployment) + + consolidationHarness, err := common.StartLatencyHarness(env) + Expect(err).ToNot(HaveOccurred()) + + consolidationReport, err := ReportConsolidation(env, + filePrefixBase+" Consolidation", 1000, 700, initialNodes, 20*time.Minute) + Expect(err).ToNot(HaveOccurred()) + + consolidationLatency, err := consolidationHarness.Stop() + Expect(err).ToNot(HaveOccurred()) + + OutputPerformanceReport(consolidationReport, consolidationPrefix) + GinkgoWriter.Printf("LatencyHarness [%s, %s]: %d histogram series, %d counter series\n", + consolidationReport.TestName, policy, len(consolidationLatency.LatencyStats), len(consolidationLatency.Counters)) + if err := common.WriteLatencySidecar(os.Getenv("OUTPUT_DIR"), consolidationPrefix, common.LatencySidecar{ + TestName: consolidationReport.TestName, + ConsolidationPolicy: string(policy), + Timestamp: time.Now(), + LatencyStats: consolidationLatency.LatencyStats, + Counters: consolidationLatency.Counters, + }); err != nil { + GinkgoWriter.Printf("LatencyHarness: %v\n", err) + } + + Expect(consolidationReport.TotalPods).To(Equal(700)) + Expect(consolidationLatency.Counters).ToNot(BeNil()) + Expect(consolidationLatency.LatencyStats).ToNot(BeNil()) + return consolidationLatency +} + +var _ = Describe("Performance", Label(debug.NoWatch), func() { + Context("Balanced Baseline", func() { + // Control arm: same 1000-pod / 700-pod fixture as basic_test.go but + // wrapped with LatencyHarness so the hero histograms + // (scheduling_decision, voluntary_disruption_decision_evaluation, + // pods_bound) are captured for both scale-out and consolidation. The + // suite-wide default policy is already WhenEmptyOrUnderutilized + // (suite_test.go:62), so this spec exists as the reference latency + // distribution the marginal Balanced runs compare against. + // Cross-policy delta analysis is offline on the paired sidecar + // JSONs; asserting deltas in-band would encode KWOK-timing flake + // into CI. + It("should capture reference latency distribution under WhenEmptyOrUnderutilized", func() { + smallDeployment, largeDeployment := buildMarginalDeployments(0) + runBaselineMarginalPhases("balanced_baseline", smallDeployment, largeDeployment) + }) + }) + + Context("Balanced Marginal Move", func() { + // Two variants share the 1000-pod fixture. The only difference is + // the small deployment's pod-deletion-cost annotation, which shifts + // the pool's total_disruption_cost denominator and pushes the + // consolidation score across the 1/k=0.5 threshold. See the RFC's + // "Marginal Move" example (designs/balanced-consolidation.md:236). + // + // Neither variant asserts an exact count from + // karpenter_consolidation_moves_total{decision=...}; KWOK timing + // blurs which candidates land in a given round. Instead the latency + // sidecar carries counters and score-histogram deltas so downstream + // analysis can compare distributions across the paired runs and + // against the baseline sidecar. + BeforeEach(func() { + nodePool.Spec.Disruption.ConsolidationPolicy = v1.ConsolidationPolicyBalanced + }) + type marginalFixture struct { + name string + filePrefix string + deletionCost int + expectedDecision string + } + fixtures := []marginalFixture{ + { + name: "just-above-threshold (score > 0.5, accepts)", + filePrefix: "balanced_marginal_accept", + deletionCost: 0, + expectedDecision: "approved", + }, + { + name: "just-below-threshold (score < 0.5, rejects)", + filePrefix: "balanced_marginal_reject", + deletionCost: 2000000000, + expectedDecision: "rejected", + }, + } + for _, fx := range fixtures { + It(fmt.Sprintf("should capture Balanced consolidation latency %s", fx.name), func() { + smallDeployment, largeDeployment := buildMarginalDeployments(fx.deletionCost) + consolidationLatency := runBaselineMarginalPhases(fx.filePrefix, smallDeployment, largeDeployment) + + // Soft directional check on the score-histogram deltas. + // karpenter_consolidation_score is labeled by decision, so + // approved and rejected observations appear as separate + // series in the delta. This checks that at least one series + // with the expected decision was recorded during the phase. + // It does not assert counts (KWOK single-round variance is + // too high to bound). + if !hasScoreSeriesForDecision(consolidationLatency.LatencyStats, fx.expectedDecision) { + GinkgoWriter.Printf( + "LatencyHarness: no karpenter_consolidation_score{decision=%s} series in delta; offline sidecar analysis carries the signal\n", + fx.expectedDecision) + } + }) + } + }) +}) + +// hasScoreSeriesForDecision reports whether the LatencyStats delta contains +// a karpenter_consolidation_score observation carrying decision=. +// Reads HistogramStats.MetricName + Labels directly rather than parsing the +// series-key string. +func hasScoreSeriesForDecision(latencyStats map[string]common.HistogramStats, decision string) bool { + for _, s := range latencyStats { + if s.MetricName == "karpenter_consolidation_score" && s.Labels["decision"] == decision { + return true + } + } + return false +} diff --git a/test/suites/performance/balanced_churn_heterogeneous_test.go b/test/suites/performance/balanced_churn_heterogeneous_test.go new file mode 100644 index 0000000000..7b9941e1ad --- /dev/null +++ b/test/suites/performance/balanced_churn_heterogeneous_test.go @@ -0,0 +1,270 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package performance + +import ( + "fmt" + "os" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" + + "sigs.k8s.io/karpenter/kwok/apis/v1alpha1" + v1 "sigs.k8s.io/karpenter/pkg/apis/v1" + "sigs.k8s.io/karpenter/pkg/test" + "sigs.k8s.io/karpenter/test/pkg/debug" + "sigs.k8s.io/karpenter/test/pkg/environment/common" +) + +// balancedPolicies enumerates the paired baseline vs Balanced iteration used +// by both spec groups. Baseline runs first so the second run starts from an +// AfterEach-clean state, keeping diff analysis stable. +var balancedPolicies = []v1.ConsolidationPolicy{ + v1.ConsolidationPolicyWhenEmptyOrUnderutilized, + v1.ConsolidationPolicyBalanced, +} + +// suiteConsolidateAfter mirrors the value pinned in suite_test.go BeforeEach. +// Update in lockstep if the suite-wide default changes. +const suiteConsolidateAfter = 30 * time.Second + +// scaleAndSettleWaitFactor sets the sleep after scale-out at 2× ConsolidateAfter +// to give the consolidation controller two full evaluation cycles before we sample metrics. +const scaleAndSettleWaitFactor = 2 + +// policyPrefix maps a ConsolidationPolicy to the short filePrefix segment +// used in artifact filenames. WhenEmptyOrUnderutilized is the reference +// baseline; Balanced is the arm under test. +func policyPrefix(p v1.ConsolidationPolicy) string { + if p == v1.ConsolidationPolicyBalanced { + return "balanced" + } + return "baseline" +} + +// buildFamilyRestrictedNodePool constructs a NodePool restricted to a single +// KWOK instance family, carrying the standard suite requirements (linux +// pinned via defaultNodePool, on-demand pinned via defaultNodePool, instance +// size clamped < 32 to match the suite-wide default). The nodePool is a +// clone of env.DefaultNodePool with the family requirement layered on. +func buildFamilyRestrictedNodePool(env *common.Environment, nodeClass *unstructured.Unstructured, family string, policy v1.ConsolidationPolicy) *v1.NodePool { + np := env.DefaultNodePool(nodeClass) + np.Name = fmt.Sprintf("%s-%s", family, np.Name) + np.Spec.Template.Labels["perf.karpenter.sh/pool"] = fmt.Sprintf("%s-pool", family) + test.ReplaceRequirements(np, + v1.NodeSelectorRequirementWithMinValues{ + Key: v1alpha1.InstanceFamilyLabelKey, + Operator: corev1.NodeSelectorOpIn, + Values: []string{family}, + }, + v1.NodeSelectorRequirementWithMinValues{ + Key: v1alpha1.InstanceSizeLabelKey, + Operator: corev1.NodeSelectorOpLt, + Values: []string{"32"}, + }, + ) + np.Spec.Limits = v1.Limits{} + np.Spec.Disruption.ConsolidationPolicy = policy + np.Spec.Disruption.ConsolidateAfter = v1.MustParseNillableDuration("30s") + np.Spec.Disruption.Budgets = []v1.Budget{{Nodes: "100%"}} + return np +} + +// scaleAndSettle updates the deployment to targetReplicas, waits for pods to +// reach that count, then sleeps two consolidateAfter cycles so the +// disruption controller has time to act before the round-end capture. +func scaleAndSettle(env *common.Environment, dep *appsv1.Deployment, targetReplicas int32, timeout time.Duration) { + replicas := targetReplicas + dep.Spec.Replicas = &replicas + env.ExpectUpdated(dep) + sel := labels.SelectorFromSet(map[string]string{test.DiscoveryLabel: "unspecified"}) + env.EventuallyExpectHealthyPodCountWithTimeout(timeout, sel, int(targetReplicas)) + time.Sleep(scaleAndSettleWaitFactor * suiteConsolidateAfter) +} + +// writeLatencySidecar emits result to OUTPUT_DIR/_latency.json via +// the shared common.WriteLatencySidecar helper. Logs summary counts to +// GinkgoWriter regardless of OUTPUT_DIR so CI logs surface the harness result. +func writeLatencySidecar(testName, filePrefix string, policy v1.ConsolidationPolicy, result *common.LatencyResult) { + if result == nil { + GinkgoWriter.Printf("LatencyHarness: nil result for %s (%s); skipping sidecar\n", testName, policy) + return + } + GinkgoWriter.Printf("LatencyHarness [%s, %s]: %d histogram series, %d counter series\n", + testName, policy, len(result.LatencyStats), len(result.Counters)) + sc := common.LatencySidecar{ + TestName: testName, + ConsolidationPolicy: string(policy), + Timestamp: time.Now(), + LatencyStats: result.LatencyStats, + Counters: result.Counters, + } + if err := common.WriteLatencySidecar(os.Getenv("OUTPUT_DIR"), filePrefix, sc); err != nil { + GinkgoWriter.Printf("LatencyHarness: %v\n", err) + } +} + +// emitPolicyRun writes both the PerformanceReport JSON and the latency +// sidecar JSON under a shared file prefix. The two artifacts always stay +// paired on disk for offline diff analysis. +func emitPolicyRun(report *PerformanceReport, filePrefix string, policy v1.ConsolidationPolicy, result *common.LatencyResult) { + OutputPerformanceReport(report, filePrefix) + writeLatencySidecar(report.TestName, filePrefix, policy, result) +} + +var _ = Describe("Performance", Label(debug.NoWatch), func() { + Context("Balanced Churn Chain", func() { + // Each It runs one policy over a 400-pod / ~40-node scale-out then + // three scale-in / scale-out churn rounds. The RFC's 4-step + // max-churn ceiling at k=2 predicts Balanced's counter deltas + // diverge from baseline's by round 3. Comparison is offline: the + // paired PerformanceReport plus latency sidecar JSONs carry + // consolidation_moves_total, nodeclaims_created_total, and + // karpenter_voluntary_disruption_decision_evaluation_duration_seconds + // deltas per policy. LatencyHarness spans the full churn window. + for _, policy := range balancedPolicies { + prefix := policyPrefix(policy) + It(fmt.Sprintf("should measure churn under %s across three scale-in / scale-out rounds", policy), func() { + By("Pinning ConsolidationPolicy for this run") + nodePool.Spec.Disruption.ConsolidationPolicy = policy + env.ExpectCreated(nodePool, nodeClass) + + By("Scaling out to the churn-chain fixture (400 pods)") + opts := test.CreateDeploymentOptions("churn-chain-app", 400, "900m", "3100Mi") + dep := test.Deployment(opts) + env.ExpectCreated(dep) + + scaleOutReport, err := ReportScaleOutWithOutput(env, + fmt.Sprintf("Balanced Churn Chain %s Scale Out", policy), + 400, 15*time.Minute, + fmt.Sprintf("balanced_churn_%s_scale_out", prefix)) + Expect(err).ToNot(HaveOccurred()) + Expect(scaleOutReport.TotalPods).To(Equal(400)) + initialNodes := scaleOutReport.TotalNodes + + By("Starting LatencyHarness for the churn window") + h, err := common.StartLatencyHarness(env) + Expect(err).ToNot(HaveOccurred()) + + By("Round 1: scale in to 200 pods") + scaleAndSettle(env, dep, 200, 10*time.Minute) + By("Round 2: scale back out to 400 pods") + scaleAndSettle(env, dep, 400, 10*time.Minute) + By("Round 3: scale in to 200 pods") + scaleAndSettle(env, dep, 200, 10*time.Minute) + + By("Capturing LatencyHarness result at end of churn window") + result, err := h.Stop() + Expect(err).ToNot(HaveOccurred()) + + By("Emitting the consolidation report and latency sidecar") + consolidationReport, err := ReportConsolidation(env, + fmt.Sprintf("Balanced Churn Chain %s", policy), + 400, 200, initialNodes, 20*time.Minute) + Expect(err).ToNot(HaveOccurred()) + emitPolicyRun(consolidationReport, + fmt.Sprintf("balanced_churn_%s_consolidation", prefix), + policy, result) + + // Soft check: harness must have observed at least one + // scrape delta. Comparison across policies is offline on + // the paired JSON artifacts; hard bounds are not asserted + // because KWOK timing variance blurs per-round counts. + Expect(result.Counters).ToNot(BeNil()) + }) + } + }) + + Context("Balanced Heterogeneous NodePools", func() { + // Two family-restricted NodePools ('c' and 'm' KWOK families) each + // carry a workload at a distinct pod density profile: a dense + // 500m/1Gi deployment on the c-pool, a sparse 2500m/8Gi deployment + // on the m-pool. Scaling both down triggers Balanced to make + // per-pool decisions (per RFC "source pool's policy governs") vs + // baseline which accepts any positive-savings move. Comparison is + // offline: paired PerformanceReport plus latency sidecar JSONs + // carry karpenter_consolidation_moves_total{nodepool}, per-pool + // disruption timing, and karpenter_nodeclaims_created_total per + // policy. + BeforeEach(func() { + if !env.IsDefaultNodeClassKWOK() { + Skip("heterogeneous NodePool fixture uses KWOK-only instance-family labels") + } + }) + for _, policy := range balancedPolicies { + prefix := policyPrefix(policy) + It(fmt.Sprintf("should split load across two heterogeneous NodePools under %s", policy), func() { + By("Building two family-restricted NodePools") + poolC := buildFamilyRestrictedNodePool(env, nodeClass, "c", policy) + poolM := buildFamilyRestrictedNodePool(env, nodeClass, "m", policy) + env.ExpectCreated(nodeClass, poolC, poolM) + + By("Deploying dense workload targeting the c-family pool") + denseOpts := test.CreateDeploymentOptions("het-dense-app", 300, "500m", "1Gi", + test.WithNodeSelector(map[string]string{"perf.karpenter.sh/pool": "c-pool"})) + denseDep := test.Deployment(denseOpts) + + By("Deploying sparse workload targeting the m-family pool") + sparseOpts := test.CreateDeploymentOptions("het-sparse-app", 100, "2500m", "8Gi", + test.WithNodeSelector(map[string]string{"perf.karpenter.sh/pool": "m-pool"})) + sparseDep := test.Deployment(sparseOpts) + + env.ExpectCreated(denseDep, sparseDep) + + scaleOutReport, err := ReportScaleOutWithOutput(env, + fmt.Sprintf("Balanced Heterogeneous %s Scale Out", policy), + 400, 15*time.Minute, + fmt.Sprintf("balanced_heterogeneous_%s_scale_out", prefix)) + Expect(err).ToNot(HaveOccurred()) + Expect(scaleOutReport.TotalPods).To(Equal(400)) + initialNodes := scaleOutReport.TotalNodes + + By("Starting LatencyHarness for the consolidation window") + h, err := common.StartLatencyHarness(env) + Expect(err).ToNot(HaveOccurred()) + + By("Scaling both deployments down to trigger cross-pool consolidation") + denseReplicas := int32(180) + sparseReplicas := int32(60) + denseDep.Spec.Replicas = &denseReplicas + sparseDep.Spec.Replicas = &sparseReplicas + env.ExpectUpdated(denseDep, sparseDep) + + By("Recording the consolidation phase") + consolidationReport, err := ReportConsolidation(env, + fmt.Sprintf("Balanced Heterogeneous %s", policy), + 400, 240, initialNodes, 25*time.Minute) + Expect(err).ToNot(HaveOccurred()) + + By("Capturing LatencyHarness result at end of consolidation") + result, err := h.Stop() + Expect(err).ToNot(HaveOccurred()) + emitPolicyRun(consolidationReport, + fmt.Sprintf("balanced_heterogeneous_%s_consolidation", prefix), + policy, result) + + Expect(consolidationReport.TotalPods).To(Equal(240)) + Expect(result.Counters).ToNot(BeNil()) + }) + } + }) +})