[CWS] Add configurable RT scheduling for ring buffer reader thread

go.mod

@@ -169,7 +169,7 @@ require (
 	github.com/DataDog/datadog-go/v5 v5.8.3
 	github.com/DataDog/datadog-operator/api v0.0.0-20260323152500-0887e50ccf73
 	github.com/DataDog/datadog-traceroute v1.0.13
-	github.com/DataDog/ebpf-manager v0.7.16
+	github.com/DataDog/ebpf-manager v0.7.17-0.20260402145614-14db7084c2aa
 	github.com/DataDog/go-sqllexer v0.2.1
 	github.com/DataDog/nikos v1.12.12
 	github.com/DataDog/sketches-go v1.4.8

pkg/config/setup/system_probe.go

@@ -306,6 +306,8 @@ func InitSystemProbeConfig(cfg pkgconfigmodel.Setup) {
 	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.event_stream.use_kprobe_fallback", true)
 	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.event_stream.buffer_size", 0)
 	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.event_stream.kretprobe_max_active", 512)
+	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.event_stream.scheduling_policy", "")
+	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.event_stream.scheduling_priority", 0)
 	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.envs_with_value", []string{"LD_PRELOAD", "LD_LIBRARY_PATH", "PATH", "HISTSIZE", "HISTFILESIZE", "GLIBC_TUNABLES", "SSH_CLIENT", "DD_SERVICE", "OTEL_SERVICE_NAME", "CLAUDECODE", "RUNNER_TRACKING_ID"})
 	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.runtime_compilation.enabled", false)
 	eventMonitorBindEnvAndSetDefault(cfg, "event_monitoring_config.network.enabled", true)

pkg/security/probe/config/config.go

@@ -7,6 +7,7 @@
 package config
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"strings"
@@ -97,6 +98,14 @@ type Config struct {
 	// EventStreamKretprobeMaxActive specifies the maximum number of active kretprobe at a given time
 	EventStreamKretprobeMaxActive int
 
+	// EventStreamSchedulingPolicy specifies the realtime scheduling policy for the ring buffer reader thread.
+	// Supported values: "", "SCHED_FIFO", "SCHED_RR". Empty means the scheduling policy is not changed (default).
+	EventStreamSchedulingPolicy string
+
+	// EventStreamSchedulingPriority specifies the realtime scheduling priority (1-99).
+	// Required when EventStreamSchedulingPolicy is set.
+	EventStreamSchedulingPriority int
+
 	// RuntimeCompilationEnabled defines if the runtime-compilation is enabled
 	RuntimeCompilationEnabled bool
 
@@ -209,6 +218,8 @@ func NewConfig() (*Config, error) {
 		EventStreamUseFentry:               getBool("event_stream.use_fentry"),
 		EventStreamUseKprobeFallback:       getBool("event_stream.use_kprobe_fallback"),
 		EventStreamKretprobeMaxActive:      getInt("event_stream.kretprobe_max_active"),
+		EventStreamSchedulingPolicy:        getString("event_stream.scheduling_policy"),
+		EventStreamSchedulingPriority:      getInt("event_stream.scheduling_priority"),
 
 		EnvsWithValue:               getStringSlice("envs_with_value"),
 		NetworkEnabled:              getBool("network.enabled"),
@@ -272,6 +283,20 @@ func (c *Config) sanitize() error {
 		return fmt.Errorf("runtime_security_config.event_stream.buffer_size must be a power of 2 and a multiple of %d", os.Getpagesize())
 	}
 
+	switch c.EventStreamSchedulingPolicy {
+	case "", "SCHED_FIFO", "SCHED_RR":
+		// valid
+	default:
+		return fmt.Errorf("invalid event_monitoring_config.event_stream.scheduling_policy %q: must be empty, \"SCHED_FIFO\", or \"SCHED_RR\"", c.EventStreamSchedulingPolicy)
+	}
+	if c.EventStreamSchedulingPolicy != "" {
+		if c.EventStreamSchedulingPriority < 1 || c.EventStreamSchedulingPriority > 99 {
+			return fmt.Errorf("event_monitoring_config.event_stream.scheduling_priority must be between 1 and 99 when scheduling_policy is set, got %d", c.EventStreamSchedulingPriority)
+		}
+	} else if c.EventStreamSchedulingPriority != 0 {
+		return errors.New("event_monitoring_config.event_stream.scheduling_priority is set but event_monitoring_config.event_stream.scheduling_policy is empty")
+	}
+
 	if !isConfigured("enable_approvers") && c.EnableKernelFilters {
 		c.EnableApprovers = true
 	}

pkg/security/probe/config/config_test.go

@@ -0,0 +1,89 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build linux
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSanitizeSchedulingConfig(t *testing.T) {
+	tests := []struct {
+		name     string
+		policy   string
+		priority int
+		wantErr  string
+	}{
+		{
+			name:     "disabled by default",
+			policy:   "",
+			priority: 0,
+		},
+		{
+			name:     "valid SCHED_FIFO",
+			policy:   "SCHED_FIFO",
+			priority: 10,
+		},
+		{
+			name:     "valid SCHED_RR",
+			policy:   "SCHED_RR",
+			priority: 1,
+		},
+		{
+			name:     "valid SCHED_FIFO max priority",
+			policy:   "SCHED_FIFO",
+			priority: 99,
+		},
+		{
+			name:     "invalid policy string",
+			policy:   "SCHED_OTHER",
+			priority: 10,
+			wantErr:  "invalid event_monitoring_config.event_stream.scheduling_policy",
+		},
+		{
+			name:     "policy set without priority",
+			policy:   "SCHED_FIFO",
+			priority: 0,
+			wantErr:  "event_monitoring_config.event_stream.scheduling_priority must be between 1 and 99",
+		},
+		{
+			name:     "priority too high",
+			policy:   "SCHED_FIFO",
+			priority: 100,
+			wantErr:  "event_monitoring_config.event_stream.scheduling_priority must be between 1 and 99",
+		},
+		{
+			name:     "priority negative",
+			policy:   "SCHED_FIFO",
+			priority: -1,
+			wantErr:  "event_monitoring_config.event_stream.scheduling_priority must be between 1 and 99",
+		},
+		{
+			name:     "priority set without policy",
+			policy:   "",
+			priority: 10,
+			wantErr:  "event_monitoring_config.event_stream.scheduling_priority is set but event_monitoring_config.event_stream.scheduling_policy is empty",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &Config{
+				EventStreamSchedulingPolicy:   tt.policy,
+				EventStreamSchedulingPriority: tt.priority,
+			}
+			err := c.sanitize()
+			if tt.wantErr != "" {
+				assert.ErrorContains(t, err, tt.wantErr)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

pkg/security/probe/eventstream/ringbuffer/ringbuffer.go

@@ -14,13 +14,26 @@ import (
 
 	manager "github.com/DataDog/ebpf-manager"
 	"github.com/cilium/ebpf/ringbuf"
+	"golang.org/x/sys/unix"
 
 	ebpfTelemetry "github.com/DataDog/datadog-agent/pkg/ebpf/telemetry"
 	"github.com/DataDog/datadog-agent/pkg/security/probe/config"
 	"github.com/DataDog/datadog-agent/pkg/security/probe/eventstream"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
 	ddsync "github.com/DataDog/datadog-agent/pkg/util/sync"
 )
 
+func parseSchedPolicy(policy string) (int, error) {
+	switch policy {
+	case "SCHED_FIFO":
+		return unix.SCHED_FIFO, nil
+	case "SCHED_RR":
+		return unix.SCHED_RR, nil
+	default:
+		return 0, fmt.Errorf("unsupported scheduling policy: %q", policy)
+	}
+}
+
 // RingBuffer implements the EventStream interface
 // using an eBPF map of type BPF_MAP_TYPE_RINGBUF
 type RingBuffer struct {
@@ -44,6 +57,26 @@ func (rb *RingBuffer) Init(mgr *manager.Manager, config *config.Config) error {
 		TelemetryEnabled: config.InternalTelemetryEnabled,
 	}
 
+	if config.EventStreamSchedulingPolicy != "" {
+		schedPolicy, err := parseSchedPolicy(config.EventStreamSchedulingPolicy)
+		if err != nil {
+			return err
+		}
+		rb.ringBuffer.RingBufferOptions.SchedPolicy = schedPolicy
+		rb.ringBuffer.RingBufferOptions.SchedPriority = config.EventStreamSchedulingPriority
+
+		errChan := make(chan error, 1)
+		rb.ringBuffer.RingBufferOptions.ErrChan = errChan
+		go func() {
+			for err := range errChan {
+				log.Warnf("ring buffer reader error: %v", err)
+			}
+		}()
+
+		log.Infof("ring buffer reader configured with %s priority %d",
+			config.EventStreamSchedulingPolicy, config.EventStreamSchedulingPriority)
+	}
+
 	ebpfTelemetry.ReportRingBufferTelemetry(rb.ringBuffer)
 	return nil
 }