Correlating eBPF Socket Event Tracing with OpenTelemetry Context in Distributed Environments — code snippets

snippet-1.txt

#include <uapi/linux/bpf.h>
#include <linux/sched.h>
#include <net/sock.h>
#include <net/inet_connection_sock.h>
#include <bcc/proto.h>

#define MAX_PAYLOAD_READ 256
#define TRACEPARENT_LEN 55

struct conn_info_t {
    u32 saddr;
    u32 daddr;
    u16 sport;
    u16 dport;
    u32 pid;
};

struct trace_match_t {
    char trace_id[32];
    char span_id[16];
    u64 timestamp_ns;
};

BPF_HASH(active_conns, u64, struct conn_info_t);
BPF_PERF_OUTPUT(socket_trace_events);

int kprobe__tcp_sendmsg(struct pt_regs *ctx, struct sock *sk, struct msghdr *msg, size_t size) {
    u64 pid_tgid = bpf_get_current_pid_tgid();
    u32 pid = pid_tgid >> 32;

    // Filter out kernel threads and uninteresting processes
    if (pid == 0) return 0;

    struct conn_info_t conn = {};
    conn.pid = pid;
    
    // Read IP addresses and ports from the struct sock
    bpf_probe_read_kernel(&conn.saddr, sizeof(conn.saddr), &sk->__sk_common.skc_rcv_saddr);
    bpf_probe_read_kernel(&conn.daddr, sizeof(conn.daddr), &sk->__sk_common.skc_daddr);
    bpf_probe_read_kernel(&conn.sport, sizeof(conn.sport), &sk->__sk_common.skc_num);
    bpf_probe_read_kernel(&conn.dport, sizeof(conn.dport), &sk->__sk_common.skc_dport);
    
    // Convert destination port from big-endian (network byte order) to CPU host byte order
    conn.dport = bpf_ntohs(conn.dport);

    // Save the connection info in a map keyed by the thread ID to retrieve during return probes
    active_conns.update(&pid_tgid, &conn);
    return 0;
}

snippet-2.txt

#define TRACEPARENT_PREFIX "traceparent:"
#define TRACEPARENT_PREFIX_LEN 12

static __always_inline int parse_traceparent(const char *buf, u32 buf_len, struct trace_match_t *match) {
    // Search for the prefix using a bounded loop to satisfy the verifier
    int prefix_idx = -1;
    #pragma unroll
    for (int i = 0; i < MAX_PAYLOAD_READ - TRACEPARENT_PREFIX_LEN; i++) {
        if (i >= buf_len) break;
        
        // Simple sequential pattern matching
        if (buf[i] == 't' && buf[i+1] == 'r' && buf[i+2] == 'a' && buf[i+3] == 'c' &&
            buf[i+4] == 'e' && buf[i+5] == 'p' && buf[i+6] == 'a' && buf[i+7] == 'r' &&
            buf[i+8] == 'e' && buf[i+9] == 'n' && buf[i+10] == 't' && buf[i+11] == ':') {
            prefix_idx = i;
            break;
        }
    }

    if (prefix_idx == -1) return -1;

    // Calculate the start of the traceparent value (skip prefix and space if present)
    u32 val_idx = prefix_idx + TRACEPARENT_PREFIX_LEN;
    if (val_idx < buf_len && buf[val_idx] == ' ') {
        val_idx++;
    }

    // Ensure we don't read past buffer boundary
    if (val_idx + TRACEPARENT_LEN > buf_len) return -1;

    // Check version prefix (should be "00-")
    if (buf[val_idx] != '0' || buf[val_idx+1] != '0' || buf[val_idx+2] != '-') {
        return -1;
    }

    u32 trace_id_start = val_idx + 3; // Skip "00-"
    
    // Copy Trace ID (32 hex characters)
    #pragma unroll
    for (int j = 0; j < 32; j++) {
        match->trace_id[j] = buf[trace_id_start + j];
    }

    u32 span_id_start = trace_id_start + 32 + 1; // Skip trace_id and the separating hyphen

    // Copy Span ID (16 hex characters)
    #pragma unroll
    for (int k = 0; k < 16; k++) {
        match->span_id[k] = buf[span_id_start + k];
    }

    match->timestamp_ns = bpf_ktime_get_ns();
    return 0;
}

