[NPM-3378] Adding eBPF-less network tracer POC

cmd/system-probe/config/adjust_npm.go

@@ -24,6 +24,8 @@ const (
 )
 
 func adjustNetwork(cfg config.Config) {
+	ebpflessEnabled := cfg.GetBool(netNS("enable_ebpf_less"))
+
 	limitMaxInt(cfg, spNS("max_conns_per_message"), maxConnsMessageBatchSize)
 
 	if cfg.GetBool(spNS("disable_tcp")) {
@@ -90,4 +92,25 @@ func adjustNetwork(cfg config.Config) {
 		log.Warn("disabling NPM connection rollups since USM connection rollups are not enabled")
 		cfg.Set(netNS("enable_connection_rollup"), false, model.SourceAgentRuntime)
 	}
+
+	// disable features that are not supported on certain
+	// configs/platforms
+	var disableConfigs []string
+	if ebpflessEnabled {
+		disableConfigs = append(disableConfigs,
+			spNS("enable_conntrack_all_namespaces"),
+			netNS("enable_protocol_classification"),
+			netNS("enable_http_monitoring"),
+			netNS("enable_https_monitoring"),
+			evNS("network_process", "enabled"),
+			netNS("enable_root_netns"),
+		)
+	}
+
+	for _, c := range disableConfigs {
+		if cfg.GetBool(c) {
+			log.Warnf("disabling %s since it is not supported for this config/platform", c)
+			cfg.Set(c, false, model.SourceAgentRuntime)
+		}
+	}
 }

cmd/system-probe/config/adjust_usm.go

@@ -9,13 +9,20 @@ import (
 	"fmt"
 
 	"github.com/DataDog/datadog-agent/pkg/config"
+	"github.com/DataDog/datadog-agent/pkg/config/model"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
 const (
 	maxHTTPFrag = 512 // matches hard limit currently imposed in NPM driver
 )
 
 func adjustUSM(cfg config.Config) {
+	if cfg.GetBool(netNS("enable_ebpf_less")) && cfg.GetBool(smNS("enabled")) {
+		log.Warn("disabling USM since ebpf-less network tracer is enabled")
+		cfg.Set(smNS("enabled"), false, model.SourceAgentRuntime)
+	}
+
 	if cfg.GetBool(smNS("enabled")) {
 		applyDefault(cfg, netNS("enable_http_monitoring"), true)
 		applyDefault(cfg, netNS("enable_https_monitoring"), true)

cmd/system-probe/modules/network_tracer_linux.go

@@ -10,14 +10,13 @@ package modules
 import (
 	"github.com/DataDog/datadog-agent/cmd/system-probe/api/module"
 	"github.com/DataDog/datadog-agent/cmd/system-probe/config"
+	"github.com/DataDog/datadog-agent/pkg/network/tracer"
 )
 
 // NetworkTracer is a factory for NPM's tracer
 var NetworkTracer = module.Factory{
 	Name:             config.NetworkTracerModule,
 	ConfigNamespaces: networkTracerModuleConfigNamespaces,
 	Fn:               createNetworkTracerModule,
-	NeedsEBPF: func() bool {
-		return true
-	},
+	NeedsEBPF:        tracer.NeedsEBPF,
 }

go.mod

@@ -1374,3 +1374,5 @@ replace (
 
 // Prevent a false-positive detection by the Google and Ikarus security vendors on VirusTotal
 exclude go.opentelemetry.io/proto/otlp v1.1.0
+
+replace github.com/google/gopacket v1.1.19 => github.com/DataDog/gopacket v0.0.0-20240626205202-4ac4cee31f14

pkg/config/setup/system_probe.go

@@ -288,6 +288,8 @@ func InitSystemProbeConfig(cfg pkgconfigmodel.Config) {
 	// connection aggregation with port rollups
 	cfg.BindEnvAndSetDefault(join(netNS, "enable_connection_rollup"), false)
 
+	cfg.BindEnvAndSetDefault(join(netNS, "enable_ebpf_less"), false)
+
 	// windows config
 	cfg.BindEnvAndSetDefault(join(spNS, "windows.enable_monotonic_count"), false)
 

pkg/ebpf/ebpftest/buildmode.go

@@ -16,13 +16,15 @@ var (
 	RuntimeCompiled BuildMode
 	CORE            BuildMode
 	Fentry          BuildMode
+	Ebpfless        BuildMode
 )
 
 func init() {
 	Prebuilt = prebuilt{}
 	RuntimeCompiled = runtimeCompiled{}
 	CORE = core{}
 	Fentry = fentry{}
+	Ebpfless = ebpfless{}
 }
 
 // BuildMode is an eBPF build mode
@@ -95,6 +97,23 @@ func (f fentry) Env() map[string]string {
 	}
 }
 
+type ebpfless struct{}
+
+func (e ebpfless) String() string {
+	return "eBPFless"
+}
+
+func (e ebpfless) Env() map[string]string {
+	return map[string]string{
+		"NETWORK_TRACER_FENTRY_TESTS":        "false",
+		"DD_ENABLE_RUNTIME_COMPILER":         "false",
+		"DD_ENABLE_CO_RE":                    "false",
+		"DD_ALLOW_RUNTIME_COMPILED_FALLBACK": "false",
+		"DD_ALLOW_PRECOMPILED_FALLBACK":      "false",
+		"DD_NETWORK_CONFIG_ENABLE_EBPF_LESS": "true",
+	}
+}
+
 // GetBuildMode returns which build mode the current environment matches, if any
 func GetBuildMode() BuildMode {
 	for _, mode := range []BuildMode{Prebuilt, RuntimeCompiled, CORE, Fentry} {

pkg/ebpf/ebpftest/buildmode_linux.go

@@ -30,6 +30,10 @@ func SupportedBuildModes() []BuildMode {
 		(runtime.GOARCH == "amd64" && (hostPlatform == "amazon" || hostPlatform == "amzn") && kv.Major() == 5 && kv.Minor() == 10) {
 		modes = append(modes, Fentry)
 	}
+	if os.Getenv("TEST_EBPF_LESS_OVERRIDE") == "true" {
+		modes = append(modes, Ebpfless)
+	}
+
 	return modes
 }
 

pkg/network/config/config.go

@@ -292,6 +292,8 @@ type Config struct {
 	// buffers (>=5.8) will result in forcing the use of Perf Maps instead.
 	EnableUSMRingBuffers bool
 
+	EnableEbpfless bool
+
 	// EnableUSMEventStream enables USM to use the event stream instead
 	// of netlink for receiving process events.
 	EnableUSMEventStream bool
@@ -393,6 +395,8 @@ func New() *Config {
 
 		EnableNPMConnectionRollup: cfg.GetBool(join(netNS, "enable_connection_rollup")),
 
+		EnableEbpfless: cfg.GetBool(join(netNS, "enable_ebpf_less")),
+
 		// Service Monitoring
 		EnableJavaTLSSupport:        cfg.GetBool(join(smjtNS, "enabled")),
 		JavaAgentDebug:              cfg.GetBool(join(smjtNS, "debug")),

pkg/network/dns/monitor_linux.go

@@ -11,8 +11,6 @@ import (
 	"fmt"
 	"math"
 
-	"golang.org/x/net/bpf"
-
 	"github.com/vishvananda/netns"
 
 	manager "github.com/DataDog/ebpf-manager"
@@ -33,6 +31,26 @@ type dnsMonitor struct {
 
 // NewReverseDNS starts snooping on DNS traffic to allow IP -> domain reverse resolution
 func NewReverseDNS(cfg *config.Config, _ telemetry.Component) (ReverseDNS, error) {
+	// Create the RAW_SOCKET inside the root network namespace
+	var (
+		packetSrc *filterpkg.AFPacketSource
+		srcErr    error
+		ns        netns.NsHandle
+	)
+	ns, err := cfg.GetRootNetNs()
+	if err != nil {
+		return nil, err
+	}
+	defer ns.Close()
+
+	err = kernel.WithNS(ns, func() error {
+		packetSrc, srcErr = filterpkg.NewPacketSource(4)
+		return srcErr
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	currKernelVersion, err := kernel.HostVersion()
 	if err != nil {
 		// if the platform couldn't be determined, treat it as new kernel case
@@ -42,12 +60,11 @@ func NewReverseDNS(cfg *config.Config, _ telemetry.Component) (ReverseDNS, error
 	pre410Kernel := currKernelVersion < kernel.VersionCode(4, 1, 0)
 
 	var p *ebpfProgram
-	var filter *manager.Probe
-	var bpfFilter []bpf.RawInstruction
-	if pre410Kernel {
-		bpfFilter, err = generateBPFFilter(cfg)
-		if err != nil {
+	if pre410Kernel || cfg.EnableEbpfless {
+		if bpfFilter, err := generateBPFFilter(cfg); err != nil {
 			return nil, fmt.Errorf("error creating bpf classic filter: %w", err)
+		} else if err = packetSrc.SetBPF(bpfFilter); err != nil {
+			return nil, fmt.Errorf("could not set BPF filter on packet source: %w", err)
 		}
 	} else {
 		p, err = newEBPFProgram(cfg)
@@ -59,35 +76,19 @@ func NewReverseDNS(cfg *config.Config, _ telemetry.Component) (ReverseDNS, error
 			return nil, fmt.Errorf("error initializing ebpf programs: %w", err)
 		}
 
-		filter, _ = p.GetProbe(manager.ProbeIdentificationPair{EBPFFuncName: probes.SocketDNSFilter, UID: probeUID})
+		filter, _ := p.GetProbe(manager.ProbeIdentificationPair{EBPFFuncName: probes.SocketDNSFilter, UID: probeUID})
 		if filter == nil {
 			return nil, fmt.Errorf("error retrieving socket filter")
 		}
-	}
-
-	// Create the RAW_SOCKET inside the root network namespace
-	var (
-		packetSrc *filterpkg.AFPacketSource
-		srcErr    error
-		ns        netns.NsHandle
-	)
-	if ns, err = cfg.GetRootNetNs(); err != nil {
-		return nil, err
-	}
-	defer ns.Close()
 
-	err = kernel.WithNS(ns, func() error {
-		packetSrc, srcErr = filterpkg.NewPacketSource(filter, bpfFilter)
-		return srcErr
-	})
-	if err != nil {
-		return nil, err
+		packetSrc.SetEbpf(filter)
 	}
 
 	snoop, err := newSocketFilterSnooper(cfg, packetSrc)
 	if err != nil {
 		return nil, err
 	}
+
 	return &dnsMonitor{
 		snoop,
 		p,

pkg/network/dns/packet_source_windows.go

@@ -31,7 +31,7 @@ func newWindowsPacketSource(telemetrycomp telemetry.Component) (packetSource, er
 	return &windowsPacketSource{di: di}, nil
 }
 
-func (p *windowsPacketSource) VisitPackets(exit <-chan struct{}, visit func([]byte, time.Time) error) error {
+func (p *windowsPacketSource) VisitPackets(exit <-chan struct{}, visit func([]byte, _ uint8, time.Time) error) error {
 	for {
 		didReadPacket, err := p.di.ReadDNSPacket(visit)
 		if err != nil {

pkg/network/dns/snooper.go

@@ -69,7 +69,7 @@ type packetSource interface {
 	// The format of the packet is dependent on the implementation of packetSource -- i.e. it may be an ethernet frame, or a IP frame.
 	// The data buffer is reused between invocations of VisitPacket and thus should not be pointed to.
 	// If the cancel channel is closed, VisitPackets will stop reading.
-	VisitPackets(cancel <-chan struct{}, visitor func(data []byte, timestamp time.Time) error) error
+	VisitPackets(cancel <-chan struct{}, visitor func(data []byte, _ uint8, timestamp time.Time) error) error
 
 	// PacketType returns the type of packet this source reads
 	PacketType() gopacket.LayerType
@@ -154,7 +154,7 @@ func (s *socketFilterSnooper) Close() {
 // The *translation is recycled and re-used in subsequent calls and it should not be accessed concurrently.
 // The second parameter `ts` is the time when the packet was captured off the wire. This is used for latency calculation
 // and much more reliable than calling time.Now() at the user layer.
-func (s *socketFilterSnooper) processPacket(data []byte, ts time.Time) error {
+func (s *socketFilterSnooper) processPacket(data []byte, _ uint8, ts time.Time) error {
 	t := s.getCachedTranslation()
 	pktInfo := dnsPacketInfo{}
 

pkg/network/dns/snooper_test.go

@@ -390,7 +390,7 @@ func TestParsingError(t *testing.T) {
 
 	reverseDNS := rdns.(*dnsMonitor)
 	// Pass a byte array of size 1 which should result in parsing error
-	err = reverseDNS.processPacket(make([]byte, 1), time.Now())
+	err = reverseDNS.processPacket(make([]byte, 1), 0, time.Now())
 	require.NoError(t, err)
 	assert.True(t, cacheTelemetry.length.Load() == 0)
 	assert.True(t, snooperTelemetry.decodingErrors.Load() == 1)

pkg/network/event_common.go

@@ -27,6 +27,9 @@ const (
 	maxByteCountChange uint64 = 375 << 30
 	// use typical small MTU size, 1300, to get max packet count
 	maxPacketCountChange uint64 = maxByteCountChange / 1300
+
+	// ConnectionByteKeyMaxLen represents the maximum size in bytes of a connection byte key
+	ConnectionByteKeyMaxLen = 41
 )
 
 // ConnectionType will be either TCP or UDP
@@ -338,6 +341,15 @@ func (c ConnectionStats) ByteKeyNAT(buf []byte) []byte {
 	return generateConnectionKey(c, buf, true)
 }
 
+// IsValid returns `true` if the connection has a valid source and dest
+// ports and IPs
+func (c ConnectionStats) IsValid() bool {
+	return c.Source.IsValid() &&
+		c.Dest.IsValid() &&
+		c.SPort > 0 &&
+		c.DPort > 0
+}
+
 const keyFmt = "p:%d|src:%s:%d|dst:%s:%d|f:%d|t:%d"
 
 // BeautifyKey returns a human readable byte key (used for debugging purposes)