fix: IP-to-FQDN resolution, OpenSSL 3.x TLS capture, and log cleanup

common/net.go

@@ -224,12 +224,6 @@ func NewDomain(fqdn string, ips []netaddr.IP) *Domain {
 }
 
 func NewDestinationKey(dst, actualDst netaddr.IPPort, domain *Domain, dstWorkload Workload, actualDestWorkload Workload) DestinationKey {
-	// Debug logging for duplicate metrics investigation
-	if dst.IP().String() == "10.43.37.74" && dst.Port() == 8000 {
-		klog.Infof("DEBUG NewDestinationKey: dst=%s actualDst=%s domain=%v dstWorkload=%s actualDestWorkload=%s",
-			dst, actualDst, domain, dstWorkload.Name, actualDestWorkload.Name)
-	}
-
 	if IsIpExternal(actualDst.IP()) && domain != nil && !domain.SpecifyIP {
 		dstWorkload.Name = domain.FQDN
 		actualDestWorkload.Name = domain.FQDN
@@ -279,6 +273,13 @@ func NormalizeFQDN(fqdn string, requestType string) string {
 	return fqdn
 }
 
+func (dk DestinationKey) WithResolvedDomain(fqdn string) DestinationKey {
+	dk.destination = HostPortWithEmptyIP(fqdn, dk.destination.Port())
+	dk.destinationWorkload.Name = fqdn
+	dk.actualDestinationWorkload.Name = fqdn
+	return dk
+}
+
 func (dk DestinationKey) GetDestinationWorkload() Workload {
 	return dk.destinationWorkload
 }

containers/http_processor.go

@@ -87,13 +87,20 @@ func (ctx *HTTPRequestContext) parseRequest() {
 		return
 	}
 
-	// Parse method and path
-	ctx.Method, ctx.Path = l7.ParseHttp(ctx.RawPayload)
-
-	// Parse full HTTP request for headers
-	if req, err := l7.ParseHTTPRequest(ctx.RawPayload); err == nil && req != nil && req.Header != nil {
-		ctx.Headers = req.Header
+	req, err := l7.ParseHTTPRequest(ctx.RawPayload)
+	if err == nil && req != nil {
+		ctx.Method = req.Method
+		if req.URL != nil {
+			ctx.Path = req.URL.RequestURI()
+		}
+		if req.Header != nil {
+			ctx.Headers = req.Header
+		} else {
+			ctx.Headers = http.Header{}
+		}
 	} else {
+		// Fallback for malformed URIs: still extract method/path
+		ctx.Method, ctx.Path = l7.ParseHttp(ctx.RawPayload)
 		ctx.Headers = http.Header{}
 	}
 }

containers/l7.go

@@ -101,6 +101,7 @@ func normalizeHttpPath(path string) string {
 }
 
 type L7Stats struct {
+	mu          sync.RWMutex
 	requests    map[l7.Protocol]*prometheus.CounterVec
 	latency     map[l7.Protocol]*prometheus.HistogramVec
 	initialized map[l7.Protocol]bool
@@ -114,7 +115,7 @@ func NewL7Stats() L7Stats {
 	}
 }
 
-func (s L7Stats) observe(protocol l7.Protocol, status, method string, duration time.Duration, key common.DestinationKey, srcWorkload common.Workload, r *l7.RequestData, traceId string) {
+func (s *L7Stats) observe(protocol l7.Protocol, status, method, path string, duration time.Duration, key common.DestinationKey, srcWorkload common.Workload, r *l7.RequestData, traceId string) {
 	s.ensureInitialized(protocol)
 
 	actualDestWorkload := key.GetActualDestinationWorkload()
@@ -125,14 +126,16 @@ func (s L7Stats) observe(protocol l7.Protocol, status, method string, duration t
 		metricsProtocol = l7.ProtocolHTTP
 	}
 
+	destWorkload := key.GetDestinationWorkload()
+
 	// Base labels that all protocols use (with string interning for memory optimization)
 	labelValues := []string{
 		labelInterner.intern(status),
 		labelInterner.intern(key.DestinationLabelValue()),
 		labelInterner.intern(key.ActualDestinationLabelValue()),
-		labelInterner.intern(key.GetDestinationWorkload().Kind),
-		labelInterner.intern(key.GetDestinationWorkload().Name),
-		labelInterner.intern(key.GetDestinationWorkload().Namespace),
+		labelInterner.intern(destWorkload.Kind),
+		labelInterner.intern(destWorkload.Name),
+		labelInterner.intern(destWorkload.Namespace),
 		labelInterner.intern(srcWorkload.Kind),
 		labelInterner.intern(srcWorkload.Name),
 		labelInterner.intern(srcWorkload.Namespace),
@@ -141,6 +144,8 @@ func (s L7Stats) observe(protocol l7.Protocol, status, method string, duration t
 		labelInterner.intern(actualDestWorkload.Namespace),
 		labelInterner.intern(srcWorkload.Region),
 		labelInterner.intern(srcWorkload.Zone),
+		labelInterner.intern(destWorkload.Region),
+		labelInterner.intern(destWorkload.Zone),
 		labelInterner.intern(actualDestWorkload.Region),
 		labelInterner.intern(actualDestWorkload.Zone),
 		labelInterner.intern(actualDestWorkload.Instance),
@@ -154,13 +159,12 @@ func (s L7Stats) observe(protocol l7.Protocol, status, method string, duration t
 	case l7.ProtocolRabbitmq, l7.ProtocolNats:
 		counterLabelValues = append(counterLabelValues, labelInterner.intern(method))
 	case l7.ProtocolHTTP:
-		parsedMethod, path := l7.ParseHttp(r.Payload)
 		if ValidUtf8([]byte(path)) {
 			counterLabelValues = append(counterLabelValues, labelInterner.intern(normalizeHttpPath(path)))
 		} else {
 			counterLabelValues = append(counterLabelValues, "")
 		}
-		counterLabelValues = append(counterLabelValues, labelInterner.intern(parsedMethod))
+		counterLabelValues = append(counterLabelValues, labelInterner.intern(method))
 	case l7.ProtocolDNS:
 		requestType, domain, _ := l7.ParseDns(r.Payload)
 		counterLabelValues = append(counterLabelValues, labelInterner.intern(requestType), labelInterner.intern(common.NormalizeFQDN(domain, requestType)))
@@ -178,17 +182,22 @@ func (s L7Stats) observe(protocol l7.Protocol, status, method string, duration t
 		histogramLabelValues = append(histogramLabelValues, labelInterner.intern(requestType), labelInterner.intern(common.NormalizeFQDN(domain, requestType)))
 	}
 
-	// Update counter
-	if counter := s.requests[protocol]; counter != nil {
+	// Map reads are safe after ensureInitialized — the protocol entry exists.
+	// Prometheus CounterVec/HistogramVec are internally thread-safe.
+	s.mu.RLock()
+	counter := s.requests[protocol]
+	histogram := s.latency[protocol]
+	s.mu.RUnlock()
+
+	if counter != nil {
 		if c, err := counter.GetMetricWithLabelValues(counterLabelValues...); err != nil {
 			klog.Warningln("Error getting counter metric:", err)
 		} else {
 			c.Inc()
 		}
 	}
 
-	// Update histogram
-	if histogram := s.latency[protocol]; histogram != nil && duration != 0 {
+	if histogram != nil && duration != 0 {
 		if h, err := histogram.GetMetricWithLabelValues(histogramLabelValues...); err != nil {
 			klog.Warningln("Error getting histogram metric:", err)
 		} else {
@@ -197,7 +206,18 @@ func (s L7Stats) observe(protocol l7.Protocol, status, method string, duration t
 	}
 }
 
-func (s L7Stats) ensureInitialized(protocol l7.Protocol) {
+func (s *L7Stats) ensureInitialized(protocol l7.Protocol) {
+	s.mu.RLock()
+	if s.initialized[protocol] {
+		s.mu.RUnlock()
+		return
+	}
+	s.mu.RUnlock()
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Double-check after acquiring write lock
 	if s.initialized[protocol] {
 		return
 	}
@@ -208,7 +228,7 @@ func (s L7Stats) ensureInitialized(protocol l7.Protocol) {
 		metricsProtocol = l7.ProtocolHTTP
 	}
 
-	// Base labels for all protocols
+	// Base labels for all protocols (aligned with TCP metric label names)
 	baseLabels := []string{
 		"status",
 		"destination",
@@ -224,9 +244,11 @@ func (s L7Stats) ensureInitialized(protocol l7.Protocol) {
 		"actual_destination_workload_namespace",
 		"src_region",
 		"src_az",
-		"destination_region",
-		"destination_az",
-		"destination_instance",
+		"destination_workload_region",
+		"destination_workload_az",
+		"actual_destination_region",
+		"actual_destination_az",
+		"actual_destination_instance",
 	}
 
 	// Initialize request counter
@@ -270,7 +292,9 @@ func (s L7Stats) ensureInitialized(protocol l7.Protocol) {
 	s.initialized[protocol] = true
 }
 
-func (s L7Stats) collect(ch chan<- prometheus.Metric) {
+func (s *L7Stats) collect(ch chan<- prometheus.Metric) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	for _, counterVec := range s.requests {
 		if counterVec != nil {
 			counterVec.Collect(ch)
@@ -283,7 +307,7 @@ func (s L7Stats) collect(ch chan<- prometheus.Metric) {
 	}
 }
 
-func (s L7Stats) delete(dst common.HostPort) {
+func (s *L7Stats) delete(dst common.HostPort) {
 	// With the new architecture, we don't need to delete per-destination metrics
 	// since all metrics are shared across destinations with different label values
 	// This method can be kept for interface compatibility but doesn't need to do anything

containers/llm_stream.go

@@ -185,7 +185,7 @@ func (t *LLMStreamTracker) OnRequestHeaders(
 
 	t.streams[key] = stream
 
-	klog.V(2).Infof("LLM_STREAM_START: key=%s provider=%s model=%s op=%s trace_id=%s",
+	klog.V(4).Infof("LLM_STREAM_START: key=%s provider=%s model=%s op=%s trace_id=%s",
 		key, provider, stream.Model, stream.Operation, traceID)
 
 	return true
@@ -255,7 +255,7 @@ func (t *LLMStreamTracker) OnDataFrame(
 	// Track first token time
 	if stream.FirstTokenTime.IsZero() {
 		stream.FirstTokenTime = now
-		klog.V(2).Infof("LLM_STREAM_FIRST_TOKEN: key=%s ttft_ms=%d",
+		klog.V(4).Infof("LLM_STREAM_FIRST_TOKEN: key=%s ttft_ms=%d",
 			key, now.Sub(stream.RequestTime).Milliseconds())
 	}
 	stream.LastDataTime = now

containers/metrics.go

@@ -61,10 +61,6 @@ var metrics = struct {
 	GpuMemoryUsagePercent *prometheus.Desc
 
 	Ip2Fqdn *prometheus.Desc
-
-	LLMRequests   *prometheus.Desc
-	LLMTokensUsed *prometheus.Desc
-	LLMLatency    *prometheus.Desc
 }{
 	ContainerInfo: metric("container_info", "Meta information about the container", "image", "systemd_triggered_by"),
 
@@ -122,10 +118,6 @@ var metrics = struct {
 
 	GpuUsagePercent:       metric("container_resources_gpu_usage_percent", "Percent of GPU compute resources used by the container", "gpu_uuid"),
 	GpuMemoryUsagePercent: metric("container_resources_gpu_memory_usage_percent", "Percent of GPU memory used by the container", "gpu_uuid"),
-
-	LLMRequests:   metric("container_llm_requests_total", "Total number of LLM API requests", "provider", "model", "host"),
-	LLMTokensUsed: metric("container_llm_tokens_total", "Total number of LLM tokens used", "provider", "model", "type", "host"),
-	LLMLatency:    metric("container_llm_requests_duration_seconds", "Average LLM API request duration", "provider", "model", "host"),
 }
 
 var (

containers/registry.go

@@ -383,7 +383,7 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 				if e.L7Request == nil {
 					continue
 				}
-				klog.V(2).Infof("L7_EVENT_REGISTRY: pid=%d fd=%d protocol=%d timestamp=%d",
+				klog.V(5).Infof("L7_EVENT_REGISTRY: pid=%d fd=%d protocol=%d timestamp=%d",
 					e.Pid, e.Fd, e.L7Request.Protocol, e.Timestamp)
 				r.processL7Event(e)
 			}
@@ -396,7 +396,7 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 // processL7Event handles an L7 event, queueing it for retry if the connection isn't found yet
 func (r *Registry) processL7Event(e ebpftracer.Event) {
 	if c := r.containersByPid[e.Pid]; c != nil {
-		klog.V(2).Infof("L7_EVENT_CONTAINER_FOUND: pid=%d container=%s", e.Pid, c.id)
+		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)
 		if result == L7RequestConnNotFound {
 			// Connection not found - queue for retry
@@ -433,11 +433,11 @@ func (r *Registry) queueL7EventForRetry(e ebpftracer.Event) {
 	// Limit queue size to prevent memory issues
 	const maxPendingEvents = 1000
 	if len(r.pendingL7Events) >= maxPendingEvents {
-		klog.V(2).Infof("L7_EVENT_QUEUE_FULL: dropping event pid=%d fd=%d", e.Pid, e.Fd)
+		klog.V(3).Infof("L7_EVENT_QUEUE_FULL: dropping event pid=%d fd=%d", e.Pid, e.Fd)
 		return
 	}
 
-	klog.V(2).Infof("L7_EVENT_QUEUED: pid=%d fd=%d protocol=%d", e.Pid, e.Fd, e.L7Request.Protocol)
+	klog.V(3).Infof("L7_EVENT_QUEUED: pid=%d fd=%d protocol=%d", e.Pid, e.Fd, e.L7Request.Protocol)
 	r.pendingL7Events = append(r.pendingL7Events, pendingL7Event{
 		event:      e,
 		addedAt:    time.Now(),
@@ -467,7 +467,7 @@ func (r *Registry) processPendingL7Events() {
 	for _, p := range pending {
 		// Expire old events
 		if now.Sub(p.addedAt) > maxAge {
-			klog.V(2).Infof("L7_EVENT_EXPIRED: pid=%d fd=%d age=%v", p.event.Pid, p.event.Fd, now.Sub(p.addedAt))
+			klog.V(3).Infof("L7_EVENT_EXPIRED: pid=%d fd=%d age=%v", p.event.Pid, p.event.Fd, now.Sub(p.addedAt))
 			continue
 		}
 
@@ -488,13 +488,13 @@ func (r *Registry) processPendingL7Events() {
 				p.retryCount++
 				stillPending = append(stillPending, p)
 			} else {
-				klog.V(2).Infof("L7_EVENT_MAX_RETRIES: pid=%d fd=%d", p.event.Pid, p.event.Fd)
+				klog.V(3).Infof("L7_EVENT_MAX_RETRIES: pid=%d fd=%d", p.event.Pid, p.event.Fd)
 			}
 			continue
 		}
 
 		// Successfully processed
-		klog.V(2).Infof("L7_EVENT_RETRY_SUCCESS: pid=%d fd=%d retries=%d", p.event.Pid, p.event.Fd, p.retryCount)
+		klog.V(3).Infof("L7_EVENT_RETRY_SUCCESS: pid=%d fd=%d retries=%d", p.event.Pid, p.event.Fd, p.retryCount)
 		r.ip2fqdnLock.Lock()
 		for ip, domain := range ip2fqdn {
 			r.ip2fqdn[ip] = domain

ebpftracer/ebpf/l7/memcached.c

@@ -50,6 +50,9 @@ int is_memcached_query(char *buf, __u64 buf_size) {
 
 static __always_inline
 int is_memcached_response(char *buf, __u64 buf_size, __s32 *status) {
+    if (buf_size < 5) {
+        return 0;
+    }
     char r[3];
     bpf_read(buf, r);
     char end[2];