snippet-3.txt

struct socket_event_t {
    struct conn_info_t conn;
    char trace_id[32];
    char span_id[16];
    u64 tcp_rtt_us;
    u32 tcp_snd_cwnd;
    u64 timestamp_ns;
};

BPF_RINGBUF_OUTPUT(events, 128); // Ring buffer configured with 128 pages

int kretprobe__tcp_sendmsg(struct pt_regs *ctx) {
    u64 pid_tgid = bpf_get_current_pid_tgid();
    struct conn_info_t *conn = active_conns.lookup(&pid_tgid);
    if (!conn) return 0; // Missed entry

    // Retrieve return code (bytes sent)
    int ret = PT_REGS_RC(ctx);
    if (ret <= 0) goto cleanup;

    // Read the user buffer parameter from the registers (rsi is the msg pointer in x86_64)
    // For tcp_sendmsg, we inspect the payload to locate HTTP/gRPC metadata
    struct msghdr *msg = (struct msghdr *)PT_REGS_PARM2(ctx);
    if (!msg) goto cleanup;

    // Use a stack buffer to read the initial segment of the payload
    char payload_buf[MAX_PAYLOAD_READ] = {};
    
    // Extract user payload buffer pointer from msg_iter
    struct iov_iter *iter = &msg->msg_iter;
    const struct iovec *iov = iter->iov;
    if (!iov) goto cleanup;

    char *user_ptr = iov->iov_base;
    if (!user_ptr) goto cleanup;

    // Safely copy the payload from user space to kernel stack
    if (bpf_probe_read_user(payload_buf, sizeof(payload_buf), user_ptr) != 0) {
        goto cleanup;
    }

    struct trace_match_t match = {};
    if (parse_traceparent(payload_buf, sizeof(payload_buf), &match) == 0) {
        // We found a valid W3C traceparent header. Record TCP metrics for correlation.
        struct socket_event_t *evt = events.ringbuf_reserve(sizeof(struct socket_event_t));
        if (evt) {
            evt->conn = *conn;
            bpf_probe_read_kernel(evt->trace_id, sizeof(evt->trace_id), match.trace_id);
            bpf_probe_read_kernel(evt->span_id, sizeof(evt->span_id), match.span_id);
            evt->timestamp_ns = match.timestamp_ns;

            // Extract real-time TCP metrics from the socket
            struct sock *sk = (struct sock *)PT_REGS_PARM1(ctx);
            struct tcp_sock *tp = (struct tcp_sock *)sk;
            
            // Read Round Trip Time (RTT) and Congestion Window
            bpf_probe_read_kernel(&evt->tcp_rtt_us, sizeof(evt->tcp_rtt_us), &tp->srtt_us);
            evt->tcp_rtt_us >>= 3; // Convert from 8ths of a microsecond to raw microseconds
            
            bpf_probe_read_kernel(&evt->tcp_snd_cwnd, sizeof(evt->tcp_snd_cwnd), &tp->snd_cwnd);

            events.ringbuf_submit(evt, 0);
        }
    }

cleanup:
    active_conns.delete(&pid_tgid);
    return 0;
}

snippet-4.go

package main

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"log"
	"net"
	"os"
	"os/signal"
	"syscall"

	"github.com/cilium/ebpf"
	"github.com/cilium/ebpf/link"
	"github.com/cilium/ebpf/ringbuf"
	"golang.org/x/sys/unix"
)

type ConnInfo struct {
	Saddr uint32
	Daddr uint32
	Sport uint16
	Dport uint16
	Pid   uint32
}

