Add measured pods feature for accurate resource measurement on isolated nodes

Implements DPTP-4613 to address the issue where pod-scaler recommendations are
skewed by node contention. When multiple pods with poor CPU configurations are
scheduled on the same node, CPU is maxed out but pods finish eventually. The
pod-scaler observes low CPU utilization (due to node contention) and incorrectly
concludes requests should not be increased, leading to a cycle of reduced limits
and tighter packing.

This change introduces a measured pods system:
- Pods are classified as 'normal' or 'measured' based on whether they need fresh
  measurement data (measured if last measurement >10 days ago or never measured)
- Measured pods use podAntiAffinity rules to run on isolated nodes with no other
  CI workloads, ensuring accurate CPU/memory utilization measurement
- BigQuery integration queries and caches max CPU/memory utilization from measured
  pod runs, refreshing daily to keep data current
- Resource recommendations are applied only to the longest-running container
  in each pod, using actual measured utilization data instead of Prometheus
  metrics that may be skewed by node contention

The feature is opt-in via --enable-measured-pods flag and requires BigQuery
configuration (--bigquery-project-id and --bigquery-dataset-id).

Author: deepsm007

cmd/pod-scaler/admission.go

@@ -33,34 +33,51 @@ import (
 	"github.com/openshift/ci-tools/pkg/steps"
 )
 
-func admit(port, healthPort int, certDir string, client buildclientv1.BuildV1Interface, loaders map[string][]*cacheReloader, mutateResourceLimits bool, cpuCap int64, memoryCap string, cpuPriorityScheduling int64, reporter results.PodScalerReporter) {
+func admit(port, healthPort int, certDir string, client buildclientv1.BuildV1Interface, loaders map[string][]*cacheReloader, mutateResourceLimits bool, cpuCap int64, memoryCap string, cpuPriorityScheduling int64, authoritativeCPURequests, authoritativeMemoryRequests bool, enableMeasuredPods bool, bigQueryProjectID, bigQueryDatasetID, bigQueryCredentialsFile string, reporter results.PodScalerReporter) {
 	logger := logrus.WithField("component", "pod-scaler admission")
 	logger.Infof("Initializing admission webhook server with %d loaders.", len(loaders))
 	health := pjutil.NewHealthOnPort(healthPort)
 	resources := newResourceServer(loaders, health)
 	decoder := admission.NewDecoder(scheme.Scheme)
 
+	var bqClient *BigQueryClient
+	if enableMeasuredPods {
+		if bigQueryProjectID == "" || bigQueryDatasetID == "" {
+			logrus.Fatal("bigquery-project-id and bigquery-dataset-id are required when enable-measured-pods is true")
+		}
+		cache := NewMeasuredPodCache(logger)
+		var err error
+		bqClient, err = NewBigQueryClient(bigQueryProjectID, bigQueryDatasetID, bigQueryCredentialsFile, cache, logger)
+		if err != nil {
+			logrus.WithError(err).Fatal("Failed to create BigQuery client for measured pods")
+		}
+		logger.Info("Measured pods feature enabled with BigQuery integration")
+	}
+
 	server := webhook.NewServer(webhook.Options{
 		Port:    port,
 		CertDir: certDir,
 	})
-	server.Register("/pods", &webhook.Admission{Handler: &podMutator{logger: logger, client: client, decoder: decoder, resources: resources, mutateResourceLimits: mutateResourceLimits, cpuCap: cpuCap, memoryCap: memoryCap, cpuPriorityScheduling: cpuPriorityScheduling, reporter: reporter}})
+	server.Register("/pods", &webhook.Admission{Handler: &podMutator{logger: logger, client: client, decoder: decoder, resources: resources, mutateResourceLimits: mutateResourceLimits, cpuCap: cpuCap, memoryCap: memoryCap, cpuPriorityScheduling: cpuPriorityScheduling, authoritativeCPURequests: authoritativeCPURequests, authoritativeMemoryRequests: authoritativeMemoryRequests, bqClient: bqClient, reporter: reporter}})
 	logger.Info("Serving admission webhooks.")
 	if err := server.Start(interrupts.Context()); err != nil {
 		logrus.WithError(err).Fatal("Failed to serve webhooks.")
 	}
 }
 
 type podMutator struct {
-	logger                *logrus.Entry
-	client                buildclientv1.BuildV1Interface
-	resources             *resourceServer
-	mutateResourceLimits  bool
-	decoder               admission.Decoder
-	cpuCap                int64
-	memoryCap             string
-	cpuPriorityScheduling int64
-	reporter              results.PodScalerReporter
+	logger                      *logrus.Entry
+	client                      buildclientv1.BuildV1Interface
+	resources                   *resourceServer
+	mutateResourceLimits        bool
+	decoder                     admission.Decoder
+	cpuCap                      int64
+	memoryCap                    string
+	cpuPriorityScheduling       int64
+	authoritativeCPURequests    bool
+	authoritativeMemoryRequests bool
+	bqClient                    *BigQueryClient
+	reporter                    results.PodScalerReporter
 }
 
 func (m *podMutator) Handle(ctx context.Context, req admission.Request) admission.Response {
@@ -97,7 +114,16 @@ func (m *podMutator) Handle(ctx context.Context, req admission.Request) admissio
 		logger.WithError(err).Error("Failed to handle rehearsal Pod.")
 		return admission.Allowed("Failed to handle rehearsal Pod, ignoring.")
 	}
-	mutatePodResources(pod, m.resources, m.mutateResourceLimits, m.cpuCap, m.memoryCap, m.reporter, logger)
+	
+	// Classify pod as normal or measured (if enabled)
+	if m.bqClient != nil {
+		ClassifyPod(pod, m.bqClient, logger)
+		AddPodAntiAffinity(pod, logger)
+		// Apply measured pod resources before regular resource mutation
+		ApplyMeasuredPodResources(pod, m.bqClient, logger)
+	}
+	
+	mutatePodResources(pod, m.resources, m.mutateResourceLimits, m.cpuCap, m.memoryCap, m.authoritativeCPURequests, m.authoritativeMemoryRequests, m.reporter, logger)
 	m.addPriorityClass(pod)
 
 	marshaledPod, err := json.Marshal(pod)
@@ -292,7 +318,7 @@ func preventUnschedulable(resources *corev1.ResourceRequirements, cpuCap int64,
 	}
 }
 
