diff --git a/containers/registry.go b/containers/registry.go index 88820dc..a17a039 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -403,6 +403,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 +496,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 +529,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() 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/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 {