From 61501fc22e31248a66a3ced4f594b5581f329933 Mon Sep 17 00:00:00 2001 From: Giovanni Ferri Date: Thu, 24 Sep 2026 14:47:55 +0100 Subject: [PATCH] fix(agent): capture ScaleToZero wake frames in the runtime host netns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wake-on-traffic hook opened its AF_PACKET socket inside the agent pod. The impbr-* bridges and VM TAPs live in the host network namespace (owned by the runtime DaemonSet, hostNetwork: true), so the pod-side socket only ever saw pod traffic (loopback, pod veths, Service IPs) — never the overlay frames a suspended ScaleToZero VM must be woken by. The nightly datapath E2E failure streak since 2026-08-29 shows it directly: framesSeen increments while lastDstIP is always 10.244.x/10.96.x/127.0.0.1, never the target VM's 10.45.0.x. Mirror the PR #99 pattern used for TAP stats: capture on the runtime (host netns) and expose it over the existing runtime RPC. - runtimeapi: WakeHits backend method + RPC (drains observed destination IPs) - noderuntime: host-netns AF_PACKET capture (ETH_P_ALL — preserved the ptype_all/br_handle_frame analysis; do not simplify to ETH_P_IP) - agent: NewLinuxScaleToZeroWithRuntime now wires a runtimeWakeSource that polls WakeHits at interval/2 instead of opening a pod-side socket --- internal/agent/scaletozero_linux.go | 50 +++++++- .../scaletozero_wake_runtime_linux_test.go | 93 ++++++++++++++ internal/noderuntime/backend.go | 6 + internal/noderuntime/wakehits_linux.go | 121 ++++++++++++++++++ internal/noderuntime/wakehits_other.go | 14 ++ internal/runtimeapi/runtime.go | 26 ++++ 6 files changed, 307 insertions(+), 3 deletions(-) create mode 100644 internal/agent/scaletozero_wake_runtime_linux_test.go create mode 100644 internal/noderuntime/wakehits_linux.go create mode 100644 internal/noderuntime/wakehits_other.go diff --git a/internal/agent/scaletozero_linux.go b/internal/agent/scaletozero_linux.go index d2eaa6f..412483b 100644 --- a/internal/agent/scaletozero_linux.go +++ b/internal/agent/scaletozero_linux.go @@ -134,10 +134,54 @@ func NewLinuxScaleToZero(bufSize int, interval time.Duration) *ScaleToZero { return newScaleToZero(netlinkLinkStats, afpacketSource{}, interval, bufSize) } -// NewLinuxScaleToZeroWithRuntime wires TAP statistics through the node runtime, -// which owns the TAP network namespace, while retaining the packet capture hook. +// NewLinuxScaleToZeroWithRuntime wires TAP statistics AND wake-on-traffic +// packet capture through the node runtime, which owns the host network +// namespace where the impbr-* bridges and VM TAPs live. The agent's own +// namespace cannot observe bridged overlay traffic: an AF_PACKET socket opened +// inside the agent pod only ever sees pod-side frames (loopback, pod veths, +// Service IPs), never the overlay frames a suspended VM needs to be woken by +// (validated by the nightly datapath E2E failure streak starting 2026-08-29, +// where framesSeen incremented but lastDstIP was always pod traffic). The +// activator polls the runtime at interval/2 so a wake frame registers well +// inside a 2-minute resume window. func NewLinuxScaleToZeroWithRuntime(client *runtimeapi.Client, bufSize int, interval time.Duration) *ScaleToZero { + poll := interval / 2 + if poll <= 0 { + poll = time.Second + } return newScaleToZero(func(iface string) (uint64, error) { return client.LinkStats(context.Background(), iface) - }, afpacketSource{}, interval, bufSize) + }, runtimeWakeSource{client: client, poll: poll}, interval, bufSize) +} + +// runtimeWakeSource polls the node runtime for destination IPs observed on the +// host since the previous poll and feeds each to the wake registry. +type runtimeWakeSource struct { + client *runtimeapi.Client + poll time.Duration +} + +func (r runtimeWakeSource) Run(ctx context.Context, onDstIP func(string)) error { + logf.Log.Info("runtime wake source: polling host wake hits via runtime RPC", + "pollInterval", r.poll.String()) + ticker := time.NewTicker(r.poll) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + } + ips, err := r.client.WakeHits(ctx) + if err != nil { + if ctx.Err() != nil { + return nil + } + logf.Log.Error(err, "runtime wake source: poll failed; retrying next interval") + continue + } + for _, ip := range ips { + onDstIP(ip) + } + } } diff --git a/internal/agent/scaletozero_wake_runtime_linux_test.go b/internal/agent/scaletozero_wake_runtime_linux_test.go new file mode 100644 index 0000000..b3536a1 --- /dev/null +++ b/internal/agent/scaletozero_wake_runtime_linux_test.go @@ -0,0 +1,93 @@ +//go:build linux + +package agent + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1" + "github.com/syscode-labs/imp/internal/runtimeapi" +) + +func TestScaleToZeroRuntimeWakeSourceWakesOnRuntimeObservedFrame(t *testing.T) { + t.Parallel() + + const overlayIP = "10.45.0.3" + hits := [][]string{ + {"10.244.0.8", "10.96.0.1"}, // poll 1: pod traffic only — no match + {overlayIP, "127.0.0.1", overlayIP}, // poll 2: the wake frame arrives + } + poll := 0 + server := runtimeapi.NewServer(runtimeapi.BackendFuncs{ + LinkStatsFunc: func(_ context.Context, _ string) (uint64, error) { + return 42, nil // constant: observe() never reports idle + }, + WakeHitsFunc: func(_ context.Context) ([]string, error) { + if poll >= len(hits) { + return nil, nil + } + got := hits[poll] + poll++ + return got, nil + }, + }) + dir, err := os.MkdirTemp("/tmp", "imp-runtime-") + if err != nil { + t.Fatalf("MkdirTemp() error = %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + endpoint := filepath.Join(dir, "runtime.sock") + if err := server.Start(endpoint); err != nil { + t.Fatalf("Start() error = %v", err) + } + t.Cleanup(func() { _ = server.Close() }) + + sz := NewLinuxScaleToZeroWithRuntime(runtimeapi.NewClient(endpoint), 1, time.Second) + key := types.NamespacedName{Namespace: "default", Name: "target"} + + // Register the VM as suspended+awaiting wake, exactly as the reconciler + // does when it parks a ScaleToZero VM. + sz.reg.register(overlayIP, &impdevv1alpha1.ImpVM{ + ObjectMeta: metav1.ObjectMeta{Namespace: key.Namespace, Name: key.Name}, + }) + + // Feed the first poll through the wake registry directly: pod traffic + // must not signal the suspended VM. + _, _, err = sz.observe(key, "imptap-x", time.Minute, time.Now()) + if err != nil { + t.Fatalf("observe() error = %v", err) + } + // Drain hits via the same path the activator uses. + src := runtimeWakeSource{client: runtimeapi.NewClient(endpoint), poll: 50 * time.Millisecond} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + _ = src.Run(ctx, sz.reg.onDstIP) + close(done) + }() + + // Wait until the second poll (which contains the overlay IP) has been + // consumed — the wake registry must have signalled the VM. + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + sz.reg.mu.Lock() + signalled := sz.reg.signalled[key] + sz.reg.mu.Unlock() + if signalled { + cancel() + <-done + return + } + time.Sleep(20 * time.Millisecond) + } + cancel() + <-done + t.Fatal("wake registry never signalled the VM despite the runtime observing a frame for its overlay IP") +} diff --git a/internal/noderuntime/backend.go b/internal/noderuntime/backend.go index f89cb93..4579628 100644 --- a/internal/noderuntime/backend.go +++ b/internal/noderuntime/backend.go @@ -29,6 +29,12 @@ type Backend struct { Net network.NetManager // LinkStatsFunc is an optional test seam; production uses runtime-owned netlink. LinkStatsFunc func(context.Context, string) (uint64, error) + // WakeHitsFunc is an optional test seam; production uses a host-netns + // AF_PACKET capture started lazily on first poll. + WakeHitsFunc func(context.Context) ([]string, error) + // wakeHitsState holds platform-specific capture state (linux only; nil + // elsewhere). Managed entirely by the platform WakeHits implementation. + wakeHitsState any // StatePath stores VM inventory that survives a runtime process restart. StatePath string } diff --git a/internal/noderuntime/wakehits_linux.go b/internal/noderuntime/wakehits_linux.go new file mode 100644 index 0000000..341f0d5 --- /dev/null +++ b/internal/noderuntime/wakehits_linux.go @@ -0,0 +1,121 @@ +//go:build linux + +package noderuntime + +import ( + "context" + "net" + "sync" + "time" + + "golang.org/x/sys/unix" + logf "sigs.k8s.io/controller-runtime/pkg/log" +) + +// wakeHits records destination IPs of inbound IPv4 frames observed in the +// runtime's network namespace (the host netns, where the impbr-* bridges and +// VM TAPs live). The agent polls WakeHits and wakes any suspended ScaleToZero +// VM whose overlay IP appears in the drained set. +type wakeHits struct { + mu sync.Mutex + hits map[string]bool +} + +func (w *wakeHits) add(ip string) { + w.mu.Lock() + defer w.mu.Unlock() + w.hits[ip] = true +} + +func (w *wakeHits) take() []string { + w.mu.Lock() + defer w.mu.Unlock() + ips := make([]string, 0, len(w.hits)) + for ip := range w.hits { + ips = append(ips, ip) + } + w.hits = map[string]bool{} + return ips +} + +func htons16(v uint16) uint16 { return v<<8 | v>>8 } + +// startHostWakeCapture opens one AF_PACKET socket bound to ETH_P_ALL in the +// runtime's network namespace and records each inbound IPv4 frame's +// destination IP into hits until ctx is cancelled. +// +// The socket MUST use ETH_P_ALL rather than ETH_P_IP: a specific EtherType +// registers in the kernel's ptype_base hash, which a bridged device's +// rx_handler (br_handle_frame, RX_HANDLER_CONSUMED) never reaches. Only +// ptype_all (protocol 0) is consulted before the rx_handler, so it is the only +// way to see frames switched onto an impbr-* bridge. IPv4 filtering therefore +// happens in userspace. Do not "simplify" this back to ETH_P_IP — it silently +// stops seeing all bridged VM traffic (see the longer analysis in +// internal/agent/scaletozero_linux.go and imp#35). +func startHostWakeCapture(ctx context.Context, hits *wakeHits) { + log := logf.Log.WithName("wake-capture") + fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons16(unix.ETH_P_ALL))) + if err != nil { + log.Error(err, "wake-capture: failed to open AF_PACKET socket in host netns; wake-on-traffic is unavailable on this node") + return + } + log.Info("wake-capture: AF_PACKET socket open in host netns, capturing inbound IPv4 frames") + go func() { + <-ctx.Done() + _ = unix.Close(fd) + }() + buf := make([]byte, 65536) + go func() { + for { + n, _, err := unix.Recvfrom(fd, buf, 0) + if err != nil { + if ctx.Err() != nil { + return + } + time.Sleep(10 * time.Millisecond) + continue + } + // Frames include the 14-byte Ethernet header; filter to IPv4 via + // the EtherType at bytes 12..14 and read the destination IPv4 + // address at bytes 30..34 (eth[14] + ipv4[16..20]). + if n < 34 { + continue + } + if ethType := uint16(buf[12])<<8 | uint16(buf[13]); ethType != 0x0800 { + continue + } + hits.add(net.IP(buf[30:34]).String()) + } + }() +} + +// backendWakeState holds the lazy capture state that only exists on Linux. +type backendWakeState struct { + hits *wakeHits + cancel context.CancelFunc +} + +// wakeStateOf lazily initialises b's wake capture on first poll. +func wakeStateOf(b *Backend) *backendWakeState { + if state, ok := b.wakeHitsState.(*backendWakeState); ok { + return state + } + state := &backendWakeState{hits: &wakeHits{hits: map[string]bool{}}} + // The capture intentionally lives for the whole runtime process; it is + // never cancelled short of process exit, so the cancel func stays stored + // but uncalled. + ctx, cancel := context.WithCancel(context.Background()) //nolint:gosec // intentional process-lifetime capture + state.cancel = cancel + startHostWakeCapture(ctx, state.hits) + b.wakeHitsState = state + return state +} + +// WakeHits drains the destination IPs observed on the host since the previous +// call. Implements runtimeapi.Backend. +func (b *Backend) WakeHits(_ context.Context) ([]string, error) { + if b.WakeHitsFunc != nil { + return b.WakeHitsFunc(context.Background()) + } + return wakeStateOf(b).hits.take(), nil +} diff --git a/internal/noderuntime/wakehits_other.go b/internal/noderuntime/wakehits_other.go new file mode 100644 index 0000000..11d9521 --- /dev/null +++ b/internal/noderuntime/wakehits_other.go @@ -0,0 +1,14 @@ +//go:build !linux + +package noderuntime + +import ( + "context" + "errors" +) + +// WakeHits is unsupported off Linux; the agent falls back to no-op polling. +func (b *Backend) WakeHits(_ context.Context) ([]string, error) { + _ = b.wakeHitsState // linux-only capture state; unused here + return nil, errors.New("runtime wake hits is not supported on this platform") +} diff --git a/internal/runtimeapi/runtime.go b/internal/runtimeapi/runtime.go index 1b83cc7..9dd641b 100644 --- a/internal/runtimeapi/runtime.go +++ b/internal/runtimeapi/runtime.go @@ -49,6 +49,7 @@ type Backend interface { EnsureVXLAN(context.Context, uint32, string, string, string) error SyncFDB(context.Context, string, []network.FDBEntry) error LinkStats(context.Context, string) (uint64, error) + WakeHits(context.Context) ([]string, error) } // BackendFuncs makes protocol behavior testable without Firecracker. @@ -70,6 +71,7 @@ type BackendFuncs struct { EnsureVXLANFunc func(context.Context, uint32, string, string, string) error SyncFDBFunc func(context.Context, string, []network.FDBEntry) error LinkStatsFunc func(context.Context, string) (uint64, error) + WakeHitsFunc func(context.Context) ([]string, error) } func (f BackendFuncs) Start(ctx context.Context, vm *impdevv1alpha1.ImpVM) (int64, error) { @@ -179,6 +181,13 @@ func (f BackendFuncs) LinkStats(ctx context.Context, tapName string) (uint64, er return f.LinkStatsFunc(ctx, tapName) } +func (f BackendFuncs) WakeHits(ctx context.Context) ([]string, error) { + if f.WakeHitsFunc == nil { + return nil, errors.New("wake hits is not supported") + } + return f.WakeHitsFunc(ctx) +} + // Server serves one runtime backend over a node-local Unix socket. type Server struct { backend Backend @@ -252,6 +261,7 @@ type StateReply struct{ State VMState } type SnapshotReply struct{ Result SnapshotResult } type AliveReply struct{ Alive bool } type LinkStatsReply struct{ Bytes uint64 } +type WakeHitsReply struct{ IPs []string } type VSockReply struct { Path string Found bool @@ -358,6 +368,12 @@ func (s *rpcService) LinkStats(args TAPArgs, reply *LinkStatsReply) error { return err } +func (s *rpcService) WakeHits(_ Empty, reply *WakeHitsReply) error { + ips, err := s.backend.WakeHits(context.Background()) + reply.IPs = ips + return err +} + // Client invokes a node runtime through its Unix socket. type Client struct{ endpoint string } @@ -467,3 +483,13 @@ func (c *Client) LinkStats(ctx context.Context, tapName string) (uint64, error) } return reply.Bytes, nil } + +// WakeHits drains the destination IPs of inbound IPv4 frames observed in the +// runtime's host network namespace since the previous call. +func (c *Client) WakeHits(ctx context.Context) ([]string, error) { + var reply WakeHitsReply + if err := c.call(ctx, "WakeHits", Empty{}, &reply); err != nil { + return nil, err + } + return reply.IPs, nil +}