Merge pull request #2063 from jotak/tolerations

Amoghrd · web-flow · commit dbce6899c19f · 2025-10-02T13:56:01.000-04:00
NETOBSERV-2434: fix default tolerations, and a few other issues
diff --git a/api/flowcollector/v1beta2/flowcollector_validation_webhook.go b/api/flowcollector/v1beta2/flowcollector_validation_webhook.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"reflect"
 	"slices"
 	"strconv"
 	"strings"
@@ -214,12 +215,29 @@ func validatePortString(s string) (uint16, error) {
 }
 
 func (v *validator) validateFLP() {
+	v.validateScheduling()
 	v.validateFLPLogTypes()
 	v.validateFLPFilters()
 	v.validateFLPAlerts()
 	v.validateFLPMetricsForAlerts()
 }
 
+func (v *validator) validateScheduling() {
+	if v.fc.DeploymentModel == DeploymentModelDirect {
+		// In direct mode, agent and FLP scheduling should be consistent, to ensure the 1-1 relation
+		var agent, flp *SchedulingConfig
+		if v.fc.Agent.EBPF.Advanced != nil {
+			agent = v.fc.Agent.EBPF.Advanced.Scheduling
+		}
+		if v.fc.Processor.Advanced != nil {
+			flp = v.fc.Processor.Advanced.Scheduling
+		}
+		if !reflect.DeepEqual(agent, flp) {
+			v.warnings = append(v.warnings, "Mismatch detected between spec.agent.ebpf.advanced.scheduling and spec.processor.advanced.scheduling. In Direct mode, it can lead to inconsistent pod scheduling that would result in errors in the flow collection process.")
+		}
+	}
+}
+
 func (v *validator) validateFLPLogTypes() {
 	if v.fc.Processor.LogTypes != nil && *v.fc.Processor.LogTypes == LogTypeAll {
 		v.warnings = append(v.warnings, "Enabling all log types (in spec.processor.logTypes) has a high impact on resources footprint")
diff --git a/api/flowcollector/v1beta2/flowcollector_validation_webhook_test.go b/api/flowcollector/v1beta2/flowcollector_validation_webhook_test.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/netobserv/network-observability-operator/internal/pkg/cluster"
 	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/ptr"
@@ -728,6 +729,165 @@ func TestValidateFLP(t *testing.T) {
 	}
 }
 
+func TestValidateScheduling(t *testing.T) {
+	mismatchWarning := "Mismatch detected between spec.agent.ebpf.advanced.scheduling and spec.processor.advanced.scheduling. In Direct mode, it can lead to inconsistent pod scheduling that would result in errors in the flow collection process."
+	tests := []struct {
+		name             string
+		fc               *FlowCollector
+		expectedError    string
+		expectedWarnings admission.Warnings
+		ocpVersion       string
+	}{
+		{
+			name: "Valid default (Direct)",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					DeploymentModel: DeploymentModelDirect,
+				},
+			},
+		},
+		{
+			name: "Invalid Agent scheduling",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					DeploymentModel: DeploymentModelDirect,
+					Agent: FlowCollectorAgent{
+						EBPF: FlowCollectorEBPF{
+							Advanced: &AdvancedAgentConfig{
+								Scheduling: &SchedulingConfig{
+									Tolerations: []corev1.Toleration{{Key: "key"}},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedWarnings: admission.Warnings{mismatchWarning},
+		},
+		{
+			name: "Invalid FLP scheduling",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					DeploymentModel: DeploymentModelDirect,
+					Processor: FlowCollectorFLP{
+						Advanced: &AdvancedProcessorConfig{
+							Scheduling: &SchedulingConfig{
+								Tolerations: []corev1.Toleration{{Key: "key"}},
+							},
+						},
+					},
+				},
+			},
+			expectedWarnings: admission.Warnings{mismatchWarning},
+		},
+		{
+			name: "Invalid FLP and Agent scheduling",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					DeploymentModel: DeploymentModelDirect,
+					Agent: FlowCollectorAgent{
+						EBPF: FlowCollectorEBPF{
+							Advanced: &AdvancedAgentConfig{
+								Scheduling: &SchedulingConfig{
+									Tolerations: []corev1.Toleration{{Key: "key1"}},
+								},
+							},
+						},
+					},
+					Processor: FlowCollectorFLP{
+						Advanced: &AdvancedProcessorConfig{
+							Scheduling: &SchedulingConfig{
+								Tolerations: []corev1.Toleration{{Key: "key2"}},
+							},
+						},
+					},
+				},
+			},
+			expectedWarnings: admission.Warnings{mismatchWarning},
+		},
+		{
+			name: "Valid FLP and Agent scheduling",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					DeploymentModel: DeploymentModelDirect,
+					Agent: FlowCollectorAgent{
+						EBPF: FlowCollectorEBPF{
+							Advanced: &AdvancedAgentConfig{
+								Scheduling: &SchedulingConfig{
+									Tolerations: []corev1.Toleration{{Key: "same_key"}},
+								},
+							},
+						},
+					},
+					Processor: FlowCollectorFLP{
+						Advanced: &AdvancedProcessorConfig{
+							Scheduling: &SchedulingConfig{
+								Tolerations: []corev1.Toleration{{Key: "same_key"}},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "Valid default (Kafka)",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					DeploymentModel: DeploymentModelKafka,
+				},
+			},
+		},
+		{
+			name: "No inconsistent scheduling with Kafka",
+			fc: &FlowCollector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "cluster",
+				},
+				Spec: FlowCollectorSpec{
+					Processor: FlowCollectorFLP{
+						Advanced: &AdvancedProcessorConfig{
+							Scheduling: &SchedulingConfig{
+								Tolerations: []corev1.Toleration{{Key: "key"}},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	CurrentClusterInfo = &cluster.Info{}
+	r := FlowCollector{}
+	for _, test := range tests {
+		CurrentClusterInfo.MockOpenShiftVersion(test.ocpVersion)
+		warnings, err := r.Validate(context.TODO(), test.fc)
+		if test.expectedError == "" {
+			assert.NoError(t, err, test.name)
+		} else {
+			assert.ErrorContains(t, err, test.expectedError, test.name)
+		}
+		assert.Equal(t, test.expectedWarnings, warnings, test.name)
+	}
+}
+
 func TestElligibleMetrics(t *testing.T) {
 	met, tot := GetElligibleMetricsForAlert(AlertPacketDropsByKernel, &AlertVariant{
 		GroupBy: GroupByNamespace,
diff --git a/internal/controller/consoleplugin/consoleplugin_test.go b/internal/controller/consoleplugin/consoleplugin_test.go
@@ -191,6 +191,19 @@ func TestContainerUpdateCheck(t *testing.T) {
 	assert.Contains(report.String(), "Volumes changed")
 	old = nEw
 
+	// new toleration
+	spec.ConsolePlugin.Advanced = &flowslatest.AdvancedPluginConfig{
+		Scheduling: &flowslatest.SchedulingConfig{
+			Tolerations: []corev1.Toleration{{Key: "dummy-key", Operator: corev1.TolerationOpExists}},
+		},
+	}
+	builder = getBuilder(&spec, &loki)
+	nEw = builder.deployment(constants.PluginName, "digest")
+	report = helper.NewChangeReport("")
+	assert.True(helper.PodChanged(&old.Spec.Template, &nEw.Spec.Template, constants.PluginName, &report))
+	assert.Contains(report.String(), "Toleration changed")
+	old = nEw
+
 	// test again no change
 	loki.LokiManualParams.TLS.CACert.Name = "cm-name-2"
 	builder = getBuilder(&spec, &loki)
diff --git a/internal/controller/ebpf/agent_controller.go b/internal/controller/ebpf/agent_controller.go
@@ -460,7 +460,7 @@ func (c *AgentController) envConfig(ctx context.Context, coll *flowslatest.FlowC
 		}
 	} else {
 		config = append(config, corev1.EnvVar{Name: envExport, Value: exportGRPC})
-		advancedConfig := helper.GetAdvancedProcessorConfig(coll.Spec.Processor.Advanced)
+		advancedConfig := helper.GetAdvancedProcessorConfig(&coll.Spec)
 		// When flowlogs-pipeline is deployed as a daemonset, each agent must send
 		// data to the pod that is deployed in the same host
 		config = append(config, corev1.EnvVar{
diff --git a/internal/controller/flp/flp_common_objects.go b/internal/controller/flp/flp_common_objects.go
@@ -35,7 +35,7 @@ const (
 
 func newGRPCPipeline(desired *flowslatest.FlowCollectorSpec) config.PipelineBuilderStage {
 	return config.NewGRPCPipeline("grpc", api.IngestGRPCProto{
-		Port: int(*helper.GetAdvancedProcessorConfig(desired.Processor.Advanced).Port),
+		Port: int(*helper.GetAdvancedProcessorConfig(desired).Port),
 	})
 }
 
@@ -80,7 +80,7 @@ func podTemplate(
 	hasHostPort, hostNetwork bool,
 	annotations map[string]string,
 ) corev1.PodTemplateSpec {
-	advancedConfig := helper.GetAdvancedProcessorConfig(desired.Processor.Advanced)
+	advancedConfig := helper.GetAdvancedProcessorConfig(desired)
 	var ports []corev1.ContainerPort
 	if hasHostPort {
 		ports = []corev1.ContainerPort{{
@@ -231,7 +231,7 @@ func metricsSettings(desired *flowslatest.FlowCollectorSpec, vol *volumes.Builde
 
 func getStaticJSONConfig(desired *flowslatest.FlowCollectorSpec, vol *volumes.Builder, promTLS *flowslatest.CertificateReference, pipeline *PipelineBuilder, dynCMName string) (string, error) {
 	metricsSettings := metricsSettings(desired, vol, promTLS)
-	advancedConfig := helper.GetAdvancedProcessorConfig(desired.Processor.Advanced)
+	advancedConfig := helper.GetAdvancedProcessorConfig(desired)
 	config := map[string]interface{}{
 		"log-level": desired.Processor.LogLevel,
 		"health": map[string]interface{}{
diff --git a/internal/controller/flp/flp_pipeline_builder.go b/internal/controller/flp/flp_pipeline_builder.go
@@ -553,7 +553,7 @@ func (b *PipelineBuilder) addConnectionTracking(lastStage config.PipelineBuilder
 	// Connection tracking stage (only if LogTypes is not FLOWS)
 	if b.desired.Processor.HasConntrack() {
 		outputRecordTypes := helper.GetRecordTypes(&b.desired.Processor)
-		advancedConfig := helper.GetAdvancedProcessorConfig(b.desired.Processor.Advanced)
+		advancedConfig := helper.GetAdvancedProcessorConfig(b.desired)
 		lastStage = lastStage.ConnTrack("extract_conntrack", api.ConnTrack{
 			KeyDefinition: api.KeyDefinition{
 				FieldGroups: []api.FieldGroup{
diff --git a/internal/controller/flp/flp_test.go b/internal/controller/flp/flp_test.go
@@ -794,3 +794,23 @@ func TestLabels(t *testing.T) {
 	assert.Equal("flowlogs-pipeline-transformer-monitor", smTrans.Name)
 	assert.Equal("flowlogs-pipeline-transformer", smTrans.Spec.Selector.MatchLabels["app"])
 }
+
+func TestToleration(t *testing.T) {
+	assert := assert.New(t)
+
+	cfg := getConfig()
+	cfgKafka := cfg
+	cfgKafka.DeploymentModel = flowslatest.DeploymentModelKafka
+	info := reconcilers.Common{Namespace: "ns", ClusterInfo: &cluster.Info{}}
+	builder, _ := newMonolithBuilder(info.NewInstance(image, status.Instance{}), &cfg, &metricslatest.FlowMetricList{}, nil)
+	tBuilder, _ := newTransfoBuilder(info.NewInstance(image, status.Instance{}), &cfgKafka, &metricslatest.FlowMetricList{}, nil)
+
+	// Deployment: no specific toleration
+	depl := tBuilder.deployment(annotate("digest"))
+	assert.Len(depl.Spec.Template.Spec.Tolerations, 0)
+
+	// DaemonSet: has toleration
+	ds := builder.daemonSet(annotate("digest"))
+	assert.Len(ds.Spec.Template.Spec.Tolerations, 1)
+	assert.Equal(corev1.Toleration{Operator: "Exists"}, ds.Spec.Template.Spec.Tolerations[0])
+}
diff --git a/internal/pkg/helper/comparators.go b/internal/pkg/helper/comparators.go
@@ -102,6 +102,12 @@ func assignationChanged(old, n *corev1.PodTemplateSpec, report *ChangeReport) bo
 		}
 		return true
 	}
+	if !deepEqual(n.Spec.Tolerations, old.Spec.Tolerations) {
+		if report != nil {
+			report.Add("Toleration changed")
+		}
+		return true
+	}
 	if !deepDerivative(n.Spec.Affinity, old.Spec.Affinity) {
 		if report != nil {
 			report.Add("Affinity changed")
diff --git a/internal/pkg/helper/flowcollector.go b/internal/pkg/helper/flowcollector.go
@@ -132,10 +132,10 @@ func GetAdvancedAgentConfig(specConfig *flowslatest.AdvancedAgentConfig) flowsla
 			cfg.Env = specConfig.Env
 		}
 		if specConfig.Scheduling != nil {
-			if len(specConfig.Scheduling.NodeSelector) > 0 {
+			if specConfig.Scheduling.NodeSelector != nil {
 				cfg.Scheduling.NodeSelector = specConfig.Scheduling.NodeSelector
 			}
-			if len(specConfig.Scheduling.Tolerations) > 0 {
+			if specConfig.Scheduling.Tolerations != nil {
 				cfg.Scheduling.Tolerations = specConfig.Scheduling.Tolerations
 			}
 			if specConfig.Scheduling.Affinity != nil {
@@ -150,7 +150,11 @@ func GetAdvancedAgentConfig(specConfig *flowslatest.AdvancedAgentConfig) flowsla
 	return cfg
 }
 
-func GetAdvancedProcessorConfig(specConfig *flowslatest.AdvancedProcessorConfig) flowslatest.AdvancedProcessorConfig {
+func GetAdvancedProcessorConfig(spec *flowslatest.FlowCollectorSpec) flowslatest.AdvancedProcessorConfig {
+	var defaultToleration []corev1.Toleration
+	if !UseKafka(spec) { // TODO: with coming-soon Service deploymentModel, this should be replaced with "UseHostNetwork" (conflict should pop up while rebasing).
+		defaultToleration = []corev1.Toleration{{Operator: corev1.TolerationOpExists}}
+	}
 	cfg := flowslatest.AdvancedProcessorConfig{
 		Env:                            map[string]string{},
 		Port:                           ptr.To(GetFieldDefaultInt32(ProcessorAdvancedPath, "port")),
@@ -162,12 +166,13 @@ func GetAdvancedProcessorConfig(specConfig *flowslatest.AdvancedProcessorConfig)
 		ConversationTerminatingTimeout: ptr.To(GetFieldDefaultDuration(ProcessorAdvancedPath, "conversationTerminatingTimeout")),
 		Scheduling: &flowslatest.SchedulingConfig{
 			NodeSelector:      map[string]string{},
-			Tolerations:       []corev1.Toleration{{Operator: corev1.TolerationOpExists}},
+			Tolerations:       defaultToleration,
 			Affinity:          nil,
 			PriorityClassName: "",
 		},
 	}
 
+	specConfig := spec.Processor.Advanced
 	if specConfig != nil {
 		if len(specConfig.Env) > 0 {
 			cfg.Env = specConfig.Env
@@ -197,10 +202,10 @@ func GetAdvancedProcessorConfig(specConfig *flowslatest.AdvancedProcessorConfig)
 			cfg.ConversationTerminatingTimeout = specConfig.ConversationTerminatingTimeout
 		}
 		if specConfig.Scheduling != nil {
-			if len(specConfig.Scheduling.NodeSelector) > 0 {
+			if specConfig.Scheduling.NodeSelector != nil {
 				cfg.Scheduling.NodeSelector = specConfig.Scheduling.NodeSelector
 			}
-			if len(specConfig.Scheduling.Tolerations) > 0 {
+			if specConfig.Scheduling.Tolerations != nil {
 				cfg.Scheduling.Tolerations = specConfig.Scheduling.Tolerations
 			}
 			if specConfig.Scheduling.Affinity != nil {
@@ -249,7 +254,7 @@ func GetAdvancedPluginConfig(specConfig *flowslatest.AdvancedPluginConfig) flows
 		Port:     ptr.To(GetFieldDefaultInt32(PluginAdvancedPath, "port")),
 		Scheduling: &flowslatest.SchedulingConfig{
 			NodeSelector:      map[string]string{},
-			Tolerations:       []corev1.Toleration{{Operator: corev1.TolerationOpExists}},
+			Tolerations:       nil,
 			Affinity:          nil,
 			PriorityClassName: "",
 		},
@@ -269,10 +274,10 @@ func GetAdvancedPluginConfig(specConfig *flowslatest.AdvancedPluginConfig) flows
 			cfg.Port = specConfig.Port
 		}
 		if specConfig.Scheduling != nil {
-			if len(specConfig.Scheduling.NodeSelector) > 0 {
+			if specConfig.Scheduling.NodeSelector != nil {
 				cfg.Scheduling.NodeSelector = specConfig.Scheduling.NodeSelector
 			}
-			if len(specConfig.Scheduling.Tolerations) > 0 {
+			if specConfig.Scheduling.Tolerations != nil {
 				cfg.Scheduling.Tolerations = specConfig.Scheduling.Tolerations
 			}
 			if specConfig.Scheduling.Affinity != nil {

Original file line number	Diff line number	Diff line change
`@@ -460,7 +460,7 @@ func (c AgentController) envConfig(ctx context.Context, coll flowslatest.FlowC`
`460`	`460`	`}`
`461`	`461`	`} else {`
`462`	`462`	`config = append(config, corev1.EnvVar{Name: envExport, Value: exportGRPC})`
`463`		`- advancedConfig := helper.GetAdvancedProcessorConfig(coll.Spec.Processor.Advanced)`
	`463`	`+ advancedConfig := helper.GetAdvancedProcessorConfig(&coll.Spec)`
`464`	`464`	`// When flowlogs-pipeline is deployed as a daemonset, each agent must send`
`465`	`465`	`// data to the pod that is deployed in the same host`
`466`	`466`	`config = append(config, corev1.EnvVar{`
Original file line number	Diff line number	Diff line change
`@@ -102,6 +102,12 @@ func assignationChanged(old, n corev1.PodTemplateSpec, report ChangeReport) bo`
`102`	`102`	`}`
`103`	`103`	`return true`
`104`	`104`	`}`
	`105`	`+ if !deepEqual(n.Spec.Tolerations, old.Spec.Tolerations) {`
	`106`	`+ if report != nil {`
	`107`	`+ report.Add("Toleration changed")`
	`108`	`+ }`
	`109`	`+ return true`
	`110`	`+ }`
`105`	`111`	`if !deepDerivative(n.Spec.Affinity, old.Spec.Affinity) {`
`106`	`112`	`if report != nil {`
`107`	`113`	`report.Add("Affinity changed")`