Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion containers/container.go
Original file line number Diff line number Diff line change
Expand Up @@ -1027,7 +1027,7 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6
destIP = ip
}
}
if provider, ok := llm.ProviderForHost(host); ok {
if provider, ok := llm.ProviderForHost(host); ok && *flags.EnableLLMCapture {
c.startLLMCapture(pid, fd, timestamp, llm.Tag{Provider: provider, Host: host})
}

Expand Down
3 changes: 2 additions & 1 deletion containers/llm_capture.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"github.com/coroot/coroot-node-agent/common"
"github.com/coroot/coroot-node-agent/ebpftracer"
"github.com/coroot/coroot-node-agent/ebpftracer/l7"
"github.com/coroot/coroot-node-agent/flags"
"github.com/coroot/coroot-node-agent/llm"
"github.com/coroot/coroot-node-agent/tracing"
"golang.org/x/sys/unix"
Expand Down Expand Up @@ -144,7 +145,7 @@ func (c *Container) onLLMData(e ebpftracer.Event) {
// tuple alone, ahead of connection tracking, which short-lived connections
// often outrun. Called with c.lock held.
func (c *Container) detectLLMEndpoint(pid uint32, fd uint64, ts uint64, r *l7.RequestData, si *ebpftracer.SocketInfo) {
if r.Protocol != l7.ProtocolHTTP {
if !*flags.EnableLLMCapture || r.Protocol != l7.ProtocolHTTP {
return
}
path, host := llm.RequestPathAndHost(r.Payload)
Expand Down
6 changes: 4 additions & 2 deletions containers/registry.go
Original file line number Diff line number Diff line change
Expand Up @@ -156,14 +156,16 @@ func NewRegistry(reg prometheus.Registerer, rawReg prometheus.Registerer, proces

processInfoCh: processInfoCh,
ip_resolver: ip_resolver,
tracer: ebpftracer.NewTracer(hostNetNs, selfNetNs, *flags.DisableL7Tracing),
tracer: ebpftracer.NewTracer(hostNetNs, selfNetNs, *flags.DisableL7Tracing, *flags.EnableLLMCapture),

gpuProcessUsageSampleChan: gpuProcessUsageSampleChan,
nodeConstLabels: NodeConstLabels{MachineID: machineId, SystemUUID: systemUuid, AZ: az, Region: region},
}
// Register LLM metrics with the same registerer used for other container metrics
RegisterL7SelfMetrics(reg)
RegisterLLMMetrics(reg)
if *flags.EnableLLMCapture {
RegisterLLMMetrics(reg)
}
if err = reg.Register(kernelCounterCollector{tracer: r.tracer}); err != nil {
return nil, err
}
Expand Down
11 changes: 11 additions & 0 deletions docs/llm-observability.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,17 @@ the traffic is captured in eBPF on the node.
This feature is specific to this fork and is not present in upstream
`coroot/coroot-node-agent`.

## Enabling

LLM capture is off by default. Enable it with `--enable-llm-capture` (or
`ENABLE_LLM_CAPTURE=true`). It requires L7 tracing, which is on unless
`--disable-l7-tracing` is set.

While it is off, the feature costs one map lookup per socket read and write
in the kernel and nothing in userspace. While it is on, the kernel also
checks the destination of each new connection once, and the traffic of LLM
API connections (only those) is copied and parsed in userspace.

## How it works

1. **Identification.** A connection is identified as an LLM API
Expand Down
20 changes: 10 additions & 10 deletions ebpftracer/ebpf.go

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion ebpftracer/ebpf/l7/l7.c
Original file line number Diff line number Diff line change
Expand Up @@ -435,7 +435,7 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size,
return 0;
}

if (!is_tls && conn != &conn_on_stack && conn->bytes_sent == 0) {
if (!is_tls && conn != &conn_on_stack && conn->bytes_sent == 0 && llm_capture_enabled()) {
llm_tag_by_destination(&cid, conn);
}
if (!is_tls) {
Expand Down
20 changes: 20 additions & 0 deletions ebpftracer/ebpf/l7/llm_capture.c
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,23 @@
// a single ordered buffer lets userspace splice those and the capture chunks
// into one stream without a gap or a reordering.

// llm_capture_config[0] is non-zero when LLM capture is enabled. Userspace
// sets it at load time; while it is zero, the read and write paths pay one
// array lookup for this feature and nothing else.
struct {
__uint(type, BPF_MAP_TYPE_ARRAY);
__uint(key_size, sizeof(__u32));
__uint(value_size, sizeof(__u32));
__uint(max_entries, 1);
} llm_capture_config SEC(".maps");

static inline __attribute__((__always_inline__))
int llm_capture_enabled(void) {
__u32 zero = 0;
__u32 *enabled = bpf_map_lookup_elem(&llm_capture_config, &zero);
return enabled && *enabled;
}

// Each read or write is copied in up to two chunks of LLM_CHUNK-1 bytes; the
// mask that bounds a copy for the verifier cannot express LLM_CHUNK itself.
// Two chunks cover a full 64KB read. Anything beyond is reported in
Expand Down Expand Up @@ -111,6 +128,9 @@ void llm_emit(struct llm_event *e, char *src, __u64 len, __u64 skip_after) {
// larger only when buf holds an iovec prefix.
static inline __attribute__((__always_inline__))
int llm_capture(struct connection_id *cid, struct connection *conn, __u8 direction, char *buf, __u64 size, __u64 total) {
if (!llm_capture_enabled()) {
return 0;
}
__u64 *conn_ts = bpf_map_lookup_elem(&llm_conns, cid);
if (!conn_ts) {
return 0;
Expand Down
24 changes: 23 additions & 1 deletion ebpftracer/tracer.go
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ const (

type Tracer struct {
disableL7Tracing bool
enableLLMCapture bool
hostNetNs netns.NsHandle
selfNetNs netns.NsHandle

Expand All @@ -126,12 +127,18 @@ type Tracer struct {
ready atomic.Bool
}

func NewTracer(hostNetNs, selfNetNs netns.NsHandle, disableL7Tracing bool) *Tracer {
func NewTracer(hostNetNs, selfNetNs netns.NsHandle, disableL7Tracing, enableLLMCapture bool) *Tracer {
if disableL7Tracing {
klog.Infoln("L7 tracing is disabled")
}
// LLM capture rides on the L7 programs.
enableLLMCapture = enableLLMCapture && !disableL7Tracing
if enableLLMCapture {
klog.Infoln("LLM capture is enabled")
}
return &Tracer{
disableL7Tracing: disableL7Tracing,
enableLLMCapture: enableLLMCapture,
hostNetNs: hostNetNs,
selfNetNs: selfNetNs,

Expand Down Expand Up @@ -191,6 +198,9 @@ func (t *Tracer) TLSCiphertextSkipped() (writes, reads uint64, ok bool) {
// LLMCaptureDrops returns how many LLM capture chunks the kernel lost because
// the llm_events ring buffer was full.
func (t *Tracer) LLMCaptureDrops() (uint64, bool) {
if !t.enableLLMCapture {
return 0, false
}
m := t.readyMap("llm_capture_drops")
if m == nil {
return 0, false
Expand All @@ -204,6 +214,9 @@ func (t *Tracer) LLMCaptureDrops() (uint64, bool) {
// events, which keeps the mark from applying to a later connection reusing the
// fd. The kernel removes the mark when the fd is closed.
func (t *Tracer) TagLLMConnection(pid uint32, fd uint64, connTimestamp uint64) error {
if !t.enableLLMCapture {
return errors.New("LLM capture is disabled")
}
m := t.readyMap("llm_conns")
if m == nil {
return errors.New("ebpf collection not loaded")
Expand All @@ -215,6 +228,9 @@ func (t *Tracer) TagLLMConnection(pid uint32, fd uint64, connTimestamp uint64) e
// NAT), as an LLM API endpoint: every new connection to it is captured from
// its first write.
func (t *Tracer) TagLLMDestination(ip netaddr.IP, port uint16) error {
if !t.enableLLMCapture {
return errors.New("LLM capture is disabled")
}
m := t.readyMap("llm_dests")
if m == nil {
return errors.New("ebpf collection not loaded")
Expand Down Expand Up @@ -360,6 +376,12 @@ func (t *Tracer) ebpf(ch chan<- Event) error {
}
t.collection = c

if t.enableLLMCapture {
if err := c.Maps["llm_capture_config"].Update(uint32(0), uint32(1), ebpf.UpdateAny); err != nil {
return fmt.Errorf("failed to enable LLM capture: %w", err)
}
}
Comment thread
mayankpande88 marked this conversation as resolved.

// Initialize socket info offsets for direct fd->socket tuple extraction
// This enables L7 event processing without dependency on TCP connection tracking
if err := t.initSocketInfoOffsets(); err != nil {
Expand Down
2 changes: 1 addition & 1 deletion ebpftracer/tracer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -327,7 +327,7 @@ func runTracer(t *testing.T, verbose bool) (func() *Event, func()) {
assert.NoError(t, common.SetKernelVersion(string(bytes.Split(uname.Release[:], []byte{0})[0])))

go func() {
tt := NewTracer(0, 0, false)
tt := NewTracer(0, 0, false, false)
err := tt.Run(events)
require.NoError(t, err)
<-done
Expand Down
7 changes: 6 additions & 1 deletion flags/flags.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,12 @@ var (
// currently consumes, and attaching the probes reads the whole ELF symbol
// table of every Node.js process's binary. Opt-in matches .NET rather than
// the always-on Python probes.
EnableNodejsTracing = kingpin.Flag("enable-nodejs-tracing", "Enable Node.js event loop tracing (attaches libuv uprobes)").Default("false").Envar("ENABLE_NODEJS_TRACING").Bool()
EnableNodejsTracing = kingpin.Flag("enable-nodejs-tracing", "Enable Node.js event loop tracing (attaches libuv uprobes)").Default("false").Envar("ENABLE_NODEJS_TRACING").Bool()
// Off by default: capturing LLM API connections in full (see
// docs/llm-observability.md) adds work to every socket read and write in
// the kernel and parses captured traffic in userspace. Clusters with no
// LLM traffic, or no interest in it, should not pay for it.
EnableLLMCapture = kingpin.Flag("enable-llm-capture", "Capture LLM API traffic and export token usage metrics").Default("false").Envar("ENABLE_LLM_CAPTURE").Bool()
DisableGPUMonitoring = kingpin.Flag("disable-gpu-monitoring", "Disable GPU monitoring (NVML)").Default("false").Envar("DISABLE_GPU_MONITORING").Bool()

ContainerAllowlist = kingpin.Flag("container-allowlist", "List of allowed containers (regex patterns)").Envar("CONTAINER_ALLOWLIST").Strings()
Expand Down