diff --git a/common/log_parser_test.go b/common/log_parser_test.go index 34924d5..af2f685 100644 --- a/common/log_parser_test.go +++ b/common/log_parser_test.go @@ -12,7 +12,11 @@ import ( func TestLogParser(t *testing.T) { const defaultPatternsPerLevel = 256 ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, nil, 1*time.Second, defaultPatternsPerLevel, false) + parser := logparser.NewParser(ch, nil, nil, 1*time.Second, defaultPatternsPerLevel, logparser.SensitiveConfig{ + Enabled: true, + MinConfidence: "high", + MaxDetections: 100, + }) ch <- logparser.LogEntry{Timestamp: time.Now(), Content: "INFO:root:AWS access key: AKIAIOSFODNN7EXAMPLE", Level: logparser.LevelInfo} diff --git a/containers/container.go b/containers/container.go index 445a6f0..d7f26fe 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 { @@ -96,6 +102,9 @@ type ActiveConnection struct { http2Parser *l7.Http2Parser postgresParser *l7.PostgresParser mysqlParser *l7.MysqlParser + + parseFailCount int + protocolOverride l7.Protocol // non-zero = override eBPF-detected protocol } type ListenDetails struct { @@ -128,6 +137,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 @@ -1081,8 +1091,16 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 trace = c.tracer.NewTrace(conn.DestinationKey.ActualDestinationIfKnown(), conn.srcWorkload, destWorkload, actualDestWorkload) } + // Protocol reclassification: if previous parse attempts failed repeatedly, + // the connection was likely misidentified by eBPF heuristics. Use the + // override to skip further parsing for this connection. + protocol := r.Protocol + if conn.protocolOverride != 0 { + protocol = conn.protocolOverride + } + // Process L7 requests and update metrics - switch r.Protocol { + switch protocol { case l7.ProtocolDNS: status := r.Status.DNS() if status == "" { @@ -1158,9 +1176,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 +1211,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 +1219,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 != "" { @@ -1272,6 +1293,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 conn.postgresParser = l7.NewPostgresParser() } query := conn.postgresParser.Parse(r.Payload) + if query == "" && r.Method != l7.MethodStatementClose { + c.trackParseFail(conn, pid, fd, r.Protocol) + } else { + conn.parseFailCount = 0 + } if trace != nil { trace.PostgresQuery(query, r.Status.Error(), r.Duration) } @@ -1321,6 +1347,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // Update stats for Clickhouse c.l7Stats.observe(r.Protocol, r.Status.String(), "", "", r.Duration, conn.DestinationKey, conn.srcWorkload, r, "") query := l7.ParseClickhouse(r.Payload) + if query == "" { + c.trackParseFail(conn, pid, fd, r.Protocol) + } else { + conn.parseFailCount = 0 + } if trace != nil { trace.ClickhouseQuery(query, r.Status.Error(), r.Duration) } @@ -1328,6 +1359,11 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 // Update stats for Zookeeper c.l7Stats.observe(r.Protocol, r.Status.Zookeeper(), "", "", r.Duration, conn.DestinationKey, conn.srcWorkload, r, "") op, arg := l7.ParseZookeeper(r.Payload) + if op == "" { + c.trackParseFail(conn, pid, fd, r.Protocol) + } else { + conn.parseFailCount = 0 + } if trace != nil { trace.ZookeeperRequest(op, arg, r.Status, r.Duration) } @@ -1341,53 +1377,93 @@ func (c *Container) onL7RequestWithResult(pid uint32, fd uint64, timestamp uint6 return nil, L7RequestProcessed } +const ( + parseFailThreshold = 3 + protocolReclassified = l7.Protocol(0xFF) // sentinel: no protocol matches, hits default case +) + +// trackParseFail tracks consecutive parse failures on a connection. After +// parseFailThreshold failures, the connection's protocol is overridden to +// skip further parsing. This handles eBPF protocol misidentification where +// weak heuristics (e.g., 3-byte ClickHouse check) tag a non-matching +// connection permanently. +func (c *Container) trackParseFail(conn *ActiveConnection, pid uint32, fd uint64, proto l7.Protocol) { + conn.parseFailCount++ + if conn.parseFailCount == parseFailThreshold { + conn.protocolOverride = protocolReclassified + klog.Warningf("reclassified connection pid=%d fd=%d from %s to unknown after %d consecutive parse failures", + pid, fd, proto, conn.parseFailCount) + } +} + // processHTTP2WithoutConnection handles HTTP/2 events when TCP connection tracking failed. // 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 +1471,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 +1479,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 +1498,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). @@ -1467,6 +1540,32 @@ type activeConnAgg struct { count int } +// activeCounterLabelKeys returns the set of labelKey() values for destinations +// that have at least one open connection, and the set of destination strings +// for destinations still in lastConnectionAttempts (for failed-connect eviction). +// Called from the event handler goroutine. +func (c *Container) activeCounterLabelKeys() (activeKeys map[string]struct{}, activeFailedDsts map[string]struct{}) { + c.lock.RLock() + defer c.lock.RUnlock() + + activeKeys = make(map[string]struct{}) + for _, conn := range c.activeConnections { + if !conn.Closed.IsZero() { + continue + } + enrichedKey := c.enrichDestinationKey(conn.DestinationKey) + labels := tcpLabels(enrichedKey, conn.srcWorkload) + activeKeys[labelKey(labels)] = struct{}{} + } + + activeFailedDsts = make(map[string]struct{}, len(c.lastConnectionAttempts)) + for dst := range c.lastConnectionAttempts { + activeFailedDsts[dst.String()] = struct{}{} + } + + return activeKeys, activeFailedDsts +} + func (c *Container) onRetransmission(src netaddr.IPPort, dst netaddr.IPPort) bool { c.lock.RLock() conn, ok := c.activeConnections[ConnectionKey{src: src, dst: dst}] @@ -1746,13 +1845,19 @@ func (c *Container) runLogParser(logPath string) { } containerId := string(c.id) + sensitiveCfg := logparser.SensitiveConfig{ + Enabled: !*flags.DisableSensitiveLogParsing, + SampleRate: *flags.SensitiveLogSampleRate, + MinConfidence: *flags.SensitiveLogMinConfidence, + MaxDetections: *flags.SensitiveLogMaxDetectionsPerContainer, + } if logPath != "" { if c.logParsers[logPath] != nil { return } ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, sensitiveCfg) reader, err := logs.NewTailReader(proc.HostPath(logPath), ch) if err != nil { klog.Warningln(err) @@ -1772,7 +1877,7 @@ func (c *Container) runLogParser(logPath string) { klog.Warningln(err) return } - parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, nil, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, sensitiveCfg) stop := func() { JournaldUnsubscribe(c.metadata.systemd.Unit) } @@ -1789,7 +1894,7 @@ func (c *Container) runLogParser(logPath string) { delete(c.logParsers, "stdout/stderr") } ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, *flags.DisableSensitiveLogParsing) + parser := logparser.NewParser(ch, c.metadata.logDecoder, logs.OtelLogEmitter(containerId), multilineCollectorTimeout, *flags.LogPatternsPerContainer, sensitiveCfg) reader, err := logs.NewTailReader(proc.HostPath(c.metadata.logPath), ch) if err != nil { klog.Warningln(err) @@ -1851,14 +1956,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/containers/registry.go b/containers/registry.go index 88820dc..6980794 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -222,10 +222,14 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { defer gcTicker.Stop() ebpfStatsTicker := time.NewTicker(MinTrafficStatsUpdateInterval) defer ebpfStatsTicker.Stop() + counterEvictTicker := time.NewTicker(gcInterval) + defer counterEvictTicker.Stop() for { select { case <-ebpfStatsTicker.C: r.updateEbpfStatsAndActiveConns() + case <-counterEvictTicker.C: + r.evictStaleCounterLabels() case now := <-gcTicker.C: for pid, c := range r.containersByPid { cg, err := proc.ReadCgroup(pid) @@ -403,6 +407,11 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) { const maxIP2FQDNEntries = 10000 func (r *Registry) processL7Event(e ebpftracer.Event) { + defer func() { + if p := recover(); p != nil { + klog.Errorf("recovered from panic in L7 event handler: pid=%d fd=%d protocol=%d: %v", e.Pid, e.Fd, e.L7Request.Protocol, p) + } + }() if c := r.containersByPid[e.Pid]; c != nil { klog.V(5).Infof("L7_EVENT_CONTAINER_FOUND: pid=%d container=%s", e.Pid, c.id) ip2fqdn, result := c.onL7RequestWithResult(e.Pid, e.Fd, e.Timestamp, e.L7Request, e.SocketInfo) @@ -491,7 +500,10 @@ func (r *Registry) processPendingL7Events() { continue } - ip2fqdn, result := c.onL7RequestWithResult(p.event.Pid, p.event.Fd, p.event.Timestamp, p.event.L7Request, p.event.SocketInfo) + ip2fqdn, result, panicked := r.safeOnL7Request(c, p.event) + if panicked { + continue + } if result == L7RequestConnNotFound { // Still not found - keep in queue if not exceeded max retries if p.retryCount < maxRetries { @@ -521,6 +533,17 @@ func (r *Registry) processPendingL7Events() { } } +func (r *Registry) safeOnL7Request(c *Container, e ebpftracer.Event) (ip2fqdn map[netaddr.IP]*common.Domain, result L7RequestResult, panicked bool) { + defer func() { + if p := recover(); p != nil { + klog.Errorf("recovered from panic in pending L7 retry: pid=%d fd=%d protocol=%d: %v", e.Pid, e.Fd, e.L7Request.Protocol, p) + panicked = true + } + }() + ip2fqdn, result = c.onL7RequestWithResult(e.Pid, e.Fd, e.Timestamp, e.L7Request, e.SocketInfo) + return +} + func (r *Registry) getOrCreateContainer(pid uint32) *Container { // Fast path: try to find existing container with read lock lockStart := time.Now() @@ -738,6 +761,22 @@ func (r *Registry) updateEbpfStatsAndActiveConns() { } } +// evictStaleCounterLabels removes CounterVec entries for destinations that no +// longer have active connections. Runs at gcInterval from the event handler goroutine. +func (r *Registry) evictStaleCounterLabels() { + r.containerLock.RLock() + containers := make([]*Container, 0, len(r.containersById)) + for _, c := range r.containersById { + containers = append(containers, c) + } + r.containerLock.RUnlock() + + for _, c := range containers { + activeKeys, activeFailedDsts := c.activeCounterLabelKeys() + c.tcpMetrics.EvictStaleLabels(activeKeys, activeFailedDsts) + } +} + func (r *Registry) getDomain(ip netaddr.IP) *common.Domain { r.ip2fqdnLock.RLock() defer r.ip2fqdnLock.RUnlock() diff --git a/containers/tcp_metrics.go b/containers/tcp_metrics.go index d3e20fd..fb955fd 100644 --- a/containers/tcp_metrics.go +++ b/containers/tcp_metrics.go @@ -1,6 +1,7 @@ package containers import ( + "strings" "sync" "github.com/coroot/coroot-node-agent/common" @@ -11,7 +12,7 @@ import ( // Event handlers update these directly (lock-free for scrapes). // Collect() just forwards pre-built metrics without holding c.lock. type TCPMetrics struct { - mu sync.RWMutex // protects lazy initialization only + mu sync.RWMutex // protects initialization and eviction vs collect successful *prometheus.CounterVec totalTime *prometheus.CounterVec @@ -23,6 +24,14 @@ type TCPMetrics struct { restarts prometheus.Counter oomKills prometheus.Counter + // knownLabels tracks label combinations pushed to the 11-label CounterVecs. + // Key is null-byte-joined label values; value is the original slice for DeleteLabelValues. + // Written only from the event handler goroutine (no extra lock needed for writes). + knownLabels map[string][]string + + // knownFailedLabels tracks label combinations pushed to the 7-label failed CounterVec. + knownFailedLabels map[string][]string + constLabels prometheus.Labels initialized bool } @@ -74,9 +83,33 @@ func (t *TCPMetrics) ensureInitialized() { t.restarts = newCounter("container_restarts_total", "Number of times the container was restarted", cl) t.oomKills = newCounter("container_oom_kills_total", "Total number of times the container was terminated by the OOM killer", cl) + t.knownLabels = make(map[string][]string) + t.knownFailedLabels = make(map[string][]string) t.initialized = true } +// labelKey produces a map key from a label slice using null-byte separator. +func labelKey(labels []string) string { + return strings.Join(labels, "\x00") +} + +// trackLabels records a label combination for the 11-label CounterVecs. +// Only called from the event handler goroutine — no extra lock needed. +func (t *TCPMetrics) trackLabels(labels []string) { + k := labelKey(labels) + if _, exists := t.knownLabels[k]; !exists { + t.knownLabels[k] = labels + } +} + +// trackFailedLabels records a label combination for the 7-label failed CounterVec. +func (t *TCPMetrics) trackFailedLabels(labels []string) { + k := labelKey(labels) + if _, exists := t.knownFailedLabels[k]; !exists { + t.knownFailedLabels[k] = labels + } +} + // tcpLabels builds the 11-label value slice for TCP metrics. func tcpLabels(key common.DestinationKey, src common.Workload) []string { dest := key.GetDestinationWorkload() @@ -93,6 +126,7 @@ func tcpLabels(key common.DestinationKey, src common.Workload) []string { func (t *TCPMetrics) ObserveConnectionOpen(key common.DestinationKey, src common.Workload, durationSeconds float64) { t.ensureInitialized() labels := tcpLabels(key, src) + t.trackLabels(labels) t.successful.WithLabelValues(labels...).Inc() t.totalTime.WithLabelValues(labels...).Add(durationSeconds) } @@ -100,23 +134,28 @@ func (t *TCPMetrics) ObserveConnectionOpen(key common.DestinationKey, src common // ObserveConnectionFailed records a failed connection attempt. func (t *TCPMetrics) ObserveConnectionFailed(dst common.HostPort, workload common.Workload) { t.ensureInitialized() - t.failed.WithLabelValues( + labels := []string{ dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind, - ).Inc() + } + t.trackFailedLabels(labels) + t.failed.WithLabelValues(labels...).Inc() } // ObserveRetransmission records a TCP retransmission. func (t *TCPMetrics) ObserveRetransmission(key common.DestinationKey, src common.Workload) { t.ensureInitialized() - t.retransmits.WithLabelValues(tcpLabels(key, src)...).Inc() + labels := tcpLabels(key, src) + t.trackLabels(labels) + t.retransmits.WithLabelValues(labels...).Inc() } // ObserveTraffic records byte count deltas for a connection. func (t *TCPMetrics) ObserveTraffic(key common.DestinationKey, src common.Workload, sentDelta, recvDelta uint64) { t.ensureInitialized() labels := tcpLabels(key, src) + t.trackLabels(labels) if sentDelta > 0 { t.bytesSent.WithLabelValues(labels...).Add(float64(sentDelta)) } @@ -153,6 +192,39 @@ func (t *TCPMetrics) ObserveOOMKill() { t.oomKills.Inc() } +// EvictStaleLabels removes CounterVec entries for label combinations that are +// no longer active. activeKeys contains labelKey() values for currently open +// connections. activeFailedDsts contains destination strings (labels[0]) for +// destinations still in lastConnectionAttempts. +// Must be called from the event handler goroutine. Takes mu.Lock to exclude +// concurrent collect() calls during eviction. +func (t *TCPMetrics) EvictStaleLabels(activeKeys map[string]struct{}, activeFailedDsts map[string]struct{}) { + t.mu.Lock() + defer t.mu.Unlock() + if !t.initialized { + return + } + + for k, labels := range t.knownLabels { + if _, active := activeKeys[k]; !active { + t.successful.DeleteLabelValues(labels...) + t.totalTime.DeleteLabelValues(labels...) + t.retransmits.DeleteLabelValues(labels...) + t.bytesSent.DeleteLabelValues(labels...) + t.bytesRecv.DeleteLabelValues(labels...) + delete(t.knownLabels, k) + } + } + + for k, labels := range t.knownFailedLabels { + // labels[0] is the destination string + if _, active := activeFailedDsts[labels[0]]; !active { + t.failed.DeleteLabelValues(labels...) + delete(t.knownFailedLabels, k) + } + } +} + // collect forwards all pre-built metrics to the channel. No c.lock needed. func (t *TCPMetrics) collect(ch chan<- prometheus.Metric) { t.mu.RLock() diff --git a/ebpftracer/ebpf/l7/l7.c b/ebpftracer/ebpf/l7/l7.c index c6a2169..9c473c4 100644 --- a/ebpftracer/ebpf/l7/l7.c +++ b/ebpftracer/ebpf/l7/l7.c @@ -408,8 +408,6 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size, req->protocol = PROTOCOL_RABBITMQ; } else if (is_amqp_method_frame(payload, size)) { req->protocol = PROTOCOL_RABBITMQ; - } else if (is_clickhouse_query(payload, size)) { - req->protocol = PROTOCOL_CLICKHOUSE; } else if (is_zk_request(payload, total_size)) { req->protocol = PROTOCOL_ZOOKEEPER; } else if (is_kafka_request(payload, size, &req->request_id)) { diff --git a/ebpftracer/l7/clickhouse.go b/ebpftracer/l7/clickhouse.go index a31a25f..626efba 100644 --- a/ebpftracer/l7/clickhouse.go +++ b/ebpftracer/l7/clickhouse.go @@ -2,10 +2,13 @@ package l7 import ( "bytes" + "io" "github.com/ClickHouse/ch-go/proto" ) +const clickhouseClientCodeQuery = 1 + func ParseClickhouse(payload []byte) (result string) { // Recover from panics caused by malformed/incomplete packets defer func() { @@ -14,7 +17,28 @@ func ParseClickhouse(payload []byte) (result string) { } }() - r := proto.NewReader(bytes.NewReader(payload)) + // Layer 2: Structural validation before invoking ch-go. + // Reject misidentified payloads early, before the library can + // decode garbage varint lengths and attempt unbounded allocations. + if len(payload) < 3 { + return "" + } + if payload[0] != clickhouseClientCodeQuery { + return "" + } + // Query ID length is varint-encoded. Single-byte varints (0-127) cover + // all practical query IDs. Multi-byte varints (>127) in this position + // indicate garbage data from a misidentified connection. + if payload[1] > 127 { + return "" + } + + // Layer 1: Bound the reader to the actual payload size. + // ch-go's proto.Reader decodes varint string lengths and pre-allocates + // that many bytes. With corrupted data, the varint can decode to TB-scale + // values, causing a fatal OOM that bypasses defer/recover. + // LimitReader caps reads at len(payload), turning the OOM into io.EOF. + r := proto.NewReader(io.LimitReader(bytes.NewReader(payload), int64(len(payload)))) var err error if _, err = r.Byte(); err != nil { return "" diff --git a/ebpftracer/l7/http2.go b/ebpftracer/l7/http2.go index 6b2169b..7c484d5 100644 --- a/ebpftracer/l7/http2.go +++ b/ebpftracer/l7/http2.go @@ -37,6 +37,16 @@ const ( // Max accumulated header block size (64KB) to prevent unbounded growth maxPendingHeaderBlockSize = 64 * 1024 + + // Max accumulated DATA payload size per stream (128KB). + // Prevents unbounded memory growth on high-throughput HTTP/2 connections. + // 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 { @@ -90,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 @@ -104,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), } } @@ -135,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 { @@ -228,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 } @@ -256,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 != "" { @@ -330,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 @@ -397,27 +446,40 @@ frameLoop: offset = frameStart break frameLoop } - dataPayload := payload[offset : offset+h.Length] - switch method { - case MethodHttp2ClientFrames: - // Client DATA frame = request payload - req := p.activeRequests[h.StreamId] - if req != nil { - 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 - req := p.activeRequests[h.StreamId] - if req != nil { - // Track first response time for TTFT - if req.firstResponseTime == 0 && len(dataPayload) > 0 { - req.firstResponseTime = kernelTime + } + + // 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.RequestPayload = append(req.RequestPayload, dataPayload...) } - req.ResponsePayload = append(req.ResponsePayload, 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...) + } } } } diff --git a/ebpftracer/l7/postgres.go b/ebpftracer/l7/postgres.go index 148a120..5d23c25 100644 --- a/ebpftracer/l7/postgres.go +++ b/ebpftracer/l7/postgres.go @@ -26,6 +26,15 @@ func (p *PostgresParser) Parse(payload []byte) string { return "" } cmd := payload[0] + // Reject unknown frame types early — Postgres detection in eBPF is weak + // (single-byte check), so misidentified connections can reach here. + // Accept all valid frontend message types per the Postgres wire protocol. + switch cmd { + case PostgresFrameQuery, PostgresFrameBind, PostgresFrameParse, PostgresFrameClose, + 'E', 'D', 'S', 'H', 'X': // Execute, Describe, Sync, Flush, Terminate + default: + return "" + } switch cmd { case PostgresFrameQuery: var query string diff --git a/ebpftracer/l7/zookeeper.go b/ebpftracer/l7/zookeeper.go index 71a04d5..0f40d5a 100644 --- a/ebpftracer/l7/zookeeper.go +++ b/ebpftracer/l7/zookeeper.go @@ -108,11 +108,25 @@ func zkReadString(r io.Reader) string { return string(res[:n]) } +func isValidZkOp(op int32) bool { + switch op { + case zkOpCreate, zkOpDelete, zkOpExists, zkOpGetData, zkOpSetData, + zkOpGetAcl, zkOpSetAcl, zkOpGetChildren, zkOpSync, zkOpPing, + zkOpGetChildren2, zkOpCheck, zkOpMulti, zkOpReconfig, + zkOpCreateContainer, zkOpCreateTTL, zkOpClose, zkOpSetAuth, zkOpSetWatches: + return true + } + return false +} + func ParseZookeeper(payload []byte) (string, string) { r := bytes.NewReader(payload) h := zkRequestHeader{} if err := binary.Read(r, binary.BigEndian, &h); err != nil { return "", "" } + if !isValidZkOp(h.OpType) { + return "", "" + } return zkParse(r, h.OpType) } diff --git a/ebpftracer/tls.go b/ebpftracer/tls.go index b5eb709..986fe90 100644 --- a/ebpftracer/tls.go +++ b/ebpftracer/tls.go @@ -10,12 +10,12 @@ import ( "os" "regexp" "strings" - "sync" "unsafe" "github.com/cilium/ebpf/link" "github.com/coroot/coroot-node-agent/common" "github.com/coroot/coroot-node-agent/proc" + lru "github.com/hashicorp/golang-lru/v2" "k8s.io/klog/v2" ) @@ -53,8 +53,9 @@ var ( // strippedGoExeCache caches exe paths that are Go binaries but have no TLS symbols. // This avoids expensive ELF scanning for the same stripped binary across many - // short-lived processes (e.g., kubectl invocations). Uses sync.Map for lock-free reads. - strippedGoExeCache = &sync.Map{} // map[string]struct{} + // short-lived processes (e.g., kubectl invocations). Capped at 1000 entries to + // prevent unbounded growth on CI/CD nodes with many unique Go binaries. + strippedGoExeCache, _ = lru.New[string, struct{}](1000) ) func (t *Tracer) AttachOpenSslUprobes(pid uint32) []link.Link { @@ -181,7 +182,7 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { // Skip binaries we already know are stripped (no TLS symbols). // This avoids expensive ELF scanning for repeated short-lived processes like kubectl. - if _, stripped := strippedGoExeCache.Load(exeName); stripped { + if strippedGoExeCache.Contains(exeName) { klog.V(3).Infof("GO_TLS_SKIP_STRIPPED: pid=%d exe=%s", pid, exeName) return nil, true // still a Go app, just stripped } @@ -258,7 +259,7 @@ func (t *Tracer) AttachGoTlsUprobes(pid uint32) ([]link.Link, bool) { log("failed to get write symbol", err) // Cache this exe as stripped to skip future attempts if exeName != "" { - strippedGoExeCache.Store(exeName, struct{}{}) + strippedGoExeCache.Add(exeName, struct{}{}) } return nil, isGolangApp } diff --git a/ebpftracer/tracer.go b/ebpftracer/tracer.go index 0f00b30..26ec9de 100644 --- a/ebpftracer/tracer.go +++ b/ebpftracer/tracer.go @@ -448,6 +448,15 @@ func safeDuration(ns uint64) time.Duration { return time.Duration(ns) } +// clampSize safely converts a uint64 size to int, capping at maxSize. +// Prevents negative int from uint64 values with the high bit set. +func clampSize(size uint64, maxSize int) int { + if size > uint64(maxSize) { + return maxSize + } + return int(size) +} + func (t *lostSamplesTracker) recordLostSamples(name string, count uint64, cpu int) { t.count.Add(count) now := time.Now().Unix() @@ -492,14 +501,8 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy } payloadSize := binary.LittleEndian.Uint64(data[40:48]) responseSize := binary.LittleEndian.Uint64(data[48:56]) - payloadLen := int(payloadSize) - if payloadLen > MaxPayloadSize { - payloadLen = MaxPayloadSize - } - responseLen := int(responseSize) - if responseLen > MaxPayloadSize { - responseLen = MaxPayloadSize - } + payloadLen := clampSize(payloadSize, MaxPayloadSize) + responseLen := clampSize(responseSize, MaxPayloadSize) payloadData := make([]byte, payloadLen) if l7EventHeaderSize+payloadLen <= len(data) { @@ -607,15 +610,8 @@ func runRingbufEventsReader(name string, r *ringbuf.Reader, ch chan<- Event) { payloadSize := binary.LittleEndian.Uint64(data[40:48]) responseSize := binary.LittleEndian.Uint64(data[48:56]) - - payloadLen := int(payloadSize) - if payloadLen > MaxPayloadSize { - payloadLen = MaxPayloadSize - } - responseLen := int(responseSize) - if responseLen > MaxPayloadSize { - responseLen = MaxPayloadSize - } + payloadLen := clampSize(payloadSize, MaxPayloadSize) + responseLen := clampSize(responseSize, MaxPayloadSize) payloadData := make([]byte, payloadLen) if l7EventHeaderSize+payloadLen <= len(data) { diff --git a/flags/flags.go b/flags/flags.go index d460b74..9a256ca 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -52,17 +52,20 @@ var ( ProfilesEndpoint = kingpin.Flag("profiles-endpoint", "The URL of the endpoint to send profiles to").Envar("PROFILES_ENDPOINT").URL() InsecureSkipVerify = kingpin.Flag("insecure-skip-verify", "whether to skip verifying the certificate or not").Envar("INSECURE_SKIP_VERIFY").Default("false").Bool() - ScrapeInterval = kingpin.Flag("scrape-interval", "How often to gather metrics from the agent").Default("15s").Envar("SCRAPE_INTERVAL").Duration() - WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/coroot-node-agent").Envar("WAL_DIR").String() - MaxSpoolSize = kingpin.Flag("max-spool-size", "Maximum size of the on-disk spool used to buffer data when it cannot be sent to collector. Supports size suffixes like KB, MB, or GB.").Default("500MB").Envar("MAX_SPOOL_SIZE").Bytes() - ResolveDns = kingpin.Flag("resolve-dns", "should resolve DNS").Default("false").Envar("RESOLVE_DNS").Bool() - IgnoreControlPlane = kingpin.Flag("ignore-control-plane", "ignore control plane like loki").Default("karpenter,loki,prometheus,grafana,kubelet,etcd,apiserver,victoria,nudgebee-agent,kube-system").Envar("IGNORE_CONTROL_PLANE").String() - SanitizeHeaders = kingpin.Flag("sanitize-headers", "should sanitize headers").Default("true").Envar("SANITIZE_HEADERS").Bool() - SensitiveHeader = kingpin.Flag("sensitive-headers", "sanitize headers using patterns").Default("Authorization, Cookie, X-Action-Token").Envar("SENSITIVE_HEADERS").String() - DisableKubeProbe = kingpin.Flag("disable-kube-probe", "disable kube probe trace").Default("true").Envar("DISABLE_KUBE_PROBE").Bool() - EnableDynamicLogTailing = kingpin.Flag("enable-dynamic-log-tailing", "Enable automatic tailing of log files opened by container processes.").Default("false").Envar("ENABLE_DYNAMIC_LOG_TAILING").Bool() - DisableSensitiveLogParsing = kingpin.Flag("disable-sensitive-log-parsing", "disable sensitive log parsing").Default("true").Envar("DISABLE_SENSITIVE_LOG_PARSING").Bool() - TraceIdHeaders = kingpin.Flag("trace-id-headers", "trace id headers").Default("Traceparent,X-Request-Id").Envar("TRACE_ID_HEADERS").String() + ScrapeInterval = kingpin.Flag("scrape-interval", "How often to gather metrics from the agent").Default("15s").Envar("SCRAPE_INTERVAL").Duration() + WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/coroot-node-agent").Envar("WAL_DIR").String() + MaxSpoolSize = kingpin.Flag("max-spool-size", "Maximum size of the on-disk spool used to buffer data when it cannot be sent to collector. Supports size suffixes like KB, MB, or GB.").Default("500MB").Envar("MAX_SPOOL_SIZE").Bytes() + ResolveDns = kingpin.Flag("resolve-dns", "should resolve DNS").Default("false").Envar("RESOLVE_DNS").Bool() + IgnoreControlPlane = kingpin.Flag("ignore-control-plane", "ignore control plane like loki").Default("karpenter,loki,prometheus,grafana,kubelet,etcd,apiserver,victoria,nudgebee-agent,kube-system").Envar("IGNORE_CONTROL_PLANE").String() + SanitizeHeaders = kingpin.Flag("sanitize-headers", "should sanitize headers").Default("true").Envar("SANITIZE_HEADERS").Bool() + SensitiveHeader = kingpin.Flag("sensitive-headers", "sanitize headers using patterns").Default("Authorization, Cookie, X-Action-Token").Envar("SENSITIVE_HEADERS").String() + DisableKubeProbe = kingpin.Flag("disable-kube-probe", "disable kube probe trace").Default("true").Envar("DISABLE_KUBE_PROBE").Bool() + EnableDynamicLogTailing = kingpin.Flag("enable-dynamic-log-tailing", "Enable automatic tailing of log files opened by container processes.").Default("false").Envar("ENABLE_DYNAMIC_LOG_TAILING").Bool() + DisableSensitiveLogParsing = kingpin.Flag("disable-sensitive-log-parsing", "disable sensitive log parsing").Default("true").Envar("DISABLE_SENSITIVE_LOG_PARSING").Bool() + SensitiveLogSampleRate = kingpin.Flag("sensitive-log-sample-rate", "Sample rate for sensitive log detection: check 1-in-N lines (0 or 1 = every line)").Default("100").Envar("SENSITIVE_LOG_SAMPLE_RATE").Int() + SensitiveLogMinConfidence = kingpin.Flag("sensitive-log-min-confidence", "Minimum confidence level for sensitive patterns: high, medium, low").Default("medium").Envar("SENSITIVE_LOG_MIN_CONFIDENCE").String() + SensitiveLogMaxDetectionsPerContainer = kingpin.Flag("sensitive-log-max-detections-per-container", "Max unique sensitive patterns tracked per container (0 = unlimited)").Default("100").Envar("SENSITIVE_LOG_MAX_DETECTIONS_PER_CONTAINER").Int() + TraceIdHeaders = kingpin.Flag("trace-id-headers", "trace id headers").Default("Traceparent,X-Request-Id").Envar("TRACE_ID_HEADERS").String() HttpPathNormalizationRules = kingpin.Flag("http-path-normalization-rules", "Custom HTTP path normalization rules in format 'pattern1:replacement1,pattern2:replacement2'").Envar("HTTP_PATH_NORMALIZATION_RULES").String() diff --git a/go.mod b/go.mod index 2e8b940..3094580 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/jpillora/backoff v1.0.0 github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14 - github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928 + github.com/nudgebee/logparser v0.0.0-20260406040008-c408d0a5c69c github.com/opencontainers/runtime-spec v1.1.0 github.com/prometheus/client_golang v1.20.5 github.com/prometheus/client_model v0.6.1 @@ -40,7 +40,7 @@ require ( go.opentelemetry.io/otel/trace v1.40.0 golang.org/x/arch v0.4.0 golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa - golang.org/x/net v0.46.0 + golang.org/x/net v0.48.0 golang.org/x/sys v0.40.0 golang.org/x/time v0.14.0 gopkg.in/alecthomas/kingpin.v2 v2.2.6 @@ -127,14 +127,14 @@ require ( github.com/josharian/intern v1.0.0 // indirect github.com/josharian/native v1.1.0 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.17.11 // indirect + github.com/klauspost/compress v1.18.0 // indirect github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect github.com/mackerelio/go-osstat v0.2.5 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mailru/easyjson v0.7.7 // indirect github.com/mdlayher/genetlink v1.3.2 // indirect github.com/mdlayher/netlink v1.7.2 // indirect - github.com/mdlayher/socket v0.4.1 // indirect + github.com/mdlayher/socket v0.5.1 // indirect github.com/mitchellh/go-wordwrap v1.0.1 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect @@ -149,7 +149,7 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/oklog/ulid v1.3.1 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect - github.com/opencontainers/image-spec v1.1.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect github.com/opencontainers/selinux v1.13.0 // indirect github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect @@ -167,8 +167,8 @@ require ( github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.11.0 // indirect github.com/spf13/cast v1.7.0 // indirect - github.com/spf13/cobra v1.8.1 // indirect - github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace // indirect + github.com/spf13/cobra v1.9.1 // indirect + github.com/spf13/pflag v1.0.7 // indirect github.com/spf13/viper v1.19.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/ulikunitz/xz v0.5.14 // indirect @@ -187,15 +187,15 @@ require ( go4.org/intern v0.0.0-20211027215823-ae77deb06f29 // indirect go4.org/netipx v0.0.0-20231129151722-fdeea329fbba // indirect go4.org/unsafe/assume-no-moving-gc v0.0.0-20230525183740-e7c30c78aeb2 // indirect - golang.org/x/oauth2 v0.33.0 // indirect - golang.org/x/sync v0.18.0 // indirect - golang.org/x/term v0.36.0 // indirect - golang.org/x/text v0.30.0 // indirect - golang.org/x/tools v0.38.0 // indirect + golang.org/x/oauth2 v0.34.0 // indirect + golang.org/x/sync v0.19.0 // indirect + golang.org/x/term v0.38.0 // indirect + golang.org/x/text v0.32.0 // indirect + golang.org/x/tools v0.39.0 // indirect google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect - google.golang.org/grpc v1.76.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect + google.golang.org/grpc v1.79.3 // indirect google.golang.org/protobuf v1.36.10 // indirect gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect diff --git a/go.sum b/go.sum index 35f7c4c..2bf2129 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -cel.dev/expr v0.24.0 h1:56OvJKSH3hDGL0ml5uSxZmz3/3Pq4tJ+fb1unVLAFcY= -cel.dev/expr v0.24.0/go.mod h1:hLPLo1W4QUmuYdA72RBX06QTs6MXw941piREPl3Yfiw= +cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= +cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= cyphar.com/go-pathrs v0.2.1 h1:9nx1vOgwVvX1mNBWDu93+vaceedpbsDqo+XuBGL40b8= cyphar.com/go-pathrs v0.2.1/go.mod h1:y8f1EMG7r+hCuFf/rXsKqMJrJAUoADZGNh5/vZPKcGc= @@ -58,8 +58,8 @@ github.com/cilium/workerpool v1.2.0 h1:Wc2iOPTvCgWKQXeq4L5tnx4QFEI+z5q1+bSpSS0cn github.com/cilium/workerpool v1.2.0/go.mod h1:GOYJhwlnIjR+jWSDNBb5kw47G1H/XA9X4WOBpgr4pQU= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= -github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443 h1:aQ3y1lwWyqYPiWZThqv1aFbZMiM9vblcSArJRf2Irls= -github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8= +github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5 h1:6xNmx7iTtyBRev0+D/Tv1FZd4SCg8axKApyNyRsAt/w= +github.com/cncf/xds/go v0.0.0-20251210132809-ee656c7534f5/go.mod h1:KdCmV+x/BuvyMxRnYBlmVaq4OLiKW6iRQfvC62cvdkI= github.com/containerd/cgroups v1.1.0 h1:v8rEWFl6EoqHB+swVNjVoCJE8o3jX7e8nqBGPLaDFBM= github.com/containerd/cgroups v1.1.0/go.mod h1:6ppBcbh/NOOUU+dMKrykgaBnK9lCIBxHqJDGwsa1mIw= github.com/containerd/containerd v1.7.29 h1:90fWABQsaN9mJhGkoVnuzEY+o1XDPbg9BTC9QTAHnuE= @@ -88,7 +88,7 @@ github.com/coroot/dotnetdiag v1.2.2 h1:PVP/By8o+xhPjfVolJYcjHLbFQInM7pkaD6/otPLc github.com/coroot/dotnetdiag v1.2.2/go.mod h1:veXCMlFzm1yNl7wwJb/ZLxO4WbzhDBoy1VG1XtkH2ls= github.com/coroot/pyroscope/ebpf v0.0.0-20250418092207-a70610b6df72 h1:NFEMXMOdUzQFG+OwRVecv/yc75/s8MmLkNquMm6hk3I= github.com/coroot/pyroscope/ebpf v0.0.0-20250418092207-a70610b6df72/go.mod h1:IepHM9FJ0n3n3k+ZV23Y7vNAfvWI7LDuLqWPO4rB6sQ= -github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o= +github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/cyphar/filepath-securejoin v0.6.0 h1:BtGB77njd6SVO6VztOHfPxKitJvd/VPT+OFBFMOi1Is= github.com/cyphar/filepath-securejoin v0.6.0/go.mod h1:A8hd4EnAeyujCJRrICiOWqjS1AX0a9kM5XL+NwKoYSc= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -112,8 +112,8 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= -github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8= -github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU= +github.com/envoyproxy/protoc-gen-validate v1.3.0 h1:TvGH1wof4H33rezVKWSpqKz5NXWg5VPuZ0uONDT6eb4= +github.com/envoyproxy/protoc-gen-validate v1.3.0/go.mod h1:HvYl7zwPa5mffgyeTUHA9zHIH36nmrm7oCbo4YKoSWA= github.com/evanphx/json-patch v5.9.0+incompatible h1:fBXyNpNMuTTDdquAq/uisOr2lShz4oaXpDTX2bLe7ls= github.com/evanphx/json-patch/v5 v5.9.0 h1:kcBlZQbplgElYIlo/n1hJbls2z/1awpXxpRi0/FOJfg= github.com/evanphx/json-patch/v5 v5.9.0/go.mod h1:VNkHZ/282BpEyt/tObQO8s5CMPmYYq14uClGH4abBuQ= @@ -254,8 +254,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc= -github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -290,8 +290,8 @@ github.com/mdlayher/netlink v1.4.1/go.mod h1:e4/KuJ+s8UhfUpO9z00/fDZZmhSrs+oxyqA github.com/mdlayher/netlink v1.7.2 h1:/UtM3ofJap7Vl4QWCPDGXY8d3GIY2UGSDbK+QWmY8/g= github.com/mdlayher/netlink v1.7.2/go.mod h1:xraEF7uJbxLhc5fpHL4cPe221LI2bdttWlU+ZGLfQSw= github.com/mdlayher/socket v0.0.0-20210307095302-262dc9984e00/go.mod h1:GAFlyu4/XV68LkQKYzKhIo/WW7j3Zi0YRAz/BOoanUc= -github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U= -github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA= +github.com/mdlayher/socket v0.5.1 h1:VZaqt6RkGkt2OE9l3GcC6nZkqD3xKeQLyfleW/uBcos= +github.com/mdlayher/socket v0.5.1/go.mod h1:TjPLHI1UgwEv5J1B5q0zTZq12A/6H7nKmtTanQE37IQ= github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14 h1:eKehnW2s+3DQYZLAa/Pm04sk1G+k8LlZt0OUDbyYmrI= github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14/go.mod h1:hDhp1SgOluLtKhnB65Wb/j3f7ghQWdOl+XIrbH9yqWc= github.com/mitchellh/go-wordwrap v1.0.1 h1:TLuKupo69TCn6TQSyGxwI1EblZZEsQ0vMlAFQflz0v0= @@ -323,8 +323,10 @@ github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= -github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928 h1:KExtH0Z1wo+yJgY2EJbZhF0YRLNcYZVASSN5gtZ513o= -github.com/nudgebee/logparser v0.0.0-20260218041043-99ea4437d928/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= +github.com/nudgebee/logparser v0.0.0-20260313095946-b22168b95d9c h1:C0Jx8+61ndlWeKrS25W7hFJW/qpxUpG96tBQdjffd5I= +github.com/nudgebee/logparser v0.0.0-20260313095946-b22168b95d9c/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= +github.com/nudgebee/logparser v0.0.0-20260406040008-c408d0a5c69c h1:drsGFCiBXp5OZm92HE0iVx8U4XhpzAxL5vVKubBGl+I= +github.com/nudgebee/logparser v0.0.0-20260406040008-c408d0a5c69c/go.mod h1:oFnM9D6YEjZzb1jy0kJ/NUkTbkZA+BgNYjpLO4n4szA= github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4= github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= @@ -334,8 +336,8 @@ github.com/onsi/gomega v1.35.1 h1:Cwbd75ZBPxFSuZ6T+rN/WCb/gOc6YgFBXLlZLhC7Ds4= github.com/onsi/gomega v1.35.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= -github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug= -github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/opencontainers/runtime-spec v1.1.0 h1:HHUyrt9mwHUjtasSbXSMvs4cyFxh+Bll4AjJ9odEGpg= github.com/opencontainers/runtime-spec v1.1.0/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0= github.com/opencontainers/selinux v1.13.0 h1:Zza88GWezyT7RLql12URvoxsbLfjFx988+LGaWfbL84= @@ -385,11 +387,11 @@ github.com/spf13/afero v1.11.0 h1:WJQKhtpdm3v2IzqG8VMqrr6Rf3UYpEF239Jy9wNepM8= github.com/spf13/afero v1.11.0/go.mod h1:GH9Y3pIexgf1MTIWtNGyogA5MwRIDXGUr+hbWNoBjkY= github.com/spf13/cast v1.7.0 h1:ntdiHjuueXFgm5nzDRdOS4yfT43P5Fnud6DH50rz/7w= github.com/spf13/cast v1.7.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= -github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM= -github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y= -github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace h1:9PNP1jnUjRhfmGMlkXHjYPishpcw4jpSt/V/xYY3FMA= -github.com/spf13/pflag v1.0.6-0.20210604193023-d5e0c0615ace/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo= +github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0= +github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/pflag v1.0.7 h1:vN6T9TfwStFPFM5XzjsvmzZkLuaLX+HS+0SeFLRgU6M= +github.com/spf13/pflag v1.0.7/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/viper v1.19.0 h1:RWq5SEjt8o25SROyN3z2OrDB9l7RPd3lwTWU8EcEdcI= github.com/spf13/viper v1.19.0/go.mod h1:GQUN9bilAbhU/jgc1bKs99f/suXKeUMct8Adx5+Ntkg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -498,18 +500,18 @@ golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4= -golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210= +golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= +golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= -golang.org/x/oauth2 v0.33.0 h1:4Q+qn+E5z8gPRJfmRy7C2gGG3T4jIprK6aSYgTXGRpo= -golang.org/x/oauth2 v0.33.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= +golang.org/x/oauth2 v0.34.0 h1:hqK/t4AKgbqWkdkcAeI8XLmbK+4m4G5YeQRrmiotGlw= +golang.org/x/oauth2 v0.34.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= -golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -540,13 +542,13 @@ golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q= -golang.org/x/term v0.36.0/go.mod h1:Qu394IJq6V6dCBRgwqshf3mPF85AqzYEzofzRdZkWss= +golang.org/x/term v0.38.0 h1:PQ5pkm/rLO6HnxFR7N2lJHOZX6Kez5Y1gDSJla6jo7Q= +golang.org/x/term v0.38.0/go.mod h1:bSEAKrOT1W+VSu9TSCMtoGEOUcKxOKgl3LE5QEF/xVg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= -golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= +golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU= +golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY= golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -558,8 +560,8 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= -golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ= -golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs= +golang.org/x/tools v0.39.0 h1:ik4ho21kwuQln40uelmciQPp9SipgNDdrafrYA4TmQQ= +golang.org/x/tools v0.39.0/go.mod h1:JnefbkDPyD8UU2kI5fuf8ZX4/yUeh9W877ZeBONxUqQ= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -573,17 +575,17 @@ google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo= google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 h1:9+tzLLstTlPTRyJTh+ah5wIMsBW5c4tQwGTN3thOW9Y= google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9/go.mod h1:mqHbVIp48Muh7Ywss/AD6I5kNVKZMmAa/QEW58Gxp2s= -google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba h1:B14OtaXuMaCQsl2deSvNkyPKIzq3BjfxQp8d00QyWx4= -google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:G5IanEx8/PgI9w6CFcYQf7jMtHQhZruvfM1i3qOqk5U= -google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba h1:UKgtfRM7Yh93Sya0Fo8ZzhDP4qBckrrxEr2oF5UIVb8= -google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= +google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 h1:fCvbg86sFXwdrl5LgVcTEvNC+2txB5mgROGmRL5mrls= +google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg= google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY= google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk= google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc= -google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A= -google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c= +google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= +google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= diff --git a/main.go b/main.go index b334db7..b200bfc 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "flag" "fmt" "log" @@ -163,16 +164,16 @@ func main() { if err != nil { klog.Errorf("Error starting resolver: %v", err) } + defer resolver.StopWatching() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() go func() { sigChannel := make(chan os.Signal, 1) - defer close(sigChannel) - signal.Notify(sigChannel, os.Interrupt, syscall.SIGTERM) <-sigChannel - klog.Infoln("Received signal, shutting down") - resolver.StopWatching() - os.Exit(0) // Ensure graceful termination + cancel() }() hostname, kv, err := uname() if err != nil { @@ -252,8 +253,21 @@ func main() { "Metrics collection timeout", )) + srv := &http.Server{Addr: *flags.ListenAddress} + go func() { + <-ctx.Done() + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + klog.Errorf("HTTP server shutdown error: %v", err) + } + }() + klog.Infoln("listening on:", *flags.ListenAddress) - klog.Errorln(http.ListenAndServe(*flags.ListenAddress, nil)) + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + klog.Errorln(err) + } + klog.Infoln("shutdown complete") } func info(name, version string) prometheus.Collector {