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

@@ -125,29 +125,45 @@ func TestPopulatePsiToCtx(t *testing.T) {
 			dir:       "./testdata/trigger-false",
 			expectErr: "",
 			expect: map[string]any{
-				"memory_full_avg10":  2.4,
-				"memory_full_avg300": 1.71,
-				"memory_full_avg60":  5.16,
-				"memory_full_total":  1.0654831e+07,
-				"memory_some_avg10":  2.82,
-				"memory_some_avg300": 1.97,
-				"memory_some_avg60":  5.95,
-				"memory_some_total":  1.217234e+07,
+				"memory_full_avg10":    2.4,
+				"memory_full_avg300":   1.71,
+				"memory_full_avg60":    5.16,
+				"memory_full_total":    1.0654831e+07,
+				"memory_some_avg10":    2.82,
+				"memory_some_avg300":   1.97,
+				"memory_some_avg60":    5.95,
+				"memory_some_total":    1.217234e+07,
+				"d_memory_full_avg10":  0.0,
+				"d_memory_full_avg300": 0.0,
+				"d_memory_full_avg60":  0.0,
+				"d_memory_full_total":  0.0,
+				"d_memory_some_avg10":  0.0,
+				"d_memory_some_avg300": 0.0,
+				"d_memory_some_avg60":  0.0,
+				"d_memory_some_total":  0.0,
 			},
 		},
 		{
 			name:      "true",
 			dir:       "./testdata/trigger-true",
 			expectErr: "",
 			expect: map[string]any{
-				"memory_full_avg10":  14.54,
-				"memory_full_avg60":  6.97,
-				"memory_full_avg300": 1.82,
-				"memory_full_total":  1.0654831e+07,
-				"memory_some_avg10":  17.06,
-				"memory_some_avg60":  8.04,
-				"memory_some_avg300": 2.1,
-				"memory_some_total":  1.217234e+07,
+				"memory_full_avg10":    14.54,
+				"memory_full_avg60":    6.97,
+				"memory_full_avg300":   1.82,
+				"memory_full_total":    1.0654831e+07,
+				"memory_some_avg10":    17.06,
+				"memory_some_avg60":    8.04,
+				"memory_some_avg300":   2.1,
+				"memory_some_total":    1.217234e+07,
+				"d_memory_full_avg10":  0.0,
+				"d_memory_full_avg300": 0.0,
+				"d_memory_full_avg60":  0.0,
+				"d_memory_full_total":  0.0,
+				"d_memory_some_avg10":  0.0,
+				"d_memory_some_avg300": 0.0,
+				"d_memory_some_avg60":  0.0,
+				"d_memory_some_total":  0.0,
 			},
 		},
 	} {
@@ -156,7 +172,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 +257,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")
 	}
@@ -77,7 +75,7 @@ func (suite *OomSuite) TestOom() {
 	suite.Require().NoError(err)
 
 	memoryBytes := memInfo.GetMessages()[0].GetMeminfo().GetMemtotal() * 1024
-	numReplicas := int((memoryBytes/1024/1024+2048-1)/2048) * numWorkers * 15
+	numReplicas := int((memoryBytes/1024/1024+2048-1)/2048) * numWorkers * 25
 
 	suite.T().Logf("detected memory: %s, workers %d => scaling to %d replicas",
 		humanize.IBytes(memoryBytes), numWorkers, numReplicas)
@@ -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/config/types/runtime/testdata/oom.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1alpha1
 kind: OOMConfig
-triggerExpression: memory_full_avg10 > 12.0 && time_since_trigger > duration("500ms")
+triggerExpression: memory_full_avg10 > 12.0 && d_memory_full_avg10 > 0.0 && time_since_trigger > duration("500ms")
 cgroupRankingExpression: 'memory_max.hasValue() ? 0.0 : ({Besteffort: 1.0, Burstable: 0.5, Guaranteed: 0.0, Podruntime: 0.0, System: 0.0}[class] * double(memory_current.orValue(0u)) / double(memory_peak.orValue(0u) - memory_current.orValue(0u)))'
 sampleInterval: 100ms

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 :

website/content/v1.12/reference/configuration/runtime/oomconfig.md

@@ -16,7 +16,7 @@ title: OOMConfig
 {{< highlight yaml >}}
 apiVersion: v1alpha1
 kind: OOMConfig
-triggerExpression: memory_full_avg10 > 12.0 && time_since_trigger > duration("500ms") # This expression defines when to trigger OOM action.
+triggerExpression: memory_full_avg10 > 12.0 && d_memory_full_avg10 > 0.0 && time_since_trigger > duration("500ms") # This expression defines when to trigger OOM action.
 cgroupRankingExpression: 'memory_max.hasValue() ? 0.0 : ({Besteffort: 1.0, Burstable: 0.5, Guaranteed: 0.0, Podruntime: 0.0, System: 0.0}[class] * double(memory_current.orValue(0u)) / double(memory_peak.orValue(0u) - memory_current.orValue(0u)))' # This expression defines how to rank cgroups for OOM handler.
 sampleInterval: 100ms # How often should the trigger expression be evaluated.
 {{< /highlight >}}