Create more test bucket to debug high cpu issues

pkg/test/ginkgo/cmd_runsuite.go

@@ -567,75 +567,26 @@ func (o *GinkgoRunSuiteOptions) Run(suite *TestSuite, clusterConfig *clusterdisc
 		return err
 	}
 
-	kubeTests, openshiftTests := splitTests(primaryTests, func(t *testCase) bool {
-		return k8sTestNames[t.name]
-	})
-
-	storageTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "[sig-storage]")
-	})
-
-	networkK8sTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "[sig-network]")
-	})
-
-	orderedNamespaceDeletionTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "OrderedNamespaceDeletion")
-	})
-
-	networkTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "[sig-network]")
-	})
-
-	netpolTests, networkK8sTests := splitTests(networkK8sTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "Netpol")
-	})
+	creator := selectBucketCreator()
+	buckets := creator.createBuckets(primaryTests, k8sTestNames, parallelism)
 
-	buildsTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "[sig-builds]")
-	})
-
-	// separate from cliTests
-	mustGatherTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
-		return strings.Contains(t.name, "[sig-cli] oc adm must-gather")
-	})
+	for _, b := range buckets {
+		logrus.Infof("Found %d %s tests (parallelism %d)", len(b.tests), b.name, b.parallelism)
+	}
 
