[WP] Add metric for dropped packets

Author: theop-dd

pkg/security/ebpf/c/include/maps.h

@@ -98,6 +98,7 @@ BPF_LRU_MAP(dns_responses_sent_to_userspace, u16, struct dns_responses_sent_to_u
 BPF_LRU_MAP(capabilities_usage, struct capabilities_usage_key_t, struct capabilities_usage_entry_t, 1) // max entries will be overridden at runtime
 BPF_LRU_MAP(sock_cookie_pid, u64, u32, 1); // max entries will be overridden at runtime
 BPF_LRU_MAP(memfd_tracking, struct memfd_key_t, u32, 1024)
+BPF_LRU_MAP(dropped_packets, u64, u64, 512)
 
 BPF_LRU_MAP_FLAGS(tasks_in_coredump, u64, u8, 64, BPF_F_NO_COMMON_LRU)
 BPF_LRU_MAP_FLAGS(syscalls, u64, struct syscall_cache_t, 1, BPF_F_NO_COMMON_LRU) // max entries will be overridden at runtime

pkg/security/ebpf/probes/rawpacket/bpffilter.go

@@ -37,6 +37,9 @@ const (
 	PolicyDrop
 )
 
+// MaxDropActionFilters is the maximum number of network drop action filters tracked in kernel.
+const MaxDropActionFilters = 512
+
 // ToTCAct converts a policy to a TCAct
 func (p Policy) ToTCAct() TCAct {
 	switch p {

pkg/security/ebpf/probes/rawpacket/pcap.go

@@ -39,6 +39,9 @@ const (
 
 	// payload size
 	structRawPacketEventDataSize = 256
+
+	dropStatsKeyStackOffset = int16(-8)
+	dropStatsValStackOffset = int16(-16)
 )
 
 // ProgOpts defines options
@@ -60,6 +63,7 @@ type ProgOpts struct {
 	ctxSaveReg            asm.Register
 	tailCallMapFd         int
 	hasGetCurrentCgroupId bool
+	dropStatsMapFd        int
 }
 
 // DefaultProgOpts default options
@@ -97,6 +101,60 @@ func (opts *ProgOpts) WithGetCurrentCgroupID(hasGetCurrentCgroupId bool) *ProgOp
 	return opts
 }
 
+// WithDropStatsMapFd sets the map fd used to count dropped packets per filter index.
+func (opts *ProgOpts) WithDropStatsMapFd(fd int) *ProgOpts {
+	opts.dropStatsMapFd = fd
+	return opts
+}
+
+func dropStatsIncrementInsts(filterIndex int, dropStatsMapFd int, nextLabel string) asm.Instructions {
+	incLabel := fmt.Sprintf("inc_drop_stat_%d", filterIndex)
+	initLabel := fmt.Sprintf("init_drop_stat_%d", filterIndex)
+
+	return asm.Instructions{
+		asm.Mov.Reg(asm.R1, asm.RFP).WithSymbol(incLabel),
+		asm.Add.Imm(asm.R1, int32(dropStatsKeyStackOffset)),
+		asm.Mov.Imm(asm.R2, int32(filterIndex)),
+		asm.StoreMem(asm.R1, 0, asm.R2, asm.DWord),
+		asm.StoreImm(asm.R1, 4, 0, asm.DWord),
+
+		asm.LoadMapPtr(asm.R1, dropStatsMapFd),
+		asm.Mov.Reg(asm.R2, asm.RFP),
+		asm.Add.Imm(asm.R2, int32(dropStatsKeyStackOffset)),
+		asm.FnMapLookupElem.Call(),
+		asm.JEq.Imm(asm.R0, 0, initLabel),
+
+		asm.Mov.Reg(asm.R5, asm.R0),
+		asm.LoadMem(asm.R6, asm.R5, 0, asm.DWord),
+		asm.Add.Imm(asm.R6, 1),
+		asm.StoreMem(asm.R5, 0, asm.R6, asm.DWord),
+
+		asm.Instruction{
+			OpCode:   asm.Ja.Op(asm.ImmSource),
+			Constant: 0,
+		}.WithReference(nextLabel),
+
+		asm.Mov.Reg(asm.R3, asm.RFP).WithSymbol(initLabel),
+		asm.Add.Imm(asm.R3, int32(dropStatsValStackOffset)),
+		asm.Mov.Imm(asm.R4, 1),
+		asm.StoreMem(asm.R3, 0, asm.R4, asm.DWord),
+		asm.StoreImm(asm.R3, 4, 0, asm.DWord),
+
+		asm.LoadMapPtr(asm.R1, dropStatsMapFd),
+		asm.Mov.Reg(asm.R2, asm.RFP),
+		asm.Add.Imm(asm.R2, int32(dropStatsKeyStackOffset)),
+		asm.Mov.Reg(asm.R3, asm.RFP),
+		asm.Add.Imm(asm.R3, int32(dropStatsValStackOffset)),
+		asm.Mov.Imm(asm.R4, 0),
+		asm.FnMapUpdateElem.Call(),
+
+		asm.Instruction{
+			OpCode:   asm.Ja.Op(asm.ImmSource),
+			Constant: 0,
+		}.WithReference(nextLabel),
+	}
+}
+
 // FilterToInsts compile a bpf filter expression
 func FilterToInsts(index int, filter Filter, opts ProgOpts) (asm.Instructions, error) {
 	pcapBPF, err := pcap.CompileBPFFilter(layers.LinkTypeEthernet, 256, filter.BPFFilter)
@@ -133,6 +191,10 @@ func FilterToInsts(index int, filter Filter, opts ProgOpts) (asm.Instructions, e
 	}
 
 	mismatchLabel := fmt.Sprintf("mismatch_%d_", index)
+	matchLabel := opts.onMatchLabel
+	if opts.dropStatsMapFd != 0 {
+		matchLabel = fmt.Sprintf("inc_drop_stat_%d", index)
+	}
 
 	if filter.Pid != 0 {
 		insts = append(insts,
@@ -142,7 +204,7 @@ func FilterToInsts(index int, filter Filter, opts ProgOpts) (asm.Instructions, e
 			// check the pid
 			// load the pid from the packet
 			asm.LoadMem(asm.R7, opts.eventPtrReg, structRawPacketEventPidOffset, asm.Word),
-			asm.JEq.Imm(asm.R7, int32(filter.Pid), opts.onMatchLabel),
+			asm.JEq.Imm(asm.R7, int32(filter.Pid), matchLabel),
 			asm.Mov.Imm(asm.R4, 0).WithSymbol(mismatchLabel), // nop instruction, just hold the symbol
 		)
 	} else if !filter.CGroupPathKey.IsNull() {
@@ -169,14 +231,19 @@ func FilterToInsts(index int, filter Filter, opts ProgOpts) (asm.Instructions, e
 
 			// check the cgroup id
 			asm.LoadImm(asm.R4, int64(filter.CGroupPathKey.Inode), asm.DWord),
-			asm.JEq.Reg(asm.R7, asm.R4, opts.onMatchLabel),
+			asm.JEq.Reg(asm.R7, asm.R4, matchLabel),
 			asm.Mov.Imm(asm.R4, 0).WithSymbol(mismatchLabel), // nop instruction, just hold the symbol
 		)
 	} else {
 		insts = append(insts,
-			asm.JNE.Imm(cbpfcOpts.Result, 0, opts.onMatchLabel).WithSymbol(resultLabel),
+			asm.JNE.Imm(cbpfcOpts.Result, 0, matchLabel).WithSymbol(resultLabel),
 		)
 	}
+
+	if opts.dropStatsMapFd != 0 {
+		insts = append(insts, dropStatsIncrementInsts(index, opts.dropStatsMapFd, opts.onMatchLabel)...)
+	}
+
 	return insts, nil
 }
 

pkg/security/ebpf/probes/rawpacket/pcap_unsupported.go

@@ -40,6 +40,11 @@ func (opts *ProgOpts) WithGetCurrentCgroupID(_ bool) *ProgOpts {
 	return opts
 }
 
+// WithDropStatsMapFd sets the map fd used to count dropped packets per filter index.
+func (opts *ProgOpts) WithDropStatsMapFd(_ int) *ProgOpts {
+	return opts
+}
+
 // FilterToInsts compile a bpf filter expression
 func FilterToInsts(_ int, _ Filter, _ ProgOpts) (asm.Instructions, error) {
 	return asm.Instructions{}, errors.New("not supported")

pkg/security/metrics/metrics.go

@@ -579,6 +579,10 @@ var (
 	// MetricEventSampleSampled is the name of the metric used to report events that were sampled in kernel
 	// Tags: event_type
 	MetricEventSampleSampled = newRuntimeMetric(".event_sample.sampled")
+
+	// MetricRawPacketDropped is the name of the metric used to count packets dropped by network_filter actions
+	// Tags: rule_id
+	MetricRawPacketDropped = newRuntimeMetric(".network.raw_packet.dropped")
 )
 
 var (

pkg/security/probe/BUILD.bazel

@@ -96,6 +96,7 @@ go_library(
             "//pkg/security/probe/monitors/discarder",
             "//pkg/security/probe/monitors/dns",
             "//pkg/security/probe/monitors/eventsample",
+            "//pkg/security/probe/monitors/rawpacketdrop",
             "//pkg/security/probe/monitors/syscalls",
             "//pkg/security/probe/procfs",
             "//pkg/security/probe/sysctl",
@@ -213,6 +214,7 @@ go_library(
             "//pkg/security/probe/monitors/discarder",
             "//pkg/security/probe/monitors/dns",
             "//pkg/security/probe/monitors/eventsample",
+            "//pkg/security/probe/monitors/rawpacketdrop",
             "//pkg/security/probe/monitors/syscalls",
             "//pkg/security/probe/procfs",
             "//pkg/security/probe/sysctl",

pkg/security/probe/monitors/rawpacketdrop/BUILD.bazel

@@ -0,0 +1,25 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "rawpacketdrop",
+    srcs = ["rawpacketdrop_monitor.go"],
+    importpath = "github.com/DataDog/datadog-agent/pkg/security/probe/monitors/rawpacketdrop",
+    visibility = ["//visibility:public"],
+    deps = select({
+        "@rules_go//go/platform:android": [
+            "//pkg/security/metrics",
+            "//pkg/security/probe/managerhelper",
+            "@com_github_cilium_ebpf//:ebpf",
+            "@com_github_datadog_datadog_go_v5//statsd",
+            "@com_github_datadog_ebpf_manager//:ebpf-manager",
+        ],
+        "@rules_go//go/platform:linux": [
+            "//pkg/security/metrics",
+            "//pkg/security/probe/managerhelper",
+            "@com_github_cilium_ebpf//:ebpf",
+            "@com_github_datadog_datadog_go_v5//statsd",
+            "@com_github_datadog_ebpf_manager//:ebpf-manager",
+        ],
+        "//conditions:default": [],
+    }),
+)

pkg/security/probe/monitors/rawpacketdrop/rawpacketdrop_monitor.go

@@ -0,0 +1,91 @@
+// 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 rawpacketdrop holds raw packet drop monitor related files
+package rawpacketdrop
+
+import (
+	"fmt"
+
+	manager "github.com/DataDog/ebpf-manager"
+	lib "github.com/cilium/ebpf"
+
+	"github.com/DataDog/datadog-go/v5/statsd"
+
+	"github.com/DataDog/datadog-agent/pkg/security/metrics"
+	"github.com/DataDog/datadog-agent/pkg/security/probe/managerhelper"
+)
+
+// RuleIDsProvider returns the current filter index to rule_id mapping.
+type RuleIDsProvider func() map[uint32]string
+
+// Monitor reports kernel-side raw packet drop counters grouped by rule_id.
+type Monitor struct {
+	statsdClient statsd.ClientInterface
+	droppedMap   *lib.Map
+	ruleIDs      RuleIDsProvider
+	lastCounts   map[string]uint64
+}
+
+// NewMonitor returns a new Monitor.
+func NewMonitor(manager *manager.Manager, statsdClient statsd.ClientInterface, ruleIDs RuleIDsProvider) (*Monitor, error) {
+	droppedMap, err := managerhelper.Map(manager, "dropped_packets")
+	if err != nil {
+		return nil, err
+	}
+
+	return &Monitor{
+		statsdClient: statsdClient,
+		droppedMap:   droppedMap,
+		ruleIDs:      ruleIDs,
+		lastCounts:   make(map[string]uint64),
+	}, nil
+}
+
+// SendStats emits deltas from the kernel dropped_packets map grouped by rule_id.
+func (m *Monitor) SendStats() error {
+	ruleIDs := m.ruleIDs()
+	if len(ruleIDs) == 0 {
+		m.lastCounts = make(map[string]uint64)
+		return nil
+	}
+
+	currentCounts := make(map[string]uint64, len(ruleIDs))
+	iterator := m.droppedMap.Iterate()
+
+	var filterIndex uint64
+	var count uint64
+	for iterator.Next(&filterIndex, &count) {
+		ruleID, ok := ruleIDs[uint32(filterIndex)]
+		if !ok || ruleID == "" {
+			continue
+		}
+		currentCounts[ruleID] += count
+	}
+
+	for ruleID, count := range currentCounts {
+		last := m.lastCounts[ruleID]
+		if count <= last {
+			continue
+		}
+
+		delta := count - last
+		tags := []string{"rule_id:" + ruleID}
+		if err := m.statsdClient.Count(metrics.MetricRawPacketDropped, int64(delta), tags, 1.0); err != nil {
+			return fmt.Errorf("failed to send raw packet dropped metric: %w", err)
+		}
+		m.lastCounts[ruleID] = count
+	}
+
+	for ruleID := range m.lastCounts {
+		if _, ok := currentCounts[ruleID]; !ok {
+			delete(m.lastCounts, ruleID)
+		}
+	}
+
+	return nil
+}

pkg/security/probe/probe_ebpf.go

@@ -187,6 +187,8 @@ type EBPFProbe struct {
 
 	// raw packet filter for actions
 	rawPacketActionFilters []rawpacket.Filter
+	dropActionRuleIDs      map[uint32]string
+	dropActionRuleIDsLock  sync.RWMutex
 
 	// remediation tracking
 	activeRemediations     map[string]*Remediation
@@ -783,10 +785,17 @@ func (p *EBPFProbe) setupRawPacketFiltersOnNewRuleset(rs *rules.RuleSet) error {
 func (p *EBPFProbe) applyRawPacketActionFilters(applyFromRuleset bool) error {
 	// TODO check cgroupv2
 
+	p.rebuildDropActionRuleIDs()
+
 	opts := rawpacket.DefaultProgOpts()
 	opts.WithProgPrefix("raw_packet_drop_action_")
 	opts.WithGetCurrentCgroupID(p.kernelVersion.HasBpfGetCurrentPidTgidForSchedCLS())
 
+	if droppedPacketsMap, err := managerhelper.Map(p.Manager, "dropped_packets"); err == nil {
+		p.clearDroppedPacketsMap(droppedPacketsMap)
+		opts.WithDropStatsMapFd(droppedPacketsMap.FD())
+	}
+
 	// adapt max instruction limits depending of the kernel version
 	if p.kernelVersion.Code >= kernel.Kernel5_2 {
 		opts.MaxProgSize = 1_000_000
@@ -831,6 +840,40 @@ func (p *EBPFProbe) addRawPacketActionFilter(actionFilter rawpacket.Filter) erro
 	return p.applyRawPacketActionFilters(false)
 }
 
+func (p *EBPFProbe) rebuildDropActionRuleIDs() {
+	ruleIDs := make(map[uint32]string, len(p.rawPacketActionFilters))
+	for i, filter := range p.rawPacketActionFilters {
+		if i >= rawpacket.MaxDropActionFilters {
+			break
+		}
+		ruleIDs[uint32(i)] = string(filter.RuleID)
+	}
+
+	p.dropActionRuleIDsLock.Lock()
+	p.dropActionRuleIDs = ruleIDs
+	p.dropActionRuleIDsLock.Unlock()
+}
+
+func (p *EBPFProbe) getDropActionRuleIDs() map[uint32]string {
+	p.dropActionRuleIDsLock.RLock()
+	defer p.dropActionRuleIDsLock.RUnlock()
+
+	out := make(map[uint32]string, len(p.dropActionRuleIDs))
+	for index, ruleID := range p.dropActionRuleIDs {
+		out[index] = ruleID
+	}
+	return out
+}
+
+func (p *EBPFProbe) clearDroppedPacketsMap(droppedPacketsMap *lib.Map) {
+	iterator := droppedPacketsMap.Iterate()
+	var key uint64
+	var value uint64
+	for iterator.Next(&key, &value) {
+		_ = droppedPacketsMap.Delete(key)
+	}
+}
+
 // Start the probe
 func (p *EBPFProbe) Start() error {
 	// Apply rules to the already stored data before starting the event stream to avoid concurrency issues
@@ -3177,6 +3220,7 @@ func NewEBPFProbe(probe *Probe, config *config.Config, hostname string, opts Opt
 		MetricNameTruncated:  atomic.NewUint64(0),
 		activeRemediations:   make(map[string]*Remediation),
 		pid:                  utils.Getpid(),
+		dropActionRuleIDs:    make(map[uint32]string),
 	}
 
 	p.onNewPCE = func(pce *model.ProcessCacheEntry, err error) {