This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-rover.git
The following commit(s) were added to refs/heads/main by this push:
new 9ea70d3 Enhance log module and enhance address resolver in access log
module (#213)
9ea70d3 is described below
commit 9ea70d387712770584a412587782ce462428763d
Author: mrproliu <[email protected]>
AuthorDate: Wed Jul 15 23:01:56 2026 +0800
Enhance log module and enhance address resolver in access log module (#213)
---
.gitignore | 4 +-
.golangci.yml | 11 +
CHANGES.md | 8 +
bpf/accesslog/ambient/ztunnel.c | 58 ++-
bpf/accesslog/syscalls/connect_conntrack.c | 6 +-
configs/rover_configs.yaml | 5 +
pkg/accesslog/collector/collector.go | 3 -
pkg/accesslog/collector/connection.go | 16 +-
pkg/accesslog/collector/ztunnel.go | 528 +++++++++++++++++++--
pkg/accesslog/collector/ztunnel_accesslog.go | 301 ++++++++++++
pkg/accesslog/collector/ztunnel_accesslog_test.go | 130 +++++
pkg/accesslog/collector/ztunnel_admin.go | 296 ++++++++++++
pkg/accesslog/collector/ztunnel_flush_test.go | 170 +++++++
pkg/accesslog/collector/ztunnel_resolution_test.go | 109 +++++
pkg/accesslog/common/connection.go | 347 +++++++++++++-
pkg/accesslog/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 | 10 +
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 | 77 +++
pkg/tools/netns/netns_test.go | 173 +++++++
27 files changed, 2575 insertions(+), 91 deletions(-)
diff --git a/.gitignore b/.gitignore
index 56cd228..8d6f6f5 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/.golangci.yml b/.golangci.yml
index d66521a..13aa755 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/CHANGES.md b/CHANGES.md
index 63f97d4..1f2ca99 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 e1fc44a..ba46060 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 8f1f5cb..7510967 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 a5a6b0c..68ede6d 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/collector.go
b/pkg/accesslog/collector/collector.go
index 5cf4d2b..0413264 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/connection.go
b/pkg/accesslog/collector/connection.go
index 29adfa9..6621db9 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 be5c168..c36a59b 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/<pid>/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
+ // "17h<hash>E" 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 0000000..9febdcc
--- /dev/null
+++ b/pkg/accesslog/collector/ztunnel_accesslog.go
@@ -0,0 +1,301 @@
+// 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
+)
+
+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
+ }
+ 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 line != "" {
+ 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.
+//
+//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): "<rfc3339-ts> <stream>
<F|P> <payload>", 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":"<payload>\n","stream":"...","time":"<ts>"}.
+ // 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), "{") {
+ // 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
+ }
+ 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, msgConnectionComplete) {
+ message = msgConnectionComplete
+ } else if strings.Contains(payload, msgConnectionOpened) {
+ message = msgConnectionOpened
+ }
+ }
+
+ // only the outbound leg carries (app src -> real target pod); inbound
has the reverse
+ if direction != "outbound" || srcAddr == "" || podAddr == "" {
+ return
+ }
+ if message != msgConnectionComplete && message != msgConnectionOpened {
+ 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 0000000..d53666b
--- /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":"<payload>\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 0000000..f23179f
--- /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) (ip string, port int, err 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. <name>_created,
<name>_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 0000000..cfbccc6
--- /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 0000000..a218717
--- /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 b00ffab..decfce9 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,170 @@ 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
+ }
+ intervalUnresolved := unresolvedCount -
lastUnresolvedCount
+ lastUnresolvedCount = unresolvedCount
+ conntrackStats := "not available"
+ if c.connectTracker != nil {
+ conntrackStats =
c.connectTracker.StatsString()
+ }
+ topPairs, byReason := c.drainUnresolvedRemotes()
+ // 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(),
intervalUnresolved, 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 +703,7 @@ func (c *ConnectionManager) buildConnection(event
*events.SocketConnectEvent, so
PID: event.PID,
Socket: socket,
LastCheckExistTime: time.Now(),
+ CreatedAt: time.Now(),
ProtocolBreak: protocolBreak,
}
}
@@ -588,7 +919,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 +940,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 0000000..4218b48
--- /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 5dce05d..6b45a33 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 018c5c4..c501bac 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.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 0000000..b14a270
--- /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", "")
+
+ 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")
+ }
+}
+
+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 7705bb9..59531ba 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 cea08f9..617f82b 100644
--- a/pkg/process/finders/kubernetes/template.go
+++ b/pkg/process/finders/kubernetes/template.go
@@ -151,6 +151,16 @@ 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 <sidecar>" 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
+ }
+ }
return false
}
diff --git a/pkg/tools/btf/linker.go b/pkg/tools/btf/linker.go
index 271a10d..f8aaee9 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 ac02ec1..b376288 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 831d2cc..e0b17b6 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 9f09309..357ec98 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 0000000..45dfd84
--- /dev/null
+++ b/pkg/tools/netns/netns.go
@@ -0,0 +1,77 @@
+// 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/<pid>/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.
+//
+// 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 {
+ 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 0000000..774a5ec
--- /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)
+ }
+}
