test: fix OOM test

Scale the number of replicas according to available resources.

The problem was that `qemu-race` has more memory available.

Signed-off-by: Andrey Smirnov <andrey.smirnov@siderolabs.com>

Author: smira

internal/integration/k8s/oom.go

@@ -14,6 +14,9 @@ import (
 	"time"
 
 	"github.com/cosi-project/runtime/pkg/state"
+	"github.com/dustin/go-humanize"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v3"
 
 	"github.com/siderolabs/talos/internal/integration/base"
 	"github.com/siderolabs/talos/pkg/machinery/client"
@@ -26,16 +29,8 @@ type OomSuite struct {
 	base.K8sSuite
 }
 
-var (
-	//go:embed testdata/oom.yaml
-	oomPodSpec []byte
-
-	//go:embed testdata/oom-50-replicas.yaml
-	oom50ReplicasPatch []byte
-
-	//go:embed testdata/oom-1-replica.yaml
-	oom1ReplicaPatch []byte
-)
+//go:embed testdata/oom.yaml
+var oomPodSpec []byte
 
 // SuiteName returns the name of the suite.
 func (suite *OomSuite) SuiteName() string {
@@ -72,40 +67,80 @@ func (suite *OomSuite) TestOom() {
 
 	suite.Require().NoError(suite.WaitForDeploymentAvailable(ctx, time.Minute, "default", "stress-mem", 2))
 
-	// Scale to 50
-	suite.PatchK8sObject(ctx, "default", "apps", "Deployment", "v1", "stress-mem", oom50ReplicasPatch)
+	// Figure out number of replicas, this is ballpark estimation of 15 replicas per 2GB of memory (per worker node)
+	numWorkers := len(suite.DiscoverNodeInternalIPsByType(ctx, machine.TypeWorker))
+	suite.Require().Greaterf(numWorkers, 0, "at least one worker node is required for the test")
+
+	memInfo, err := suite.Client.Memory(client.WithNode(ctx, suite.RandomDiscoveredNodeInternalIP(machine.TypeWorker)))
+	suite.Require().NoError(err)
+
+	memoryBytes := memInfo.GetMessages()[0].GetMeminfo().GetMemtotal() * 1024
+	numReplicas := int((memoryBytes/1024/1024+2048-1)/2048) * numWorkers * 15
+
+	suite.T().Logf("detected total memory: %s, workers %d => scaling to %d replicas",
+		humanize.IBytes(memoryBytes), numWorkers, numReplicas)
+
+	// Scale to discovered number of replicas
+	suite.PatchK8sObject(ctx, "default", "apps", "Deployment", "v1", "stress-mem", patchToReplicas(suite.T(), numReplicas))
 
 	// Expect at least one OOM kill of stress-ng within 15 seconds
-	suite.Assert().True(suite.waitForOOMKilled(ctx, 15*time.Second, "stress-ng"))
+	suite.Assert().True(suite.waitForOOMKilled(ctx, 15*time.Second, 2*time.Minute, "stress-ng"))
 
 	// Scale to 1, wait for deployment to scale down, proving system is operational
-	suite.PatchK8sObject(ctx, "default", "apps", "Deployment", "v1", "stress-mem", oom1ReplicaPatch)
+	suite.PatchK8sObject(ctx, "default", "apps", "Deployment", "v1", "stress-mem", patchToReplicas(suite.T(), 1))
 	suite.Require().NoError(suite.WaitForDeploymentAvailable(ctx, time.Minute, "default", "stress-mem", 1))
 
 	suite.APISuite.AssertClusterHealthy(ctx)
 }
 
+func patchToReplicas(t *testing.T, replicas int) []byte {
+	spec := map[string]any{
+		"spec": map[string]any{
+			"replicas": replicas,
+		},
+	}
+
+	patch, err := yaml.Marshal(spec)
+	require.NoError(t, err)
+
+	return patch
+}
+
 // Waits for a period of time and return returns whether or not OOM events containing a specified process have been observed.
-func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeout time.Duration, substr string) bool {
+//
+//nolint:gocyclo
+func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeToObserve, timeout time.Duration, substr string) bool {
 	startTime := time.Now()
 
 	watchCh := make(chan state.Event)
-	workerNode := suite.RandomDiscoveredNodeInternalIP(machine.TypeWorker)
-	workerCtx := client.WithNode(ctx, workerNode)
-
-	suite.Assert().NoError(suite.Client.COSI.WatchKind(
-		workerCtx,
-		runtime.NewOOMActionSpec(runtime.NamespaceName, "").Metadata(),
-		watchCh,
-	))
+	workerNodes := suite.DiscoverNodeInternalIPsByType(ctx, machine.TypeWorker)
+
+	// start watching OOM events on all worker nodes
+	for _, workerNode := range workerNodes {
+		suite.Assert().NoError(suite.Client.COSI.WatchKind(
+			client.WithNode(ctx, workerNode),
+			runtime.NewOOMActionSpec(runtime.NamespaceName, "").Metadata(),
+			watchCh,
+		))
+	}
 
 	timeoutCh := time.After(timeout)
-	ret := false
+	timeToObserveCh := time.After(timeToObserve)
+	numOOMObserved := 0
 
 	for {
 		select {
 		case <-timeoutCh:
-			return ret
+			suite.T().Logf("observed %d OOM events containing process substring %q", numOOMObserved, substr)
+
+			return numOOMObserved > 0
+		case <-timeToObserveCh:
+			if numOOMObserved > 0 {
+				// if we already observed some OOM events, consider it a success
+				suite.T().Logf("observed %d OOM events containing process substring %q", numOOMObserved, substr)
+
+				return true
+			}
 		case ev := <-watchCh:
 			if ev.Type != state.Created || ev.Resource.Metadata().Created().Before(startTime) {
 				continue
@@ -115,7 +150,7 @@ func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeout time.Durati
 
 			for _, proc := range res.Processes {
 				if strings.Contains(proc, substr) {
-					ret = true
+					numOOMObserved++
 				}
 			}
 		}