in buffer batching with perf buffers for NPM

.github/CODEOWNERS

@@ -456,10 +456,12 @@
 /pkg/util/crio/                         @DataDog/container-integrations
 /pkg/util/docker/                       @DataDog/container-integrations
 /pkg/util/ecs/                          @DataDog/container-integrations
+/pkg/util/encoding/                     @DataDog/ebpf-platform
 /pkg/util/funcs/                        @DataDog/ebpf-platform
 /pkg/util/gpu/                          @DataDog/container-platform
 /pkg/util/kernel/                       @DataDog/ebpf-platform
 /pkg/util/safeelf/                      @DataDog/ebpf-platform
+/pkg/util/slices/                       @DataDog/ebpf-platform
 /pkg/util/ktime                         @DataDog/agent-security
 /pkg/util/kubernetes/                   @DataDog/container-integrations @DataDog/container-platform @DataDog/container-app
 /pkg/util/podman/                       @DataDog/container-integrations

cmd/system-probe/config/adjust_npm.go

@@ -26,6 +26,10 @@ const (
 func adjustNetwork(cfg model.Config) {
 	ebpflessEnabled := cfg.GetBool(netNS("enable_ebpfless"))
 
+	deprecateInt(cfg, spNS("closed_connection_flush_threshold"), netNS("closed_connection_flush_threshold"))
+	deprecateInt(cfg, spNS("closed_channel_size"), netNS("closed_channel_size"))
+	applyDefault(cfg, netNS("closed_channel_size"), 500)
+
 	limitMaxInt(cfg, spNS("max_conns_per_message"), maxConnsMessageBatchSize)
 
 	if cfg.GetBool(spNS("disable_tcp")) {

pkg/config/setup/system_probe.go

@@ -195,8 +195,11 @@ func InitSystemProbeConfig(cfg pkgconfigmodel.Config) {
 	cfg.BindEnvAndSetDefault(join(spNS, "max_tracked_connections"), 65536)
 	cfg.BindEnv(join(spNS, "max_closed_connections_buffered"))
 	cfg.BindEnv(join(netNS, "max_failed_connections_buffered"))
-	cfg.BindEnvAndSetDefault(join(spNS, "closed_connection_flush_threshold"), 0)
-	cfg.BindEnvAndSetDefault(join(spNS, "closed_channel_size"), 500)
+	cfg.BindEnv(join(spNS, "closed_connection_flush_threshold"))
+	cfg.BindEnv(join(netNS, "closed_connection_flush_threshold"))
+	cfg.BindEnv(join(spNS, "closed_channel_size"))
+	cfg.BindEnv(join(netNS, "closed_channel_size"))
+	cfg.BindEnvAndSetDefault(join(netNS, "closed_buffer_wakeup_count"), 4)
 	cfg.BindEnvAndSetDefault(join(spNS, "max_connection_state_buffered"), 75000)
 
 	cfg.BindEnvAndSetDefault(join(spNS, "disable_dns_inspection"), false, "DD_DISABLE_DNS_INSPECTION")
@@ -212,6 +215,7 @@ func InitSystemProbeConfig(cfg pkgconfigmodel.Config) {
 	cfg.BindEnvAndSetDefault(join(spNS, "enable_conntrack_all_namespaces"), true, "DD_SYSTEM_PROBE_ENABLE_CONNTRACK_ALL_NAMESPACES")
 	cfg.BindEnvAndSetDefault(join(netNS, "enable_protocol_classification"), true, "DD_ENABLE_PROTOCOL_CLASSIFICATION")
 	cfg.BindEnvAndSetDefault(join(netNS, "enable_ringbuffers"), true, "DD_SYSTEM_PROBE_NETWORK_ENABLE_RINGBUFFERS")
+	cfg.BindEnvAndSetDefault(join(netNS, "enable_custom_batching"), false, "DD_SYSTEM_PROBE_NETWORK_ENABLE_CUSTOM_BATCHING")
 	cfg.BindEnvAndSetDefault(join(netNS, "enable_tcp_failed_connections"), true, "DD_SYSTEM_PROBE_NETWORK_ENABLE_FAILED_CONNS")
 	cfg.BindEnvAndSetDefault(join(netNS, "ignore_conntrack_init_failure"), false, "DD_SYSTEM_PROBE_NETWORK_IGNORE_CONNTRACK_INIT_FAILURE")
 	cfg.BindEnvAndSetDefault(join(netNS, "conntrack_init_timeout"), 10*time.Second)

pkg/ebpf/c/bpf_helpers_custom.h

@@ -39,4 +39,17 @@ unsigned long long load_half(void *skb,
 unsigned long long load_word(void *skb,
     unsigned long long off) asm("llvm.bpf.load.word");
 
+// declare our own versions of these enums, because they don't exist on <5.8
+enum {
+	DD_BPF_RB_NO_WAKEUP = 1,
+	DD_BPF_RB_FORCE_WAKEUP = 2,
+};
+
+enum {
+	DD_BPF_RB_AVAIL_DATA = 0,
+	DD_BPF_RB_RING_SIZE = 1,
+	DD_BPF_RB_CONS_POS = 2,
+	DD_BPF_RB_PROD_POS = 3,
+};
+
 #endif

pkg/ebpf/manager.go

@@ -100,3 +100,40 @@ func (m *Manager) InitWithOptions(bytecode io.ReaderAt, opts *manager.Options) e
 	}
 	return nil
 }
+
+type modifierPreStart interface {
+	PreStart() error
+}
+
+// Start is a wrapper around ebpf-manager.Manager.Start
+func (m *Manager) Start() error {
+	for _, mod := range m.EnabledModifiers {
+		if ps, ok := mod.(modifierPreStart); ok {
+			if err := ps.PreStart(); err != nil {
+				return fmt.Errorf("prestart %s manager modifier: %w", mod, err)
+			}
+		}
+	}
+	return m.Manager.Start()
+}
+
+type modifierAfterStop interface {
+	AfterStop(manager.MapCleanupType) error
+}
+
+// Stop is a wrapper around ebpf-manager.Manager.Stop
+func (m *Manager) Stop(ct manager.MapCleanupType) error {
+	if err := m.Manager.Stop(ct); err != nil {
+		return err
+	}
+
+	for _, mod := range m.EnabledModifiers {
+		if as, ok := mod.(modifierAfterStop); ok {
+			if err := as.AfterStop(ct); err != nil {
+				return fmt.Errorf("afterstop %s manager modifier: %w", mod, err)
+			}
+		}
+	}
+
+	return nil
+}