fix: improve OOM controller stability and make test strict on false positives

- Add d_* PSI derivative values to the trigger expression context
- Only trigger OOM action while PSI is rising
- Make OOM test fail if controller kills a cgroup without stress-ng

Signed-off-by: Dmitrii Sharshakov <[email protected]>

Author: dsseng

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

@@ -10,6 +10,7 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/google/cel-go/common/types"
 	"go.uber.org/zap"
@@ -56,12 +57,7 @@ func (cgroup *RankedCgroup) CalculateScore(expr *cel.Expression) (float64, error
 
 // EvaluateTrigger is a method obtaining data and evaluating the trigger expression.
 // When the result is true, designated OOM action is to be executed.
-func EvaluateTrigger(triggerExpr cel.Expression, evalContext map[string]any, cgroup string) (bool, error) {
-	err := PopulatePsiToCtx(cgroup, evalContext)
-	if err != nil {
-		return false, fmt.Errorf("cannot populate PSI context: %w", err)
-	}
-
+func EvaluateTrigger(triggerExpr cel.Expression, evalContext map[string]any) (bool, error) {
 	trigger, err := triggerExpr.EvalBool(celenv.OOMTrigger(), evalContext)
 	if err != nil {
 		return false, fmt.Errorf("cannot evaluate expression: %w", err)
@@ -71,7 +67,7 @@ func EvaluateTrigger(triggerExpr cel.Expression, evalContext map[string]any, cgr
 }
 
 // PopulatePsiToCtx populates the context with PSI data from a cgroup.
-func PopulatePsiToCtx(cgroup string, evalContext map[string]any) error {
+func PopulatePsiToCtx(cgroup string, evalContext map[string]any, psi map[string]float64, sampleInterval time.Duration) error {
 	node, err := cgroups.GetCgroupProperty(cgroup, "memory.pressure")
 	if err != nil {
 		return fmt.Errorf("cannot read memory pressure: %w", err)
@@ -93,7 +89,15 @@ func PopulatePsiToCtx(cgroup string, evalContext map[string]any) error {
 				return fmt.Errorf("PSI is not defined")
 			}
 
+			diff := 0.
+
+			if oldValue, ok := psi["memory_"+psiType+"_"+span]; ok {
+				diff = (value.Float64() - oldValue) / sampleInterval.Seconds()
+			}
+
+			evalContext["d_memory_"+psiType+"_"+span] = diff
 			evalContext["memory_"+psiType+"_"+span] = value.Float64()
+			psi["memory_"+psiType+"_"+span] = value.Float64()
 		}
 	}
 

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

@@ -156,7 +156,7 @@ func TestPopulatePsiToCtx(t *testing.T) {
 
 			ctx := map[string]any{}
 
-			err := oom.PopulatePsiToCtx(test.dir, ctx)
+			err := oom.PopulatePsiToCtx(test.dir, ctx, make(map[string]float64), 0)
 
 			if test.expectErr == "" {
 				require.NoError(t, err)
@@ -241,7 +241,10 @@ func TestEvaluateTrigger(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 
-			trigger, err := oom.EvaluateTrigger(test.triggerExpr, test.ctx, test.dir)
+			err := oom.PopulatePsiToCtx(test.dir, test.ctx, make(map[string]float64), 0)
+			require.NoError(t, err)
+
+			trigger, err := oom.EvaluateTrigger(test.triggerExpr, test.ctx)
 
 			assert.Equal(t, test.expect, trigger)
 

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

@@ -46,6 +46,7 @@ type OOMController struct {
 	V1Alpha1Mode    runtime.Mode
 	actionLog       []actionLogItem
 	idSeq           int
+	psi             map[string]float64
 }
 
 // Name implements controller.Controller interface.
@@ -117,6 +118,7 @@ func (ctrl *OOMController) Run(ctx context.Context, r controller.Runtime, logger
 	triggerExpr := defaultTriggerExpr()
 	scoringExpr := defaultScoringExpr()
 	sampleInterval := defaultSampleInterval
+	ctrl.psi = make(map[string]float64)
 
 	ticker := time.NewTicker(sampleInterval)
 	tickerC := ticker.C
@@ -150,7 +152,14 @@ func (ctrl *OOMController) Run(ctx context.Context, r controller.Runtime, logger
 			"time_since_trigger": time.Since(ctrl.ActionTriggered),
 		}
 
-		trigger, err := oom.EvaluateTrigger(triggerExpr, evalContext, ctrl.CgroupRoot)
+		err := oom.PopulatePsiToCtx(ctrl.CgroupRoot, evalContext, ctrl.psi, sampleInterval)
+		if err != nil {
+			logger.Error("cannot populate PSI context", zap.Error(err))
+
+			continue
+		}
+
+		trigger, err := oom.EvaluateTrigger(triggerExpr, evalContext)
 		if err != nil {
 			logger.Error("cannot evaluate OOM trigger expression", zap.Error(err))
 

internal/integration/k8s/oom.go

@@ -39,8 +39,6 @@ func (suite *OomSuite) SuiteName() string {
 
 // TestOom verifies that system remains stable after handling an OOM event.
 func (suite *OomSuite) TestOom() {
-	suite.T().Skip("skip the test until https://github.com/siderolabs/talos/issues/12077 is resolved")
-
 	if suite.Cluster == nil {
 		suite.T().Skip("without full cluster state reaching out to the node IP is not reliable")
 	}
@@ -86,12 +84,16 @@ func (suite *OomSuite) TestOom() {
 	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, 2*time.Minute, "stress-ng"))
+	suite.Assert().True(suite.waitForOOMKilled(ctx, 15*time.Second, 2*time.Minute, "stress-ng", 1))
 
 	// Scale to 1, wait for deployment to scale down, proving system is operational
 	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))
 
+	// Monitor OOM kills for 15 seconds and make sure no kills other than stress-ng happen
+	// Allow 0 as well: ideally that'd be the case, but fail on anything not containing stress-ng
+	suite.Assert().True(suite.waitForOOMKilled(ctx, 15*time.Second, 2*time.Minute, "stress-ng", 0))
+
 	suite.APISuite.AssertClusterHealthy(ctx)
 }
 
@@ -111,7 +113,7 @@ func patchToReplicas(t *testing.T, replicas int) []byte {
 // Waits for a period of time and return returns whether or not OOM events containing a specified process have been observed.
 //
 //nolint:gocyclo
-func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeToObserve, timeout time.Duration, substr string) bool {
+func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeToObserve, timeout time.Duration, substr string, n int) bool {
 	startTime := time.Now()
 
 	watchCh := make(chan state.Event)
@@ -135,9 +137,9 @@ func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeToObserve, time
 		case <-timeoutCh:
 			suite.T().Logf("observed %d OOM events containing process substring %q", numOOMObserved, substr)
 
-			return numOOMObserved > 0
+			return numOOMObserved >= n
 		case <-timeToObserveCh:
-			if numOOMObserved > 0 {
+			if numOOMObserved >= n {
 				// if we already observed some OOM events, consider it a success
 				suite.T().Logf("observed %d OOM events containing process substring %q", numOOMObserved, substr)
 
@@ -150,11 +152,20 @@ func (suite *OomSuite) waitForOOMKilled(ctx context.Context, timeToObserve, time
 
 			res := ev.Resource.(*runtime.OOMAction).TypedSpec()
 
+			found := false
+
 			for _, proc := range res.Processes {
 				if strings.Contains(proc, substr) {
+					found = true
 					numOOMObserved++
 				}
 			}
+
+			if !found {
+				suite.T().Logf("observed an OOM event not containing process substring %q: %q", substr, res.Processes)
+
+				return false
+			}
 		}
 	}
 }

pkg/machinery/cel/celenv/celenv.go

@@ -98,6 +98,14 @@ var OOMTrigger = sync.OnceValue(func() *cel.Env {
 					cel.Variable("memory_full_avg60", types.DoubleType),
 					cel.Variable("memory_full_avg300", types.DoubleType),
 					cel.Variable("memory_full_total", types.DoubleType),
+					cel.Variable("d_memory_some_avg10", types.DoubleType),
+					cel.Variable("d_memory_some_avg60", types.DoubleType),
+					cel.Variable("d_memory_some_avg300", types.DoubleType),
+					cel.Variable("d_memory_some_total", types.DoubleType),
+					cel.Variable("d_memory_full_avg10", types.DoubleType),
+					cel.Variable("d_memory_full_avg60", types.DoubleType),
+					cel.Variable("d_memory_full_avg300", types.DoubleType),
+					cel.Variable("d_memory_full_total", types.DoubleType),
 					cel.Variable("time_since_trigger", types.DurationType),
 					cel.OptionalTypes(),
 				},

pkg/machinery/constants/constants.go

@@ -1314,7 +1314,7 @@ const (
 	ContainerMarkerFilePath = "/usr/etc/in-container"
 
 	// DefaultOOMTriggerExpression is the default CEL expression used to determine whether to trigger OOM.
-	DefaultOOMTriggerExpression = `memory_full_avg10 > 12.0 && time_since_trigger > duration("500ms")`
+	DefaultOOMTriggerExpression = `memory_full_avg10 > 12.0 && d_memory_full_avg10 > 0.0 && time_since_trigger > duration("500ms")`
 
 	// DefaultOOMCgroupRankingExpression is the default CEL expression used to rank cgroups for OOM killer.
 	DefaultOOMCgroupRankingExpression = `memory_max.hasValue() ? 0.0 :