type SocketEvent struct {
	Conn       ConnInfo
	TraceID    [32]byte
	SpanID     [16]byte
	TcpRttUs   uint64
	TcpSndCwnd uint32
	Timestamp  uint64
}

func main() {
	// Stop execution cleanly when interrupted
	stopper := make(chan os.Signal, 1)
	signal.Notify(stopper, os.Interrupt, syscall.SIGTERM)

	// Set resource limit (RLIMIT_MEMLOCK) for eBPF map allocations
	if err := unix.Setrlimit(unix.RLIMIT_MEMLOCK, &unix.Rlimit{
		Cur: unix.RLIM_INFINITY,
		Max: unix.RLIM_INFINITY,
	}); err != nil {
		log.Fatalf("failed to set memlock limit: %v", err)
	}

	// Load the compiled eBPF ELF binary
	spec, err := ebpf.LoadCollectionSpec("/opt/ebpf/socket_tracer.o")
	if err != nil {
		log.Fatalf("failed to load eBPF collection: %v", err)
	}

	var objs struct {
		ActiveConns *ebpf.Map  `ebpf:"active_conns"`
		Events      *ebpf.Map  `ebpf:"events"`
		TcpSendmsg  *ebpf.Program `ebpf:"kprobe__tcp_sendmsg"`
		TcpSendmsgRet *ebpf.Program `ebpf:"kretprobe__tcp_sendmsg"`
	}

	if err := spec.LoadAndAssign(&objs, nil); err != nil {
		log.Fatalf("failed to load eBPF maps and programs: %v", err)
	}
	defer objs.ActiveConns.Close()
	defer objs.Events.Close()
	defer objs.TcpSendmsg.Close()
	defer objs.TcpSendmsgRet.Close()

	// Attach kprobe to tcp_sendmsg
	kp, err := link.Kprobe("tcp_sendmsg", objs.TcpSendmsg, nil)
	if err != nil {
		log.Fatalf("failed to attach kprobe: %v", err)
	}
	defer kp.Close()

	// Attach return probe to tcp_sendmsg
	krp, err := link.Kretprobe("tcp_sendmsg", objs.TcpSendmsgRet, nil)
	if err != nil {
		log.Fatalf("failed to attach kretprobe: %v", err)
	}
	defer krp.Close()

	// Open the BPF Ring Buffer to consume events
	rd, err := ringbuf.NewReader(objs.Events)
	if err != nil {
		log.Fatalf("failed to open ring buffer: %v", err)
	}
	defer rd.Close()

	go func() {
		<-stopper
		rd.Close()
	}()

	log.Printf("Listening for eBPF socket events...")

	var event SocketEvent
	for {
		record, err := rd.Read()
		if err != nil {
			if errors.Is(err, ringbuf.ErrClosed) {
				log.Println("Ring buffer closed. Exiting.")
				return
			}
			log.Printf("error reading from ring buffer: %v", err)
			continue
		}

		// Decode binary event structure
		if err := binary.Read(bytes.NewReader(record.RawSample), binary.LittleEndian, &event); err != nil {
			log.Printf("failed to parse event structure: %v", err)
			continue
		}

		saddr := intToIP(event.Conn.Saddr)
		daddr := intToIP(event.Conn.Daddr)
		
		fmt.Printf("[PID %d] TRACEID: %s SPANID: %s | TCP Link: %s:%d -> %s:%d | RTT: %dus | CWND: %d pkts\n",
			event.Conn.Pid,
			string(event.TraceID[:]),
			string(event.SpanID[:]),
			saddr, event.Conn.Sport,
			daddr, event.Conn.Dport,
			event.TcpRttUs,
			event.TcpSndCwnd,
		)
	}
}

func intToIP(nn uint32) net.IP {
	ip := make(net.IP, 4)
	binary.LittleEndian.PutUint32(ip, nn)
	return ip
}

snippet-5.go

package main

import (
	"context"
	"sync"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/sdk/trace"
)

