feat(network): attribute proxy sockets to the pod owning their netns

cmd/system-probe/modules/network_tracer.go

@@ -55,6 +55,10 @@ func createNetworkTracerModule(_ *sysconfigtypes.Config, deps module.FactoryDepe
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
+	if err := enableNetNSContainerAttribution(ctx, t, deps); err != nil {
+		log.Warnf("network namespace container attribution disabled: %s", err)
+	}
+
 	var connsSender sender.Sender
 	if ncfg.DirectSend {
 		connsSender, err = sender.New(ctx, t, sender.Dependencies{

cmd/system-probe/modules/network_tracer_darwin.go

@@ -8,6 +8,9 @@
 package modules
 
 import (
+	"context"
+
+	"github.com/DataDog/datadog-agent/pkg/network/tracer"
 	"github.com/DataDog/datadog-agent/pkg/system-probe/api/module"
 	"github.com/DataDog/datadog-agent/pkg/system-probe/config"
 )
@@ -26,3 +29,7 @@ func (nt *networkTracer) platformRegister(_ *module.Router) error {
 	// No platform-specific endpoints for Darwin yet
 	return nil
 }
+
+func enableNetNSContainerAttribution(_ context.Context, _ *tracer.Tracer, _ module.FactoryDependencies) error {
+	return nil
+}

cmd/system-probe/modules/network_tracer_linux.go

@@ -57,3 +57,7 @@ func getNetworkID(ctx context.Context) (string, error) {
 	}
 	return id, nil
 }
+
+func enableNetNSContainerAttribution(ctx context.Context, t *tracer.Tracer, deps module.FactoryDependencies) error {
+	return t.EnableNetNSContainerAttribution(ctx, deps.WMeta)
+}

cmd/system-probe/modules/network_tracer_windows.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	httpprotocol "github.com/DataDog/datadog-agent/pkg/network/protocols/http"
+	"github.com/DataDog/datadog-agent/pkg/network/tracer"
 	"github.com/DataDog/datadog-agent/pkg/system-probe/api/module"
 	"github.com/DataDog/datadog-agent/pkg/system-probe/config"
 	"github.com/DataDog/datadog-agent/pkg/system-probe/utils"
@@ -75,3 +76,7 @@ func (nt *networkTracer) platformRegister(httpMux *module.Router) error {
 
 	return nil
 }
+
+func enableNetNSContainerAttribution(_ context.Context, _ *tracer.Tracer, _ module.FactoryDependencies) error {
+	return nil
+}

pkg/config/schema/yaml/system-probe_schema.yaml

@@ -683,6 +683,10 @@ properties:
         default: false
         env_vars:
         - DD_SYSTEM_PROBE_NETWORK_ENABLE_HTTPS_MONITORING
+      enable_netns_container_attribution:
+        node_type: setting
+        type: boolean
+        default: false
       enable_protocol_classification:
         node_type: setting
         type: boolean
@@ -750,6 +754,13 @@ properties:
         node_type: setting
         type: integer
         default: 1024
+      netns_container_attribution_owner_images:
+        node_type: setting
+        type: array
+        default:
+        - ztunnel
+        items:
+          type: string
   service_monitoring_config:
     $ref: system-probe-usm.yaml
   ping:

pkg/network/config/config.go

@@ -190,6 +190,15 @@ type Config struct {
 	// for things like creating netlink sockets for conntrack updates, etc.
 	EnableRootNetNs bool
 
+	// EnableNetNSContainerAttribution attributes a socket to the container that owns its network
+	// namespace, rather than the container of its owning process, when the owner is a proxy that
+	// creates sockets inside other pods (for example Istio ambient's ztunnel). Requires EnableProcessEventMonitoring.
+	EnableNetNSContainerAttribution bool
+
+	// NetNSContainerAttributionOwnerImages lists the short image names of the owning containers
+	// whose sockets may be reattributed.
+	NetNSContainerAttributionOwnerImages []string
+
 	// ProtocolClassificationEnabled specifies whether the tracer should enhance connection data with protocols names by
 	// classifying the L7 protocols being used.
 	ProtocolClassificationEnabled bool
@@ -302,6 +311,9 @@ func New() *Config {
 
 		EnableRootNetNs: cfg.GetBool(sysconfig.FullKeyPath(netNS, "enable_root_netns")),
 
+		EnableNetNSContainerAttribution:      cfg.GetBool(sysconfig.FullKeyPath(netNS, "enable_netns_container_attribution")),
+		NetNSContainerAttributionOwnerImages: cfg.GetStringSlice(sysconfig.FullKeyPath(netNS, "netns_container_attribution_owner_images")),
+
 		EnableNPMConnectionRollup: cfg.GetBool(sysconfig.FullKeyPath(netNS, "enable_connection_rollup")),
 
 		EnableCORETracer: cfg.GetBool(sysconfig.FullKeyPath(netNS, "enable_co_re")),

pkg/network/event_common.go

@@ -280,8 +280,11 @@ type ConnectionStats struct {
 	ContainerID   struct {
 		Source, Dest *intern.Value
 	}
-	CertInfo unique.Handle[CertInfo]
-	DNSStats map[dns.Hostname]map[dns.QueryType]dns.Stats
+	// NetNSOriginalContainerID is the container of the owning process when ContainerID.Source was
+	// reattributed to the container owning the socket's network namespace; nil otherwise.
+	NetNSOriginalContainerID *intern.Value
+	CertInfo                 unique.Handle[CertInfo]
+	DNSStats                 map[dns.Hostname]map[dns.QueryType]dns.Stats
 	// TCPFailures stores the number of failures for a POSIX error code
 	TCPFailures map[uint16]uint32
 

pkg/network/sender/encode.go

@@ -147,12 +147,17 @@ func (d *directSender) encodeContainerForPID(connsChunk []network.ConnectionStat
 	// every connection with a ContainerID has a unique PID
 	writtenPids := make(map[uint32]struct{}, cidCount)
 	for _, conn := range connsChunk {
-		if conn.ContainerID.Source != nil {
+		// the PID's own container, not the one a reattributed socket was moved to
+		cid := conn.ContainerID.Source
+		if conn.NetNSOriginalContainerID != nil {
+			cid = conn.NetNSOriginalContainerID
+		}
+		if cid != nil {
 			if _, ok := writtenPids[conn.Pid]; !ok {
 				writtenPids[conn.Pid] = struct{}{}
 				builder.AddContainerForPid(func(w *model.CollectorConnections_ContainerForPidEntryBuilder) {
 					w.SetKey(int32(conn.Pid))
-					w.SetValue(getInternedString(conn.ContainerID.Source))
+					w.SetValue(getInternedString(cid))
 				})
 			}
 		}

pkg/network/sender/sender_test.go

@@ -693,3 +693,38 @@ func TestCollectLoopIgnoresPendingTickAfterCancel(t *testing.T) {
 
 	require.Zero(t, source.calls.Load(), "collectLoop collected after the context was cancelled")
 }
+
+func TestNetworkConnectionNetNSReattributed(t *testing.T) {
+	// two sockets owned by the same proxy process: one in its own container, one that the
+	// tracer attributed to the pod owning the socket's network namespace
+	p := makeConnections(2)
+	p[1].Pid = p[0].Pid
+	p[1].SPort++
+	proxyCID := p[0].ContainerID.Source
+	p[1].ContainerID.Source = intern.GetByString("app-container")
+	p[1].NetNSOriginalContainerID = proxyCID
+
+	fakeTagger := taggerfxmock.SetupFakeTagger(t)
+	fakeTagger.SetTags(taggertypes.NewEntityID(taggertypes.Process, strconv.Itoa(int(p[0].Pid))), "process", nil, nil, []string{"service:ztunnel"}, nil)
+
+	d := mockDirectSender(t, nil)
+	d.maxConnsPerMessage = 10
+	d.tagger = fakeTagger
+	payloads := slices.Collect(d.batches(&network.Connections{BufferedData: network.BufferedData{Conns: p}}, 1))
+	require.Len(t, payloads, 1)
+	m, err := model.DecodeMessage(payloads[0])
+	require.NoError(t, err)
+	cc, ok := m.Body.(*model.CollectorConnections)
+	require.True(t, ok)
+
+	assert.Equal(t, proxyCID.Get().(string), cc.ContainerForPid[int32(p[0].Pid)], "the PID must keep mapping to the process's own container")
+	require.Len(t, cc.Connections, 2)
+	for _, c := range cc.Connections {
+		tags := cc.GetConnectionsTags(c.TagsIdx)
+		if c.Laddr.ContainerId == "app-container" {
+			assert.NotContains(t, tags, "service:ztunnel", "the proxy's process tags must not move onto the pod")
+		} else {
+			assert.Contains(t, tags, "service:ztunnel")
+		}
+	}
+}

pkg/network/sender/tags.go

@@ -104,19 +104,26 @@ func (d *directSender) addTags(builder *model.ConnectionBuilder, nc network.Conn
 
 	tagsStr := tagsSet.Subset(tagIndexes)
 	if nc.Pid > 0 {
+		// a socket reattributed to its network namespace's container is still owned by the proxy
+		// process, so the PID's service context and process tags describe the proxy, not the pod
+		reattributed := nc.NetNSOriginalContainerID != nil
 		var serviceTags []string
 		if dsc := directSenderConsumerInstance.Load(); dsc != nil {
-			serviceTags = dsc.extractor.GetServiceContext(int32(nc.Pid))
+			if !reattributed {
+				serviceTags = dsc.extractor.GetServiceContext(int32(nc.Pid))
+			}
 			if processName := dsc.processNameExtractor.GetProcessName(int32(nc.Pid)); processName != "" {
 				tagsStr = append(tagsStr, "process_name:"+processName)
 			}
 		}
 		tagsStr = append(tagsStr, serviceTags...)
-		processEntityID := types.NewEntityID(types.Process, strconv.Itoa(int(nc.Pid)))
-		if processTags, err := d.tagger.Tag(processEntityID, types.HighCardinality); err != nil {
-			log.Debugf("error getting tags for process %v: %v", nc.Pid, err)
-		} else {
-			tagsStr = append(tagsStr, processTags...)
+		if !reattributed {
+			processEntityID := types.NewEntityID(types.Process, strconv.Itoa(int(nc.Pid)))
+			if processTags, err := d.tagger.Tag(processEntityID, types.HighCardinality); err != nil {
+				log.Debugf("error getting tags for process %v: %v", nc.Pid, err)
+			} else {
+				tagsStr = append(tagsStr, processTags...)
+			}
 		}
 	}
 	if len(tagsStr) > 0 {

pkg/network/tracer/BUILD.bazel

@@ -12,6 +12,7 @@ go_library(
         "ebpf_conntracker.go",
         "interfaceclassifier_other.go",
         "interfaceclassifier_windows.go",
+        "netns_containers.go",
         "process_cache.go",
         "process_cache_linux.go",
         "process_cache_windows.go",
@@ -33,6 +34,7 @@ go_library(
         "@rules_go//go/platform:android": [
             "//comp/core/telemetry/def",
             "//comp/core/telemetry/impl",
+            "//comp/core/workloadmeta/def",
             "//pkg/config/setup",
             "//pkg/ebpf",
             "//pkg/ebpf/bytecode",
@@ -110,6 +112,7 @@ go_library(
         "@rules_go//go/platform:linux": [
             "//comp/core/telemetry/def",
             "//comp/core/telemetry/impl",
+            "//comp/core/workloadmeta/def",
             "//pkg/config/setup",
             "//pkg/ebpf",
             "//pkg/ebpf/bytecode",
@@ -195,6 +198,7 @@ dd_agent_go_test(
         "ebpf_conntracker_test.go",
         "interfaceclassifier_windows_test.go",
         "mock_gateway_lookup_test.go",
+        "netns_containers_test.go",
         "offsetguess_test.go",
         "process_cache_test.go",
         "tracer_darwin_test.go",
@@ -216,6 +220,7 @@ dd_agent_go_test(
     ] + select({
         "@rules_go//go/platform:android": [
             "//comp/core/telemetry/impl",
+            "//comp/core/workloadmeta/def",
             "//pkg/config/env",
             "//pkg/config/mock",
             "//pkg/ebpf",
@@ -293,6 +298,7 @@ dd_agent_go_test(
         ],
         "@rules_go//go/platform:linux": [
             "//comp/core/telemetry/impl",
+            "//comp/core/workloadmeta/def",
             "//pkg/config/env",
             "//pkg/config/mock",
             "//pkg/ebpf",