diff --git a/containers/container.go b/containers/container.go index d25a4a5..6a87b19 100644 --- a/containers/container.go +++ b/containers/container.go @@ -877,7 +877,7 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst, actualDst // This is used when TCP connection tracking fails (common for Go TLS due to goroutine thread switching) // but we have socket tuple info extracted directly from the fd func (c *Container) createConnectionFromSocketInfo(pid uint32, fd uint64, timestamp uint64, socketInfo *ebpftracer.SocketInfo) (conn *ActiveConnection, filtered bool) { - connection, filtered := c.connectionFromSocketInfo(pid, fd, timestamp, socketInfo) + connection, filtered := c.connectionFromSocketInfo(pid, fd, timestamp, socketInfo, true) if connection == nil { return nil, filtered } @@ -898,8 +898,9 @@ func (c *Container) createConnectionFromSocketInfo(pid uint32, fd uint64, timest // connectionFromSocketInfo builds the connection an L7 event's socket tuple // describes, without tracking it. filtered is true for a connection the agent -// does not track. -func (c *Container) connectionFromSocketInfo(pid uint32, fd uint64, timestamp uint64, socketInfo *ebpftracer.SocketInfo) (conn *ActiveConnection, filtered bool) { +// does not track. tcp is false for a socket that may be UDP: the kernel's +// address translations are recorded for TCP connections only. +func (c *Container) connectionFromSocketInfo(pid uint32, fd uint64, timestamp uint64, socketInfo *ebpftracer.SocketInfo, tcp bool) (conn *ActiveConnection, filtered bool) { if socketInfo == nil || !socketInfo.Valid { return nil, false } @@ -928,10 +929,13 @@ func (c *Container) connectionFromSocketInfo(pid uint32, fd uint64, timestamp ui // The socket holds the address the application connected to, before any // NAT: a service's ClusterIP, not the pod behind it. The kernel records the // translation per local address from conntrack, as the open event's - // actual destination does. + // actual destination does. The table is keyed by local address alone, so + // for a UDP socket it can only hold an earlier TCP connection's, from + // whatever socket last had the same local port: DNS queries were labelled + // with that connection's server. var actualDst netaddr.IPPort - if c.registry.tracer != nil { - actualDst, _ = c.registry.tracer.ActualDestination(src) + if tcp { + actualDst, _ = c.registry.actualDestination(src) } // Same filters and labels as a connection seen opening: without them, // traffic to ignored destinations was tracked through this path alone. @@ -1254,8 +1258,10 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // fd, usually the connection to the address just resolved. For the // same reason an entry on the fd that is not this socket is an // earlier socket's. The query takes its connection from its own tuple. + // The tuple does not say UDP or TCP, and DNS over TCP that was not + // seen connecting is rare enough to go without its translation. var filtered bool - if conn, filtered = c.connectionFromSocketInfo(pid, fd, timestamp, socketInfo); conn == nil { + if conn, filtered = c.connectionFromSocketInfo(pid, fd, timestamp, socketInfo, false); conn == nil { if !filtered { dropL7Event(c.id, "unknown_connection", pid, fd, r, socketInfo) } diff --git a/containers/registry.go b/containers/registry.go index 4366c70..3ff73f2 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -73,6 +73,10 @@ type Registry struct { tracer *ebpftracer.Tracer events chan ebpftracer.Event + // actualDestination is the tracer's post-NAT destination lookup for a + // TCP connection's local address (ebpftracer.Tracer.ActualDestination). + // NewRegistry always sets it. + actualDestination func(src netaddr.IPPort) (netaddr.IPPort, bool) containersById map[ContainerID]*Container containersByCgroupId map[string]*Container @@ -161,6 +165,7 @@ func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, proces gpuProcessUsageSampleChan: gpuProcessUsageSampleChan, nodeConstLabels: NodeConstLabels{MachineID: machineId, SystemUUID: systemUuid, AZ: az, Region: region}, } + r.actualDestination = r.tracer.ActualDestination // Register LLM metrics with the same registerer used for other container metrics RegisterL7SelfMetrics(reg) if *flags.EnableLLMCapture { diff --git a/containers/socket_connection_test.go b/containers/socket_connection_test.go index 3cd0f51..3595d2e 100644 --- a/containers/socket_connection_test.go +++ b/containers/socket_connection_test.go @@ -41,7 +41,10 @@ func newSocketTestContainer(t *testing.T, names map[string]string) *Container { *flags.IgnoreControlPlane = "loki,victoria" t.Cleanup(func() { *flags.IgnoreControlPlane = prev }) return &Container{ - registry: &Registry{ip2fqdn: common.NewFQDNCache()}, + registry: &Registry{ + ip2fqdn: common.NewFQDNCache(), + actualDestination: func(netaddr.IPPort) (netaddr.IPPort, bool) { return netaddr.IPPort{}, false }, + }, ip_resolver: stubResolver{names: names}, processes: map[uint32]*Process{}, connectionsByPidFd: map[PidFd]*ActiveConnection{}, @@ -180,7 +183,7 @@ func TestDNSResolverKeepsItsNameAcrossReusedFd(t *testing.T) { if d := c.registry.getDomain(netaddr.MustParseIP(resolver)); d != nil { t.Errorf("resolver %s named %q", resolver, d.FQDN) } - if got := c.dnsDestinations(t); len(got) != 1 || !got[resolver+":53"] { + if got := c.dnsLabelValues(t, "destination"); len(got) != 1 || !got[resolver+":53"] { t.Errorf("DNS destinations = %v, want only %s:53", got, resolver) } } @@ -207,13 +210,37 @@ func TestEarlierSocketOnFdIsNotUsed(t *testing.T) { t.Errorf("earlier connection renamed to %q", got) } - c.connectionsByPidFd[PidFd{Pid: 1, Fd: 7}], _ = c.connectionFromSocketInfo(1, 7, 100, socketInfo("10.0.0.2", 40001, "203.0.113.9", 443)) + c.connectionsByPidFd[PidFd{Pid: 1, Fd: 7}], _ = c.connectionFromSocketInfo(1, 7, 100, socketInfo("10.0.0.2", 40001, "203.0.113.9", 443), true) c.handleL7(t, 1, 7, 0, dnsResponse(t, "www.example.org", "198.51.100.7"), socketInfo("10.0.0.2", 40002, "192.0.2.53", 53)) - if got := c.dnsDestinations(t); len(got) != 1 || !got["192.0.2.53:53"] { + if got := c.dnsLabelValues(t, "destination"); len(got) != 1 || !got["192.0.2.53:53"] { t.Errorf("DNS destinations = %v, want only 192.0.2.53:53", got) } } +// The kernel's address translations are recorded for TCP connections only, +// keyed by local address. A UDP query whose local port an earlier TCP +// connection had used found that connection's server there, and was labelled +// with it: one more set of series for every server the container had reached. +// A TCP connection built from its tuple still takes its translation. +func TestDNSQueryTakesNoTCPTranslation(t *testing.T) { + c := newL7TestContainer(t) + tcpServer := netaddr.MustParseIPPort("198.51.100.80:8080") + c.registry.actualDestination = func(netaddr.IPPort) (netaddr.IPPort, bool) { return tcpServer, true } + + c.handleL7(t, 1, 7, 0, dnsResponse(t, "api.example.com", "203.0.113.9"), socketInfo("10.0.0.2", 40000, "192.0.2.53", 53)) + if got := c.dnsLabelValues(t, "actual_destination"); len(got) != 1 || !got["192.0.2.53:53"] { + t.Errorf("DNS actual destinations = %v, want only 192.0.2.53:53", got) + } + + conn, _ := c.createConnectionFromSocketInfo(1, 8, 100, socketInfo("10.0.0.2", 40001, "192.0.2.80", 80)) + if conn == nil { + t.Fatal("no connection for the TCP socket") + } + if got := conn.DestinationKey.ActualDestinationLabelValue(); got != tcpServer.String() { + t.Errorf("TCP actual destination = %q, want %s", got, tcpServer) + } +} + func newL7TestContainer(t *testing.T) *Container { t.Helper() // Without a parsed command line no public network is tracked. @@ -242,8 +269,8 @@ func (c *Container) handleL7(t *testing.T, pid uint32, fd uint64, ts uint64, r * return ip2fqdn } -// dnsDestinations returns the destination label values of the DNS counter. -func (c *Container) dnsDestinations(t *testing.T) map[string]bool { +// dnsLabelValues returns the values of label name on the DNS counter. +func (c *Container) dnsLabelValues(t *testing.T, name string) map[string]bool { t.Helper() ch := make(chan prometheus.Metric, 100) c.l7Stats.requests[l7.ProtocolDNS].Collect(ch) @@ -255,7 +282,7 @@ func (c *Container) dnsDestinations(t *testing.T) map[string]bool { t.Fatal(err) } for _, l := range pb.GetLabel() { - if l.GetName() == "destination" { + if l.GetName() == name { got[l.GetValue()] = true } }