chore: test to prod

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}
 

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)

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()

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,30 +126,36 @@ 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)
 }
 
 // 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()

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)) {

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 ""