-func mutatePodResources(pod *corev1.Pod, server *resourceServer, mutateResourceLimits bool, cpuCap int64, memoryCap string, reporter results.PodScalerReporter, logger *logrus.Entry) {
+func mutatePodResources(pod *corev1.Pod, server *resourceServer, mutateResourceLimits bool, cpuCap int64, memoryCap string, authoritativeCPU, authoritativeMemory bool, reporter results.PodScalerReporter, logger *logrus.Entry) {
 	mutateResources := func(containers []corev1.Container) {
 		for i := range containers {
 			meta := podscaler.MetadataFor(pod.ObjectMeta.Labels, pod.ObjectMeta.Name, containers[i].Name)

cmd/pod-scaler/admission_test.go

@@ -554,7 +554,7 @@ func TestMutatePodResources(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 			original := testCase.pod.DeepCopy()
-			mutatePodResources(testCase.pod, testCase.server, testCase.mutateResourceLimits, 10, "20Gi", &defaultReporter, logrus.WithField("test", testCase.name))
+			mutatePodResources(testCase.pod, testCase.server, testCase.mutateResourceLimits, 10, "20Gi", false, false, &defaultReporter, logrus.WithField("test", testCase.name))
 			diff := cmp.Diff(original, testCase.pod)
 			// In some cases, cmp.Diff decides to use non-breaking spaces, and it's not
 			// particularly deterministic about this. We don't care.

cmd/pod-scaler/main.go

@@ -67,6 +67,13 @@ type consumerOptions struct {
 	cpuCap                int64
 	memoryCap             string
 	cpuPriorityScheduling int64
+	
+	// Measured pods options - when enabled, pods are classified as "normal" or "measured"
+	// Measured pods run on isolated nodes to get accurate CPU/memory utilization data
+	enableMeasuredPods          bool
+	bigQueryProjectID           string
+	bigQueryDatasetID           string
+	bigQueryCredentialsFile     string
 }
 
 func bindOptions(fs *flag.FlagSet) *options {
@@ -89,6 +96,10 @@ func bindOptions(fs *flag.FlagSet) *options {
 	fs.Int64Var(&o.cpuCap, "cpu-cap", 10, "The maximum CPU request value, ex: 10")
 	fs.StringVar(&o.memoryCap, "memory-cap", "20Gi", "The maximum memory request value, ex: '20Gi'")
 	fs.Int64Var(&o.cpuPriorityScheduling, "cpu-priority-scheduling", 8, "Pods with CPU requests at, or above, this value will be admitted with priority scheduling")
+	fs.BoolVar(&o.enableMeasuredPods, "enable-measured-pods", false, "Enable measured pods feature. When enabled, pods are classified as 'normal' or 'measured' and measured pods run on isolated nodes to get accurate CPU/memory utilization data.")
+	fs.StringVar(&o.bigQueryProjectID, "bigquery-project-id", "", "Google Cloud project ID for BigQuery queries (required if enable-measured-pods is true)")
+	fs.StringVar(&o.bigQueryDatasetID, "bigquery-dataset-id", "", "BigQuery dataset ID for pod metrics (required if enable-measured-pods is true)")
+	fs.StringVar(&o.bigQueryCredentialsFile, "bigquery-credentials-file", "", "Path to Google Cloud credentials file for BigQuery access")
 	o.resultsOptions.Bind(fs)
 	return &o
 }
@@ -268,7 +279,7 @@ func mainAdmission(opts *options, cache Cache) {
 		logrus.WithError(err).Fatal("Failed to create pod-scaler reporter.")
 	}
 
-	go admit(opts.port, opts.instrumentationOptions.HealthPort, opts.certDir, client, loaders(cache), opts.mutateResourceLimits, opts.cpuCap, opts.memoryCap, opts.cpuPriorityScheduling, reporter)
+	go admit(opts.port, opts.instrumentationOptions.HealthPort, opts.certDir, client, loaders(cache), opts.mutateResourceLimits, opts.cpuCap, opts.memoryCap, opts.cpuPriorityScheduling, false, false, opts.enableMeasuredPods, opts.bigQueryProjectID, opts.bigQueryDatasetID, opts.bigQueryCredentialsFile, reporter)
 }
 
 func loaders(cache Cache) map[string][]*cacheReloader {