Set GOMEMLIMIT on FLP (#2447)

* Set GOMEMLIMIT on FLP

(Same as done on the agent)

* Fix tests; do not set limit=0

* Make sure pods are restarted on changed args/envs

Author: jotak

internal/controller/ebpf/agent_controller.go

@@ -56,7 +56,6 @@ const (
 	envKafkaSASLIDPath            = "KAFKA_SASL_CLIENT_ID_PATH"
 	envKafkaSASLSecretPath        = "KAFKA_SASL_CLIENT_SECRET_PATH"
 	envLogLevel                   = "LOG_LEVEL"
-	envGoMemLimit                 = "GOMEMLIMIT"
 	envEnablePktDrop              = "ENABLE_PKT_DROPS"
 	envEnableDNSTracking          = "ENABLE_DNS_TRACKING"
 	envEnableFlowRTT              = "ENABLE_RTT"
@@ -730,16 +729,8 @@ func getEnvConfig(coll *flowslatest.FlowCollector, cinfo *cluster.Info) []corev1
 		})
 	}
 
-	// set GOMEMLIMIT which allows specifying a soft memory cap to force GC when resource limit is reached
-	// to prevent OOM
-	if coll.Spec.Agent.EBPF.Resources.Limits.Memory() != nil {
-		if memLimit, ok := coll.Spec.Agent.EBPF.Resources.Limits.Memory().AsInt64(); ok {
-			// we will set the GOMEMLIMIT to current memlimit - 10% as a headroom to account for
-			// memory sources the Go runtime is unaware of
-			memLimit -= int64(float64(memLimit) * 0.1)
-			config = append(config, corev1.EnvVar{Name: envGoMemLimit, Value: fmt.Sprint(memLimit)})
-		}
-	}
+	// set GOMEMLIMIT which allows specifying a soft memory cap to force GC when resource limit is reached to prevent OOM
+	config = helper.EnvFromReqsLimits(config, &coll.Spec.Agent.EBPF.Resources)
 
 	if coll.Spec.Agent.EBPF.IsPktDropEnabled() {
 		config = append(config, corev1.EnvVar{

internal/controller/ebpf/agent_controller_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 )
@@ -98,7 +99,6 @@ func TestGetEnvConfig_Default(t *testing.T) {
 
 	env := getEnvConfig(&fc, &cluster.Info{})
 	assert.Equal(t, []corev1.EnvVar{
-		{Name: "GOMEMLIMIT", Value: "0"},
 		{Name: "METRICS_ENABLE", Value: "true"},
 		{Name: "METRICS_SERVER_PORT", Value: "9400"},
 		{Name: "METRICS_PREFIX", Value: "netobserv_agent_"},
@@ -129,6 +129,9 @@ func TestGetEnvConfig_WithOverrides(t *testing.T) {
 							"TC_ATTACH_MODE":                     "any",
 						},
 					},
+					Resources: corev1.ResourceRequirements{
+						Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("800Mi")},
+					},
 					Metrics: flowslatest.EBPFMetrics{
 						Enable: ptr.To(false),
 					},
@@ -148,7 +151,7 @@ func TestGetEnvConfig_WithOverrides(t *testing.T) {
 
 	env := getEnvConfig(&fc, &cluster.Info{})
 	assert.Equal(t, []corev1.EnvVar{
-		{Name: "GOMEMLIMIT", Value: "0"},
+		{Name: "GOMEMLIMIT", Value: "754974720"},
 		{Name: "FLOW_FILTER_RULES", Value: `[{"ip_cidr":"0.0.0.0/0","action":"Accept"}]`},
 		{Name: "AGENT_IP", Value: "",
 			ValueFrom: &corev1.EnvVarSource{
@@ -177,7 +180,6 @@ func TestGetEnvConfig_OCP4_14(t *testing.T) {
 	info.Mock("4.14.5", "")
 	env := getEnvConfig(&fc, &info)
 	assert.Equal(t, []corev1.EnvVar{
-		{Name: "GOMEMLIMIT", Value: "0"},
 		{Name: "METRICS_ENABLE", Value: "true"},
 		{Name: "METRICS_SERVER_PORT", Value: "9400"},
 		{Name: "METRICS_PREFIX", Value: "netobserv_agent_"},

internal/controller/flp/flp_common_objects.go

@@ -161,6 +161,8 @@ func podTemplate(
 	}
 	envs = append(envs, constants.EnvNoHTTP2)
 
+	envs = helper.EnvFromReqsLimits(envs, &desired.Processor.Resources)
+
 	container := corev1.Container{
 		Name:            constants.FLPName,
 		Image:           imageName,

internal/controller/flp/flp_controller_test.go

@@ -255,7 +255,9 @@ func ControllerSpecs() {
 					Protocol:      "TCP",
 				}))
 				g.Expect(cnt.Env).To(Equal([]v1.EnvVar{
-					{Name: "GOGC", Value: "400"}, {Name: "GOMAXPROCS", Value: "33"}, {Name: "GODEBUG", Value: "http2server=0"},
+					{Name: "GOGC", Value: "400"},
+					{Name: "GOMAXPROCS", Value: "33"},
+					{Name: "GODEBUG", Value: "http2server=0"},
 				}))
 
 				By("Allocating the proper toleration to allow its placement in the master nodes")

internal/pkg/helper/comparators.go

@@ -130,8 +130,9 @@ func volumesChanged(old, n *corev1.PodTemplateSpec, report *ChangeReport) bool {
 func containerChanged(old, n *corev1.Container, report *ChangeReport) bool {
 	return report.Check("Image changed", n.Image != old.Image) ||
 		report.Check("Pull policy changed", n.ImagePullPolicy != old.ImagePullPolicy) ||
-		report.Check("Args changed", !deepDerivative(n.Args, old.Args)) ||
+		report.Check("Args changed", !deepEqual(n.Args, old.Args)) ||
 		report.Check("Resources req/limit changed", !deepDerivative(n.Resources, old.Resources)) ||
+		report.Check("Env changed", !deepEqual(n.Env, old.Env)) ||
 		report.Check("Liveness probe changed", probeChanged(n.LivenessProbe, old.LivenessProbe)) ||
 		report.Check("Startup probe changed", probeChanged(n.StartupProbe, old.StartupProbe))
 }

internal/pkg/helper/env.go

@@ -1,6 +1,8 @@
 package helper
 
 import (
+	"fmt"
+
 	corev1 "k8s.io/api/core/v1"
 )
 
@@ -24,3 +26,16 @@ func BuildEnvFromDefaults(config, defaults map[string]string) []corev1.EnvVar {
 	}
 	return result
 }
+
+func EnvFromReqsLimits(envs []corev1.EnvVar, reqs *corev1.ResourceRequirements) []corev1.EnvVar {
+	// set GOMEMLIMIT which allows specifying a soft memory cap to force GC when resource limit is reached to prevent OOM
+	if reqs.Limits.Memory() != nil {
+		if memLimit, ok := reqs.Limits.Memory().AsInt64(); ok && memLimit > 0 {
+			// we will set the GOMEMLIMIT to current memlimit - 10% as a headroom to account for
+			// memory sources the Go runtime is unaware of
+			memLimit -= int64(float64(memLimit) * 0.1)
+			envs = append(envs, corev1.EnvVar{Name: "GOMEMLIMIT", Value: fmt.Sprint(memLimit)})
+		}
+	}
+	return envs
+}