Merge pull request #8 from coroot/ebpf_l7

Capturing L7 protocols at the eBPF level

Author: apetruhin

.dockerignore

@@ -0,0 +1,2 @@
+**/.git/
+**/.idea/

containers/container.go

@@ -3,6 +3,7 @@ package containers
 import (
 	"github.com/coroot/coroot-node-agent/cgroup"
 	"github.com/coroot/coroot-node-agent/common"
+	"github.com/coroot/coroot-node-agent/ebpftracer"
 	"github.com/coroot/coroot-node-agent/flags"
 	"github.com/coroot/coroot-node-agent/logs"
 	"github.com/coroot/coroot-node-agent/node"
@@ -14,6 +15,7 @@ import (
 	"inet.af/netaddr"
 	"k8s.io/klog/v2"
 	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -57,6 +59,17 @@ type AddrPair struct {
 	dst netaddr.IPPort
 }
 
+type ActiveConnection struct {
+	ActualDest netaddr.IPPort
+	Pid        uint32
+	Fd         uint64
+}
+
+type L7Stats struct {
+	Requests *prometheus.CounterVec
+	Latency  prometheus.Histogram
+}
+
 type Container struct {
 	cgroup   *cgroup.Cgroup
 	metadata *ContainerMetadata
@@ -73,11 +86,13 @@ type Container struct {
 
 	listens map[netaddr.IPPort]map[uint32]time.Time // listen addr -> pid -> close time
 
-	connectsSuccessful map[AddrPair]int             // dst:actual_dst -> count
-	connectsFailed     map[netaddr.IPPort]int       // dst -> count
+	connectsSuccessful map[AddrPair]int64           // dst:actual_dst -> count
+	connectsFailed     map[netaddr.IPPort]int64     // dst -> count
 	connectLastAttempt map[netaddr.IPPort]time.Time // dst -> time
-	connectionsActive  map[AddrPair]netaddr.IPPort  // src:dst -> actual_dst
-	retransmits        map[AddrPair]int             // dst:actual_dst -> count
+	connectionsActive  map[AddrPair]ActiveConnection
+	retransmits        map[AddrPair]int64 // dst:actual_dst -> count
+
+	l7Stats map[ebpftracer.L7Protocol]map[AddrPair]*L7Stats // protocol -> dst:actual_dst -> stats
 
 	oomKills int
 
@@ -101,11 +116,12 @@ func NewContainer(cg *cgroup.Cgroup, md *ContainerMetadata) *Container {
 
 		listens: map[netaddr.IPPort]map[uint32]time.Time{},
 
-		connectsSuccessful: map[AddrPair]int{},
-		connectsFailed:     map[netaddr.IPPort]int{},
+		connectsSuccessful: map[AddrPair]int64{},
+		connectsFailed:     map[netaddr.IPPort]int64{},
 		connectLastAttempt: map[netaddr.IPPort]time.Time{},
-		connectionsActive:  map[AddrPair]netaddr.IPPort{},
-		retransmits:        map[AddrPair]int{},
+		connectionsActive:  map[AddrPair]ActiveConnection{},
+		retransmits:        map[AddrPair]int64{},
+		l7Stats:            map[ebpftracer.L7Protocol]map[AddrPair]*L7Stats{},
 
 		mountIds: map[string]struct{}{},
 
@@ -147,6 +163,12 @@ func (c *Container) Describe(ch chan<- *prometheus.Desc) {
 	for _, m := range metricsList {
 		ch <- m
 	}
+	for _, protoStats := range c.l7Stats {
+		for _, s := range protoStats {
+			s.Requests.Describe(ch)
+			s.Latency.Describe(ch)
+		}
+	}
 }
 
 func (c *Container) Collect(ch chan<- prometheus.Metric) {
@@ -242,8 +264,8 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
 	}
 
 	connections := map[AddrPair]int{}
-	for c, actualDst := range c.connectionsActive {
-		connections[AddrPair{src: c.dst, dst: actualDst}]++
+	for c, conn := range c.connectionsActive {
+		connections[AddrPair{src: c.dst, dst: conn.ActualDest}]++
 	}
 	for d, count := range connections {
 		ch <- gauge(metrics.NetConnectionsActive, float64(count), d.src.String(), d.dst.String())
@@ -271,7 +293,14 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) {
 		ch <- gauge(metrics.ApplicationType, 1, appType)
 	}
 
-	if !*flags.NoPingUpstreams {
+	for _, protoStats := range c.l7Stats {
+		for _, s := range protoStats {
+			s.Requests.Collect(ch)
+			s.Latency.Collect(ch)
+		}
+	}
+
+	if !*flags.DisablePinger {
 		for ip, rtt := range c.ping(netNs) {
 			ch <- gauge(metrics.NetLatency, rtt, ip.String())
 		}
@@ -316,7 +345,7 @@ func (c *Container) onProcessExit(pid uint32, oomKill bool) {
 	}
 }
 
-func (c *Container) onFileOpen(pid uint32, fd uint32) {
+func (c *Container) onFileOpen(pid uint32, fd uint64) {
 	mntId, logPath := resolveFd(pid, fd)
 	c.lock.Lock()
 	defer c.lock.Unlock()
@@ -349,7 +378,7 @@ func (c *Container) onListenClose(pid uint32, addr netaddr.IPPort) {
 	}
 }
 
-func (c *Container) onConnectionOpen(pid uint32, src, dst netaddr.IPPort, failed bool) {
+func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPPort, failed bool) {
 	if dst.IP().IsLoopback() {
 		netNs, err := proc.GetNetNs(pid)
 		isHostNs := err == nil && hostNetNsId == netNs.UniqueId()
@@ -376,7 +405,11 @@ func (c *Container) onConnectionOpen(pid uint32, src, dst netaddr.IPPort, failed
 	} else {
 		actualDst := ConntrackGetActualDestination(src, dst)
 		c.connectsSuccessful[AddrPair{src: dst, dst: actualDst}]++
-		c.connectionsActive[AddrPair{src: src, dst: dst}] = actualDst
+		c.connectionsActive[AddrPair{src: src, dst: dst}] = ActiveConnection{
+			ActualDest: actualDst,
+			Pid:        pid,
+			Fd:         fd,
+		}
 	}
 	c.connectLastAttempt[dst] = time.Now()
 }
@@ -391,14 +424,65 @@ func (c *Container) onConnectionClose(srcDst AddrPair) bool {
 	return true
 }
 
+func (c *Container) onL7Request(pid uint32, fd uint64, r *ebpftracer.L7Request) {
+	for dest, conn := range c.connectionsActive {
+		if conn.Pid == pid && conn.Fd == fd {
+			key := AddrPair{src: dest.dst, dst: conn.ActualDest}
+			stats := c.l7Stats[r.Protocol]
+			if stats == nil {
+				stats = map[AddrPair]*L7Stats{}
+				c.l7Stats[r.Protocol] = stats
+			}
+			s := stats[key]
+			if s == nil {
+				constLabels := map[string]string{"destination": key.src.String(), "actual_destination": key.dst.String()}
+				cOpts, ok := L7Requests[r.Protocol]
+				if !ok {
+					klog.Warningln("cannot find metric description for L7 protocol: %s", r.Protocol.String())
+					return
+				}
+				hOpts, ok := L7Latency[r.Protocol]
+				if !ok {
+					klog.Warningln("cannot find metric description for L7 protocol: %s", r.Protocol.String())
+					return
+				}
+				s = &L7Stats{
+					Requests: prometheus.NewCounterVec(
+						prometheus.CounterOpts{Name: cOpts.Name, Help: cOpts.Help, ConstLabels: constLabels},
+						[]string{"status"},
+					),
+					Latency: prometheus.NewHistogram(
+						prometheus.HistogramOpts{Name: hOpts.Name, Help: hOpts.Help, ConstLabels: constLabels},
+					),
+				}
+				stats[key] = s
+			}
+			status := ""
+			switch r.Protocol {
+			case ebpftracer.L7ProtocolHTTP:
+				status = strconv.Itoa(r.Status)
+			default:
+				if r.Status == 500 {
+					status = "failed"
+				} else {
+					status = "ok"
+				}
+			}
+			s.Requests.WithLabelValues(status).Inc()
+			s.Latency.Observe(r.Duration.Seconds())
+			return
+		}
+	}
+}
+
 func (c *Container) onRetransmit(srcDst AddrPair) bool {
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	actualDst, ok := c.connectionsActive[srcDst]
+	conn, ok := c.connectionsActive[srcDst]
 	if !ok {
 		return false
 	}
-	c.retransmits[AddrPair{src: srcDst.dst, dst: actualDst}]++
+	c.retransmits[AddrPair{src: srcDst.dst, dst: conn.ActualDest}]++
 	return true
 }
 
@@ -566,7 +650,7 @@ func (c *Container) ping(netNs netns.NsHandle) map[netaddr.IP]float64 {
 }
 
 func (c *Container) runLogParser(logPath string) {
-	if *flags.NoParseLogs {
+	if *flags.DisableLogParsing {
 		return
 	}
 
@@ -667,6 +751,13 @@ func (c *Container) gc(now time.Time) {
 					delete(c.retransmits, d)
 				}
 			}
+			for _, protoStats := range c.l7Stats {
+				for d := range protoStats {
+					if d.src == dst {
+						delete(protoStats, d)
+					}
+				}
+			}
 		}
 	}
 }
@@ -738,7 +829,7 @@ func (c *Container) revalidateListens(now time.Time, actualListens map[netaddr.I
 	}
 }
 
-func resolveFd(pid uint32, fd uint32) (mntId string, logPath string) {
+func resolveFd(pid uint32, fd uint64) (mntId string, logPath string) {
 	info := proc.GetFdInfo(pid, fd)
 	if info == nil {
 		return

containers/metrics.go

@@ -1,6 +1,7 @@
 package containers
 
 import (
+	"github.com/coroot/coroot-node-agent/ebpftracer"
 	"github.com/prometheus/client_golang/prometheus"
 	"reflect"
 )
@@ -71,6 +72,25 @@ var metrics = struct {
 	ApplicationType: metric("container_application_type", "Type of the application running in the container (e.g. memcached, postgres, mysql)", "application_type"),
 }
 
+var (
+	L7Requests = map[ebpftracer.L7Protocol]prometheus.CounterOpts{
+		ebpftracer.L7ProtocolHTTP:      {Name: "container_http_requests_total", Help: "Total number of outbound HTTP requests"},
+		ebpftracer.L7ProtocolPostgres:  {Name: "container_postgres_queries_total", Help: "Total number of outbound Postgres queries"},
+		ebpftracer.L7ProtocolRedis:     {Name: "container_redis_queries_total", Help: "Total number of outbound Redis queries"},
+		ebpftracer.L7ProtocolMemcached: {Name: "container_memcached_queries_total", Help: "Total number of outbound Memcached queries"},
+		ebpftracer.L7ProtocolMysql:     {Name: "container_mysql_queries_total", Help: "Total number of outbound Mysql queries"},
+		ebpftracer.L7ProtocolMongo:     {Name: "container_mongo_queries_total", Help: "Total number of outbound Mongo queries"},
+	}
+	L7Latency = map[ebpftracer.L7Protocol]prometheus.HistogramOpts{
+		ebpftracer.L7ProtocolHTTP:      {Name: "container_http_request_duration_seconds_total", Help: "Histogram of the response time for each outbound HTTP request"},
+		ebpftracer.L7ProtocolPostgres:  {Name: "container_postgres_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Postgres query"},
+		ebpftracer.L7ProtocolRedis:     {Name: "container_redis_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Redis query"},
+		ebpftracer.L7ProtocolMemcached: {Name: "container_memcached_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Memcached query"},
+		ebpftracer.L7ProtocolMysql:     {Name: "container_mysql_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Mysql query"},
+		ebpftracer.L7ProtocolMongo:     {Name: "container_mongo_queries_duration_seconds_total", Help: "Histogram of the execution time for each outbound Mongo query"},
+	}
+)
+
 func metric(name, help string, labels ...string) *prometheus.Desc {
 	return prometheus.NewDesc(name, help, labels, nil)
 }

containers/registry.go

@@ -5,6 +5,7 @@ import (
 	"github.com/coroot/coroot-node-agent/cgroup"
 	"github.com/coroot/coroot-node-agent/common"
 	"github.com/coroot/coroot-node-agent/ebpftracer"
+	"github.com/coroot/coroot-node-agent/flags"
 	"github.com/coroot/coroot-node-agent/proc"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/vishvananda/netns"
@@ -78,7 +79,7 @@ func NewRegistry(reg prometheus.Registerer, kernelVersion string) (*Registry, er
 	}
 
 	go cs.handleEvents(cs.events)
-	t, err := ebpftracer.NewTracer(cs.events, kernelVersion)
+	t, err := ebpftracer.NewTracer(cs.events, kernelVersion, *flags.DisableL7Tracing)
 	if err != nil {
 		close(cs.events)
 		return nil, err
@@ -177,13 +178,13 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 
 			case ebpftracer.EventTypeConnectionOpen:
 				if c := r.getOrCreateContainer(e.Pid); c != nil {
-					c.onConnectionOpen(e.Pid, e.SrcAddr, e.DstAddr, false)
+					c.onConnectionOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, false)
 				} else {
 					klog.Infoln("TCP connection from unknown container", e)
 				}
 			case ebpftracer.EventTypeConnectionError:
 				if c := r.getOrCreateContainer(e.Pid); c != nil {
-					c.onConnectionOpen(e.Pid, e.SrcAddr, e.DstAddr, true)
+					c.onConnectionOpen(e.Pid, e.Fd, e.SrcAddr, e.DstAddr, true)
 				} else {
 					klog.Infoln("TCP connection error from unknown container", e)
 				}
@@ -201,6 +202,13 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 						break
 					}
 				}
+			case ebpftracer.EventTypeL7Request:
+				if e.L7Request == nil {
+					continue
+				}
+				if c := r.containersByPid[e.Pid]; c != nil {
+					c.onL7Request(e.Pid, e.Fd, e.L7Request)
+				}
 			}
 		}
 	}

ebpftracer/Dockerfile

@@ -2,8 +2,8 @@ FROM alpine:3.13
 
 RUN apk add llvm clang libbpf-dev linux-headers
 
-COPY ebpf/* /tmp/
-WORKDIR /tmp
+COPY ebpf /tmp/ebpf
+WORKDIR /tmp/ebpf
 
 RUN clang -g -O2 -target bpf -D__KERNEL=416 -c ebpf.c -o ebpf416.o && llvm-strip --strip-debug ebpf416.o
 RUN clang -g -O2 -target bpf -D__KERNEL=420 -c ebpf.c -o ebpf420.o && llvm-strip --strip-debug ebpf420.o

ebpftracer/Makefile

@@ -1,7 +1,8 @@
 build:
 	@echo ===BUILDING===
-	docker build -t ebpftracer .
-	docker cp $(shell docker create --rm ebpftracer):/tmp/ebpf.go ./ebpf.go
+	docker rmi -f ebpftracer
+	docker build -t ebpftracer --progress plain .
+	docker run --rm --name ebpftracer ebpftracer cat /tmp/ebpf/ebpf.go > ./ebpf.go
 	@echo
 
 test: test_vm1 test_vm2 test_vm3 test_vm4

ebpftracer/ebpf.go

@@ -2,6 +2,7 @@
 #include <bpf/bpf_helpers.h>
 #include <bpf/bpf_core_read.h>
 #include <bpf/bpf_tracing.h>
+#include <bpf/bpf_endian.h>
 
 #define EVENT_TYPE_PROCESS_START	1
 #define EVENT_TYPE_PROCESS_EXIT		2
@@ -17,7 +18,8 @@
 
 #include "proc.c"
 #include "file.c"
-#include "tcp_state.c"
-#include "tcp_retransmit.c"
+#include "tcp/state.c"
+#include "tcp/retransmit.c"
+#include "l7/l7.c"
 
 char _license[] SEC("license") = "GPL";

ebpftracer/ebpf/ebpf.c

@@ -3,7 +3,7 @@
 struct file_event {
 	__u32 type;
 	__u32 pid;
-	__u32 fd;
+	__u64 fd;
 };
 
 struct {