-	logrus.Infof("Found %d openshift tests", len(openshiftTests))
-	logrus.Infof("Found %d kubernetes tests", len(kubeTests))
-	logrus.Infof("Found %d storage tests", len(storageTests))
-	logrus.Infof("Found %d network k8s tests", len(networkK8sTests))
-	logrus.Infof("Found %d ordered namespace deletion k8s tests", len(orderedNamespaceDeletionTests))
-	logrus.Infof("Found %d network tests", len(networkTests))
-	logrus.Infof("Found %d netpol tests", len(netpolTests))
-	logrus.Infof("Found %d builds tests", len(buildsTests))
-	logrus.Infof("Found %d must-gather tests", len(mustGatherTests))
-
-	// If user specifies a count, duplicate the kube and openshift tests that many times.
+	// If user specifies a count, duplicate the tests in each bucket.
 	expectedTestCount := len(early) + len(late)
 	if count != -1 {
-		originalKube := kubeTests
-		originalOpenshift := openshiftTests
-		originalStorage := storageTests
-		originalNetworkK8s := networkK8sTests
-		originalOrderedNamespaceDeletionTests := orderedNamespaceDeletionTests
-		originalNetwork := networkTests
-		originalNetpol := netpolTests
-		originalBuilds := buildsTests
-		originalMustGather := mustGatherTests
-
-		for i := 1; i < count; i++ {
-			kubeTests = append(kubeTests, copyTests(originalKube)...)
-			openshiftTests = append(openshiftTests, copyTests(originalOpenshift)...)
-			storageTests = append(storageTests, copyTests(originalStorage)...)
-			networkK8sTests = append(networkK8sTests, copyTests(originalNetworkK8s)...)
-			orderedNamespaceDeletionTests = append(orderedNamespaceDeletionTests, copyTests(originalOrderedNamespaceDeletionTests)...)
-			networkTests = append(networkTests, copyTests(originalNetwork)...)
-			netpolTests = append(netpolTests, copyTests(originalNetpol)...)
-			buildsTests = append(buildsTests, copyTests(originalBuilds)...)
-			mustGatherTests = append(mustGatherTests, copyTests(originalMustGather)...)
+		for i := range buckets {
+			original := buckets[i].tests
+			for j := 1; j < count; j++ {
+				buckets[i].tests = append(buckets[i].tests, copyTests(original)...)
+			}
 		}
 	}
-	expectedTestCount += len(openshiftTests) + len(kubeTests) + len(storageTests) + len(networkK8sTests) + len(orderedNamespaceDeletionTests) + len(networkTests) + len(netpolTests) + len(buildsTests) + len(mustGatherTests)
+	for _, b := range buckets {
+		expectedTestCount += len(b.tests)
+	}
 
 	abortFn := neverAbort
 	testCtx := ctx
@@ -656,78 +607,19 @@ func (o *GinkgoRunSuiteOptions) Run(suite *TestSuite, clusterConfig *clusterdisc
 	// TODO: will move to the monitor
 	pc.SetEvents([]string{upgradeEvent})
 
-	// Run kube, storage, openshift, and must-gather tests. If user specified a count of -1,
-	// we loop indefinitely.
+	// Run test buckets. If user specified a count of -1, we loop indefinitely.
 	for i := 0; (i < 1 || count == -1) && testCtx.Err() == nil; i++ {
-
-		kubeTestsCopy := copyTests(kubeTests)
-		kubeIntervalID, kubeStartTime := recordTestBucketInterval(monitorEventRecorder, "Kubernetes")
-		q.Execute(testCtx, kubeTestsCopy, parallelism, testOutputConfig, abortFn)
-		monitorEventRecorder.EndInterval(kubeIntervalID, time.Now())
-		logrus.Infof("Completed Kubernetes test bucket in %v", time.Since(kubeStartTime))
-		tests = append(tests, kubeTestsCopy...)
-
-		// I thought about randomizing the order of the kube, storage, and openshift tests, but storage dominates our e2e runs, so it doesn't help much.
-		storageTestsCopy := copyTests(storageTests)
-		storageIntervalID, storageStartTime := recordTestBucketInterval(monitorEventRecorder, "Storage")
-		q.Execute(testCtx, storageTestsCopy, max(1, parallelism/2), testOutputConfig, abortFn) // storage tests only run at half the parallelism, so we can avoid cloud provider quota problems.
-		monitorEventRecorder.EndInterval(storageIntervalID, time.Now())
-		logrus.Infof("Completed Storage test bucket in %v", time.Since(storageStartTime))
-		tests = append(tests, storageTestsCopy...)
-
-		networkK8sTestsCopy := copyTests(networkK8sTests)
-		networkK8sIntervalID, networkK8sStartTime := recordTestBucketInterval(monitorEventRecorder, "NetworkK8s")
-		q.Execute(testCtx, networkK8sTestsCopy, max(1, parallelism/2), testOutputConfig, abortFn) // run network tests separately.
-		monitorEventRecorder.EndInterval(networkK8sIntervalID, time.Now())
-		logrus.Infof("Completed NetworkK8s test bucket in %v", time.Since(networkK8sStartTime))
-		tests = append(tests, networkK8sTestsCopy...)
-
-		orderedNamespaceDeletionTestsCopy := copyTests(orderedNamespaceDeletionTests)
-		orderedNamespaceDeletionIntervalID, orderedNamespaceDeletionStartTime := recordTestBucketInterval(monitorEventRecorder, "OrderedNamespaceDeletion")
-		q.Execute(testCtx, orderedNamespaceDeletionTestsCopy, 1, testOutputConfig, abortFn) // Run ordered namespace deletion tests one at a time. They are sensitive to backlogs in the namespace controller's workqueue that have been observed during high test paralellism as many ephemeral test namespaces are being finalized concurrently.
-		monitorEventRecorder.EndInterval(orderedNamespaceDeletionIntervalID, time.Now())
-		logrus.Infof("Completed OrderedNamespaceDeletion test bucket in %v", time.Since(orderedNamespaceDeletionStartTime))
-		tests = append(tests, orderedNamespaceDeletionTestsCopy...)
-
-		networkTestsCopy := copyTests(networkTests)
-		networkIntervalID, networkStartTime := recordTestBucketInterval(monitorEventRecorder, "Network")
-		q.Execute(testCtx, networkTestsCopy, max(1, parallelism/2), testOutputConfig, abortFn) // run network tests separately.
-		monitorEventRecorder.EndInterval(networkIntervalID, time.Now())
-		logrus.Infof("Completed Network test bucket in %v", time.Since(networkStartTime))
-		tests = append(tests, networkTestsCopy...)
-
-		// k8s netpol tests are known to be heavy and there are cases when run in parallel it can overload
-		// a cluster causing negative side effects (e.g., https://issues.redhat.com/browse/OCPBUGS-57665)
-		// https://github.com/openshift/origin/pull/26775/changes#diff-998be43366fe821c61ca242aa34949870c9c6df2572cc060000e4cd990a72bebL58-L62
-		// this will only run 2 in parallel at once
-		netpolTestsCopy := copyTests(netpolTests)
-		netpolIntervalID, netpolStartTime := recordTestBucketInterval(monitorEventRecorder, "Netpol")
-		q.Execute(testCtx, netpolTestsCopy, 2, testOutputConfig, abortFn)
-		monitorEventRecorder.EndInterval(netpolIntervalID, time.Now())
-		logrus.Infof("Completed Netpol test bucket in %v", time.Since(netpolStartTime))
-		tests = append(tests, netpolTestsCopy...)
-
-		buildsTestsCopy := copyTests(buildsTests)
-		buildsIntervalID, buildsStartTime := recordTestBucketInterval(monitorEventRecorder, "Builds")
-		q.Execute(testCtx, buildsTestsCopy, max(1, parallelism/2), testOutputConfig, abortFn) // builds tests only run at half the parallelism, so we can avoid high cpu problems.
-		monitorEventRecorder.EndInterval(buildsIntervalID, time.Now())
-		logrus.Infof("Completed Builds test bucket in %v", time.Since(buildsStartTime))
-		tests = append(tests, buildsTestsCopy...)
-
-		openshiftTestsCopy := copyTests(openshiftTests)
-		openshiftIntervalID, openshiftStartTime := recordTestBucketInterval(monitorEventRecorder, "OpenShift")
-		q.Execute(testCtx, openshiftTestsCopy, parallelism, testOutputConfig, abortFn)
-		monitorEventRecorder.EndInterval(openshiftIntervalID, time.Now())
-		logrus.Infof("Completed OpenShift test bucket in %v", time.Since(openshiftStartTime))
-		tests = append(tests, openshiftTestsCopy...)
-
-		// run the must-gather tests after parallel tests to reduce resource contention
-		mustGatherTestsCopy := copyTests(mustGatherTests)
-		mustGatherIntervalID, mustGatherStartTime := recordTestBucketInterval(monitorEventRecorder, "MustGather")
-		q.Execute(testCtx, mustGatherTestsCopy, parallelism, testOutputConfig, abortFn)
-		monitorEventRecorder.EndInterval(mustGatherIntervalID, time.Now())
-		logrus.Infof("Completed MustGather test bucket in %v", time.Since(mustGatherStartTime))
-		tests = append(tests, mustGatherTestsCopy...)
+		for _, b := range buckets {
+			if len(b.tests) == 0 {
+				continue
+			}
+			testsCopy := copyTests(b.tests)
+			intervalID, startTime := recordTestBucketInterval(monitorEventRecorder, b.name)
+			q.Execute(testCtx, testsCopy, b.parallelism, testOutputConfig, abortFn)
+			monitorEventRecorder.EndInterval(intervalID, time.Now())
+			logrus.Infof("Completed %s test bucket in %v", b.name, time.Since(startTime))
+			tests = append(tests, testsCopy...)
+		}
 	}
 
 	// TODO: will move to the monitor

pkg/test/ginkgo/test_bucket.go

@@ -0,0 +1,172 @@
+package ginkgo
+
+import (
+	"os"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+)
+
+// testBucket represents a named group of tests with a specific parallelism level.
+type testBucket struct {
+	name        string
+	tests       []*testCase
+	parallelism int
+}
+
+// testBucketCreator splits tests into ordered buckets for execution.
+// Different implementations can apply different bucketing strategies
+// based on job characteristics (e.g., heavy jobs with FIPS + TechPreview).
+type testBucketCreator interface {
+	createBuckets(tests []*testCase, k8sTestNames map[string]bool, parallelism int) []testBucket
+}
+
+// isHeavyJob returns true if the job is both FIPS-enabled and TechPreview,
+// which requires CPU-heavy test bucketing to avoid test failures from CPU pressure.
+func isHeavyJob() bool {
+	jobName := strings.ToLower(os.Getenv("JOB_NAME"))
+	return strings.Contains(jobName, "fips") && strings.Contains(jobName, "techpreview")
+}
+
+// defaultBucketCreator implements the original test bucketing strategy.
+// Tests are split into functional groups with standard parallelism levels.
+type defaultBucketCreator struct{}
+
+func (d *defaultBucketCreator) createBuckets(tests []*testCase, k8sTestNames map[string]bool, parallelism int) []testBucket {
+	kubeTests, openshiftTests := splitTests(tests, func(t *testCase) bool {
+		return k8sTestNames[t.name]
+	})
+
+	storageTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-storage]")
+	})
+
+	networkK8sTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-network]")
+	})
+
+	orderedNamespaceDeletionTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "OrderedNamespaceDeletion")
+	})
+
+	networkTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-network]")
+	})
+
+	netpolTests, networkK8sTests := splitTests(networkK8sTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "Netpol")
+	})
+
+	buildsTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-builds]")
+	})
+
+	mustGatherTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-cli] oc adm must-gather")
+	})
+
+	return []testBucket{
+		{name: "Kubernetes", tests: kubeTests, parallelism: parallelism},
+		{name: "Storage", tests: storageTests, parallelism: max(1, parallelism/2)},
+		{name: "NetworkK8s", tests: networkK8sTests, parallelism: max(1, parallelism/2)},
+		{name: "OrderedNamespaceDeletion", tests: orderedNamespaceDeletionTests, parallelism: 1},
+		{name: "Network", tests: networkTests, parallelism: max(1, parallelism/2)},
+		{name: "Netpol", tests: netpolTests, parallelism: 2},
+		{name: "Builds", tests: buildsTests, parallelism: max(1, parallelism/2)},
+		{name: "OpenShift", tests: openshiftTests, parallelism: parallelism},
+		{name: "MustGather", tests: mustGatherTests, parallelism: parallelism},
+	}
+}
+
+// heavyJobBucketCreator implements CPU-aware bucketing for heavy jobs
+// (FIPS + TechPreview) that experience high CPU pressure.
+// CPU-intensive test groups are separated into dedicated buckets with
+// reduced parallelism to prevent API server / etcd overload.
+type heavyJobBucketCreator struct{}
+
+func (h *heavyJobBucketCreator) createBuckets(tests []*testCase, k8sTestNames map[string]bool, parallelism int) []testBucket {
+	kubeTests, openshiftTests := splitTests(tests, func(t *testCase) bool {
+		return k8sTestNames[t.name]
+	})
+
+	storageTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-storage]")
+	})
+
+	networkK8sTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-network]")
+	})
+
+	orderedNamespaceDeletionTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "OrderedNamespaceDeletion")
+	})
+
+	kubeCPUHeavyTests, kubeTests := splitTests(kubeTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[DRA]") ||
+			strings.Contains(t.name, "InPlace Resize") ||
+			strings.Contains(t.name, "Probing container") || strings.Contains(t.name, "Probing restartable init container") ||
+			strings.Contains(t.name, "Pod Generation") ||
+			strings.Contains(t.name, "CustomResourcePublishOpenAPI") ||
+			strings.Contains(t.name, "CustomResourceValidationRules") ||
+			strings.Contains(t.name, "CustomResourceConversionWebhook") ||
+			strings.Contains(t.name, "CustomResourceDefinition Watch") ||
+			strings.Contains(t.name, "CustomResourceFieldSelectors") ||
+			strings.Contains(t.name, "CRDValidationRatcheting") ||
+			strings.Contains(t.name, "FieldValidation")
+	})
+
+	netpolTests, networkK8sTests := splitTests(networkK8sTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "Netpol")
+	})
+
+	buildsTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-builds]")
+	})
+
+	mustGatherTests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-cli] oc adm must-gather")
+	})
+
+	// Tier 1: heaviest per-test CPU impact — infrastructure-intensive operations
+	openshiftCPUHeavyTier1Tests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[sig-olmv1]") ||
+			(strings.Contains(t.name, "[sig-network-edge]") &&
+				(strings.Contains(t.name, "Router") || strings.Contains(t.name, "GatewayAPI"))) ||
+			strings.Contains(t.name, "cloud-provider-aws") ||
+			strings.Contains(t.name, "[sig-network]")
+	})
+
+	// Tier 2: sustained load but less per-test impact
+	openshiftCPUHeavyTier2Tests, openshiftTests := splitTests(openshiftTests, func(t *testCase) bool {
+		return strings.Contains(t.name, "[Feature:DeploymentConfig]") ||
+			strings.Contains(t.name, "poddisruptionbudgets") ||
+			strings.Contains(t.name, "[sig-node] [Jira:Node/Kubelet]") ||
+			strings.Contains(t.name, "[sig-node] Probe configuration [OTP]") ||
+			strings.Contains(t.name, "[sig-cli] Workloads ")
+	})
+
+	return []testBucket{
+		{name: "Kubernetes", tests: kubeTests, parallelism: parallelism},
+		{name: "KubeCPUHeavy", tests: kubeCPUHeavyTests, parallelism: max(1, parallelism/2)},
+		{name: "Storage", tests: storageTests, parallelism: max(1, parallelism*2/3)},
+		{name: "NetworkK8s", tests: networkK8sTests, parallelism: max(1, parallelism*2/3)},
+		{name: "OrderedNamespaceDeletion", tests: orderedNamespaceDeletionTests, parallelism: 1},
+		{name: "Netpol", tests: netpolTests, parallelism: 2},
+		{name: "Builds", tests: buildsTests, parallelism: max(1, parallelism*2/3)},
+		{name: "OpenShift", tests: openshiftTests, parallelism: parallelism},
+		{name: "OpenshiftCPUHeavyTier1", tests: openshiftCPUHeavyTier1Tests, parallelism: max(1, parallelism/3)},
+		{name: "OpenshiftCPUHeavyTier2", tests: openshiftCPUHeavyTier2Tests, parallelism: max(1, parallelism*2/3)},
+		{name: "MustGather", tests: mustGatherTests, parallelism: parallelism},
+	}
+}
+
+// selectBucketCreator returns the appropriate bucket creator based on job characteristics.
+func selectBucketCreator() testBucketCreator {
+	jobName := os.Getenv("JOB_NAME")
+	if isHeavyJob() {
+		logrus.Infof("Heavy job detected (JOB_NAME=%s) - using CPU-heavy bucketing strategy", jobName)
+		return &heavyJobBucketCreator{}
+	}
+	logrus.Infof("Standard job (JOB_NAME=%s) - using default bucketing strategy", jobName)
+	return &defaultBucketCreator{}
+}