type EbpfMetricCache struct {
	mu     sync.RWMutex
	store  map[string]*SocketEvent
	expiry time.Duration
}

func NewEbpfMetricCache(expiry time.Duration) *EbpfMetricCache {
	c := &EbpfMetricCache{
		store:  make(map[string]*SocketEvent),
		expiry: expiry,
	}
	// Run background eviction loop to prevent memory leaks
	go c.evictionLoop()
	return c
}

func (c *EbpfMetricCache) Put(key string, evt *SocketEvent) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.store[key] = evt
}

func (c *EbpfMetricCache) Get(key string) (*SocketEvent, bool) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	val, ok := c.store[key]
	return val, ok
}

func (c *EbpfMetricCache) evictionLoop() {
	ticker := time.NewTicker(c.expiry)
	for range ticker.C {
		c.mu.Lock()
		now := uint64(time.Now().UnixNano())
		for k, v := range c.store {
			// Evict events older than cache lifetime (e.g. 30 seconds)
			if now - v.Timestamp > uint64(c.expiry.Nanoseconds()) {
				delete(c.store, k)
			}
		}
		c.mu.Unlock()
	}
}

// EbpfSpanProcessor implements the OpenTelemetry SpanProcessor interface
type EbpfSpanProcessor struct {
	cache *EbpfMetricCache
	next  trace.SpanProcessor
}

func NewEbpfSpanProcessor(cache *EbpfMetricCache, next trace.SpanProcessor) *EbpfSpanProcessor {
	return &EbpfSpanProcessor{
		cache: cache,
		next:  next,
	}
}

func (sp *EbpfSpanProcessor) OnStart(parent context.Context, s trace.ReadWriteSpan) {
	sp.next.OnStart(parent, s)
}

func (sp *EbpfSpanProcessor) OnEnd(s trace.ReadOnlySpan) {
	traceID := s.SpanContext().TraceID().String()
	spanID := s.SpanContext().SpanID().String()
	
	// Create lookup key
	key := traceID + ":" + spanID
	
	if evt, found := sp.cache.Get(key); found {
		// Cast read-only span to writeable span to append metric attributes
		if rws, ok := s.(trace.ReadWriteSpan); ok {
			rws.SetAttributes(
				attribute.Int64("network.kernel.rtt_us", int64(evt.TcpRttUs)),
				attribute.Int64("network.kernel.snd_cwnd_packets", int64(evt.TcpSndCwnd)),
				attribute.Int64("network.kernel.pid", int64(evt.Conn.Pid)),
				attribute.String("network.kernel.peer_ip", intToIP(evt.Conn.Daddr).String()),
			)
		}
	}
	sp.next.OnEnd(s)
}

func (sp *EbpfSpanProcessor) Shutdown(ctx context.Context) error {
	return sp.next.Shutdown(ctx)
}

func (sp *EbpfSpanProcessor) ForceFlush(ctx context.Context) error {
	return sp.next.ForceFlush(ctx)
}

snippet-6.txt

struct filter_key_t {
    u32 daddr;
    u16 dport;
};

// Map containing target IP subnets or ports we want to trace
BPF_HASH(trace_filter_map, struct filter_key_t, u8);

static __always_inline int should_trace_packet(u32 daddr, u16 dport) {
    // 1. Fast check for standard services (e.g. API ports)
    if (dport != 80 && dport != 443 && dport != 8080 && dport != 50051) {
        // If it doesn't match standard ports, check if the specific destination is in our filter map
        struct filter_key_t key = {
            .daddr = daddr,
            .dport = dport
        };
        u8 *active = trace_filter_map.lookup(&key);
        if (!active) {
            return 0; // Skip tracing
        }
    }
    
    // 2. Avoid tracing loopback traffic if unnecessary
    if (daddr == 16777343) { // 127.0.0.1 in integer format
        return 0;
    }

    return 1; // Proceed with tracing
}