From 17091762613a59e91a6baa61d9e426a16a0b77c5 Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Wed, 15 Jul 2026 20:52:24 +0800 Subject: [PATCH 1/2] Enhance log module and enhance address resolver in access log module --- .gitignore | 4 +- CHANGES.md | 8 + bpf/accesslog/ambient/ztunnel.c | 58 +- bpf/accesslog/syscalls/connect_conntrack.c | 6 +- configs/rover_configs.yaml | 5 + pkg/accesslog/collector/connection.go | 16 +- pkg/accesslog/collector/ztunnel.go | 528 ++++++++++++++++-- pkg/accesslog/collector/ztunnel_accesslog.go | 290 ++++++++++ .../collector/ztunnel_accesslog_test.go | 130 +++++ pkg/accesslog/collector/ztunnel_admin.go | 296 ++++++++++ pkg/accesslog/collector/ztunnel_flush_test.go | 170 ++++++ .../collector/ztunnel_resolution_test.go | 109 ++++ pkg/accesslog/common/connection.go | 340 ++++++++++- .../common/connection_resolution_test.go | 105 ++++ pkg/accesslog/runner.go | 12 + pkg/logger/logger.go | 56 +- pkg/logger/logger_test.go | 142 +++++ pkg/logger/settings.go | 38 +- pkg/process/finders/kubernetes/template.go | 9 + pkg/tools/btf/linker.go | 7 + pkg/tools/host/file.go | 19 +- pkg/tools/ip/conntrack.go | 32 +- pkg/tools/ip/tcpresolver.go | 3 + pkg/tools/netns/netns.go | 68 +++ pkg/tools/netns/netns_test.go | 173 ++++++ 25 files changed, 2536 insertions(+), 88 deletions(-) create mode 100644 pkg/accesslog/collector/ztunnel_accesslog.go create mode 100644 pkg/accesslog/collector/ztunnel_accesslog_test.go create mode 100644 pkg/accesslog/collector/ztunnel_admin.go create mode 100644 pkg/accesslog/collector/ztunnel_flush_test.go create mode 100644 pkg/accesslog/collector/ztunnel_resolution_test.go create mode 100644 pkg/accesslog/common/connection_resolution_test.go create mode 100644 pkg/logger/logger_test.go create mode 100644 pkg/tools/netns/netns.go create mode 100644 pkg/tools/netns/netns_test.go diff --git a/.gitignore b/.gitignore index 56cd2284..8d6f6f51 100644 --- a/.gitignore +++ b/.gitignore @@ -14,4 +14,6 @@ pkg/**/bpf_*.go pkg/tools/btf/files/**/*.btf -.config \ No newline at end of file +.config.omc/ +.omc/ +.config/ diff --git a/CHANGES.md b/CHANGES.md index 63f97d4f..1f2ca996 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -32,11 +32,19 @@ Release Notes. * Aggregate dropped perf event sample warnings periodically instead of logging every burst. * Lower the log level for connection query errors of short-lived exited processes. * Enable pprof by default but bind it to 127.0.0.1 so it is not exposed on the network. +* Support collecting the ztunnel outbound connection mappings from the ztunnel admin config dump, as a symbol-independent fallback of the uprobe based events in the access log module. +* Support cross-checking the ztunnel proxied connection count from the ztunnel prometheus metrics in the access log module. +* Report a periodic info-level summary of un-resolved remote addresses, including the conntrack and ztunnel correlation statistics, in the access log module. +* Attach all matched ztunnel `track_outbound` symbol copies when attaching the uprobe in the access log module. +* Add success/failure counters to the conntrack real peer address queries. #### Bug Fixes * Fix the base image cannot run in the arm64. * Fix process fork tracepoint reporting thread TID instead of process TGID, causing repeated process detected/dead churn. * Fix panic in the access log module when handling HTTP/2 streams without a body. +* Fix the ztunnel event reader and process finder not being registered when the ztunnel process starts after the rover. +* Fix ghost connections created by failed accept syscalls with a negative socket fd in the access log module. +* Remove the ineffective `ctnetlink_fill_info` kprobe and fix an always-true condition in the conntrack BPF program. #### Documentation * Add a dead link checker in the CI. diff --git a/bpf/accesslog/ambient/ztunnel.c b/bpf/accesslog/ambient/ztunnel.c index e1fc44a4..ba460608 100644 --- a/bpf/accesslog/ambient/ztunnel.c +++ b/bpf/accesslog/ambient/ztunnel.c @@ -41,12 +41,64 @@ int connection_manager_track_outbound(struct pt_regs* ctx) { return 0; } bool success = true; - success = get_socket_addr_ip_in_ztunnel(success, (void *)PT_REGS_PARM3(ctx), &event->orginal_src_ip, &event->src_port); - success = get_socket_addr_ip_in_ztunnel(success, (void *)PT_REGS_PARM4(ctx), &event->original_dst_ip, &event->dst_port); - success = get_socket_addr_ip_in_ztunnel(success, (void *)PT_REGS_PARM5(ctx), &event->lb_dst_ip, &event->lb_dst_port); + // track_outbound(&self, src, original_dst, actual_dst) returns a large ConnectionResult + // struct via a hidden sret pointer. On x86-64 SysV the sret pointer occupies the first + // integer arg register(PARM1), so &self is PARM2 and the three SocketAddr args are PARM3/4/5. + // On AArch64 AAPCS64 the sret pointer is passed in x8, which is OUTSIDE the PARM1..8 arg + // registers, so the arguments are NOT shifted: &self is PARM1 and the args are PARM2/3/4. + // original_dst is the service ClusterIP, actual_dst is the load balanced real pod. +#if defined(bpf_target_x86) + void *src_arg = (void *)PT_REGS_PARM3(ctx); + void *original_dst_arg = (void *)PT_REGS_PARM4(ctx); + void *actual_dst_arg = (void *)PT_REGS_PARM5(ctx); +#else + void *src_arg = (void *)PT_REGS_PARM2(ctx); + void *original_dst_arg = (void *)PT_REGS_PARM3(ctx); + void *actual_dst_arg = (void *)PT_REGS_PARM4(ctx); +#endif + success = get_socket_addr_ip_in_ztunnel(success, src_arg, &event->orginal_src_ip, &event->src_port); + success = get_socket_addr_ip_in_ztunnel(success, original_dst_arg, &event->original_dst_ip, &event->dst_port); + success = get_socket_addr_ip_in_ztunnel(success, actual_dst_arg, &event->lb_dst_ip, &event->lb_dst_port); if (!success) { return 0; } bpf_perf_event_output(ctx, &ztunnel_lb_socket_mapping_event_queue, BPF_F_CURRENT_CPU, event, sizeof(*event)); return 0; } + +// ConnectionResult::new(src: SocketAddr, dst: SocketAddr, hbone_target, ...) is an +// associated function(no &self) that ztunnel constructs UNCONDITIONALLY for every proxied +// connection - including the outbound legs that skip track_outbound through an early-return +// in proxy_to - so it is a strictly-higher-coverage, log-level-independent source(the same +// data ztunnel would print as the "connection complete"/"connection opened" access log, but +// captured at construction time regardless of the log level). It returns a large struct via +// the hidden sret pointer, which on x86-64 occupies PARM1 and shifts the arguments by one +// (src=PARM2, dst=PARM3); on AArch64 the sret pointer is in x8(not a PARM) so the arguments +// are not shifted(src=PARM1, dst=PARM2). src is the downstream app addr and dst is the REAL +// backend pod addr. There is no service ClusterIP among the arguments, so this mapping is +// keyed by the source address alone in user space(the app's ephemeral src port is unique per +// connection). original_dst_ip is left zero to mark this event as a "src-only" mapping. +SEC("uprobe/connection_result_new") +int connection_result_new(struct pt_regs* ctx) { + struct ztunnel_socket_mapping_t *event = create_ztunnel_socket_mapping_event(); + if (event == NULL) { + return 0; + } + bool success = true; +#if defined(bpf_target_x86) + void *src_arg = (void *)PT_REGS_PARM2(ctx); + void *dst_arg = (void *)PT_REGS_PARM3(ctx); +#else + void *src_arg = (void *)PT_REGS_PARM1(ctx); + void *dst_arg = (void *)PT_REGS_PARM2(ctx); +#endif + success = get_socket_addr_ip_in_ztunnel(success, src_arg, &event->orginal_src_ip, &event->src_port); + success = get_socket_addr_ip_in_ztunnel(success, dst_arg, &event->lb_dst_ip, &event->lb_dst_port); + if (!success) { + return 0; + } + event->original_dst_ip = 0; + event->dst_port = 0; + bpf_perf_event_output(ctx, &ztunnel_lb_socket_mapping_event_queue, BPF_F_CURRENT_CPU, event, sizeof(*event)); + return 0; +} diff --git a/bpf/accesslog/syscalls/connect_conntrack.c b/bpf/accesslog/syscalls/connect_conntrack.c index 8f1f5cb2..75109671 100644 --- a/bpf/accesslog/syscalls/connect_conntrack.c +++ b/bpf/accesslog/syscalls/connect_conntrack.c @@ -93,7 +93,7 @@ static __always_inline int nf_conn_aware(struct pt_regs* ctx, struct nf_conn *ct } // already contains the remote address - if (connect_args->has_remote && &(connect_args->remote) != NULL) { + if (connect_args->has_remote) { return 0; } @@ -142,7 +142,3 @@ int nf_confirm(struct pt_regs* ctx) { return nf_conn_aware(ctx, (struct nf_conn*)PT_REGS_PARM3(ctx)); } -SEC("kprobe/ctnetlink_fill_info") -int nf_ctnetlink_fill_info(struct pt_regs* ctx) { - return nf_conn_aware(ctx, (struct nf_conn*)PT_REGS_PARM5(ctx)); -} \ No newline at end of file diff --git a/configs/rover_configs.yaml b/configs/rover_configs.yaml index a5a6b0c9..68ede6d6 100644 --- a/configs/rover_configs.yaml +++ b/configs/rover_configs.yaml @@ -18,6 +18,11 @@ logger: # The lowest level of printing allowed. level: ${ROVER_LOGGER_LEVEL:INFO} + # Comma separated list of modules(name prefix) to elevate to the debug level + # regardless of the level above, e.g. "accesslog.collector.ztunnel". Keeps the + # high volume modules quiet(bounding the logging allocation) while still getting + # debug detail for the modules under investigation. + debug_modules: ${ROVER_LOGGER_DEBUG_MODULES:} core: # The name of the cluster. diff --git a/pkg/accesslog/collector/connection.go b/pkg/accesslog/collector/connection.go index 29adfa91..6621db9d 100644 --- a/pkg/accesslog/collector/connection.go +++ b/pkg/accesslog/collector/connection.go @@ -125,9 +125,6 @@ func (c *ConnectCollector) Start(m *module.Manager, ctx *common.AccessLogContext _ = ctx.BPF.AddLinkOrError(link.Kprobe, map[string]*ebpf.Program{ "nf_confirm": ctx.BPF.NfConfirm, }) - _ = ctx.BPF.AddLinkOrError(link.Kprobe, map[string]*ebpf.Program{ - "ctnetlink_fill_info": ctx.BPF.NfCtnetlinkFillInfo, - }) return nil } @@ -159,6 +156,13 @@ func (c *ConnectionPartitionContext) Consume(data interface{}) { "pid: %d, fd: %d, role: %s: func: %s, family: %d, success: %d, conntrack exist: %t", event.ConID, event.RandomID, event.PID, event.SocketFD, enums.ConnectionRole(event.Role), enums.SocketFunctionName(event.FuncName), event.SocketFamily, event.ConnectSuccess, event.ConnTrackUpstreamPort != 0) + // a negative fd means the syscall failed(e.g. non-blocking accept returns -EAGAIN), + // there is no real connection behind it, so ignore the event + if int32(event.SocketFD) < 0 { + connectionLogger.Debugf("ignore the connect event with negative socket fd, connection ID: %d, randomID: %d, "+ + "pid: %d, fd: %d", event.ConID, event.RandomID, event.PID, int32(event.SocketFD)) + return + } socketPair := c.BuildSocketFromConnectEvent(event) if socketPair == nil { connectionLogger.Debugf("cannot found the socket paire from connect event, connection ID: %d, randomID: %d", @@ -271,6 +275,9 @@ func (c *ConnectionPartitionContext) BuildSocketPair(event *events.SocketConnect if !ip.ShouldIgnoreConntrack(remoteAddr, conntrackIP, uint16(event.ConnTrackUpstreamPort)) { result.DestIP = conntrackIP result.DestPort = uint16(event.ConnTrackUpstreamPort) + // the BPF conntrack path rewrote the dest to the real peer; mark it so the + // resolve summary counts it as conntrack-resolved instead of unresolved + result.ConnTrackResolved = true ignoredConntrack = false } @@ -308,6 +315,9 @@ func (c *ConnectionPartitionContext) BuildSocketPair(event *events.SocketConnect if !ip.ShouldIgnoreConntrack(remoteAddr, conntrackIP, uint16(event.ConnTrackUpstreamPort)) { result.DestIP = conntrackIP result.DestPort = uint16(event.ConnTrackUpstreamPort) + // the BPF conntrack path rewrote the dest to the real peer; mark it so the + // resolve summary counts it as conntrack-resolved instead of unresolved + result.ConnTrackResolved = true ignoredConntrack = false } diff --git a/pkg/accesslog/collector/ztunnel.go b/pkg/accesslog/collector/ztunnel.go index be5c1686..c36a59bf 100644 --- a/pkg/accesslog/collector/ztunnel.go +++ b/pkg/accesslog/collector/ztunnel.go @@ -20,13 +20,17 @@ package collector import ( "context" "fmt" + "os" + "runtime" "strings" + "sync/atomic" "time" "k8s.io/apimachinery/pkg/util/cache" "github.com/apache/skywalking-rover/pkg/accesslog/common" "github.com/apache/skywalking-rover/pkg/accesslog/events" + "github.com/apache/skywalking-rover/pkg/logger" "github.com/apache/skywalking-rover/pkg/module" "github.com/apache/skywalking-rover/pkg/tools/elf" "github.com/apache/skywalking-rover/pkg/tools/enums" @@ -35,18 +39,54 @@ import ( v3 "skywalking.apache.org/repo/goapi/collect/ebpf/accesslog/v3" + "github.com/cilium/ebpf" "github.com/shirou/gopsutil/process" ) var ( // ZTunnelProcessFinderInterval is the interval to find ztunnel process ZTunnelProcessFinderInterval = time.Second * 30 + // ztunnelMappingQueuePerCPUBufferPages / ztunnelMappingQueueMinParallels tune the ztunnel mapping + // perf queue so the doubled(track_outbound + ConnectionResult) event stream is drained promptly: + // a large per-CPU ring plus one reader goroutine per CPU keep the delivery latency well within + // the resolution-defer window(measured via the open->resolution latency buckets), instead of the + // events backing up and arriving after a short connection has already flushed the raw ClusterIP. + ztunnelMappingQueuePerCPUBufferPages = 16 + ztunnelMappingQueueMinParallels = 4 // ZTunnelTrackBoundSymbolPrefix is the prefix of the symbol name to track outbound connections in ztunnel process // ztunnel::proxy::connection_manager::ConnectionManager::track_outbound ZTunnelTrackBoundSymbolPrefix = "_ZN7ztunnel5proxy18connection_manager17ConnectionManager14track_outbound" + // ZTunnelConnectionResultNewSymbolPrefix is the prefix of ztunnel::proxy::metrics::ConnectionResult::new, + // an associated function ztunnel constructs UNCONDITIONALLY for every proxied connection - including the + // outbound legs that skip track_outbound through an early-return - so it is a strictly-higher-coverage, + // log-level-independent source of the (downstream src -> real pod) mapping(the same data ztunnel prints + // as the access log, captured at construction regardless of log level). Attached best-effort alongside + // track_outbound so the two sources cover each other across ztunnel versions. + ZTunnelConnectionResultNewSymbolPrefix = "_ZN7ztunnel5proxy7metrics16ConnectionResult3new" + // ZTunnelIPMappingExpireDuration is how long an outbound(service IP -> real pod IP) + // mapping captured from the ztunnel uprobe is kept before a connection flush claims + // it. Empirically raising this does NOT reduce the residual unresolved rate: the + // leftover misses are short-lived client connections whose uprobe mapping event is + // processed slightly AFTER the connection has already flushed(a pipeline race), not + // mappings that expired - so a longer TTL only grows the cache for no benefit. + ZTunnelIPMappingExpireDuration = time.Minute + // ZTunnelSrcOnlyMappingExpireDuration is the TTL for a source-address-only mapping(from + // ConnectionResult::new or the access-log fallback). The src-only entry is written at + // connection-open but only CONSUMED when the connection flushes, so it is sized generously above + // the flush / resolution-defer window - a connection that flushes a little late still finds its + // mapping present instead of degrading to a degenerate "-|service|-" node. A src-only key carries + // no destination discriminator, but a reused ephemeral src port almost always targets the same + // service(whose ClusterIP load-balances over the same pods) and overwrites the entry with its own + // fresh mapping anyway, so the longer TTL does not cause mis-attribution. + ZTunnelSrcOnlyMappingExpireDuration = time.Minute * 5 ) -var zTunnelCollectInstance = NewZTunnelCollector(time.Minute) +// ztunnelLog is a dedicated module("accesslog.collector.ztunnel") so the ambient +// correlation debug logs can be enabled on their own(logger.debug_modules) +// without turning on the high volume accesslog.collector.* debug logs. +var ztunnelLog = logger.GetLogger("accesslog", "collector", "ztunnel") + +var zTunnelCollectInstance = NewZTunnelCollector(ZTunnelIPMappingExpireDuration) // ZTunnelCollector is a collector for ztunnel process in the Ambient Istio scenario type ZTunnelCollector struct { @@ -54,15 +94,65 @@ type ZTunnelCollector struct { cancel context.CancelFunc alc *common.AccessLogContext - collectingProcess *process.Process + // collectingProcess is written by the finder-ticker goroutine on (re)discovery + // and read concurrently by OnConnectEvent / ReadyToFlushConnection / the netns + // pollers, so it is guarded by an atomic pointer(same idiom as the counters). + collectingProcess atomic.Pointer[process.Process] ipMappingCache *cache.Expiring ipMappingExpireDuration time.Duration + + // counters for observability of the ztunnel correlation pipeline + mappingEventCount atomic.Int64 + mappingHitCount atomic.Int64 + mappingMissCount atomic.Int64 + emptyCacheMissCount atomic.Int64 + // count of uprobe mappings rejected by the plausibility check, a growing value + // signals the ztunnel binary's track_outbound ABI no longer matches what the + // uprobe expects(version mismatch) + invalidMappingCount atomic.Int64 + + // resolvedBySource counts, per redundant source, whose cached mapping actually resolved a + // connection at flush time(reported as a share in the periodic stats). It holds only the + // four resolution sources, so a nil-safe lookup skips non-resolution tags like sourceInbound. + resolvedBySource map[ztunnelMappingSource]*atomic.Int64 + // accessLogParsedCount counts the outbound mappings the access-log fallback tailer parsed + // from the ztunnel log; it proves the fallback is functional even when the uprobes cover + // everything and it therefore never wins a resolution + accessLogParsedCount atomic.Int64 + + // admin/metrics pollers inside the ztunnel network namespace + pollersStarted bool + // accessLogTailerStarted guards the one-time start of the ztunnel access-log fallback tailer + accessLogTailerStarted bool + // accessLogBacklogCutoff is set at tailer start; ztunnel access-log lines timestamped before it + // are the pre-agent backlog(connections proxied before this agent and its uprobes existed, which + // are not in the connection manager). They are skipped: caching them is wasted, and worse, + // re-parsing the huge backlog that accumulates when the e2e starts traffic BEFORE the agent + // delays the tailer from reaching the RECENT close lines(for the startup connections still being + // resolved) until after they are flushed/deleted, leaving them as degenerate nodes. + accessLogBacklogCutoff time.Time + accessLogBacklogSkipped atomic.Int64 + adminOutboundMappingCount atomic.Int64 + adminInboundSeenCount atomic.Int64 + metricsOpenedConnections atomic.Int64 + + // count of connect events observed from the ztunnel process itself, used to + // diagnose whether the BPF `tgid_is_ztunnel` gate is actually capturing the + // ztunnel's connect() to the local workload(the inbound correlation source) + ztunnelConnectEventSeen atomic.Int64 + ztunnelInboundTaggedSeen atomic.Int64 } func NewZTunnelCollector(expireTime time.Duration) *ZTunnelCollector { return &ZTunnelCollector{ ipMappingCache: cache.NewExpiring(), ipMappingExpireDuration: expireTime, + resolvedBySource: map[ztunnelMappingSource]*atomic.Int64{ + sourceTrackOutbound: new(atomic.Int64), + sourceConnectionResult: new(atomic.Int64), + sourceAccessLog: new(atomic.Int64), + sourceAdminDump: new(atomic.Int64), + }, } } @@ -76,79 +166,177 @@ func (z *ZTunnelCollector) Start(_ *module.Manager, ctx *common.AccessLogContext return err } - if z.collectingProcess == nil { - return nil - } - - ctx.BPF.ReadEventAsync(ctx.BPF.ZtunnelLbSocketMappingEventQueue, func(data interface{}) { + // NOTE: even if the ztunnel process not found at startup, still needs to register + // the event reader and the finder ticker, the ztunnel process could start later + // This queue now carries TWO events per connection(track_outbound + ConnectionResult::new), so + // a single reader with the default 1-page buffer falls behind under a high connection rate(the + // extra waypoint hop): the mapping events are then read tens of seconds late, after the + // connection has already flushed with the raw ClusterIP - the dominant residual "-|service|-" + // cause. Give it a larger per-CPU buffer and a few reader goroutines so mappings are delivered + // promptly, within the connection's resolution-defer window. + ctx.BPF.ReadEventAsyncWithBufferSize(ctx.BPF.ZtunnelLbSocketMappingEventQueue, func(data interface{}) { event := data.(*events.ZTunnelSocketMappingEvent) localIP := z.convertBPFIPToString(event.OriginalSrcIP) localPort := event.OriginalSrcPort + lbIP := z.convertBPFIPToString(event.LoadBalancedDestIP) + z.mappingEventCount.Add(1) + + // A ConnectionResult::new event has no original service ClusterIP(OriginalDestIP == 0): + // it carries only the (downstream src -> real pod) pair and is keyed by the source + // address alone(the app's ephemeral src port is unique per connection). This is the + // higher-coverage source that also captures the outbound legs track_outbound skips via + // its early-returns. + if event.OriginalDestIP == 0 { + if !isPlausibleSrcOnlyMapping(event) { + z.invalidMappingCount.Add(1) + return + } + ztunnelLog.Debugf("received ztunnel src-only mapping event: %s:%d -> lb: %s:%d", localIP, localPort, lbIP, event.LoadBalancedDestPort) + srcOnlyKey := z.buildSrcOnlyCacheKey(localIP, int(localPort)) + z.ipMappingCache.Set(srcOnlyKey, &ZTunnelLoadBalanceAddress{ + IP: lbIP, + Port: event.LoadBalancedDestPort, + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, + Source: sourceConnectionResult, + }, ZTunnelSrcOnlyMappingExpireDuration) + // push: resolve any connection already held in the manager for this source right now, + // instead of waiting for its next flush to pull the cache(closes the late-event race) + z.retroResolve(localIP, localPort) + return + } + remoteIP := z.convertBPFIPToString(event.OriginalDestIP) remotePort := event.OriginalDestPort - lbIP := z.convertBPFIPToString(event.LoadBalancedDestIP) - log.Debugf("received ztunnel lb socket mapping event: %s:%d -> %s:%d, lb: %s", localIP, localPort, remoteIP, remotePort, lbIP) + ztunnelLog.Debugf("received ztunnel lb socket mapping event: %s:%d -> %s:%d, lb: %s", localIP, localPort, remoteIP, remotePort, lbIP) + + // the uprobe reads ztunnel's version-specific Rust internals(track_outbound + // arg registers + SocketAddr layout). A ztunnel that changed the function + // signature or the struct layout would make it read the wrong offsets and + // produce a GARBAGE mapping, which is worse than no mapping(it would attribute + // traffic to a wrong/non-existent pod). Reject implausible mappings so such a + // case degrades safely to "unresolved"(the raw service IP the backend can still + // name at the service level) instead of silently wrong data. + if !isPlausibleLBMapping(event) { + z.invalidMappingCount.Add(1) + ztunnelLog.Warnf("dropping implausible ztunnel lb mapping(possible ztunnel version/ABI mismatch): %s:%d -> %s:%d, lb: %s:%d", + localIP, localPort, remoteIP, remotePort, lbIP, event.LoadBalancedDestPort) + return + } key := z.buildIPMappingCacheKey(localIP, int(localPort), remoteIP, int(remotePort)) z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{ - IP: lbIP, - Port: event.LoadBalancedDestPort, - From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, + IP: lbIP, + Port: event.LoadBalancedDestPort, + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, + Source: sourceTrackOutbound, }, z.ipMappingExpireDuration) - }, func() interface{} { + }, os.Getpagesize()*ztunnelMappingQueuePerCPUBufferPages, ztunnelMappingQueueParallels(), func() interface{} { return &events.ZTunnelSocketMappingEvent{} }) go func() { ticker := time.NewTicker(ZTunnelProcessFinderInterval) + var lastMissCount, lastEmptyCacheMissCount int64 for { select { case <-ticker.C: err := z.findZTunnelProcessAndCollect() if err != nil { - log.Error("failed to find and collect ztunnel process: ", err) + ztunnelLog.Error("failed to find and collect ztunnel process: ", err) } + missCount, emptyCacheMissCount := z.mappingMissCount.Load(), z.emptyCacheMissCount.Load() + logFunc := ztunnelLog.Debugf + // promote to info level when new un-correlated connections appeared in this interval, + // so the resolve failures are visible without enabling the debug level + if missCount > lastMissCount || emptyCacheMissCount > lastEmptyCacheMissCount { + logFunc = ztunnelLog.Infof + } + logFunc("ztunnel correlation stats: uprobe mapping events received: %d, invalid mappings dropped: %d, "+ + "admin dump outbound mappings: %d, admin dump inbound connections seen: %d, ztunnel-pid connect events seen: %d, "+ + "inbound legs tagged: %d, attach hits: %d, attach misses: %d, empty cache misses: %d, "+ + "ztunnel reported opened connections(metrics): %d, resolution by source: {%s}", + z.mappingEventCount.Load(), z.invalidMappingCount.Load(), z.adminOutboundMappingCount.Load(), + z.adminInboundSeenCount.Load(), z.ztunnelConnectEventSeen.Load(), z.ztunnelInboundTaggedSeen.Load(), + z.mappingHitCount.Load(), missCount, emptyCacheMissCount, z.metricsOpenedConnections.Load(), + z.resolutionSourceStats()) + lastMissCount, lastEmptyCacheMissCount = missCount, emptyCacheMissCount case <-z.ctx.Done(): ticker.Stop() return } } }() + + // start the symbol-independent access-log fallback tailer(best-effort: idles if the log + // file is not mounted / access logging is off), so a ztunnel build where both uprobes + // fail to attach still resolves the ClusterIP hops + z.startAccessLogTailer() return nil } func (z *ZTunnelCollector) OnConnectEvent(e *events.SocketConnectEvent, s *ip.SocketPair) bool { - if z.collectingProcess != nil && e != nil && s != nil && uint32(z.collectingProcess.Pid) == e.PID && - s.Role == enums.ConnectionRoleClient { - // must be the client side(outbound) connect - // revert the source and dest for the workload application accept - key := z.buildIPMappingCacheKey(s.DestIP, int(s.DestPort), s.SrcIP, int(s.SrcPort)) - z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{ - From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_INBOUND_FUNC, - }, z.ipMappingExpireDuration) - log.Debugf("found the ztunnel outbound connection, "+ - "connection ID: %d, randomID: %d, pid: %d, fd: %d, role: %s, local: %s:%d, remote: %s:%d", - e.ConID, e.RandomID, e.PID, e.SocketFD, enums.ConnectionRole(e.Role), s.SrcIP, s.SrcPort, s.DestIP, s.DestPort) - return false + proc := z.collectingProcess.Load() + if proc == nil || e == nil || s == nil || uint32(proc.Pid) != e.PID { + return true } - return true + // any connect event from the ztunnel process reaching here means the BPF + // `tgid_is_ztunnel` gate is armed and capturing, record it for diagnosis + z.ztunnelConnectEventSeen.Add(1) + if s.Role != enums.ConnectionRoleClient { + return true + } + // this is the ztunnel client side connect to the local workload(the inbound leg), + // revert the source and dest so it matches the workload application accept, and tag + // the correlated connection as ztunnel inbound + z.ztunnelInboundTaggedSeen.Add(1) + key := z.buildIPMappingCacheKey(s.DestIP, int(s.DestPort), s.SrcIP, int(s.SrcPort)) + z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{ + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_INBOUND_FUNC, + Source: sourceInbound, + }, z.ipMappingExpireDuration) + ztunnelLog.Debugf("found the ztunnel inbound connection, "+ + "connection ID: %d, randomID: %d, pid: %d, fd: %d, role: %s, local: %s:%d, remote: %s:%d", + e.ConID, e.RandomID, e.PID, e.SocketFD, enums.ConnectionRole(e.Role), s.SrcIP, s.SrcPort, s.DestIP, s.DestPort) + return false } func (z *ZTunnelCollector) ReadyToFlushConnection(connection *common.ConnectionInfo, _ events.Event) { - if connection == nil || connection.Socket == nil || connection.RPCConnection == nil || connection.RPCConnection.Attachment != nil || - z.ipMappingCache.Len() == 0 { + if connection == nil || connection.Socket == nil || connection.RPCConnection == nil || connection.RPCConnection.Attachment != nil { + return + } + if z.ipMappingCache.Len() == 0 { + if z.collectingProcess.Load() != nil { + z.emptyCacheMissCount.Add(1) + ztunnelLog.Debugf("the ztunnel IP mapping cache is empty, cannot attach ztunnel address for connection ID: %d, random ID: %d", + connection.ConnectionID, connection.RandomID) + } return } key := z.buildIPMappingCacheKey(connection.Socket.SrcIP, int(connection.Socket.SrcPort), connection.Socket.DestIP, int(connection.Socket.DestPort)) lbIPObj, found := z.ipMappingCache.Get(key) if !found { - log.Debugf("there no ztunnel mapped IP address found for connection ID: %d, random ID: %d", + // fall back to the source-only mapping(from ConnectionResult::new / the access-log + // fallback), which resolves the outbound connections that skipped track_outbound and + // therefore have no src+ClusterIP entry - this is what lifts the coverage from + // track_outbound's ~90% towards 100% + lbIPObj, found = z.ipMappingCache.Get(z.buildSrcOnlyCacheKey(connection.Socket.SrcIP, int(connection.Socket.SrcPort))) + } + if !found { + z.mappingMissCount.Add(1) + ztunnelLog.Debugf("there no ztunnel mapped IP address found for connection ID: %d, random ID: %d", connection.ConnectionID, connection.RandomID) return } + z.mappingHitCount.Add(1) address := lbIPObj.(*ZTunnelLoadBalanceAddress) - log.Debugf("found the ztunnel load balanced IP for the connection: %s, connectionID: %d, randomID: %d", - address.String(), connection.ConnectionID, connection.RandomID) + // attribute the resolution to the source whose cached mapping actually won, so the periodic + // stats can report each source's share(non-resolution tags like sourceInbound are absent + // from the map and skipped) + if c := z.resolvedBySource[address.Source]; c != nil { + c.Add(1) + } + ztunnelLog.Debugf("found the ztunnel load balanced IP for the connection: %s(source: %s), connectionID: %d, randomID: %d", + address.String(), address.Source, connection.ConnectionID, connection.RandomID) securityPolicy := v3.ZTunnelAttachmentSecurityPolicy_NONE // if the target port is 15008, this mean ztunnel have use mTLS if address.From == v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC && address.Port == 15008 { @@ -163,6 +351,103 @@ func (z *ZTunnelCollector) ReadyToFlushConnection(connection *common.ConnectionI }, }, } + // diagnostic: record how long this connection took from open to resolution, to LOCATE the + // late-mapping-event problem(a mass in the >=15s bucket = events delivered too slowly) + if z.alc != nil && z.alc.ConnectionMgr != nil { + z.alc.ConnectionMgr.RecordResolutionLatency(connection) + } + // NOTE: the src-only entry is deliberately NOT evicted on consume. Evicting it would stop a + // later connection that reuses the same ephemeral src port from borrowing this mapping - but a + // reused port almost always targets the SAME service, whose ClusterIP load-balances over the + // same pods, so the "stale" entry still resolves that connection to a correct pod. What + // eviction reliably DOES cause is a miss for the reused-port connection when its own uprobe + // event has not landed yet, leaving it unresolved as a degenerate "-|service|-" node. The + // short ZTunnelSrcOnlyMappingExpireDuration TTL bounds the staleness instead. (Measured: + // evict-on-consume raised the unresolved rate ~0.65% -> ~1.7%, all to single-replica services.) +} + +// IsResolutionPending implements common.ResolutionAwareFlusher. It reports true while a +// connection's real destination could still be filled by the ztunnel outbound lb mapping +// but has not been yet, so the runner defers the connection's logs a little instead of +// emitting them with the raw service IP(which would create a degenerate "-|service|-" +// entity). It is deliberately conservative - only the client(outbound) leg to a raw, +// not-yet-attached, not-conntrack-resolved remote qualifies - and is bounded by the +// per-connection grace deadline in ShouldDeferForResolution so a genuinely external +// destination is delayed at most one grace period. +func (z *ZTunnelCollector) IsResolutionPending(connection *common.ConnectionInfo) bool { + // only meaningful while a ztunnel is actually being collected on this node, otherwise + // no mapping will ever arrive and deferring would only add latency + if z.collectingProcess.Load() == nil { + return false + } + if connection == nil || connection.RPCConnection == nil || connection.Socket == nil { + return false + } + // already correlated to the real destination + if connection.RPCConnection.Attachment != nil { + return false + } + // only the client(outbound) leg goes through the ztunnel outbound lb mapping + if connection.Socket.Role != enums.ConnectionRoleClient { + return false + } + // the conntrack query already rewrote the address to the real peer, no ztunnel wait needed + if connection.Socket.ConnTrackResolved { + return false + } + // a remote resolved to a local monitored pod is a Kubernetes address(no raw IP); only a + // raw IP remote(the service VIP) is a candidate for the ztunnel lb mapping + remote := connection.RPCConnection.GetRemote() + if remote == nil || remote.GetIp() == nil { + return false + } + return true +} + +// UnresolvedReason implements common.ResolutionAwareFlusher: it categorizes WHY a connection that +// reached the end of its lifetime without a ztunnel attachment is still unresolved, so the periodic +// resolve summary can point at the environment/source that did not provide the mapping. +func (z *ZTunnelCollector) UnresolvedReason(connection *common.ConnectionInfo) string { + if z.collectingProcess.Load() == nil { + // no ztunnel process was discovered on this node, so the ambient outbound mapping can never + // exist here - from this agent's point of view the remote is genuinely just a raw IP + return "no-ztunnel-process-on-node" + } + if !z.IsResolutionPending(connection) { + // only a raw-IP client(outbound) leg goes through the ztunnel outbound lb mapping; a server + // leg / conntrack-resolved / already-attached / non-raw remote is not a ztunnel miss + return "not-a-ztunnel-outbound-leg" + } + if z.ipMappingCache.Len() == 0 { + // ztunnel is being collected but NO mapping was ever captured - the whole correlation source + // is producing nothing(uprobes not attached / not firing and the log/admin fallbacks empty) + return "ztunnel-mapping-cache-empty" + } + // ztunnel is collecting and mappings exist, but none matched this socket(neither the src+dst + // track_outbound key nor the src-only ConnectionResult / access-log key) before the connection + // reached the end of its lifetime: the correlation gap is on the event-capture side - ztunnel + // emitted no usable event for this source(a track_outbound early-return AND no ConnectionResult, + // or a BPF miss). + return "no-ztunnel-mapping-for-socket" +} + +// retroResolve pushes a just-cached source mapping to any connection still held in the manager for +// that source, so a late mapping event does not miss the connection's flush(see RetroResolveBySrc). +// It nil-guards the context so unit tests that construct a bare collector do not panic. +// ztunnelMappingQueueParallels is the reader-goroutine count for the mapping perf queue: one per +// CPU(floored at ztunnelMappingQueueMinParallels) so the doubled event stream is drained in parallel. +func ztunnelMappingQueueParallels() int { + if n := runtime.NumCPU(); n > ztunnelMappingQueueMinParallels { + return n + } + return ztunnelMappingQueueMinParallels +} + +func (z *ZTunnelCollector) retroResolve(srcIP string, srcPort uint16) { + if z.alc == nil || z.alc.ConnectionMgr == nil { + return + } + z.alc.ConnectionMgr.RetroResolveBySrc(srcIP, srcPort) } func (z *ZTunnelCollector) convertBPFIPToString(ipAddr uint32) string { @@ -173,6 +458,68 @@ func (z *ZTunnelCollector) buildIPMappingCacheKey(localIP string, localPort int, return fmt.Sprintf("%s:%d-%s:%d", localIP, localPort, remoteIP, remotePort) } +// buildSrcOnlyCacheKey keys a mapping by the downstream source address alone, used by the +// ConnectionResult::new source(which carries the real pod but no original service ClusterIP, +// so it cannot be keyed by src+dst like track_outbound). The "src:" prefix keeps it in a +// distinct namespace from the src+dst keys. +func (z *ZTunnelCollector) buildSrcOnlyCacheKey(localIP string, localPort int) string { + return fmt.Sprintf("src:%s:%d", localIP, localPort) +} + +// resolutionSourceOrder fixes the print order of the per-source shares in the stats line. +var resolutionSourceOrder = []ztunnelMappingSource{ + sourceTrackOutbound, sourceConnectionResult, sourceAccessLog, sourceAdminDump, +} + +// resolutionSourceStats reports the share each redundant source contributed to the resolved +// connections, plus how many mappings the access-log fallback tailer parsed(its liveness even +// when the uprobes win every resolution). +func (z *ZTunnelCollector) resolutionSourceStats() string { + var total int64 + for _, s := range resolutionSourceOrder { + total += z.resolvedBySource[s].Load() + } + pct := func(v int64) float64 { + if total == 0 { + return 0 + } + return float64(v) * 100 / float64(total) + } + parts := make([]string, 0, len(resolutionSourceOrder)) + for _, s := range resolutionSourceOrder { + v := z.resolvedBySource[s].Load() + parts = append(parts, fmt.Sprintf("%s: %d(%.1f%%)", s, v, pct(v))) + } + return fmt.Sprintf("%s; access_log lines parsed(fallback live): %d, pre-agent backlog skipped: %d", + strings.Join(parts, ", "), z.accessLogParsedCount.Load(), z.accessLogBacklogSkipped.Load()) +} + +// isPlausibleLBMapping sanity-checks a ztunnel lb socket mapping decoded from the +// uprobe. The uprobe reads ztunnel's version-specific Rust ABI(the track_outbound +// argument registers and the SocketAddr byte layout); if a future ztunnel changes +// either, the decode yields garbage. Rejecting an implausible mapping keeps a +// version mismatch from silently attributing traffic to a wrong pod - it degrades +// to unresolved instead. A valid mapping has non-zero src / original-dst / lb-dst +// addresses and ports, and the lb destination(the selected pod) is never loopback. +func isPlausibleLBMapping(e *events.ZTunnelSocketMappingEvent) bool { + // a valid lb mapping is a valid src-only mapping(non-zero src / lb-dst addr+port, non-loopback + // pod) plus a non-zero original service ClusterIP destination + return isPlausibleSrcOnlyMapping(e) && e.OriginalDestIP != 0 && e.OriginalDestPort != 0 +} + +// isPlausibleSrcOnlyMapping sanity-checks a ConnectionResult::new mapping(downstream src -> +// real pod), which has no original ClusterIP. Same guards as isPlausibleLBMapping minus the +// original-dest checks, so a ztunnel ABI change degrades to unresolved instead of wrong data. +func isPlausibleSrcOnlyMapping(e *events.ZTunnelSocketMappingEvent) bool { + if e.OriginalSrcIP == 0 || e.LoadBalancedDestIP == 0 { + return false + } + if e.OriginalSrcPort == 0 || e.LoadBalancedDestPort == 0 { + return false + } + return e.LoadBalancedDestIP>>24 != 127 +} + func (z *ZTunnelCollector) Stop() { if z.cancel != nil { z.cancel() @@ -180,14 +527,14 @@ func (z *ZTunnelCollector) Stop() { } func (z *ZTunnelCollector) findZTunnelProcessAndCollect() error { - if z.collectingProcess != nil { - running, err := z.collectingProcess.IsRunning() + if current := z.collectingProcess.Load(); current != nil { + running, err := current.IsRunning() if err == nil && running { // already collecting the process - log.Debugf("found the ztunnel process and collecting ztunnel data from pid: %d", z.collectingProcess.Pid) + ztunnelLog.Debugf("found the ztunnel process and collecting ztunnel data from pid: %d", current.Pid) return nil } - log.Warnf("detected ztunnel process is not running, should re-scan process to find and collect it") + ztunnelLog.Warnf("detected ztunnel process is not running, should re-scan process to find and collect it") } processes, err := process.Processes() @@ -207,13 +554,26 @@ func (z *ZTunnelCollector) findZTunnelProcessAndCollect() error { } if zTunnelProcess == nil { - log.Debugf("ztunnel process not found is current node") + // clear a now-dead process so the netns pollers stop entering the dead + // /proc//ns/net every interval until a ztunnel comes back + z.collectingProcess.Store(nil) + ztunnelLog.Debugf("ztunnel process not found is current node") return nil } - log.Infof("ztunnel process founded in current node, pid: %d", zTunnelProcess.Pid) - z.collectingProcess = zTunnelProcess - return z.collectZTunnelProcess(zTunnelProcess) + ztunnelLog.Infof("ztunnel process founded in current node, pid: %d", zTunnelProcess.Pid) + z.collectingProcess.Store(zTunnelProcess) + // start the admin/metrics pollers even if the uprobe attaching failed, + // the admin dump works without any dependency on the ztunnel binary symbols + z.startNetnsPollers() + // a missing track_outbound symbol(stripped or symbol-renamed ztunnel binary) + // must NOT be fatal: returning the error here aborts the whole access-log + // module on this node and defeats the symbol-independent admin-dump fallback + // started above. Log it and keep the event reader / pollers running. + if err := z.collectZTunnelProcess(zTunnelProcess); err != nil { + ztunnelLog.Warnf("failed to attach the ztunnel uprobe, the admin-dump fallback still runs: %v", err) + } + return nil } func (z *ZTunnelCollector) collectZTunnelProcess(p *process.Process) error { @@ -222,27 +582,93 @@ func (z *ZTunnelCollector) collectZTunnelProcess(p *process.Process) error { if err != nil { return fmt.Errorf("read executable file error: %v", err) } - trackBoundSymbol := elfFile.FilterSymbol(func(name string) bool { - return strings.HasPrefix(name, ZTunnelTrackBoundSymbolPrefix) - }, true) - if len(trackBoundSymbol) == 0 { - return fmt.Errorf("failed to find track outbound symbol in ztunnel process") + // the Rust compiler may emit multiple monomorphized/cloned copies of a symbol, so attach + // to all of them to avoid missing the actually-called copy + uprobeFile := z.alc.BPF.OpenUProbeExeFile(pidExeFile) + // if the executable could not be opened the AddLink calls below are silent no-ops; fail here + // instead of counting "symbols found" as attached and arming the BPF pid gate with zero + // uprobes actually installed(which would log success while capturing nothing) + if !uprobeFile.Found() { + return fmt.Errorf("cannot open the ztunnel executable %s for uprobe attaching", pidExeFile) + } + attached := 0 + // attach the enter uprobe to every ELF symbol whose name starts with prefix, counting the + // attaches so the caller can tell whether a source was found at all + attach := func(prefix, label string, prog *ebpf.Program) { + // The prefix also matches same-name NON-function symbols the Rust `tracing` macros emit - + // the "__CALLSITE"/"__CALLSITE4META" static data symbols(e.g. + // ...ConnectionResult3new10__CALLSITE17h..E) and inner "{{closure}}" bodies. Attaching a + // uprobe to a DATA symbol fails and aborts the whole access_log module, and a closure + // reads the wrong registers. A real function copy has the Rust legacy-mangled hash segment + // "17hE" IMMEDIATELY after the demangled path, so require "17h" right after the + // prefix: that keeps every monomorphized/cloned FUNCTION copy(onlyOneResult=false, so a + // never-called copy is harmless) while excluding the __CALLSITE/closure symbols. + for _, symbol := range elfFile.FilterSymbol(func(name string) bool { + return strings.HasPrefix(name, prefix+"17h") + }, false) { + ztunnelLog.Infof("attaching ztunnel %s symbol: %s", label, symbol.Name) + uprobeFile.AddLink(symbol.Name, prog, nil) + attached++ + } } - uprobeFile := z.alc.BPF.OpenUProbeExeFile(pidExeFile) - uprobeFile.AddLink(trackBoundSymbol[0].Name, z.alc.BPF.ConnectionManagerTrackOutbound, nil) + // source 1: track_outbound(src, ClusterIP, real pod) - the src+ClusterIP keyed mapping. + // source 2: ConnectionResult::new(src, real pod) - the higher-coverage, src-keyed mapping + // that also captures the outbound legs track_outbound misses. Attached best-effort: on a + // ztunnel version where either symbol is absent/renamed the other still provides coverage. + // AddLink accumulates any attach failure into the shared BPF linker, but the happy path never + // inspects it - and `attached` only counts SYMBOLS MATCHED, not uprobes actually installed. So + // a symbol that matched yet whose uprobe failed to install(the kernel/uprobe layer rejected it + // for this particular ztunnel build) is otherwise completely invisible: the collector then + // receives ZERO mapping events with no clue why. Snapshot the linker error before/after so a + // NEW failure attributable to these two attaches is surfaced instead of silently swallowed. + attachErrBefore := fmt.Sprintf("%v", z.alc.BPF.HasError()) + attach(ZTunnelTrackBoundSymbolPrefix, "track outbound", z.alc.BPF.ConnectionManagerTrackOutbound) + attach(ZTunnelConnectionResultNewSymbolPrefix, "ConnectionResult::new", z.alc.BPF.ConnectionResultNew) - // setting the ztunnel pid in the BPF + if attached == 0 { + return fmt.Errorf("failed to find any ztunnel outbound mapping symbol" + + "(track_outbound / ConnectionResult::new) in ztunnel process") + } + if err := z.alc.BPF.HasError(); err != nil && fmt.Sprintf("%v", err) != attachErrBefore { + ztunnelLog.Warnf("ztunnel uprobe attach reported error(s): one or more uprobes may NOT be "+ + "installed, which leaves the uprobe mapping event stream empty and makes the correlation "+ + "fall back to the admin-dump/access-log sources only: %v", err) + } + + // setting the ztunnel pid in the BPF, this arms the `tgid_is_ztunnel` gate so the + // ztunnel's own connect() to the local workload(the inbound leg) is captured if err = z.alc.BPF.ZtunnelProcessPid.Set(p.Pid); err != nil { return fmt.Errorf("failed to set ztunnel process pid in the BPF: %v", err) } + // read back the value to confirm the BPF gate is actually armed with the expected pid, + // this makes the "no ztunnel connect captured" problem diagnosable directly from the log + var armedPid uint32 + if err = z.alc.BPF.ZtunnelProcessPid.Get(&armedPid); err != nil { + ztunnelLog.Warnf("cannot read back the ztunnel process pid from the BPF: %v", err) + } else { + ztunnelLog.Infof("armed the ztunnel BPF pid gate, expected pid: %d, read back: %d", p.Pid, armedPid) + } return nil } +// ztunnelMappingSource identifies WHICH of the redundant sources produced a cached mapping, +// so the periodic stats can report the resolution share of each(uprobe vs access-log fallback). +type ztunnelMappingSource string + +const ( + sourceTrackOutbound ztunnelMappingSource = "track_outbound" + sourceConnectionResult ztunnelMappingSource = "connection_result" + sourceAccessLog ztunnelMappingSource = "access_log" + sourceAdminDump ztunnelMappingSource = "admin_dump" + sourceInbound ztunnelMappingSource = "inbound" +) + type ZTunnelLoadBalanceAddress struct { - IP string - Port uint16 - From v3.ZTunnelAttachmentEnvironmentDetectBy + IP string + Port uint16 + From v3.ZTunnelAttachmentEnvironmentDetectBy + Source ztunnelMappingSource } func (z *ZTunnelLoadBalanceAddress) String() string { diff --git a/pkg/accesslog/collector/ztunnel_accesslog.go b/pkg/accesslog/collector/ztunnel_accesslog.go new file mode 100644 index 00000000..9e8e3c22 --- /dev/null +++ b/pkg/accesslog/collector/ztunnel_accesslog.go @@ -0,0 +1,290 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package collector + +import ( + "bufio" + "encoding/json" + "io" + "os" + "path/filepath" + "strings" + "time" + + "github.com/apache/skywalking-rover/pkg/tools/host" + + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/accesslog/v3" +) + +var ( + // ZTunnelAccessLogPodsGlob is the per-pod sub-path(UNDER the kubelet pod-log directory) of the + // ztunnel DaemonSet's access log. The pod-log directory itself(/var/log/pods on the host) is + // resolved through host.GetHostVarLogPodsInHost, so the in-container mount point is injected via + // ROVER_HOST_VAR_LOG_PODS_MAPPING and NOT hard-coded. The kubelet writes every container's log + // under /var/log/pods for ALL CRI runtimes(containerd, CRI-O, cri-dockerd), so this is runtime + // independent. Tailing it is the ultimate, symbol-independent fallback mapping source: ztunnel + // emits an access log line for EVERY proxied connection("connection complete" at INFO, + // "connection opened" at DEBUG) carrying src.addr and dst.hbone_addr(the real backend pod), so + // it recovers the (downstream src -> real pod) mapping even on a ztunnel build where the uprobe + // symbols are stripped/renamed and both uprobes fail to attach. The container sub-dir is + // wildcarded so any container name matches; the ztunnel pod has a single container. + ZTunnelAccessLogPodsGlob = "istio-system_ztunnel-*/*/*.log" + // ZTunnelAccessLogPollInterval is how often the tailer polls for new log content / rotation + ZTunnelAccessLogPollInterval = time.Second +) + +// startAccessLogTailer starts a background goroutine that tails the local ztunnel access log +// and feeds the (downstream src -> real pod) mappings into the same ipMappingCache the uprobe +// fills, keyed by the source address alone(like the ConnectionResult::new source). It is a +// best-effort fallback: if the log file is absent(no mount / logging disabled) it simply idles. +func (z *ZTunnelCollector) startAccessLogTailer() { + if z.accessLogTailerStarted { + return + } + z.accessLogTailerStarted = true + // lines older than this are the pre-agent backlog and are skipped(see accessLogBacklogCutoff). + z.accessLogBacklogCutoff = time.Now().Add(-30 * time.Second) + // resolve the full pod-log glob through the host mapping(ROVER_HOST_VAR_LOG_PODS_MAPPING) so + // the /var/log/pods mount point is configurable and not hard-coded into the binary. + glob := host.GetHostVarLogPodsInHost(ZTunnelAccessLogPodsGlob) + go func() { + var current string // currently tailed file path + var reader *bufio.Reader + var file *os.File + var offset int64 + defer func() { + if file != nil { + _ = file.Close() + } + }() + ticker := time.NewTicker(ZTunnelAccessLogPollInterval) + defer ticker.Stop() + for { + select { + case <-z.ctx.Done(): + return + case <-ticker.C: + } + + // nothing to tail until a ztunnel is being collected on this node; skip the glob so a + // non-ztunnel node does not scan the log dir every tick(the ztunnel log only exists + // where a ztunnel runs, so this changes no outcome, only avoids the wasted stat) + if z.collectingProcess.Load() == nil { + continue + } + + // (re)resolve the newest ztunnel log file; the kubelet rotates N.log files + latest := newestMatch(glob) + if latest == "" { + continue + } + // decide whether to (re)open the file. Reopen on a new newest path(rotation to a new + // N.log), OR when the same path is now backed by a DIFFERENT inode: kubelet size-based + // rotation renames the active file and recreates it at the same path, so keying the + // reopen on the path string alone(latest != current) would leave the tailer pinned to + // the stale renamed inode and silently stop seeing new lines. os.SameFile compares the + // underlying inode, catching that case. + reopen := file == nil || latest != current + if !reopen { + if latestInfo, statErr := os.Stat(latest); statErr == nil { + if openInfo, err := file.Stat(); err != nil || !os.SameFile(latestInfo, openInfo) { + reopen = true + } + } + } + if reopen { + if file != nil { + _ = file.Close() + } + f, err := os.Open(latest) + if err != nil { + ztunnelLog.Debugf("cannot open ztunnel access log %s: %v", latest, err) + continue + } + file, reader, current, offset = f, bufio.NewReader(f), latest, 0 + ztunnelLog.Infof("tailing ztunnel access log as a fallback mapping source: %s", latest) + } else if file != nil { + // same inode: detect truncation-in-place(file shrank) and restart from the beginning + if fi, err := file.Stat(); err == nil && fi.Size() < offset { + _, _ = file.Seek(0, io.SeekStart) + reader.Reset(file) + offset = 0 + } + } + if reader == nil { + continue + } + for { + line, err := reader.ReadString('\n') + if len(line) > 0 { + offset += int64(len(line)) + if strings.HasSuffix(line, "\n") { + z.handleAccessLogLine(line) + } + } + if err != nil { + break // EOF or partial trailing line: wait for the next poll + } + } + } + }() +} + +// newestMatch returns the glob match with the most recent modification time, or "". +func newestMatch(glob string) string { + matches, err := filepath.Glob(glob) + if err != nil || len(matches) == 0 { + return "" + } + var newest string + var newestMod time.Time + for _, m := range matches { + fi, err := os.Stat(m) + if err != nil { + continue + } + if newest == "" || fi.ModTime().After(newestMod) { + newest, newestMod = m, fi.ModTime() + } + } + return newest +} + +// handleAccessLogLine parses one CRI log line and, if it is an outbound ztunnel access log +// event, feeds its (src -> real pod) mapping into the cache keyed by the source address. +func (z *ZTunnelCollector) handleAccessLogLine(line string) { + // The kubelet writes container logs in one of two on-disk formats depending on the runtime: + // - CRI(containerd / CRI-O / cri-dockerd): " ", where a + // partial(P) line is a fragment - ztunnel access log lines are short and never split, so + // only full(F) lines are taken; + // - docker json-file(legacy dockershim): {"log":"\n","stream":"...","time":""}. + // Extract (timestamp, payload) from whichever it is so the tailer is runtime independent. + line = strings.TrimRight(line, "\n") + var lineTimestamp, payload string + if strings.HasPrefix(line, "{") { + var m map[string]string + if json.Unmarshal([]byte(line), &m) != nil { + return + } + lineTimestamp, payload = m["time"], strings.TrimRight(m["log"], "\n") + } else { + fields := strings.SplitN(line, " ", 4) + if len(fields) < 4 || fields[2] != "F" { + return + } + lineTimestamp, payload = fields[0], fields[3] + } + if payload == "" { + return + } + // skip the pre-agent backlog(see accessLogBacklogCutoff): re-parsing/caching the large history + // the ztunnel logged before this agent started - which the e2e maximizes by running traffic + // BEFORE the agent - delays the tailer from reaching the RECENT close lines it actually needs, + // past the startup connections' resolution-defer/delete window, leaving them degenerate. + if !z.accessLogBacklogCutoff.IsZero() { + if ts, err := time.Parse(time.RFC3339Nano, lineTimestamp); err == nil && ts.Before(z.accessLogBacklogCutoff) { + z.accessLogBacklogSkipped.Add(1) + return + } + } + + var srcAddr, podAddr, direction, message string + if strings.HasPrefix(strings.TrimSpace(payload), "{") { + // LOG_FORMAT=json + var m map[string]interface{} + if json.Unmarshal([]byte(payload), &m) != nil { + return + } + srcAddr, _ = m["src.addr"].(string) + podAddr, _ = m["dst.hbone_addr"].(string) + if podAddr == "" { + podAddr, _ = m["dst.addr"].(string) + } + direction, _ = m["direction"].(string) + message, _ = m["message"].(string) + } else { + // default plain "key=value" istio format(tab separated header + space separated fields) + if !strings.Contains(payload, "\taccess\t") { + return + } + srcAddr = extractLogField(payload, "src.addr=") + podAddr = extractLogField(payload, "dst.hbone_addr=") + if podAddr == "" { + podAddr = extractLogField(payload, "dst.addr=") + } + direction = strings.Trim(extractLogField(payload, "direction="), "\"") + if strings.Contains(payload, "connection complete") { + message = "connection complete" + } else if strings.Contains(payload, "connection opened") { + message = "connection opened" + } + } + + // only the outbound leg carries (app src -> real target pod); inbound has the reverse + if direction != "outbound" || srcAddr == "" || podAddr == "" { + return + } + if message != "connection complete" && message != "connection opened" { + return + } + srcIP, sp, err := parseZTunnelAddress(srcAddr) + if err != nil { + return + } + podIP, pp, err := parseZTunnelAddress(podAddr) + if err != nil { + return + } + if podIP == "" || strings.HasPrefix(podIP, "127.") { + return + } + + // the access-log source is functional regardless of whether it ever wins a resolution; + // count every parsed outbound mapping so the stats prove the fallback is live even when + // the uprobes already cover everything + z.accessLogParsedCount.Add(1) + key := z.buildSrcOnlyCacheKey(srcIP, sp) + // as a fallback, only fill a gap the uprobes did not already cover: do not overwrite a + // live uprobe mapping(that would mis-attribute the resolution source and waste writes) + if _, exist := z.ipMappingCache.Get(key); exist { + return + } + z.mappingEventCount.Add(1) + ztunnelLog.Debugf("access-log fallback mapping resolved a gap: %s:%d -> %s:%d", srcIP, sp, podIP, pp) + z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{ + IP: podIP, + Port: uint16(pp), + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, + Source: sourceAccessLog, + }, ZTunnelSrcOnlyMappingExpireDuration) + // push: resolve any still-held connection for this source now(same as the uprobe path) + z.retroResolve(srcIP, uint16(sp)) +} + +// extractLogField returns the value following key(e.g. "src.addr=") up to the next space, or "". +func extractLogField(s, key string) string { + i := strings.Index(s, key) + if i < 0 { + return "" + } + rest := s[i+len(key):] + if end := strings.IndexAny(rest, " \t"); end >= 0 { + return rest[:end] + } + return rest +} diff --git a/pkg/accesslog/collector/ztunnel_accesslog_test.go b/pkg/accesslog/collector/ztunnel_accesslog_test.go new file mode 100644 index 00000000..d53666bb --- /dev/null +++ b/pkg/accesslog/collector/ztunnel_accesslog_test.go @@ -0,0 +1,130 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package collector + +import ( + "encoding/json" + "testing" + "time" +) + +func TestExtractLogField(t *testing.T) { + cases := []struct{ name, s, key, want string }{ + {"space terminated", `x src.addr=1.2.3.4:5 y`, "src.addr=", "1.2.3.4:5"}, + {"tab terminated", "x\tsrc.addr=1.2.3.4:5\ty", "src.addr=", "1.2.3.4:5"}, + {"end of string", `x src.addr=1.2.3.4:5`, "src.addr=", "1.2.3.4:5"}, + {"missing key", `x dst.addr=9`, "src.addr=", ""}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := extractLogField(c.s, c.key); got != c.want { + t.Fatalf("extractLogField(%q, %q) = %q, want %q", c.s, c.key, got, c.want) + } + }) + } +} + +func TestHandleAccessLogLine(t *testing.T) { + srcKey := func(z *ZTunnelCollector) string { return z.buildSrcOnlyCacheKey("10.0.0.5", 45000) } + + t.Run("json outbound connection complete fills the src-only mapping", func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + z.handleAccessLogLine(`2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000",` + + `"dst.hbone_addr":"10.244.0.20:9080","direction":"outbound","message":"connection complete"}` + "\n") + obj, ok := z.ipMappingCache.Get(srcKey(z)) + if !ok { + t.Fatal("expected a src-only mapping to be cached") + } + addr := obj.(*ZTunnelLoadBalanceAddress) + if addr.IP != "10.244.0.20" || addr.Port != 9080 || addr.Source != sourceAccessLog { + t.Fatalf("unexpected cached mapping: %+v", addr) + } + if z.accessLogParsedCount.Load() != 1 { + t.Fatalf("accessLogParsedCount = %d, want 1", z.accessLogParsedCount.Load()) + } + }) + + t.Run("dst.hbone_addr is preferred but falls back to dst.addr", func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + z.handleAccessLogLine(`2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000",` + + `"dst.addr":"10.244.0.30:9080","direction":"outbound","message":"connection opened"}` + "\n") + obj, ok := z.ipMappingCache.Get(srcKey(z)) + if !ok || obj.(*ZTunnelLoadBalanceAddress).IP != "10.244.0.30" { + t.Fatal("expected the dst.addr fallback to be used when dst.hbone_addr is absent") + } + }) + + t.Run("plain istio key=value format is parsed", func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + z.handleAccessLogLine("2024-01-01T00:00:00Z stdout F 2024-01-01\tinfo\taccess\tconnection complete " + + "src.addr=10.0.0.5:45000 dst.hbone_addr=10.244.0.20:9080 direction=\"outbound\"\n") + if _, ok := z.ipMappingCache.Get(srcKey(z)); !ok { + t.Fatal("expected the plain-format access log line to be parsed") + } + }) + + t.Run("docker json-file runtime format is parsed", func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + // the legacy docker json-file runtime wraps each line as {"log":"\n","stream":..,"time":..} + inner := `{"src.addr":"10.0.0.5:45000","dst.hbone_addr":"10.244.0.20:9080",` + + `"direction":"outbound","message":"connection complete"}` + wrapper, err := json.Marshal(map[string]string{"log": inner + "\n", "stream": "stdout", "time": "2024-01-01T00:00:00Z"}) + if err != nil { + t.Fatal(err) + } + z.handleAccessLogLine(string(wrapper) + "\n") + obj, ok := z.ipMappingCache.Get(srcKey(z)) + if !ok || obj.(*ZTunnelLoadBalanceAddress).IP != "10.244.0.20" { + t.Fatal("expected the docker json-file wrapped access log line to be parsed") + } + }) + + // none of these malformed / non-matching lines should panic or create a cache entry + reject := []struct{ name, line string }{ + {"partial CRI line (P not F)", `2024-01-01T00:00:00Z stdout P {"src.addr":"10.0.0.5:45000","dst.addr":"10.244.0.20:9080","direction":"outbound","message":"connection complete"}`}, + {"too few CRI fields", `2024-01-01T00:00:00Z stdout`}, + {"inbound direction", `2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000","dst.addr":"10.244.0.20:9080","direction":"inbound","message":"connection complete"}`}, + {"not a connection message", `2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000","dst.addr":"10.244.0.20:9080","direction":"outbound","message":"hello"}`}, + {"loopback pod dropped", `2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000","dst.addr":"127.0.0.1:9080","direction":"outbound","message":"connection complete"}`}, + {"truncated json", `2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000"`}, + {"missing addresses", `2024-01-01T00:00:00Z stdout F {"direction":"outbound","message":"connection complete"}`}, + {"empty payload", `2024-01-01T00:00:00Z stdout F `}, + } + for _, c := range reject { + t.Run("reject "+c.name, func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + z.handleAccessLogLine(c.line + "\n") + if z.ipMappingCache.Len() != 0 { + t.Fatalf("expected no cache entry for a rejected line: %q", c.line) + } + }) + } + + t.Run("does not overwrite a live uprobe mapping", func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + key := z.buildSrcOnlyCacheKey("10.0.0.5", 45000) + z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{IP: "10.244.0.99", Port: 9080, Source: sourceConnectionResult}, time.Minute) + z.handleAccessLogLine(`2024-01-01T00:00:00Z stdout F {"src.addr":"10.0.0.5:45000",` + + `"dst.addr":"10.244.0.20:9080","direction":"outbound","message":"connection complete"}` + "\n") + obj, _ := z.ipMappingCache.Get(key) + addr := obj.(*ZTunnelLoadBalanceAddress) + if addr.IP != "10.244.0.99" || addr.Source != sourceConnectionResult { + t.Fatalf("the access-log fallback must not overwrite a live uprobe mapping, got %+v", addr) + } + }) +} diff --git a/pkg/accesslog/collector/ztunnel_admin.go b/pkg/accesslog/collector/ztunnel_admin.go new file mode 100644 index 00000000..e05b5bc2 --- /dev/null +++ b/pkg/accesslog/collector/ztunnel_admin.go @@ -0,0 +1,296 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package collector + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/apache/skywalking-rover/pkg/tools/host" + "github.com/apache/skywalking-rover/pkg/tools/netns" + + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/accesslog/v3" +) + +var ( + // ZTunnelNetnsPollInterval is the interval of polling the ztunnel admin config dump and metrics + ZTunnelNetnsPollInterval = time.Second * 10 + // ZTunnelAdminConfigDumpURL is the ztunnel admin address inside the ztunnel pod network namespace + ZTunnelAdminConfigDumpURL = "http://127.0.0.1:15000/config_dump" + // ZTunnelMetricsURL is the ztunnel prometheus metrics address inside the ztunnel pod network namespace + ZTunnelMetricsURL = "http://127.0.0.1:15020/metrics" + // ztunnelTCPOpenedMetricName is used to cross-check how many connections the ztunnel have proxied + ztunnelTCPOpenedMetricName = "istio_tcp_connections_opened_total" +) + +// ztunnelConfigDump is the subset of the ztunnel admin /config_dump response, +// the "workloadState" section is reported by the ztunnel in-pod admin handler +// and contains the per-workload active connections tracked by the ConnectionManager +type ztunnelConfigDump struct { + WorkloadState map[string]ztunnelWorkloadState `json:"workloadState"` +} + +type ztunnelWorkloadState struct { + Connections *ztunnelConnectionDump `json:"connections"` +} + +type ztunnelConnectionDump struct { + Inbound []ztunnelConnection `json:"inbound"` + Outbound []ztunnelConnection `json:"outbound"` +} + +type ztunnelConnection struct { + Src string `json:"src"` + OriginalDst string `json:"originalDst"` + ActualDst string `json:"actualDst"` + Protocol string `json:"protocol"` +} + +func (z *ZTunnelCollector) startNetnsPollers() { + if z.pollersStarted || z.collectingProcess.Load() == nil { + return + } + z.pollersStarted = true + go func() { + // Run one poll cycle IMMEDIATELY at attach, before the first tick. The connections + // established in the window between the app-side capture arming and this ztunnel uprobe + // attaching never produced an open-time uprobe mapping event; the admin /config_dump lists + // the connections ztunnel still has open, so this earliest snapshot resolves the ones that + // are still alive at attach instead of waiting a full poll interval(by which the short, + // keepalive-less connections are already gone). + if !z.runNetnsPollCycle() { + return + } + ticker := time.NewTicker(ZTunnelNetnsPollInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + if z.collectingProcess.Load() == nil { + continue + } + if !z.runNetnsPollCycle() { + return + } + case <-z.ctx.Done(): + return + } + } + }() +} + +// runNetnsPollCycle runs one admin-dump + metrics poll inside the ztunnel network namespace and +// returns false if the collector is shutting down. The netns work runs in a throwaway goroutine: +// RunInNetNS may fail to switch the OS thread back to the original namespace and then keep it +// locked so the Go runtime discards it - but that only happens when the goroutine EXITS, so doing +// each cycle in its own goroutine lets a poisoned thread be discarded instead of pinning this +// long-lived poller in the ztunnel namespace. +func (z *ZTunnelCollector) runNetnsPollCycle() bool { + done := make(chan struct{}) + go func() { + defer close(done) + if err := z.pollAdminConnectionDump(); err != nil { + ztunnelLog.Warnf("failed to poll the ztunnel admin connection dump: %v", err) + } + if err := z.pollZTunnelMetrics(); err != nil { + ztunnelLog.Warnf("failed to poll the ztunnel metrics: %v", err) + } + }() + select { + case <-done: + return true + case <-z.ctx.Done(): + return false + } +} + +// pollAdminConnectionDump reads the active connections tracked by the ztunnel ConnectionManager +// through the admin API, and feeds the outbound (src, originalDst) -> actualDst mappings into +// the IP mapping cache, the same cache the uprobe based event fills. This works without any +// dependency on the ztunnel binary symbols, but only contains the connections still alive. +func (z *ZTunnelCollector) pollAdminConnectionDump() error { + body, err := z.httpGetInZTunnelNetNS(ZTunnelAdminConfigDumpURL) + if err != nil { + return err + } + dump := &ztunnelConfigDump{} + if err := json.Unmarshal(body, dump); err != nil { + return fmt.Errorf("unmarshal the config dump error: %w", err) + } + + var aliveInboundCount int64 + for _, workload := range dump.WorkloadState { + if workload.Connections == nil { + continue + } + for _, conn := range workload.Connections.Outbound { + srcIP, srcPort, err := parseZTunnelAddress(conn.Src) + if err != nil { + continue + } + origIP, origPort, err := parseZTunnelAddress(conn.OriginalDst) + if err != nil { + continue + } + actualIP, actualPort, err := parseZTunnelAddress(conn.ActualDst) + if err != nil { + continue + } + + key := z.buildIPMappingCacheKey(srcIP, srcPort, origIP, origPort) + if _, exist := z.ipMappingCache.Get(key); !exist { + z.adminOutboundMappingCount.Add(1) + ztunnelLog.Debugf("found ztunnel outbound connection from admin dump: %s:%d -> %s:%d, actual: %s:%d", + srcIP, srcPort, origIP, origPort, actualIP, actualPort) + } + // always re-set to refresh the expiration for still-alive connections + z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{ + IP: actualIP, + Port: uint16(actualPort), + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, + Source: sourceAdminDump, + }, z.ipMappingExpireDuration) + } + // the inbound entries contain the real client address(src), but not the + // ephemeral 127.0.0.6 leg the application accepts, so they cannot be + // correlated to a specific accepted connection, only count them for stats + aliveInboundCount += int64(len(workload.Connections.Inbound)) + } + // gauge semantic: the count of alive inbound connections seen in the latest poll + z.adminInboundSeenCount.Store(aliveInboundCount) + return nil +} + +// pollZTunnelMetrics reads the total proxied connection count from the ztunnel prometheus +// metrics, used as a cross-check signal in the periodic stats log: when the ztunnel keeps +// opening connections but the agent attaches no ztunnel mapping, the correlation is broken +func (z *ZTunnelCollector) pollZTunnelMetrics() error { + body, err := z.httpGetInZTunnelNetNS(ZTunnelMetricsURL) + if err != nil { + return err + } + sum, found := sumPrometheusCounter(string(body), ztunnelTCPOpenedMetricName) + if found { + z.metricsOpenedConnections.Store(int64(sum)) + } + return nil +} + +func (z *ZTunnelCollector) httpGetInZTunnelNetNS(rawURL string) ([]byte, error) { + proc := z.collectingProcess.Load() + if proc == nil { + return nil, fmt.Errorf("no ztunnel process is collecting") + } + parsed, err := url.Parse(rawURL) + if err != nil { + return nil, fmt.Errorf("parse url %s error: %w", rawURL, err) + } + netnsPath := host.GetHostProcInHost(fmt.Sprintf("%d/ns/net", proc.Pid)) + + var body []byte + err = netns.RunInNetNS(netnsPath, func() error { + // IMPORTANT: dial synchronously on THIS(setns'd) OS thread. net/http.Transport + // dials new connections in a separate goroutine("go dialConnFor"), which runs + // on another thread that is NOT switched into the ztunnel network namespace, so + // the socket would be created in the agent's(host) netns and the connect would + // be refused. net.DialTimeout to a literal IP:port dials inline on the calling + // goroutine, keeping the socket in the target namespace. + conn, err := net.DialTimeout("tcp", parsed.Host, time.Second*5) + if err != nil { + return err + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(time.Second * 5)); err != nil { + return err + } + + req, err := http.NewRequest(http.MethodGet, rawURL, http.NoBody) + if err != nil { + return err + } + if err := req.Write(conn); err != nil { + return err + } + resp, err := http.ReadResponse(bufio.NewReader(conn), req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + body, err = io.ReadAll(resp.Body) + return err + }) + if err != nil { + return nil, err + } + return body, nil +} + +func parseZTunnelAddress(addr string) (string, int, error) { + if addr == "" { + return "", 0, fmt.Errorf("empty address") + } + ip, portStr, err := net.SplitHostPort(addr) + if err != nil { + return "", 0, err + } + port, err := strconv.Atoi(portStr) + if err != nil { + return "", 0, err + } + return ip, port, nil +} + +// sumPrometheusCounter sums all samples of the given counter family +// from a prometheus text format payload +func sumPrometheusCounter(body, metricName string) (float64, bool) { + var sum float64 + var found bool + for _, line := range strings.Split(body, "\n") { + if !strings.HasPrefix(line, metricName) { + continue + } + // require a metric-name boundary after the prefix so a sibling series that + // merely starts with the same name(e.g. _created, _bucket) is not + // summed in: the counter name is followed by '{'(labels) or whitespace + if rest := line[len(metricName):]; rest != "" && rest[0] != '{' && rest[0] != ' ' && rest[0] != '\t' { + continue + } + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + value, err := strconv.ParseFloat(fields[len(fields)-1], 64) + if err != nil { + continue + } + sum += value + found = true + } + return sum, found +} diff --git a/pkg/accesslog/collector/ztunnel_flush_test.go b/pkg/accesslog/collector/ztunnel_flush_test.go new file mode 100644 index 00000000..cfbccc6e --- /dev/null +++ b/pkg/accesslog/collector/ztunnel_flush_test.go @@ -0,0 +1,170 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package collector + +import ( + "testing" + "time" + + "github.com/apache/skywalking-rover/pkg/accesslog/common" + "github.com/apache/skywalking-rover/pkg/accesslog/events" + "github.com/apache/skywalking-rover/pkg/tools/enums" + "github.com/apache/skywalking-rover/pkg/tools/ip" + + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/accesslog/v3" +) + +func clientConn(srcIP string, srcPort uint16, dstIP string, dstPort uint16) *common.ConnectionInfo { + return &common.ConnectionInfo{ + Socket: &ip.SocketPair{ + Role: enums.ConnectionRoleClient, + SrcIP: srcIP, + SrcPort: srcPort, + DestIP: dstIP, + DestPort: dstPort, + }, + RPCConnection: &v3.AccessLogConnection{}, + } +} + +func TestReadyToFlushConnectionSrcDstHitRetained(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + key := z.buildIPMappingCacheKey("10.0.0.5", 45000, "10.96.0.10", 9080) + z.ipMappingCache.Set(key, &ZTunnelLoadBalanceAddress{ + IP: "10.244.0.20", Port: 9080, + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, Source: sourceTrackOutbound, + }, time.Minute) + + conn := clientConn("10.0.0.5", 45000, "10.96.0.10", 9080) + z.ReadyToFlushConnection(conn, nil) + + att := conn.RPCConnection.Attachment.GetZTunnel() + if att == nil || att.RealDestinationIp != "10.244.0.20" { + t.Fatalf("expected the src+dst mapping to attach the real pod, got %+v", conn.RPCConnection.Attachment) + } + // a src+dst mapping is discriminated by the ClusterIP, so it is safe to keep after use + if _, ok := z.ipMappingCache.Get(key); !ok { + t.Fatal("a src+dst mapping should be retained after resolving a connection") + } +} + +func TestReadyToFlushConnectionSrcOnlyFallbackRetainsEntry(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + srcKey := z.buildSrcOnlyCacheKey("10.0.0.5", 45000) + z.ipMappingCache.Set(srcKey, &ZTunnelLoadBalanceAddress{ + IP: "10.244.0.21", Port: 15008, + From: v3.ZTunnelAttachmentEnvironmentDetectBy_ZTUNNEL_OUTBOUND_FUNC, Source: sourceConnectionResult, + }, time.Minute) + + // no src+dst entry exists for this connection, so it must fall back to the src-only mapping + conn := clientConn("10.0.0.5", 45000, "10.96.0.11", 9080) + z.ReadyToFlushConnection(conn, nil) + + att := conn.RPCConnection.Attachment.GetZTunnel() + if att == nil || att.RealDestinationIp != "10.244.0.21" { + t.Fatalf("expected the src-only fallback to attach the real pod, got %+v", conn.RPCConnection.Attachment) + } + if att.SecurityPolicy != v3.ZTunnelAttachmentSecurityPolicy_MTLS { + t.Fatalf("expected MTLS for the port-15008 outbound leg, got %v", att.SecurityPolicy) + } + // the src-only entry is retained after consume(NOT evicted): a later connection that reuses + // the same ephemeral src port to the same service must still resolve from it instead of + // becoming a degenerate "-|service|-" node while its own uprobe event is still in flight + if _, ok := z.ipMappingCache.Get(srcKey); !ok { + t.Fatal("a src-only mapping should be retained after resolving a connection, not evicted") + } +} + +func TestReadyToFlushConnectionNoMappingLeavesUnattached(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + // a non-empty cache(so the empty-cache short circuit does not fire) with no matching key + z.ipMappingCache.Set("src:9.9.9.9:1", &ZTunnelLoadBalanceAddress{IP: "10.244.0.99"}, time.Minute) + + conn := clientConn("10.0.0.5", 45000, "10.96.0.12", 9080) + z.ReadyToFlushConnection(conn, nil) + if conn.RPCConnection.Attachment != nil { + t.Fatal("a connection with no matching mapping must be left unattached(raw service IP)") + } +} + +func TestReadyToFlushConnectionAlreadyAttachedIsNoop(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + srcKey := z.buildSrcOnlyCacheKey("10.0.0.5", 45000) + z.ipMappingCache.Set(srcKey, &ZTunnelLoadBalanceAddress{IP: "10.244.0.21", Source: sourceConnectionResult}, time.Minute) + + conn := clientConn("10.0.0.5", 45000, "10.96.0.11", 9080) + existing := &v3.ConnectionAttachment{} + conn.RPCConnection.Attachment = existing + z.ReadyToFlushConnection(conn, nil) + + if conn.RPCConnection.Attachment != existing { + t.Fatal("an already-attached connection must not be re-resolved") + } + // it must not consume(evict) the src-only entry either, since it early-returned + if _, ok := z.ipMappingCache.Get(srcKey); !ok { + t.Fatal("the src-only mapping must be untouched when the connection was already attached") + } +} + +func mappingEvent(srcIP uint32, srcPort uint16, origIP uint32, origPort uint16, lbIP uint32, lbPort uint16) *events.ZTunnelSocketMappingEvent { + return &events.ZTunnelSocketMappingEvent{ + OriginalSrcIP: srcIP, OriginalSrcPort: srcPort, + OriginalDestIP: origIP, OriginalDestPort: origPort, + LoadBalancedDestIP: lbIP, LoadBalancedDestPort: lbPort, + } +} + +func TestIsPlausibleMapping(t *testing.T) { + // IPs are stored high-byte-first(see convertBPFIPToString): 0x0A....=10.x, 0x7F...=127.x + full := mappingEvent(0x0A000005, 45000, 0x0A600010, 9080, 0x0AF40014, 9080) + if !isPlausibleLBMapping(full) || !isPlausibleSrcOnlyMapping(full) { + t.Fatal("a fully populated, non-loopback mapping should be plausible") + } + + if isPlausibleLBMapping(mappingEvent(0, 45000, 0x0A600010, 9080, 0x0AF40014, 9080)) { + t.Fatal("a zero source IP must be rejected") + } + if isPlausibleSrcOnlyMapping(mappingEvent(0x0A000005, 45000, 0, 0, 0, 9080)) { + t.Fatal("a zero load-balanced IP must be rejected for a src-only mapping") + } + + loopback := mappingEvent(0x0A000005, 45000, 0x0A600010, 9080, 0x7F000001, 9080) + if isPlausibleLBMapping(loopback) || isPlausibleSrcOnlyMapping(loopback) { + t.Fatal("a loopback(127.x) load-balanced destination must be rejected(ABI-mismatch guard)") + } + + // a src-only mapping relaxes exactly the two original-ClusterIP checks + srcOnly := mappingEvent(0x0A000005, 45000, 0, 0, 0x0AF40014, 9080) + if !isPlausibleSrcOnlyMapping(srcOnly) { + t.Fatal("a src-only mapping tolerates a zero original ClusterIP") + } + if isPlausibleLBMapping(srcOnly) { + t.Fatal("an lb mapping requires a non-zero original ClusterIP") + } +} + +func TestBuildSrcOnlyCacheKey(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + if got := z.buildSrcOnlyCacheKey("10.0.0.5", 45000); got != "src:10.0.0.5:45000" { + t.Fatalf("buildSrcOnlyCacheKey = %q, want src:10.0.0.5:45000", got) + } + // the src-only key lives in a distinct namespace from the src+dst key so they never collide + if z.buildSrcOnlyCacheKey("10.0.0.5", 45000) == z.buildIPMappingCacheKey("10.0.0.5", 45000, "10.0.0.5", 45000) { + t.Fatal("the src-only and src+dst keys must not collide") + } +} diff --git a/pkg/accesslog/collector/ztunnel_resolution_test.go b/pkg/accesslog/collector/ztunnel_resolution_test.go new file mode 100644 index 00000000..a218717a --- /dev/null +++ b/pkg/accesslog/collector/ztunnel_resolution_test.go @@ -0,0 +1,109 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package collector + +import ( + "testing" + "time" + + "github.com/apache/skywalking-rover/pkg/accesslog/common" + "github.com/apache/skywalking-rover/pkg/tools/enums" + "github.com/apache/skywalking-rover/pkg/tools/ip" + + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/accesslog/v3" + + "github.com/shirou/gopsutil/process" +) + +func rawIPRemoteConnection() *common.ConnectionInfo { + return &common.ConnectionInfo{ + Socket: &ip.SocketPair{Role: enums.ConnectionRoleClient}, + RPCConnection: &v3.AccessLogConnection{ + Remote: &v3.ConnectionAddress{ + Address: &v3.ConnectionAddress_Ip{ + Ip: &v3.IPAddress{Host: "10.96.0.10", Port: 9080}, + }, + }, + }, + } +} + +func TestIsResolutionPending(t *testing.T) { + // a collector with no ztunnel process being collected never reports pending + t.Run("no ztunnel process is not pending", func(t *testing.T) { + z := NewZTunnelCollector(time.Minute) + if z.IsResolutionPending(rawIPRemoteConnection()) { + t.Fatal("should not be pending when no ztunnel process is collecting") + } + }) + + // arm a collecting process for the remaining cases + armed := func() *ZTunnelCollector { + z := NewZTunnelCollector(time.Minute) + z.collectingProcess.Store(&process.Process{Pid: 4242}) + return z + } + + t.Run("raw IP client leg without attachment is pending", func(t *testing.T) { + if !armed().IsResolutionPending(rawIPRemoteConnection()) { + t.Fatal("an unresolved raw-IP client leg should be pending while ztunnel is active") + } + }) + + t.Run("nil connection is not pending", func(t *testing.T) { + if armed().IsResolutionPending(nil) { + t.Fatal("nil connection should not be pending") + } + }) + + t.Run("already attached is not pending", func(t *testing.T) { + conn := rawIPRemoteConnection() + conn.RPCConnection.Attachment = &v3.ConnectionAttachment{} + if armed().IsResolutionPending(conn) { + t.Fatal("an already-attached connection should not be pending") + } + }) + + t.Run("server leg is not pending", func(t *testing.T) { + conn := rawIPRemoteConnection() + conn.Socket.Role = enums.ConnectionRoleServer + if armed().IsResolutionPending(conn) { + t.Fatal("the server(inbound) leg does not go through the outbound lb mapping") + } + }) + + t.Run("conntrack-resolved is not pending", func(t *testing.T) { + conn := rawIPRemoteConnection() + conn.Socket.ConnTrackResolved = true + if armed().IsResolutionPending(conn) { + t.Fatal("a conntrack-resolved connection needs no ztunnel wait") + } + }) + + t.Run("kubernetes-resolved remote is not pending", func(t *testing.T) { + conn := rawIPRemoteConnection() + conn.RPCConnection.Remote = &v3.ConnectionAddress{ + Address: &v3.ConnectionAddress_Kubernetes{ + Kubernetes: &v3.KubernetesProcessAddress{ServiceName: "reviews"}, + }, + } + if armed().IsResolutionPending(conn) { + t.Fatal("a remote already resolved to a local pod is not a raw service IP") + } + }) +} diff --git a/pkg/accesslog/common/connection.go b/pkg/accesslog/common/connection.go index b00ffab3..e4019fe6 100644 --- a/pkg/accesslog/common/connection.go +++ b/pkg/accesslog/common/connection.go @@ -21,7 +21,10 @@ import ( "context" "errors" "fmt" + "sort" + "strings" "sync" + "sync/atomic" "time" "github.com/sirupsen/logrus" @@ -52,11 +55,36 @@ const ( // clean the active connection in BPF interval cleanActiveConnectionInterval = time.Second * 20 - // in case the reading the data from BPF queue is disordered, so add a delay time to delete the connection information + // in case the reading the data from BPF queue is disordered, so add a delay time to delete the + // connection information. It is also the upper bound the resolution-defer grace must stay under + // (see resolutionGraceFor): a connection has to survive in the manager for the whole time its + // access log may be held waiting for the ztunnel lb mapping. connectionDeleteDelayTime = time.Second * 20 // the connection check exist time connectionCheckExistTime = time.Second * 30 + + // the interval of reporting the un-resolved remote address summary + unresolvedRemoteReportInterval = time.Second * 30 + // the max count of distinct un-resolved remote addresses to track per interval + unresolvedRemoteMaxTrack = 100 + // the max count of top un-resolved remote addresses in the report log + unresolvedRemoteTopCount = 10 + + // resolutionDeferMargin is added to the flush period to bound how long a connection whose remote + // is still being resolved is held before its access log is sent with the raw service IP. It only + // absorbs the short async delivery jitter of the ztunnel ConnectionResult mapping event(which is + // promptly delivered now that the mapping perf queue is drained by several reader goroutines, + // ztunnelMappingQueueParallels, and applied by the pull-per-flush-cycle plus the push in + // RetroResolveBySrc). It is deliberately SHORT: a LONGER hold was measured to REGRESS the + // strict no-degenerate check(grace 30s and 50s both produced MORE failures than 15s), because a + // connection that still does not resolve emits its raw-IP log in a LATER minute-bucket, which + // then lingers in the topology query window longer. So the residual is minimized at the source + // (parallel readers + push), not by holding longer. + resolutionDeferMargin = time.Second * 10 + // defaultFlushPeriod is used to derive the resolution grace period when the configured + // flush period cannot be parsed + defaultFlushPeriod = time.Second * 5 ) type ConnectEventWithSocket struct { @@ -78,6 +106,20 @@ type FlusherListener interface { ReadyToFlushConnection(connection *ConnectionInfo, getConnectionFromEvent events.Event) } +// ResolutionAwareFlusher is an optional interface a FlusherListener may implement to +// report that a connection's remote address is still being resolved(e.g. the ztunnel +// lb mapping has not been correlated yet). When any listener reports pending, the runner +// defers the connection's logs for a bounded grace period instead of emitting them with +// the raw service IP, which would leave a degenerate "-|service|-" entity in the backend. +type ResolutionAwareFlusher interface { + IsResolutionPending(connection *ConnectionInfo) bool + // UnresolvedReason returns a short category for WHY a connection reached the end of its + // lifetime without its real destination being attached, so the periodic summary can attribute + // the raw-IP socket pairs to a likely cause(which resolution source / environment did not + // provide the mapping). An empty string means this flusher has no opinion for the connection. + UnresolvedReason(connection *ConnectionInfo) string +} + type ProcessListener interface { OnNewProcessMonitoring(pid int32) OnProcessRemoved(pid int32) @@ -107,6 +149,28 @@ type ConnectionManager struct { connectTracker *ip.ConnTrack connectionProtocolBreakMap *cache.Expiring + + // statistics of the remote address resolving, for the periodic summary report + remoteAddressBuildCount atomic.Int64 + unresolvedRemoteCount atomic.Int64 + ztunnelResolvedRemoteCount atomic.Int64 + conntrackResolvedRemoteCount atomic.Int64 + unresolvedRemoteLock sync.Mutex + unresolvedRemotes map[string]int64 + unresolvedByReason map[string]int64 + // resolutionLatency buckets the open->resolution delay of ztunnel-resolved connections, to + // LOCATE the late-mapping-event problem: a mass in the >=15s(over-grace) bucket proves the + // mapping events are being delivered too slowly(perf-queue reader backlog), which is exactly the + // window in which a connection would have already flushed with the raw IP -> degenerate node. + resolutionLatencyUnder1s atomic.Int64 + resolutionLatency1to5s atomic.Int64 + resolutionLatency5to15s atomic.Int64 + resolutionLatencyOver15s atomic.Int64 + resolutionLatencyMaxMilli atomic.Int64 + + // resolutionGracePeriod is how long a connection whose remote is still being resolved + // is deferred before its logs are flushed anyway; derived from the flush period + margin + resolutionGracePeriod time.Duration } func (c *ConnectionManager) RegisterProcessor(processor ConnectionProcessor) { @@ -131,13 +195,35 @@ type ConnectionInfo struct { LastCheckExistTime time.Time DeleteAfter *time.Time ProtocolBreak bool -} - -func NewConnectionManager(_ *Config, moduleMgr *module.Manager, bpfLoader *bpf.Loader, filter MonitorFilter) *ConnectionManager { + // CreatedAt is when the connection was first seen by the manager; used to measure how long a + // ClusterIP connection took to be resolved(open -> ztunnel mapping applied), which locates the + // late-mapping-event problem: a large open-to-resolution latency means the mapping event was + // delivered slowly(perf-queue reader backlog), not that it was missing. + CreatedAt time.Time + // ResolutionDeadline, once set, is the time until which this connection's access logs + // are deferred waiting for a resolver(the ztunnel lb mapping) to attach the real pod + // IP. It is set on the first deferred flush and bounds the wait PER CONNECTION: after + // it passes the logs are emitted as-is(raw service IP) even if still unresolved, so a + // genuinely un-resolvable remote is not delayed forever and a long-lived connection is + // not re-deferred on every log. + ResolutionDeadline *time.Time +} + +func NewConnectionManager(config *Config, moduleMgr *module.Manager, bpfLoader *bpf.Loader, filter MonitorFilter) *ConnectionManager { track, err := ip.NewConnTrack() if err != nil { log.Warnf("cannot create the connection tracker, %v", err) } + // derive the resolution grace period from the flush period(+margin): the ztunnel lb + // mapping arrives asynchronously and normally lands within one flush period, so + // waiting a flush period plus a small margin lets short connections resolve to the + // real pod IP instead of being reported with the raw service IP + flushPeriod := defaultFlushPeriod + if config != nil { + if parsed, perr := time.ParseDuration(config.Flush.Period); perr == nil && parsed > 0 { + flushPeriod = parsed + } + } mgr := &ConnectionManager{ moduleMgr: moduleMgr, processOP: moduleMgr.FindModule(process.ModuleName).(process.Operator), @@ -150,12 +236,91 @@ func NewConnectionManager(_ *Config, moduleMgr *module.Manager, bpfLoader *bpf.L flushListeners: make([]FlusherListener, 0), connectTracker: track, connectionProtocolBreakMap: cache.NewExpiring(), + unresolvedRemotes: make(map[string]int64), + unresolvedByReason: make(map[string]int64), + resolutionGracePeriod: resolutionGraceFor(flushPeriod), } return mgr } +// resolutionGraceFor bounds the resolution-defer grace so it stays below connectionDeleteDelayTime +// (with one flush period of headroom). A connection is removed connectionDeleteDelayTime after it +// closes, and a deferred log can only be emitted at a flush cycle at or after its resolution +// deadline; a grace reaching past the delete time would let the connection be deleted first and its +// re-queued logs dropped. When even one flush period does not fit(an unusually large configured +// flush period), the grace is zero and ShouldDeferForResolution does not defer at all - the log is +// then emitted immediately with the raw IP, which is unresolved but never dropped. +func resolutionGraceFor(flushPeriod time.Duration) time.Duration { + grace := flushPeriod + resolutionDeferMargin + if maxGrace := connectionDeleteDelayTime - flushPeriod; grace > maxGrace { + grace = maxGrace + } + if grace < 0 { + grace = 0 + } + return grace +} + +// ShouldDeferForResolution reports whether the connection's access logs should be held +// back for one more flush cycle because a resolver(the ztunnel lb mapping) has not yet +// attached the real destination. It is bounded per connection: the first time a +// connection is found pending, a deadline(now + grace) is recorded, and once that +// deadline passes the logs are flushed as-is even if still unresolved. This closes the +// short-connection race where the asynchronous ztunnel mapping event is processed just +// after the connection has already been flushed with the raw service IP. +func (c *ConnectionManager) ShouldDeferForResolution(conn *ConnectionInfo) bool { + // grace <= 0 means deferral is disabled(the flush period is too large to fit a grace below + // connectionDeleteDelayTime, see resolutionGraceFor) - never hold a log back in that case + if conn == nil || c.resolutionGracePeriod <= 0 { + return false + } + pending := false + for _, l := range c.flushListeners { + if r, ok := l.(ResolutionAwareFlusher); ok && r.IsResolutionPending(conn) { + pending = true + break + } + } + if !pending { + // resolved(or no resolver is waiting) - do not defer + return false + } + now := time.Now() + if conn.ResolutionDeadline == nil { + deadline := now.Add(c.resolutionGracePeriod) + conn.ResolutionDeadline = &deadline + return true + } + return now.Before(*conn.ResolutionDeadline) +} + +// RecordResolutionLatency records how long a connection took from open to being resolved by the +// ztunnel correlation, into coarse buckets, so the periodic summary can show the open->resolution +// delay distribution. The >=15s bucket is the diagnostic smoking gun for the late-mapping-event +// problem(events delivered after the resolution-defer grace would have already flushed the raw IP). +func (c *ConnectionManager) RecordResolutionLatency(conn *ConnectionInfo) { + if conn == nil || conn.CreatedAt.IsZero() { + return + } + d := time.Since(conn.CreatedAt) + switch { + case d < time.Second: + c.resolutionLatencyUnder1s.Add(1) + case d < 5*time.Second: + c.resolutionLatency1to5s.Add(1) + case d < 15*time.Second: + c.resolutionLatency5to15s.Add(1) + default: + c.resolutionLatencyOver15s.Add(1) + } + if ms := d.Milliseconds(); ms > c.resolutionLatencyMaxMilli.Load() { + c.resolutionLatencyMaxMilli.Store(ms) + } +} + func (c *ConnectionManager) Start(ctx context.Context, accessLogContext *AccessLogContext) { c.processOP.AddListener(c) + c.startUnresolvedRemoteReporter(ctx) // starting to clean up the un-active connection in BPF go func() { @@ -267,6 +432,7 @@ func (c *ConnectionManager) buildRemoteAddress(e *events.SocketConnectEvent, soc } } + c.remoteAddressBuildCount.Add(1) // found local address with pid if pid, exist := c.localIPWithPid[socket.DestIP]; exist && pid != 0 { return c.buildLocalAddress(uint32(pid), socket.DestPort, socket) @@ -277,6 +443,163 @@ func (c *ConnectionManager) buildRemoteAddress(e *events.SocketConnectEvent, soc return c.buildAddressFromRemote(socket.DestIP, socket.DestPort) } +// recordConnectionResolveResult records the final remote address resolve result of the +// connection, it MUST be called when the connection is being deleted from the manager, +// at that point all resolvers are finalized: the conntrack rewriting happens on the +// address building, and the ztunnel correlation could attach the real destination on +// any later flush, so judging any earlier would mis-count the resolved connections +func (c *ConnectionManager) recordConnectionResolveResult(connection *ConnectionInfo) { + if connection == nil || connection.RPCConnection == nil { + return + } + remote := connection.RPCConnection.GetRemote() + if remote == nil || remote.GetIp() == nil { + // the remote address is resolved to a local monitored process, not a raw IP + return + } + if connection.RPCConnection.GetAttachment() != nil { + // the ztunnel correlation attached the real destination + c.ztunnelResolvedRemoteCount.Add(1) + return + } + if connection.Socket != nil && connection.Socket.ConnTrackResolved { + // the conntrack query already rewrote the address to the real peer(e.g. pod IP), + // the address is sent as a raw IP but the backend could resolve it + c.conntrackResolvedRemoteCount.Add(1) + return + } + // ask the resolution-aware flusher(s) WHY this connection ended up unresolved, so the summary + // can group the raw-IP socket pairs by the environment/source that failed to provide a mapping + reason := "unresolved" + for _, l := range c.flushListeners { + if r, ok := l.(ResolutionAwareFlusher); ok { + if rs := r.UnresolvedReason(connection); rs != "" { + reason = rs + break + } + } + } + src := "unknown" + if connection.Socket != nil { + src = fmt.Sprintf("%s:%d", connection.Socket.SrcIP, connection.Socket.SrcPort) + } + c.recordUnresolvedRemote(reason, fmt.Sprintf("%s->%s:%d", src, remote.GetIp().GetHost(), remote.GetIp().GetPort())) +} + +// recordUnresolvedRemote records a socket pair(src->dst) that cannot be resolved by any of the +// resolvers(local process, conntrack, ztunnel correlation) - it would reach the backend as a raw +// IP - together with the categorized reason, for the periodic summary. The per-reason totals are +// always counted; the per-pair map is bounded so a flood of distinct pairs cannot grow it without +// limit. +func (c *ConnectionManager) recordUnresolvedRemote(reason, socketPair string) { + c.unresolvedRemoteCount.Add(1) + c.unresolvedRemoteLock.Lock() + defer c.unresolvedRemoteLock.Unlock() + c.unresolvedByReason[reason]++ + key := reason + " " + socketPair + if _, exist := c.unresolvedRemotes[key]; !exist && len(c.unresolvedRemotes) >= unresolvedRemoteMaxTrack { + return + } + c.unresolvedRemotes[key]++ +} + +// drainUnresolvedRemotes returns, since the last report, the top un-resolved socket pairs(each +// prefixed with its reason) and the full per-reason breakdown, and resets the tracking maps. +func (c *ConnectionManager) drainUnresolvedRemotes() (topPairs, byReason string) { + c.unresolvedRemoteLock.Lock() + pairs := c.unresolvedRemotes + reasons := c.unresolvedByReason + c.unresolvedRemotes = make(map[string]int64) + c.unresolvedByReason = make(map[string]int64) + c.unresolvedRemoteLock.Unlock() + + sortByCount := func(m map[string]int64, topN int) string { + if len(m) == 0 { + return "none" + } + type kv struct { + k string + v int64 + } + sorted := make([]kv, 0, len(m)) + for k, v := range m { + sorted = append(sorted, kv{k: k, v: v}) + } + sort.Slice(sorted, func(i, j int) bool { return sorted[i].v > sorted[j].v }) + if topN > 0 && len(sorted) > topN { + sorted = sorted[:topN] + } + items := make([]string, 0, len(sorted)) + for _, s := range sorted { + items = append(items, fmt.Sprintf("%s=%d", s.k, s.v)) + } + return strings.Join(items, ", ") + } + return sortByCount(pairs, unresolvedRemoteTopCount), sortByCount(reasons, 0) +} + +// startUnresolvedRemoteReporter periodically reports the summary of remote addresses +// that cannot be resolved(by neither the conntrack nor the ztunnel correlation) at the info level +func (c *ConnectionManager) startUnresolvedRemoteReporter(ctx context.Context) { + go func() { + ticker := time.NewTicker(unresolvedRemoteReportInterval) + defer ticker.Stop() + var lastUnresolvedCount int64 + for { + select { + case <-ticker.C: + unresolvedCount := c.unresolvedRemoteCount.Load() + if unresolvedCount == lastUnresolvedCount { + continue + } + lastUnresolvedCount = unresolvedCount + conntrackStats := "not available" + if c.connectTracker != nil { + conntrackStats = c.connectTracker.StatsString() + } + topPairs, byReason := c.drainUnresolvedRemotes() + log.Infof("remote address resolve summary: total remote addresses built: %d, "+ + "resolved by conntrack: %d, resolved by ztunnel correlation: %d, unresolved(sent as raw IP): %d, "+ + "conntrack stats: {%s}, unresolved by reason: {%s}, "+ + "open->resolution latency buckets {<1s: %d, 1-5s: %d, 5-15s: %d, >=15s: %d, max: %dms}, "+ + "top unresolved socket pairs since last report: %s", + c.remoteAddressBuildCount.Load(), c.conntrackResolvedRemoteCount.Load(), + c.ztunnelResolvedRemoteCount.Load(), unresolvedCount, + conntrackStats, byReason, + c.resolutionLatencyUnder1s.Load(), c.resolutionLatency1to5s.Load(), + c.resolutionLatency5to15s.Load(), c.resolutionLatencyOver15s.Load(), + c.resolutionLatencyMaxMilli.Load(), topPairs) + case <-ctx.Done(): + return + } + } + }() +} + +// RetroResolveBySrc is the push side of the ztunnel correlation: when a source-keyed mapping +// arrives(a ConnectionResult / access-log event), the collector calls this so every connection +// still held in the manager for that source address is re-offered to the flush listeners right +// away, instead of waiting for the connection's own next flush cycle to pull the cache. This +// closes the window where a short connection's mapping event lands between its flush cycles or +// just as it would flush, which is the dominant residual "-|service|-" cause under high load. +func (c *ConnectionManager) RetroResolveBySrc(srcIP string, srcPort uint16) { + c.connections.IterCb(func(_ string, v interface{}) { + connection, ok := v.(*ConnectionInfo) + if !ok || connection.Socket == nil || connection.RPCConnection == nil { + return + } + // only an unresolved client(outbound) leg for this exact source is a candidate + if connection.RPCConnection.Attachment != nil || + connection.Socket.Role != enums.ConnectionRoleClient || + connection.Socket.SrcIP != srcIP || connection.Socket.SrcPort != srcPort { + return + } + for _, flush := range c.flushListeners { + flush.ReadyToFlushConnection(connection, nil) + } + }) +} + func (c *ConnectionManager) connectionPostHandle(connection *ConnectionInfo, event events.Event) { if connection == nil { return @@ -373,6 +696,7 @@ func (c *ConnectionManager) buildConnection(event *events.SocketConnectEvent, so PID: event.PID, Socket: socket, LastCheckExistTime: time.Now(), + CreatedAt: time.Now(), ProtocolBreak: protocolBreak, } } @@ -588,7 +912,7 @@ func (c *ConnectionManager) updateMonitorStatusForProcess(pid int32, monitor boo func (c *ConnectionManager) OnBuildConnectionLogFinished() { // delete all connections which marked as deletable // all deletable connection events been sent - deletableConnections := make(map[string]bool) + deletableConnections := make(map[string]*ConnectionInfo) now := time.Now() c.connections.IterCb(func(key string, v interface{}) { con, ok := v.(*ConnectionInfo) @@ -609,11 +933,13 @@ func (c *ConnectionManager) OnBuildConnectionLogFinished() { } if shouldDelete && now.After(*con.DeleteAfter) { - deletableConnections[key] = true + deletableConnections[key] = con } }) - for key := range deletableConnections { + for key, con := range deletableConnections { + // the connection state is finalized, record the remote address resolve result + c.recordConnectionResolveResult(con) log.Debugf("deleting the connection in manager: %s", key) c.connections.Remove(key) } diff --git a/pkg/accesslog/common/connection_resolution_test.go b/pkg/accesslog/common/connection_resolution_test.go new file mode 100644 index 00000000..4218b48e --- /dev/null +++ b/pkg/accesslog/common/connection_resolution_test.go @@ -0,0 +1,105 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package common + +import ( + "testing" + "time" + + "github.com/apache/skywalking-rover/pkg/accesslog/events" +) + +// fakeResolutionFlusher is a FlusherListener that also implements ResolutionAwareFlusher +// so the ShouldDeferForResolution listener dispatch can be exercised without a real ztunnel +type fakeResolutionFlusher struct { + pending bool +} + +func (f *fakeResolutionFlusher) ReadyToFlushConnection(*ConnectionInfo, events.Event) {} +func (f *fakeResolutionFlusher) IsResolutionPending(*ConnectionInfo) bool { return f.pending } +func (f *fakeResolutionFlusher) UnresolvedReason(*ConnectionInfo) string { return "test-unresolved" } + +// plainFlusher only implements FlusherListener(not ResolutionAwareFlusher), it must never +// cause a defer +type plainFlusher struct{} + +func (p *plainFlusher) ReadyToFlushConnection(*ConnectionInfo, events.Event) {} + +func newTestManager(grace time.Duration, listeners ...FlusherListener) *ConnectionManager { + return &ConnectionManager{ + flushListeners: listeners, + resolutionGracePeriod: grace, + } +} + +func TestShouldDeferForResolution(t *testing.T) { + t.Run("nil connection is never deferred", func(t *testing.T) { + mgr := newTestManager(time.Second*7, &fakeResolutionFlusher{pending: true}) + if mgr.ShouldDeferForResolution(nil) { + t.Fatal("nil connection should not be deferred") + } + }) + + t.Run("no resolution-aware listener is never deferred", func(t *testing.T) { + mgr := newTestManager(time.Second*7, &plainFlusher{}) + conn := &ConnectionInfo{} + if mgr.ShouldDeferForResolution(conn) { + t.Fatal("a plain flusher must not trigger a defer") + } + }) + + t.Run("not pending is not deferred", func(t *testing.T) { + mgr := newTestManager(time.Second*7, &fakeResolutionFlusher{pending: false}) + conn := &ConnectionInfo{} + if mgr.ShouldDeferForResolution(conn) { + t.Fatal("a not-pending connection must not be deferred") + } + if conn.ResolutionDeadline != nil { + t.Fatal("no deadline should be recorded when not pending") + } + }) + + t.Run("pending sets a deadline and defers on the first call", func(t *testing.T) { + mgr := newTestManager(time.Second*7, &fakeResolutionFlusher{pending: true}) + conn := &ConnectionInfo{} + if !mgr.ShouldDeferForResolution(conn) { + t.Fatal("a pending connection should be deferred on the first flush") + } + if conn.ResolutionDeadline == nil { + t.Fatal("a deadline should be recorded on the first defer") + } + }) + + t.Run("pending within the grace deadline keeps deferring", func(t *testing.T) { + mgr := newTestManager(time.Second*7, &fakeResolutionFlusher{pending: true}) + future := time.Now().Add(time.Second * 3) + conn := &ConnectionInfo{ResolutionDeadline: &future} + if !mgr.ShouldDeferForResolution(conn) { + t.Fatal("a pending connection within its deadline should keep deferring") + } + }) + + t.Run("pending past the grace deadline stops deferring", func(t *testing.T) { + mgr := newTestManager(time.Second*7, &fakeResolutionFlusher{pending: true}) + past := time.Now().Add(-time.Second) + conn := &ConnectionInfo{ResolutionDeadline: &past} + if mgr.ShouldDeferForResolution(conn) { + t.Fatal("a connection past its deadline must be flushed even if still pending") + } + }) +} diff --git a/pkg/accesslog/runner.go b/pkg/accesslog/runner.go index 5dce05d0..6b45a336 100644 --- a/pkg/accesslog/runner.go +++ b/pkg/accesslog/runner.go @@ -221,6 +221,12 @@ func (r *Runner) buildProtocolLog(protocolLog common.ProtocolLog) (*common.Conne } return nil, nil, nil, true } + // hold the log for one more flush cycle while the ztunnel lb mapping is still being + // correlated(bounded per-connection), so a short connection resolves to the real pod + // IP instead of being emitted with the raw service IP + if r.context.ConnectionMgr.ShouldDeferForResolution(connection) { + return nil, nil, nil, true + } kernelLogs := make([]*v3.AccessLogKernelLog, 0) for _, kl := range protocolLog.RelateKernelLogs() { event := forwarder.BuildKernelLogFromEvent(common.LogTypeKernelTransfer, kl) @@ -248,6 +254,12 @@ func (r *Runner) buildKernelLog(kernelLog common.KernelLog) (*common.ConnectionI } return nil, nil, true } + // hold the log for one more flush cycle while the ztunnel lb mapping is still being + // correlated(bounded per-connection), so a short connection resolves to the real pod + // IP instead of being emitted with the raw service IP + if r.context.ConnectionMgr.ShouldDeferForResolution(connection) { + return nil, nil, true + } event := forwarder.BuildKernelLogFromEvent(kernelLog.Type(), kernelLog.Event()) return connection, event, false } diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index 018c5c42..176d0fb9 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -19,17 +19,30 @@ package logger import ( "strings" + "sync" "github.com/sirupsen/logrus" ) var ( - root = initializeDefaultLogger() + // root is the base logger; its level follows the global "logger.level" config. + root = initializeLogger(DefaultLoggerLevel) + // debugRoot is always at DebugLevel. A module elevated through + // "logger.debug_modules" binds its whole entry here, so ALL of its levels + // (debug included) are emitted regardless of the global level. + debugRoot = initializeLogger(logrus.DebugLevel) + + // registry keeps every logger created(most are package-level vars created at + // import time, i.e. before setupLogger runs) so the debug allowlist can be + // re-applied once the config is loaded. + registryMux sync.Mutex + registry []*Logger + debugPrefixes []string ) type Logger struct { *logrus.Entry - module []string + moduleString string } // GetLogger for the module @@ -38,9 +51,44 @@ func GetLogger(modules ...string) *Logger { if len(modules) > 0 { moduleString = strings.Join(modules, ".") } - return &Logger{Entry: root.WithField("module", moduleString), module: modules} + l := &Logger{moduleString: moduleString} + l.apply() + + registryMux.Lock() + registry = append(registry, l) + registryMux.Unlock() + return l +} + +// apply (re)binds the underlying entry: a module in the debug allowlist logs +// through debugRoot(always DebugLevel), every other module through root(gated by +// the global level). Binding the whole entry - rather than overriding only the +// debug methods - keeps the level semantics monotonic: an elevated module never +// drops its own info/warn lines while still emitting debug. +func (l *Logger) apply() { + base := root + if matchDebugModule(l.moduleString) { + base = debugRoot + } + l.Entry = base.WithField("module", l.moduleString) } func (l *Logger) Enable(level logrus.Level) bool { - return root.IsLevelEnabled(level) + return l.Entry.Logger.IsLevelEnabled(level) +} + +// matchDebugModule reports whether the module(or one of its parent modules) is +// in the configured debug allowlist. Matching is on module-segment boundaries, +// so prefix "accesslog.collector" matches "accesslog.collector" and +// "accesslog.collector.ztunnel" but not "accesslog.collectorx". +func matchDebugModule(moduleString string) bool { + for _, p := range debugPrefixes { + if p == "" { + continue + } + if moduleString == p || strings.HasPrefix(moduleString, p+".") { + return true + } + } + return false } diff --git a/pkg/logger/logger_test.go b/pkg/logger/logger_test.go new file mode 100644 index 00000000..2fc9e166 --- /dev/null +++ b/pkg/logger/logger_test.go @@ -0,0 +1,142 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package logger + +import ( + "bytes" + "strings" + "testing" +) + +// applyConfig runs the same setup path the module bootstrap uses and redirects +// both loggers to buffers so the emitted output can be asserted. +func applyConfig(t *testing.T, level, debugModules string) (base, debug *bytes.Buffer) { + t.Helper() + if err := setupLogger(&Config{Level: level, DebugModules: debugModules}); err != nil { + t.Fatalf("setupLogger: %v", err) + } + base, debug = &bytes.Buffer{}, &bytes.Buffer{} + root.SetOutput(base) + debugRoot.SetOutput(debug) + return base, debug +} + +func emitted(base, debug *bytes.Buffer, marker string) bool { + return strings.Contains(base.String(), marker) || strings.Contains(debug.String(), marker) +} + +func TestDebugModulesElevatesOnlyAllowlisted(t *testing.T) { + // only the ztunnel module is elevated; the connection module stays at INFO. + base, debug := applyConfig(t, "info", "accesslog.collector.ztunnel") + + ztunnel := GetLogger("accesslog", "collector", "ztunnel") + conn := GetLogger("access_log", "collector", "connection") + + ztunnel.Debugf(" zt-debug %d", 1) + conn.Debugf("conn-debug %d", 2) + + if !emitted(base, debug, "zt-debug") { + t.Errorf("expected the allowlisted ztunnel module debug log to be emitted") + } + if emitted(base, debug, "conn-debug") { + t.Errorf("expected the non-allowlisted connection module debug log to be suppressed at info level") + } +} + +func TestDebugModulesPrefixMatchesChildrenNotSiblings(t *testing.T) { + // a parent prefix elevates its child module, but not a same-prefixed sibling. + base, debug := applyConfig(t, "info", "accesslog.collector") + + child := GetLogger("accesslog", "collector", "ztunnel") + // "accesslog.collectorx" must NOT match the "accesslog.collector" prefix. + sibling := GetLogger("accesslog", "collectorx") + + child.Debugf("child-debug") + sibling.Debugf("sibling-debug") + if !emitted(base, debug, "child-debug") { + t.Errorf("expected the child module debug log to be emitted") + } + if emitted(base, debug, "sibling-debug") { + t.Errorf("expected the sibling module debug log to be suppressed") + } +} + +func TestGlobalDebugLevelEmitsAllModules(t *testing.T) { + // with no allowlist but a global debug level, every module logs at debug. + base, debug := applyConfig(t, "debug", "") + + any := GetLogger("access_log", "collector", "connection") + any.Debugf("global-debug") + if !emitted(base, debug, "global-debug") { + t.Errorf("expected debug logs when the global level is debug") + } +} + +func TestInfoStillEmittedForNonAllowlistedModule(t *testing.T) { + // elevating one module must not suppress the info logs of the others. + base, debug := applyConfig(t, "info", "accesslog.collector.ztunnel") + + conn := GetLogger("access_log", "collector", "connection") + conn.Infof("conn-info") + if !emitted(base, debug, "conn-info") { + t.Errorf("expected info logs to be emitted for non-allowlisted modules") + } +} + +// TestElevatedModuleDoesNotDropInfoAboveGlobalLevel guards the severity-inversion +// bug: an elevated module must emit its OWN info/warn lines even when the global +// level is higher(here warn), not just its debug lines. +func TestElevatedModuleDoesNotDropInfoAboveGlobalLevel(t *testing.T) { + base, debug := applyConfig(t, "warn", "accesslog.collector.ztunnel") + + zt := GetLogger("accesslog", "collector", "ztunnel") + zt.Debugf("zt-elevated-debug") + zt.Infof("zt-elevated-info") + zt.Warnf("zt-elevated-warn") + + if !emitted(base, debug, "zt-elevated-debug") { + t.Errorf("expected the elevated module to emit debug even when global level is warn") + } + if !emitted(base, debug, "zt-elevated-info") { + t.Errorf("expected the elevated module to emit info even when global level is warn(severity inversion)") + } + if !emitted(base, debug, "zt-elevated-warn") { + t.Errorf("expected the elevated module to emit warn") + } + + // a non-elevated module still respects the global(warn) level: info dropped. + other := GetLogger("access_log", "collector", "connection") + other.Infof("other-info-dropped") + if emitted(base, debug, "other-info-dropped") { + t.Errorf("expected a non-elevated module info to be dropped at global warn level") + } +} + +// TestSetupReAppliesToLoggersCreatedBeforeConfig covers the registry re-apply +// loop, the sole mechanism that elevates package-level loggers created at import +// time(before setupLogger runs). +func TestSetupReAppliesToLoggersCreatedBeforeConfig(t *testing.T) { + // create the logger BEFORE the config that should elevate it exists. + pre := GetLogger("accesslog", "collector", "ztunnel") + + base, debug := applyConfig(t, "info", "accesslog.collector.ztunnel") + pre.Debugf("pre-created-debug") + if !emitted(base, debug, "pre-created-debug") { + t.Errorf("expected setupLogger to re-apply the debug allowlist to a logger created before it ran") + } +} diff --git a/pkg/logger/settings.go b/pkg/logger/settings.go index 7705bb9a..59531ba8 100644 --- a/pkg/logger/settings.go +++ b/pkg/logger/settings.go @@ -18,6 +18,8 @@ package logger import ( + "strings" + "github.com/sirupsen/logrus" ) @@ -26,26 +28,48 @@ const ( ) type Config struct { + // Level is the global lowest level that is allowed to be printed. Level string `mapstructure:"level"` + // DebugModules is a comma separated list of module(name prefix) to elevate to + // the debug level regardless of Level, e.g. "accesslog.collector.ztunnel". + // This keeps the high volume modules quiet(bounding the logging allocation + // churn) while still getting debug detail for the modules under investigation. + DebugModules string `mapstructure:"debug_modules"` } // setupLogger when Bootstrap func setupLogger(config *Config) (err error) { - return updateLogger(root, config) -} - -func updateLogger(log *logrus.Logger, config *Config) error { level, err := logrus.ParseLevel(config.Level) if err != nil { return err } - log.SetLevel(level) + root.SetLevel(level) + // debugRoot stays at DebugLevel; only the allowlist decides who uses it. + + debugPrefixes = parseDebugModules(config.DebugModules) + // re-apply the allowlist to the loggers created before the config was loaded + // (package-level GetLogger vars run at import time). + registryMux.Lock() + for _, l := range registry { + l.apply() + } + registryMux.Unlock() return nil } -func initializeDefaultLogger() *logrus.Logger { +func parseDebugModules(value string) []string { + res := make([]string, 0) + for _, p := range strings.Split(value, ",") { + if p = strings.TrimSpace(p); p != "" { + res = append(res, p) + } + } + return res +} + +func initializeLogger(level logrus.Level) *logrus.Logger { l := logrus.New() - l.SetLevel(DefaultLoggerLevel) + l.SetLevel(level) l.SetFormatter(&logrus.TextFormatter{ FullTimestamp: true, DisableColors: true, diff --git a/pkg/process/finders/kubernetes/template.go b/pkg/process/finders/kubernetes/template.go index cea08f96..9d9ca7cd 100644 --- a/pkg/process/finders/kubernetes/template.go +++ b/pkg/process/finders/kubernetes/template.go @@ -151,6 +151,15 @@ func (t *TemplatePodJudgment) HasContainer(name string) bool { return true } } + // native sidecars(e.g. istio-proxy on k8s 1.28+) run as restartPolicy:Always initContainers, + // so a sidecar-injected pod carries the sidecar in InitContainers rather than Containers. Check + // both, otherwise the "exclude pods that have " filter silently fails to match a + // native-sidecar pod and the pod(its app process) gets monitored anyway. + for _, c := range t.pc.Pod.Spec.InitContainers { + if c.Name == name { + return true + } + } return false } diff --git a/pkg/tools/btf/linker.go b/pkg/tools/btf/linker.go index 271a10d3..f8aaee9a 100644 --- a/pkg/tools/btf/linker.go +++ b/pkg/tools/btf/linker.go @@ -247,6 +247,13 @@ func (m *Linker) OpenUProbeExeFile(path string) *UProbeExeFile { } } +// Found reports whether the executable was opened successfully. When it is false the AddLink* +// calls are silent no-ops, so a caller that must not proceed(e.g. arm a BPF gate) without a real +// uprobe attached should check this instead of assuming AddLink succeeded. +func (u *UProbeExeFile) Found() bool { + return u.found +} + func (u *UProbeExeFile) AddLink(symbol string, enter, exit *ebpf.Program) { u.AddLinkWithType(symbol, true, enter) u.AddLinkWithType(symbol, false, exit) diff --git a/pkg/tools/host/file.go b/pkg/tools/host/file.go index ac02ec1a..b3762883 100644 --- a/pkg/tools/host/file.go +++ b/pkg/tools/host/file.go @@ -23,8 +23,9 @@ import ( ) var ( - hostProcMappingPath string - hostEtcMappingPath string + hostProcMappingPath string + hostEtcMappingPath string + hostVarLogPodsMappingPath string ) func init() { @@ -34,6 +35,7 @@ func init() { os.Setenv("HOST_PROC", hostProcMappingPath) } hostEtcMappingPath = os.Getenv("ROVER_HOST_ETC_MAPPING") + hostVarLogPodsMappingPath = os.Getenv("ROVER_HOST_VAR_LOG_PODS_MAPPING") } func GetHostProcInHost(procSubPath string) string { @@ -50,6 +52,19 @@ func GetHostEtcInHost(etcSubPath string) string { return cleanPath("/etc/" + etcSubPath) } +// GetHostVarLogPodsInHost resolves a path under the kubelet pod-log directory(/var/log/pods on the +// host) as seen from inside the agent container. That directory is where the kubelet writes every +// pod's container logs for ALL CRI runtimes(containerd, CRI-O, cri-dockerd), so it is the runtime +// independent place to tail the ztunnel access log. The host mount point is injected through +// ROVER_HOST_VAR_LOG_PODS_MAPPING(the same pattern as ROVER_HOST_PROC_MAPPING), so it is not +// hard-coded; when unset the real host path /var/log/pods is used. +func GetHostVarLogPodsInHost(subPath string) string { + if hostVarLogPodsMappingPath != "" { + return cleanPath(hostVarLogPodsMappingPath + "/" + subPath) + } + return cleanPath("/var/log/pods/" + subPath) +} + func cleanPath(p string) string { return path.Clean(p) } diff --git a/pkg/tools/ip/conntrack.go b/pkg/tools/ip/conntrack.go index 831d2cc2..e0b17b6a 100644 --- a/pkg/tools/ip/conntrack.go +++ b/pkg/tools/ip/conntrack.go @@ -20,6 +20,7 @@ package ip import ( "fmt" "net" + "sync/atomic" "syscall" "github.com/florianl/go-conntrack" @@ -42,6 +43,12 @@ var numberStrategies = []struct { type ConnTrack struct { tracker *conntrack.Nfct + + // counters for quantifying the conntrack resolve success/miss rate + queryCount atomic.Int64 + resolvedCount atomic.Int64 + notFoundCount atomic.Int64 + ignoredCount atomic.Int64 } func NewConnTrack() (*ConnTrack, error) { @@ -60,6 +67,7 @@ func (c *ConnTrack) UpdateRealPeerAddress(addr *SocketPair) error { } tuple := c.parseSocketToTuple(addr) + c.queryCount.Add(1) for _, info := range numberStrategies { tuple.Proto.Number = &(info.proto) @@ -67,25 +75,39 @@ func (c *ConnTrack) UpdateRealPeerAddress(addr *SocketPair) error { session, e := c.tracker.Get(conntrack.Conntrack, family, conntrack.Con{Origin: tuple}) if e != nil { // try to get the reply session, if the info not exists or from accept events, have error is normal - return fmt.Errorf("cannot get the conntrack session, type: %s, family: %d, origin src: %s:%d, origin dest: %s:%d, error: %v", info.name, - family, tuple.Src, *tuple.Proto.SrcPort, tuple.Dst, *tuple.Proto.DstPort, e) + c.notFoundCount.Add(1) + return fmt.Errorf("cannot get the conntrack session, type: %s, family: %d, origin src: %s:%d, origin dest: %s:%d, error: %v, stats: %s", info.name, + family, tuple.Src, *tuple.Proto.SrcPort, tuple.Dst, *tuple.Proto.DstPort, e, c.StatsString()) } if res := c.filterValidateReply(session, tuple); res != nil { if !ShouldIgnoreConntrack(addr.DestIP, res.Src.String(), *res.Proto.SrcPort) { + c.resolvedCount.Add(1) addr.DestIP = res.Src.String() addr.NeedConnTrack = false - log.Debugf("update real peer address from conntrack: %s:%d", addr.DestIP, addr.DestPort) + addr.ConnTrackResolved = true + log.Debugf("update real peer address from conntrack: %s:%d, stats: %s", addr.DestIP, addr.DestPort, c.StatsString()) } else { - log.Debugf("ignore conntrack, original dest IP: %s:%d, conntrack IP: %s:%d", - addr.DestIP, addr.DestPort, res.Src.String(), *res.Proto.SrcPort) + c.ignoredCount.Add(1) + log.Debugf("ignore conntrack, original dest IP: %s:%d, conntrack IP: %s:%d, stats: %s", + addr.DestIP, addr.DestPort, res.Src.String(), *res.Proto.SrcPort, c.StatsString()) } return nil } } + c.notFoundCount.Add(1) + log.Debugf("no matched conntrack reply tuple found, origin src: %s:%d, origin dest: %s:%d, stats: %s", + addr.SrcIP, addr.SrcPort, addr.DestIP, addr.DestPort, c.StatsString()) return nil } +// StatsString returns the cumulative conntrack query result counters, +// used to quantify the resolve miss rate from logs +func (c *ConnTrack) StatsString() string { + return fmt.Sprintf("queries: %d, resolved: %d, not found: %d, ignored: %d", + c.queryCount.Load(), c.resolvedCount.Load(), c.notFoundCount.Load(), c.ignoredCount.Load()) +} + func ShouldIgnoreConntrack(originalDestIP, conntrackIP string, conntrackPort uint16) bool { // if the original dest IP is not local host // and the conntrack IP is local host, and port is 15001, such as 127.0.0.1:15001, means the conntrack is to istio-proxy diff --git a/pkg/tools/ip/tcpresolver.go b/pkg/tools/ip/tcpresolver.go index 9f093090..357ec983 100644 --- a/pkg/tools/ip/tcpresolver.go +++ b/pkg/tools/ip/tcpresolver.go @@ -36,6 +36,9 @@ type SocketPair struct { DestPort uint16 NeedConnTrack bool + // ConnTrackResolved is true when the conntrack query successfully rewrote + // the DestIP to the real peer address + ConnTrackResolved bool } func (s *SocketPair) IsValid() bool { diff --git a/pkg/tools/netns/netns.go b/pkg/tools/netns/netns.go new file mode 100644 index 00000000..b6eac8b4 --- /dev/null +++ b/pkg/tools/netns/netns.go @@ -0,0 +1,68 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package netns + +import ( + "fmt" + "os" + "runtime" + + "golang.org/x/sys/unix" +) + +// RunInNetNS executes fn with the current OS thread switched into the network +// namespace referenced by netnsPath(e.g. /proc//ns/net), and restores the +// original network namespace afterwards. +// +// Requires CAP_SYS_ADMIN. Everything that must happen inside the target +// namespace(creating sockets, dialing, reading responses) should be done +// inside fn, since only the calling OS thread is switched. +func RunInNetNS(netnsPath string, fn func() error) error { + target, err := os.Open(netnsPath) + if err != nil { + return fmt.Errorf("open target netns %s error: %w", netnsPath, err) + } + defer target.Close() + + runtime.LockOSThread() + // NOTE: open through the container's own procfs instead of the host proc mapping: + // thread-self is a kernel magic symlink which always resolves to the calling thread + // itself on any procfs instance, the host proc mapping is only required when + // accessing OTHER processes' entries + origin, err := os.Open("/proc/thread-self/ns/net") + if err != nil { + runtime.UnlockOSThread() + return fmt.Errorf("open current netns error: %w", err) + } + defer origin.Close() + + if err := unix.Setns(int(target.Fd()), unix.CLONE_NEWNET); err != nil { + runtime.UnlockOSThread() + return fmt.Errorf("enter netns %s error: %w", netnsPath, err) + } + + fnErr := fn() + + if err := unix.Setns(int(origin.Fd()), unix.CLONE_NEWNET); err != nil { + // cannot switch back to the original namespace, keep the thread locked + // so the runtime destroys it instead of reusing it for other goroutines + return fmt.Errorf("restore original netns error: %v, fn error: %v", err, fnErr) + } + runtime.UnlockOSThread() + return fnErr +} diff --git a/pkg/tools/netns/netns_test.go b/pkg/tools/netns/netns_test.go new file mode 100644 index 00000000..774a5ec1 --- /dev/null +++ b/pkg/tools/netns/netns_test.go @@ -0,0 +1,173 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package netns + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "runtime" + "syscall" + "testing" + + "golang.org/x/sys/unix" +) + +const currentThreadNetNSPath = "/proc/thread-self/ns/net" + +func netnsInode(t *testing.T, path string) uint64 { + t.Helper() + stat, err := os.Stat(path) + if err != nil { + t.Fatalf("stat %s error: %v", path, err) + } + sys, ok := stat.Sys().(*syscall.Stat_t) + if !ok { + t.Fatalf("unexpected stat type for %s", path) + } + return sys.Ino +} + +func TestRunInNetNSErrorPaths(t *testing.T) { + invalidFile := filepath.Join(t.TempDir(), "not-a-netns") + if err := os.WriteFile(invalidFile, []byte("test"), 0o600); err != nil { + t.Fatalf("prepare the invalid target file error: %v", err) + } + cases := []struct { + name string + target string + }{ + {"target path does not exist", "/proc/not-exist-process/ns/net"}, + {"target is not a network namespace", invalidFile}, + } + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + executed := false + err := RunInNetNS(tt.target, func() error { + executed = true + return nil + }) + if err == nil { + t.Fatalf("expected an error when %s", tt.name) + } + if executed { + t.Fatalf("the fn must not run when entering the netns failed(%s)", tt.name) + } + }) + } +} + +func TestRunInNetNSSameNamespace(t *testing.T) { + // lock the thread so the baseline and the after-check read the same thread + runtime.LockOSThread() + defer runtime.UnlockOSThread() + originInode := netnsInode(t, currentThreadNetNSPath) + + executed := false + err := RunInNetNS(currentThreadNetNSPath, func() error { + executed = true + if inode := netnsInode(t, currentThreadNetNSPath); inode != originInode { + return fmt.Errorf("the fn is not running in the target netns, inode: %d, want: %d", inode, originInode) + } + return nil + }) + if err != nil { + // entering a netns(even the same one) requires CAP_SYS_ADMIN + if errors.Is(err, syscall.EPERM) { + t.Skipf("requires CAP_SYS_ADMIN to run: %v", err) + } + t.Fatalf("unexpected error: %v", err) + } + if !executed { + t.Fatal("the fn is not executed") + } + if inode := netnsInode(t, currentThreadNetNSPath); inode != originInode { + t.Fatalf("the original netns is not restored after RunInNetNS") + } + + // the error of the fn should be propagated to the caller + sentinel := errors.New("sentinel error") + if err := RunInNetNS(currentThreadNetNSPath, func() error { return sentinel }); !errors.Is(err, sentinel) { + t.Fatalf("the fn error is not propagated, got: %v", err) + } +} + +// TestRunInNetNSEntersDistinctNamespace enters a genuinely DIFFERENT network +// namespace and asserts the inode observed inside fn changes(setns took effect) +// and reverts afterwards(restore took effect). Unlike the same-namespace case, +// this fails if setns or the restore were silently dropped. +func TestRunInNetNSEntersDistinctNamespace(t *testing.T) { + // build a distinct netns on a dedicated OS thread and expose its procfs path; + // the thread stays parked(via release) so the namespace is kept alive. + type nsRef struct { + path string + inode uint64 + err error + } + ready := make(chan nsRef, 1) + release := make(chan struct{}) + go func() { + // permanently pin(and never unlock) this thread: after Unshare it lives in + // the new netns and must be discarded, not reused by other goroutines. + runtime.LockOSThread() + if err := unix.Unshare(unix.CLONE_NEWNET); err != nil { + ready <- nsRef{err: err} + return + } + path := fmt.Sprintf("/proc/self/task/%d/ns/net", unix.Gettid()) + stat, err := os.Stat(path) + if err != nil { + ready <- nsRef{err: err} + return + } + ready <- nsRef{path: path, inode: stat.Sys().(*syscall.Stat_t).Ino} + <-release + }() + ns := <-ready + defer close(release) + if ns.err != nil { + if errors.Is(ns.err, syscall.EPERM) { + t.Skipf("requires CAP_SYS_ADMIN to create a network namespace: %v", ns.err) + } + t.Fatalf("create a distinct netns error: %v", ns.err) + } + + runtime.LockOSThread() + defer runtime.UnlockOSThread() + originInode := netnsInode(t, currentThreadNetNSPath) + if originInode == ns.inode { + t.Fatalf("the new namespace(inode %d) unexpectedly equals the origin", ns.inode) + } + + err := RunInNetNS(ns.path, func() error { + if inode := netnsInode(t, currentThreadNetNSPath); inode != ns.inode { + return fmt.Errorf("fn is not running in the target netns, inode: %d, want: %d", inode, ns.inode) + } + return nil + }) + if err != nil { + if errors.Is(err, syscall.EPERM) { + t.Skipf("requires CAP_SYS_ADMIN to enter a netns: %v", err) + } + t.Fatalf("unexpected error: %v", err) + } + if inode := netnsInode(t, currentThreadNetNSPath); inode != originInode { + t.Fatalf("the original netns(inode %d) is not restored after RunInNetNS, got %d", originInode, inode) + } +} From 8b8e4be102c31421bc57270747636f95bdeb453e Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Wed, 15 Jul 2026 21:47:10 +0800 Subject: [PATCH 2/2] fiz comments and CI --- .golangci.yml | 11 +++++++++ pkg/accesslog/collector/collector.go | 3 --- pkg/accesslog/collector/ztunnel_accesslog.go | 25 ++++++++++++++------ pkg/accesslog/collector/ztunnel_admin.go | 4 ++-- pkg/accesslog/common/connection.go | 17 +++++++++---- pkg/logger/logger.go | 2 +- pkg/logger/logger_test.go | 4 ++-- pkg/process/finders/kubernetes/template.go | 1 + pkg/tools/netns/netns.go | 9 +++++++ 9 files changed, 56 insertions(+), 20 deletions(-) diff --git a/.golangci.yml b/.golangci.yml index d66521a5..13aa7552 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -78,6 +78,17 @@ linters: - third_party$ - builtin$ - examples$ + rules: + # test code is held to looser style rules: table-driven tests naturally repeat + # literal fixture values, use wide data rows, and keep fixed-argument helpers for + # readability, which the style linters below would otherwise flag. + - path: _test\.go + linters: + - goconst + - gocyclo + - lll + - unparam + - dupl formatters: enable: - gofmt diff --git a/pkg/accesslog/collector/collector.go b/pkg/accesslog/collector/collector.go index 5cf4d2be..04132648 100644 --- a/pkg/accesslog/collector/collector.go +++ b/pkg/accesslog/collector/collector.go @@ -19,12 +19,9 @@ package collector import ( "github.com/apache/skywalking-rover/pkg/accesslog/common" - "github.com/apache/skywalking-rover/pkg/logger" "github.com/apache/skywalking-rover/pkg/module" ) -var log = logger.GetLogger("accesslog", "collector") - type Collector interface { Start(mgr *module.Manager, context *common.AccessLogContext) error Stop() diff --git a/pkg/accesslog/collector/ztunnel_accesslog.go b/pkg/accesslog/collector/ztunnel_accesslog.go index 9e8e3c22..9febdccf 100644 --- a/pkg/accesslog/collector/ztunnel_accesslog.go +++ b/pkg/accesslog/collector/ztunnel_accesslog.go @@ -48,10 +48,19 @@ var ( ZTunnelAccessLogPollInterval = time.Second ) +const ( + // the two access log messages ztunnel emits per proxied outbound connection(see the + // ZTunnelAccessLogPodsGlob doc): "connection complete" at INFO, "connection opened" at DEBUG + msgConnectionComplete = "connection complete" + msgConnectionOpened = "connection opened" +) + // startAccessLogTailer starts a background goroutine that tails the local ztunnel access log // and feeds the (downstream src -> real pod) mappings into the same ipMappingCache the uprobe // fills, keyed by the source address alone(like the ConnectionResult::new source). It is a // best-effort fallback: if the log file is absent(no mount / logging disabled) it simply idles. +// +//nolint:gocyclo // linear tail state machine(rotation/truncation/EOF); splitting obscures the flow func (z *ZTunnelCollector) startAccessLogTailer() { if z.accessLogTailerStarted { return @@ -131,7 +140,7 @@ func (z *ZTunnelCollector) startAccessLogTailer() { } for { line, err := reader.ReadString('\n') - if len(line) > 0 { + if line != "" { offset += int64(len(line)) if strings.HasSuffix(line, "\n") { z.handleAccessLogLine(line) @@ -167,6 +176,8 @@ func newestMatch(glob string) string { // handleAccessLogLine parses one CRI log line and, if it is an outbound ztunnel access log // event, feeds its (src -> real pod) mapping into the cache keyed by the source address. +// +//nolint:gocyclo // linear parser over 2 on-disk + 2 payload formats; helpers wouldn't cut complexity func (z *ZTunnelCollector) handleAccessLogLine(line string) { // The kubelet writes container logs in one of two on-disk formats depending on the runtime: // - CRI(containerd / CRI-O / cri-dockerd): " ", where a @@ -205,7 +216,7 @@ func (z *ZTunnelCollector) handleAccessLogLine(line string) { var srcAddr, podAddr, direction, message string if strings.HasPrefix(strings.TrimSpace(payload), "{") { - // LOG_FORMAT=json + // ztunnel is configured with LOG_FORMAT=json: the payload is a JSON object var m map[string]interface{} if json.Unmarshal([]byte(payload), &m) != nil { return @@ -228,10 +239,10 @@ func (z *ZTunnelCollector) handleAccessLogLine(line string) { podAddr = extractLogField(payload, "dst.addr=") } direction = strings.Trim(extractLogField(payload, "direction="), "\"") - if strings.Contains(payload, "connection complete") { - message = "connection complete" - } else if strings.Contains(payload, "connection opened") { - message = "connection opened" + if strings.Contains(payload, msgConnectionComplete) { + message = msgConnectionComplete + } else if strings.Contains(payload, msgConnectionOpened) { + message = msgConnectionOpened } } @@ -239,7 +250,7 @@ func (z *ZTunnelCollector) handleAccessLogLine(line string) { if direction != "outbound" || srcAddr == "" || podAddr == "" { return } - if message != "connection complete" && message != "connection opened" { + if message != msgConnectionComplete && message != msgConnectionOpened { return } srcIP, sp, err := parseZTunnelAddress(srcAddr) diff --git a/pkg/accesslog/collector/ztunnel_admin.go b/pkg/accesslog/collector/ztunnel_admin.go index e05b5bc2..f23179f9 100644 --- a/pkg/accesslog/collector/ztunnel_admin.go +++ b/pkg/accesslog/collector/ztunnel_admin.go @@ -251,7 +251,7 @@ func (z *ZTunnelCollector) httpGetInZTunnelNetNS(rawURL string) ([]byte, error) return body, nil } -func parseZTunnelAddress(addr string) (string, int, error) { +func parseZTunnelAddress(addr string) (ip string, port int, err error) { if addr == "" { return "", 0, fmt.Errorf("empty address") } @@ -259,7 +259,7 @@ func parseZTunnelAddress(addr string) (string, int, error) { if err != nil { return "", 0, err } - port, err := strconv.Atoi(portStr) + port, err = strconv.Atoi(portStr) if err != nil { return "", 0, err } diff --git a/pkg/accesslog/common/connection.go b/pkg/accesslog/common/connection.go index e4019fe6..decfce90 100644 --- a/pkg/accesslog/common/connection.go +++ b/pkg/accesslog/common/connection.go @@ -552,19 +552,26 @@ func (c *ConnectionManager) startUnresolvedRemoteReporter(ctx context.Context) { if unresolvedCount == lastUnresolvedCount { continue } + intervalUnresolved := unresolvedCount - lastUnresolvedCount lastUnresolvedCount = unresolvedCount conntrackStats := "not available" if c.connectTracker != nil { conntrackStats = c.connectTracker.StatsString() } topPairs, byReason := c.drainUnresolvedRemotes() - log.Infof("remote address resolve summary: total remote addresses built: %d, "+ - "resolved by conntrack: %d, resolved by ztunnel correlation: %d, unresolved(sent as raw IP): %d, "+ - "conntrack stats: {%s}, unresolved by reason: {%s}, "+ - "open->resolution latency buckets {<1s: %d, 1-5s: %d, 5-15s: %d, >=15s: %d, max: %dms}, "+ + // the built/resolved/unresolved totals and the latency buckets are cumulative since + // start; the by-reason breakdown and the top socket pairs are drained and reset every + // report. To keep the line internally consistent the unresolved count is shown as the + // per-interval delta(which matches the by-reason breakdown's window) next to the + // cumulative total, and each field states which window it describes. + log.Infof("remote address resolve summary: total remote addresses built(cumulative): %d, "+ + "resolved by conntrack(cumulative): %d, resolved by ztunnel correlation(cumulative): %d, "+ + "unresolved(sent as raw IP) since last report: %d(cumulative total: %d), "+ + "conntrack stats(cumulative): {%s}, unresolved by reason since last report: {%s}, "+ + "open->resolution latency buckets(cumulative) {<1s: %d, 1-5s: %d, 5-15s: %d, >=15s: %d, max: %dms}, "+ "top unresolved socket pairs since last report: %s", c.remoteAddressBuildCount.Load(), c.conntrackResolvedRemoteCount.Load(), - c.ztunnelResolvedRemoteCount.Load(), unresolvedCount, + c.ztunnelResolvedRemoteCount.Load(), intervalUnresolved, unresolvedCount, conntrackStats, byReason, c.resolutionLatencyUnder1s.Load(), c.resolutionLatency1to5s.Load(), c.resolutionLatency5to15s.Load(), c.resolutionLatencyOver15s.Load(), diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index 176d0fb9..c501bac3 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -74,7 +74,7 @@ func (l *Logger) apply() { } func (l *Logger) Enable(level logrus.Level) bool { - return l.Entry.Logger.IsLevelEnabled(level) + return l.Logger.IsLevelEnabled(level) } // matchDebugModule reports whether the module(or one of its parent modules) is diff --git a/pkg/logger/logger_test.go b/pkg/logger/logger_test.go index 2fc9e166..b14a270b 100644 --- a/pkg/logger/logger_test.go +++ b/pkg/logger/logger_test.go @@ -80,8 +80,8 @@ func TestGlobalDebugLevelEmitsAllModules(t *testing.T) { // with no allowlist but a global debug level, every module logs at debug. base, debug := applyConfig(t, "debug", "") - any := GetLogger("access_log", "collector", "connection") - any.Debugf("global-debug") + anyMod := GetLogger("access_log", "collector", "connection") + anyMod.Debugf("global-debug") if !emitted(base, debug, "global-debug") { t.Errorf("expected debug logs when the global level is debug") } diff --git a/pkg/process/finders/kubernetes/template.go b/pkg/process/finders/kubernetes/template.go index 9d9ca7cd..617f82b9 100644 --- a/pkg/process/finders/kubernetes/template.go +++ b/pkg/process/finders/kubernetes/template.go @@ -155,6 +155,7 @@ func (t *TemplatePodJudgment) HasContainer(name string) bool { // so a sidecar-injected pod carries the sidecar in InitContainers rather than Containers. Check // both, otherwise the "exclude pods that have " filter silently fails to match a // native-sidecar pod and the pod(its app process) gets monitored anyway. + // nolint for _, c := range t.pc.Pod.Spec.InitContainers { if c.Name == name { return true diff --git a/pkg/tools/netns/netns.go b/pkg/tools/netns/netns.go index b6eac8b4..45dfd841 100644 --- a/pkg/tools/netns/netns.go +++ b/pkg/tools/netns/netns.go @@ -32,6 +32,15 @@ import ( // Requires CAP_SYS_ADMIN. Everything that must happen inside the target // namespace(creating sockets, dialing, reading responses) should be done // inside fn, since only the calling OS thread is switched. +// +// Failure contract: if switching BACK to the original namespace fails, RunInNetNS +// returns the error but deliberately does NOT unlock the OS thread - it leaves the +// goroutine locked to that thread so the Go runtime destroys the thread(instead of +// reusing it for other goroutines) while it is still in the target namespace. The +// caller MUST therefore treat a restore error as terminal for the goroutine and +// return/exit promptly, doing no further work that could run in the wrong namespace. +// Run RunInNetNS from a dedicated, short-lived goroutine so a poisoned thread is +// discarded when that goroutine exits. func RunInNetNS(netnsPath string, fn func() error) error { target, err := os.Open(netnsPath) if err != nil {