test: integration test for OOM controller

- Improve docs
- Disable OOM controller in container mode
- Log OOM events
- Add an integration test verifying the system can sustain an OOM event

Signed-off-by: Dmitrii Sharshakov <dmitry.sharshakov@siderolabs.com>

Author: dsseng

api/resource/definitions/runtime/runtime.proto

@@ -123,6 +123,13 @@ message MountStatusSpec {
   repeated string encryption_providers = 6;
 }
 
+// OOMActionSpec describes the OOM action record resource properties.
+message OOMActionSpec {
+  string trigger_context = 1;
+  double score = 2;
+  repeated string processes = 3;
+}
+
 // PlatformMetadataSpec describes platform metadata properties.
 message PlatformMetadataSpec {
   string platform = 1;

internal/app/machined/pkg/controllers/runtime/oom.go

@@ -5,11 +5,14 @@
 package runtime
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"math"
 	"os"
 	"path/filepath"
+	"strconv"
 	"sync"
 	"syscall"
 	"time"
@@ -22,16 +25,27 @@ import (
 	"golang.org/x/sys/unix"
 
 	"github.com/siderolabs/talos/internal/app/machined/pkg/controllers/runtime/internal/oom"
+	"github.com/siderolabs/talos/internal/app/machined/pkg/runtime"
 	"github.com/siderolabs/talos/pkg/machinery/cel"
 	"github.com/siderolabs/talos/pkg/machinery/cel/celenv"
 	"github.com/siderolabs/talos/pkg/machinery/constants"
 	"github.com/siderolabs/talos/pkg/machinery/resources/config"
+	runtimeres "github.com/siderolabs/talos/pkg/machinery/resources/runtime"
 )
 
+type actionLogItem struct {
+	runtimeres.OOMActionSpec
+
+	ID int
+}
+
 // OOMController is a controller that monitors memory PSI and handles near-OOM situations.
 type OOMController struct {
 	CgroupRoot      string
 	ActionTriggered time.Time
+	V1Alpha1Mode    runtime.Mode
+	actionLog       []actionLogItem
+	idSeq           int
 }
 
 // Name implements controller.Controller interface.
@@ -53,7 +67,12 @@ func (ctrl *OOMController) Inputs() []controller.Input {
 
 // Outputs implements controller.Controller interface.
 func (ctrl *OOMController) Outputs() []controller.Output {
-	return nil
+	return []controller.Output{
+		{
+			Type: runtimeres.OOMActionType,
+			Kind: controller.OutputExclusive,
+		},
+	}
 }
 
 var defaultTriggerExpr = sync.OnceValue(func() cel.Expression {
@@ -91,6 +110,10 @@ const defaultSampleInterval = 500 * time.Millisecond
 //
 //nolint:gocyclo
 func (ctrl *OOMController) Run(ctx context.Context, r controller.Runtime, logger *zap.Logger) error {
+	if ctrl.V1Alpha1Mode.InContainer() {
+		return nil
+	}
+
 	triggerExpr := defaultTriggerExpr()
 	scoringExpr := defaultScoringExpr()
 	sampleInterval := defaultSampleInterval
@@ -134,10 +157,48 @@ func (ctrl *OOMController) Run(ctx context.Context, r controller.Runtime, logger
 			continue
 		}
 
+		r.StartTrackingOutputs()
+
+		for _, action := range ctrl.actionLog {
+			if err := safe.WriterModify(ctx, r, runtimeres.NewOOMActionSpec(runtimeres.NamespaceName, strconv.Itoa(action.ID)),
+				func(item *runtimeres.OOMAction) error {
+					*item.TypedSpec() = action.OOMActionSpec
+
+					return nil
+				}); err != nil {
+				return fmt.Errorf("failed to create OOM action log: %w", err)
+			}
+		}
+
+		if err = safe.CleanupOutputs[*runtimeres.OOMAction](ctx, r); err != nil {
+			return err
+		}
+
 		// TODO: evaluate on different cgroups, not only root. E.g. action when podruntime experiences high PSI.
 		if trigger {
+			score, processes := ctrl.OomAction(logger, ctrl.CgroupRoot, scoringExpr)
+
+			ctxString, err := json.Marshal(evalContext)
+			if err != nil {
+				return fmt.Errorf("failed to marshal trigger context: %w", err)
+			}
+
+			ctrl.actionLog = append(ctrl.actionLog, actionLogItem{
+				ID: ctrl.idSeq,
+				OOMActionSpec: runtimeres.OOMActionSpec{
+					TriggerContext: string(ctxString),
+					Processes:      processes,
+					Score:          score,
+				},
+			})
+
+			ctrl.idSeq++
+
+			if len(ctrl.actionLog) > 10 {
+				ctrl.actionLog = ctrl.actionLog[len(ctrl.actionLog)-10:]
+			}
+
 			ctrl.ActionTriggered = time.Now()
-			ctrl.OomAction(logger, ctrl.CgroupRoot, scoringExpr)
 		}
 	}
 }
@@ -177,13 +238,13 @@ func (*OOMController) getConfig(cfg *config.MachineConfig) (cel.Expression, cel.
 }
 
 // OomAction handles out of memory conditions by selecting and killing cgroups based on memory usage data.
-func (ctrl *OOMController) OomAction(logger *zap.Logger, root string, scoringExpr cel.Expression) {
+func (ctrl *OOMController) OomAction(logger *zap.Logger, root string, scoringExpr cel.Expression) (float64, []string) {
 	logger.Info("OOM controller triggered")
 
 	ranking := oom.RankCgroups(logger, root, scoringExpr)
 
 	if len(ranking) == 0 {
-		return
+		return 0, []string{}
 	}
 
 	var (
@@ -198,13 +259,15 @@ func (ctrl *OOMController) OomAction(logger *zap.Logger, root string, scoringExp
 		}
 	}
 
-	err := reapCg(logger, cgroupToKill.Path)
+	processes, err := reapCg(logger, cgroupToKill.Path)
 	if err != nil {
 		logger.Error("cannot reap cgroup", zap.String("cgroup", cgroupToKill.Path), zap.Error(err))
 	}
+
+	return maxScore, processes
 }
 
-func reapCg(logger *zap.Logger, cgroupPath string) error {
+func reapCg(logger *zap.Logger, cgroupPath string) ([]string, error) {
 	logger.Warn("Sending SIGKILL to cgroup", zap.String("cgroup", cgroupPath))
 
 	processes := oom.ListCgroupProcs(cgroupPath)
@@ -213,8 +276,17 @@ func reapCg(logger *zap.Logger, cgroupPath string) error {
 	// Open pidfd's of all the processes in cgroup to accelerate kernel
 	// garbage-collecting those processes via mrelease.
 	pidfds := []int{}
+	cmdlines := []string{}
 
 	for _, pid := range processes {
+		cmdBytes, err := os.ReadFile(filepath.Join("/proc", strconv.Itoa(pid), "cmdline"))
+		if err == nil {
+			cmdlines = append(
+				cmdlines,
+				string(bytes.ReplaceAll(bytes.TrimRight(cmdBytes, "\x00"), []byte{0}, []byte{' '})),
+			)
+		}
+
 		// pidfd is always opened with CLOEXEC:
 		// https://github.com/torvalds/linux/blob/bf40f4b87761e2ec16efc8e49b9ca0d81f4115d8/kernel/pid.c#L637
 		pidfd, err := unix.PidfdOpen(pid, 0)
@@ -232,7 +304,7 @@ func reapCg(logger *zap.Logger, cgroupPath string) error {
 	if err != nil {
 		logger.Error("failed to send SIGKILL", zap.String("cgroup", cgroupPath), zap.Error(err))
 
-		return err
+		return cmdlines, err
 	}
 
 	for _, pidfd := range pidfds {
@@ -245,5 +317,5 @@ func reapCg(logger *zap.Logger, cgroupPath string) error {
 		}
 	}
 
-	return nil
+	return cmdlines, nil
 }

internal/app/machined/pkg/runtime/v1alpha2/v1alpha2_controller.go

@@ -444,7 +444,9 @@ func (ctrl *Controller) Run(ctx context.Context, drainer *runtime.Drainer) error
 		&runtimecontrollers.VersionController{},
 		&runtimecontrollers.WatchdogTimerConfigController{},
 		&runtimecontrollers.WatchdogTimerController{},
-		&runtimecontrollers.OOMController{},
+		&runtimecontrollers.OOMController{
+			V1Alpha1Mode: ctrl.v1alpha1Runtime.State().Platform().Mode(),
+		},
 		&secrets.APICertSANsController{},
 		&secrets.APIController{},
 		&secrets.EncryptionSaltController{},

internal/app/machined/pkg/runtime/v1alpha2/v1alpha2_state.go

@@ -225,6 +225,7 @@ func NewState() (*State, error) {
 		&runtime.MetaKey{},
 		&runtime.MetaLoaded{},
 		&runtime.MountStatus{},
+		&runtime.OOMAction{},
 		&runtime.PlatformMetadata{},
 		&runtime.SBOMItem{},
 		&runtime.SecurityState{},

internal/integration/base/k8s.go

@@ -25,6 +25,7 @@ import (
 	"github.com/siderolabs/gen/xslices"
 	"github.com/siderolabs/go-pointer"
 	"github.com/siderolabs/go-retry/retry"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	eventsv1 "k8s.io/api/events/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -514,6 +515,41 @@ func (k8sSuite *K8sSuite) WaitForPodToBeRunning(ctx context.Context, timeout tim
 	}
 }
 
+// WaitForDeploymentAvailable waits for the deployment with the given namespace and name to be running with the requested replicas.
+func (k8sSuite *K8sSuite) WaitForDeploymentAvailable(ctx context.Context, timeout time.Duration, namespace, deplName string, replicas int32) error {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	watcher, err := k8sSuite.Clientset.AppsV1().Deployments(namespace).Watch(ctx, metav1.ListOptions{
+		FieldSelector: fields.OneTermEqualSelector("metadata.name", deplName).String(),
+	})
+	if err != nil {
+		return err
+	}
+
+	defer watcher.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case event := <-watcher.ResultChan():
+			if event.Type == watch.Error {
+				return fmt.Errorf("error watching deployment: %v", event.Object)
+			}
+
+			deployment, ok := event.Object.(*appsv1.Deployment)
+			if !ok {
+				continue
+			}
+
+			if deployment.Name == deplName && deployment.Status.AvailableReplicas == replicas {
+				return nil
+			}
+		}
+	}
+}
+
 // LogPodLogsByLabel logs the logs of the pod with the given namespace and label.
 func (k8sSuite *K8sSuite) LogPodLogsByLabel(ctx context.Context, namespace, label, value string) {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)

internal/integration/k8s/oom.go

@@ -0,0 +1,127 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+//go:build integration_k8s
+
+package k8s
+
+import (
+	"context"
+	_ "embed"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/cosi-project/runtime/pkg/state"
+
+	"github.com/siderolabs/talos/internal/integration/base"
+	"github.com/siderolabs/talos/pkg/machinery/client"
+	"github.com/siderolabs/talos/pkg/machinery/config/machine"
+	"github.com/siderolabs/talos/pkg/machinery/resources/runtime"
+)
+
+// OomSuite verifies that userspace OOM handler will kill excessive replicas of a heavy memory consumer deployment.
+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
+)
+
+// SuiteName returns the name of the suite.
+func (suite *OomSuite) SuiteName() string {
+	return "k8s.OomSuite"
+}
+
+// TestOom verifies that system remains stable after handling an OOM event.
+func (suite *OomSuite) TestOom() {
+	if suite.Cluster == nil {
+		suite.T().Skip("without full cluster state reaching out to the node IP is not reliable")
+	}
+
+	if testing.Short() {
+		suite.T().Skip("skipping in short mode")
+	}
+
+	if suite.Cluster.Provisioner() != base.ProvisionerQEMU {
+		suite.T().Skip("skipping OOM test since provisioner is not qemu")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	suite.T().Cleanup(cancel)
+
+	oomPodManifest := suite.ParseManifests(oomPodSpec)
+
+	suite.T().Cleanup(func() {
+		cleanUpCtx, cleanupCancel := context.WithTimeout(context.Background(), time.Minute)
+		defer cleanupCancel()
+
+		suite.DeleteManifests(cleanUpCtx, oomPodManifest)
+	})
+
+	suite.ApplyManifests(ctx, oomPodManifest)
+
+	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)
+
+	// Expect at least one OOM kill of stress-ng within 15 seconds
+	suite.Assert().True(suite.waitForOOMKilled(ctx, 15*time.Second, "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.Require().NoError(suite.WaitForDeploymentAvailable(ctx, time.Minute, "default", "stress-mem", 1))
+
+	suite.APISuite.AssertClusterHealthy(ctx)
+}
+
+// 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 {
+	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,
+	))
+
+	timeoutCh := time.After(timeout)
+	ret := false
+
+	for {
+		select {
+		case <-timeoutCh:
+			return ret
+		case ev := <-watchCh:
+			if ev.Type != state.Created || ev.Resource.Metadata().Created().Before(startTime) {
+				continue
+			}
+
+			res := ev.Resource.(*runtime.OOMAction).TypedSpec()
+
+			for _, proc := range res.Processes {
+				if strings.Contains(proc, substr) {
+					ret = true
+				}
+			}
+		}
+	}
+}
+
+func init() {
+	allSuites = append(allSuites, new(OomSuite))
+}