diff --git a/containers/container.go b/containers/container.go index bb74553..7f6b361 100644 --- a/containers/container.go +++ b/containers/container.go @@ -38,6 +38,12 @@ var ( gpuStatsWindow = 15 * time.Second ) +const ( + // Max per-connection HTTP/2 parsers per container. + // Each parser holds HPACK decoders and active request state. + maxHTTP2ParsersPerContainer = 50 +) + type ContainerID string type ContainerNetwork struct { @@ -128,6 +134,7 @@ type Container struct { activeConnections map[ConnectionKey]*ActiveConnection connectionsByPidFd map[PidFd]*ActiveConnection googleHTTP2Parsers map[PidFd]*l7.Http2Parser // Per-connection HTTP/2 parsers (keyed by pid:fd for correct HPACK state) + http2SkipSet map[PidFd]struct{} // pid:fd pairs confirmed as non-LLM — skip all future HTTP/2 events l7Stats L7Stats tcpMetrics *TCPMetrics @@ -1158,9 +1165,14 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 } pidFd := PidFd{Pid: pid, Fd: fd} if c.googleHTTP2Parsers[pidFd] == nil { - c.googleHTTP2Parsers[pidFd] = l7.NewHttp2Parser() + if len(c.googleHTTP2Parsers) >= maxHTTP2ParsersPerContainer { + return nil, L7RequestProcessed + } + p := l7.NewHttp2Parser() + p.Lightweight = true + p.LLMHostChecker = isLLMRelevantHost + c.googleHTTP2Parsers[pidFd] = p } - // Use per-connection parser (each HTTP/2 connection has its own HPACK dynamic table) parser := c.googleHTTP2Parsers[pidFd] conn.http2Parser = parser // Keep reference on connection for compatibility requests := parser.Parse(r.Method, r.Payload, uint64(r.Duration)) @@ -1188,7 +1200,7 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // Feed active streams to LLM stream tracker for SSE-based completion detection // This handles streaming LLM responses that don't send END_STREAM until complete - if c.llmStreamTracker != nil { + if c.llmStreamTracker != nil && !parser.Lightweight { // Resolve host for LLM detection host := conn.DestinationKey.GetDestinationWorkload().Name if isIPAddress(host) { @@ -1196,18 +1208,16 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 host = domain.FQDN } } + activeStreams := parser.GetActiveStreamsForLLM() // Also try :authority header if host is still an IP if host == "" || isIPAddress(host) { - for _, update := range parser.GetActiveStreamsForLLM() { + for _, update := range activeStreams { if update.Authority != "" && !isIPAddress(update.Authority) { host = update.Authority break } } } - - // Process active streams for LLM tracking - activeStreams := parser.GetActiveStreamsForLLM() for _, update := range activeStreams { streamHost := host if streamHost == "" && update.Authority != "" { @@ -1345,49 +1355,70 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // This is common for Go TLS connections where goroutines switch threads between // TCP connect and TLS write, causing fd_by_pid_tgid lookup to fail in eBPF. // -// We can still extract useful info from HTTP/2 frames: -// - :authority pseudo-header for host resolution (LLM detection) -// - :path for request path matching -// - Request/response payloads for LLM token extraction -// -// Each HTTP/2 connection has its own HPACK dynamic table, so we use per-fd parsers -// to avoid header decoding corruption when a process has multiple connections. -// -// Note: Metrics that require destination IP will be skipped since we don't have it. +// Since this path produces NO L7 metrics (no connection = no DestinationKey), the only +// purpose is LLM detection. To avoid wasting CPU on non-LLM traffic (gRPC, K8s API, etc.), +// we use a two-phase approach: +// 1. If dstIP is known, check DNS cache first — skip if not LLM-relevant. +// 2. Otherwise, parse only the first event per pid:fd to extract :authority. +// If not LLM-relevant, add to skip-set and never parse that fd again. func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.RequestData, dstIP ...netaddr.IP) (map[netaddr.IP]*common.Domain, L7RequestResult) { - // Use per-connection parser: each HTTP/2 connection has its own HPACK dynamic table, - // so sharing a parser across connections corrupts header decoding. + pidFd := PidFd{Pid: pid, Fd: fd} + + // Fast path: already determined this fd is not LLM-relevant + if c.http2SkipSet != nil { + if _, skip := c.http2SkipSet[pidFd]; skip { + return nil, L7RequestProcessed + } + } + + // If we have a destination IP, check DNS cache before parsing + if len(dstIP) > 0 && !dstIP[0].IsZero() { + host := "" + if domain := c.registry.getDomain(dstIP[0]); domain != nil { + host = domain.FQDN + } + if host == "" || !isLLMRelevantHost(host) { + // Not an LLM destination — skip this and all future events + c.addToHTTP2SkipSet(pidFd) + return nil, L7RequestProcessed + } + } + + // Parse the HTTP/2 frames to extract :authority for LLM detection if c.googleHTTP2Parsers == nil { c.googleHTTP2Parsers = make(map[PidFd]*l7.Http2Parser) } - - pidFd := PidFd{Pid: pid, Fd: fd} parser := c.googleHTTP2Parsers[pidFd] if parser == nil { + if len(c.googleHTTP2Parsers) >= maxHTTP2ParsersPerContainer { + return nil, L7RequestProcessed + } parser = l7.NewHttp2Parser() + parser.Lightweight = true + parser.LLMHostChecker = isLLMRelevantHost c.googleHTTP2Parsers[pidFd] = parser } requests := parser.Parse(r.Method, r.Payload, uint64(r.Duration)) - if len(requests) == 0 { - klog.V(3).Infof("HTTP2_CONNECTIONLESS_PARSE: pid=%d fd=%d method=%d payloadLen=%d completed=0 active=%d", - pid, fd, r.Method, len(r.Payload), parser.ActiveRequestCount()) + // If parser is still in lightweight mode after parsing, no LLM host was found. + // Once we've seen at least one completed request (or enough HEADERS to decide), + // add to skip-set. We check ActiveRequestCount==0 to ensure we've seen a full + // request-response cycle (not just the first DATA frame before any HEADERS). + if parser.Lightweight && len(requests) > 0 { + // Parsed completed requests but none were LLM — this fd is non-LLM + c.addToHTTP2SkipSet(pidFd) + delete(c.googleHTTP2Parsers, pidFd) + return nil, L7RequestProcessed } for _, req := range requests { - // Extract host from :authority header (set by HTTP/2 parser) host := req.Authority if host == "" && len(dstIP) > 0 && !dstIP[0].IsZero() { - // Fallback: resolve destination IP via DNS cache. - // This handles mid-stream HPACK join where :authority is in the - // dynamic table and can't be decoded. if domain := c.registry.getDomain(dstIP[0]); domain != nil { host = domain.FQDN - klog.V(4).Infof("HTTP2_CONNECTIONLESS: resolved host from DNS cache: ip=%s host=%s path=%s", dstIP[0], host, req.Path) } } - // Detect LLM provider: try host first, then path, then response structure provider := ProviderUnknown if host != "" { provider = DetectLLMProvider(host) @@ -1395,8 +1426,6 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R if provider == ProviderUnknown && req.Path != "" { provider, _ = DetectLLMProviderFromPath(req.Path) } - // Last resort: analyze request/response structure for LLM patterns. - // Skip for gRPC (protobuf data can contain strings that match JSON patterns). isGRPC := strings.HasPrefix(req.ContentType, "application/grpc") if provider == ProviderUnknown && !isGRPC && len(req.ResponsePayload) > 0 { provider = detectProviderFromResponseStructure(req.ResponsePayload) @@ -1405,26 +1434,18 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R provider = detectProviderFromRequestStructure(req.RequestPayload) } - if host == "" && provider == ProviderUnknown { - klog.V(3).Infof("HTTP2_CONNECTIONLESS: no authority/path and no LLM detected for pid=%d fd=%d", pid, fd) - continue - } - if provider != ProviderUnknown { - // Set a default server address if host is unknown (detected from payload structure) if host == "" { host = providerDefaultHost(provider) } - - klog.V(4).Infof("HTTP2_CONNECTIONLESS_LLM_DETECTED: pid=%d fd=%d provider=%s host=%s path=%s reqLen=%d respLen=%d", - pid, fd, provider, host, req.Path, len(req.RequestPayload), len(req.ResponsePayload)) + klog.V(4).Infof("HTTP2_CONNECTIONLESS_LLM_DETECTED: pid=%d fd=%d provider=%s host=%s path=%s", + pid, fd, provider, host, req.Path) requestPayloadBase64 := base64.StdEncoding.EncodeToString(req.RequestPayload) responsePayloadBase64 := "" if len(req.ResponsePayload) > 0 { responsePayloadBase64 = base64.StdEncoding.EncodeToString(req.ResponsePayload) } - c.trackLLMRequest(provider, host, req.Path, requestPayloadBase64, responsePayloadBase64, req.Duration) } } @@ -1432,6 +1453,13 @@ func (c *Container) processHTTP2WithoutConnection(pid uint32, fd uint64, r *l7.R return nil, L7RequestProcessed } +func (c *Container) addToHTTP2SkipSet(pidFd PidFd) { + if c.http2SkipSet == nil { + c.http2SkipSet = make(map[PidFd]struct{}) + } + c.http2SkipSet[pidFd] = struct{}{} +} + // refreshActiveConnections snapshots active connections under c.lock and // pushes the gauge values to TCPMetrics. Called periodically from the // event handler goroutine (not from Collect). @@ -1857,14 +1885,38 @@ func (c *Container) gc(now time.Time) { // Clean up HTTP/2 parsers for closed/dead connections. // Parsers hold HPACK decoders, partial frame buffers, and active request maps // that accumulate memory over time if not cleaned up. + // Two cases: (1) connection-tracked parsers — delete when connection is gone, + // (2) connectionless parsers (processHTTP2WithoutConnection) — delete when pid dies. if c.googleHTTP2Parsers != nil { for pidFd := range c.googleHTTP2Parsers { - if _, alive := c.connectionsByPidFd[pidFd]; !alive { + if _, hasConn := c.connectionsByPidFd[pidFd]; hasConn { + continue // connection still alive + } + if _, hasProc := c.processes[pidFd.Pid]; !hasProc { + delete(c.googleHTTP2Parsers, pidFd) // pid dead — connectionless parser cleanup + continue + } + // Pid is alive but no connection — could be a connectionless parser + // (still needed) or a closed connection's parser (stale). + // Check if this parser was created via the connectionless path by + // seeing if there was ever a tracked connection for this pidFd. + // If activeConnections had it at some point, it's stale. + // Use a simple heuristic: if the parser has zero active requests + // and no partial frame data, it's safe to clean up. + p := c.googleHTTP2Parsers[pidFd] + if p.ActiveRequestCount() == 0 && !p.HasPartialData() { delete(c.googleHTTP2Parsers, pidFd) } } } + // Clean up skip-set entries for dead pids + for pidFd := range c.http2SkipSet { + if _, hasProc := c.processes[pidFd.Pid]; !hasProc { + delete(c.http2SkipSet, pidFd) + } + } + for dst, at := range c.lastConnectionAttempts { _, active := establishedDst[dst] if !active && !at.IsZero() && now.Sub(at) > gcInterval { diff --git a/containers/llm.go b/containers/llm.go index 1e04a0f..47a9e72 100644 --- a/containers/llm.go +++ b/containers/llm.go @@ -96,6 +96,12 @@ func providerDefaultHost(provider LLMProvider) string { } } +// isLLMRelevantHost returns true if the host matches a known LLM provider. +// Used by Http2Parser to auto-upgrade from lightweight to full mode. +func isLLMRelevantHost(host string) bool { + return DetectLLMProvider(host) != ProviderUnknown +} + // DetectLLMProvider identifies if a hostname belongs to an LLM provider func DetectLLMProvider(hostname string) LLMProvider { hostname = strings.ToLower(hostname) diff --git a/ebpftracer/l7/http2.go b/ebpftracer/l7/http2.go index fb8871d..7c484d5 100644 --- a/ebpftracer/l7/http2.go +++ b/ebpftracer/l7/http2.go @@ -43,6 +43,10 @@ const ( // Payloads are only used for LLM provider detection and trace spans, // which need at most a few KB of data. maxDataPayloadSize = 128 * 1024 + + // Max concurrent HTTP/2 streams tracked per connection. + // Prevents unbounded memory growth when responses never complete (orphan streams). + maxActiveRequests = 100 ) type Http2FrameHeader struct { @@ -96,6 +100,10 @@ type Http2Parser struct { activeRequests map[uint32]*Http2Request lastGcTime uint64 + // Reusable maps cleared on each Parse() call to avoid per-call allocation + statuses map[uint32]Status + grpcStatuses map[uint32]Status + // Buffers for partial frame reassembly across Read() calls // Needed because S2A/TLS returns small chunks that may split HTTP/2 frames clientPartialFrame []byte @@ -110,13 +118,26 @@ type Http2Parser struct { // Static table headers still work; dynamic table rebuilds over time. clientDecoderDegraded bool serverDecoderDegraded bool + + // Lightweight mode skips DATA payload accumulation and RequestHeaders map. + // Used for non-LLM HTTP/2 traffic (gRPC, K8s API, etc.) where only status + // codes are needed for L7 metrics. Auto-upgrades to full mode when an + // LLM-relevant :authority header is detected. + Lightweight bool + + // LLMHostChecker is called with :authority values to determine if this + // connection serves LLM traffic. When it returns true, Lightweight is + // set to false and full payload capture begins. + LLMHostChecker func(host string) bool } func NewHttp2Parser() *Http2Parser { return &Http2Parser{ clientDecoder: hpack.NewDecoder(4096, nil), serverDecoder: hpack.NewDecoder(4096, nil), - activeRequests: map[uint32]*Http2Request{}, + activeRequests: make(map[uint32]*Http2Request), + statuses: make(map[uint32]Status), + grpcStatuses: make(map[uint32]Status), } } @@ -141,6 +162,13 @@ func (p *Http2Parser) ActiveRequestCount() int { return len(p.activeRequests) } +// HasPartialData returns true if the parser has buffered partial frame data +// that would be lost if the parser were garbage collected. +func (p *Http2Parser) HasPartialData() bool { + return len(p.clientPartialFrame) > 0 || len(p.serverPartialFrame) > 0 || + p.clientPendingHeaders != nil || p.serverPendingHeaders != nil +} + // Http2StreamUpdate contains information about an active HTTP/2 stream // Used for notifying LLM stream tracker about streaming responses type Http2StreamUpdate struct { @@ -234,14 +262,20 @@ func (p *Http2Parser) decodeHeaderBlock( case MethodHttp2ClientFrames: req := p.activeRequests[streamId] if req == nil { + if len(p.activeRequests) >= maxActiveRequests { + // Too many active streams; skip to prevent unbounded growth + break + } req = &Http2Request{ - kernelTime: kernelTime, - RequestHeaders: make(map[string]string), + kernelTime: kernelTime, + } + if !p.Lightweight { + req.RequestHeaders = make(map[string]string) } p.activeRequests[streamId] = req } decoder.SetEmitFunc(func(hf hpack.HeaderField) { - // Store all headers for trace correlation + // Store all headers for trace correlation (full mode only) if req.RequestHeaders != nil { req.RequestHeaders[hf.Name] = hf.Value } @@ -262,6 +296,13 @@ func (p *Http2Parser) decodeHeaderBlock( case ":authority": if req.Authority == "" && hf.Value != "" { req.Authority = hf.Value + // Auto-upgrade: if authority matches an LLM host, switch to full mode + if p.Lightweight && p.LLMHostChecker != nil && p.LLMHostChecker(hf.Value) { + p.Lightweight = false + if req.RequestHeaders == nil { + req.RequestHeaders = make(map[string]string) + } + } } case "content-type": if req.ContentType == "" && hf.Value != "" { @@ -336,8 +377,10 @@ func (p *Http2Parser) Parse(method Method, payload []byte, kernelTime uint64) [] } var decoder *hpack.Decoder - statuses := map[uint32]Status{} - grpcStatuses := map[uint32]Status{} + clear(p.statuses) + clear(p.grpcStatuses) + statuses := p.statuses + grpcStatuses := p.grpcStatuses // Prepend any saved partial frame data from previous Parse() call // This handles HTTP/2 frames split across multiple S2A/TLS Read() calls @@ -403,37 +446,40 @@ frameLoop: offset = frameStart break frameLoop } - dataPayload := payload[offset : offset+h.Length] - - switch method { - case MethodHttp2ClientFrames: - // Client DATA frame = request payload (capped to prevent unbounded growth) - req := p.activeRequests[h.StreamId] - if req != nil && len(req.RequestPayload) < maxDataPayloadSize { - remaining := maxDataPayloadSize - len(req.RequestPayload) - if len(dataPayload) > remaining { - dataPayload = dataPayload[:remaining] - } - req.RequestPayload = append(req.RequestPayload, dataPayload...) + + // Track END_STREAM on server DATA frames unconditionally (needed for both modes) + if method == MethodHttp2ServerFrames && h.Flags&http2FlagEndStream != 0 { + if req := p.activeRequests[h.StreamId]; req != nil { + req.responseEndStream = true } - case MethodHttp2ServerFrames: - // Server DATA frame = response payload (capped to prevent unbounded growth) - req := p.activeRequests[h.StreamId] - if req != nil { - // Track first response time for TTFT - if req.firstResponseTime == 0 && len(dataPayload) > 0 { - req.firstResponseTime = kernelTime - } - if len(req.ResponsePayload) < maxDataPayloadSize { - remaining := maxDataPayloadSize - len(req.ResponsePayload) + } + + // Payload accumulation only in full mode + if !p.Lightweight { + dataPayload := payload[offset : offset+h.Length] + switch method { + case MethodHttp2ClientFrames: + req := p.activeRequests[h.StreamId] + if req != nil && len(req.RequestPayload) < maxDataPayloadSize { + remaining := maxDataPayloadSize - len(req.RequestPayload) if len(dataPayload) > remaining { dataPayload = dataPayload[:remaining] } - req.ResponsePayload = append(req.ResponsePayload, dataPayload...) + req.RequestPayload = append(req.RequestPayload, dataPayload...) } - // Check for END_STREAM flag on DATA frame - if h.Flags&http2FlagEndStream != 0 { - req.responseEndStream = true + case MethodHttp2ServerFrames: + req := p.activeRequests[h.StreamId] + if req != nil { + if req.firstResponseTime == 0 && len(dataPayload) > 0 { + req.firstResponseTime = kernelTime + } + if len(req.ResponsePayload) < maxDataPayloadSize { + remaining := maxDataPayloadSize - len(req.ResponsePayload) + if len(dataPayload) > remaining { + dataPayload = dataPayload[:remaining] + } + req.ResponsePayload = append(req.ResponsePayload, dataPayload...) + } } } }