fix(agent): fix AF_PACKET wake hook so scale-to-zero actually wakes

.github/workflows/ci.yml

@@ -5,6 +5,7 @@ on:
     branches: [main]
   pull_request:
     branches: [main]
+  workflow_dispatch:
 
 env:
   GO_VERSION: "1.26"
@@ -124,27 +125,82 @@ jobs:
         with:
           files: coverage.out
 
-  e2e:
-    name: E2E (Talos)
-    if: vars.E2E_RUNNER_LABEL != ''
-    runs-on: ${{ vars.E2E_RUNNER_LABEL }}
-    needs: test
+  e2e-datapath:
+    name: E2E Datapath (Kind, real agent)
+    # Boots a real Firecracker agent via nested KVM — slower and less proven than
+    # the Kind smoke job, so it's manual-dispatch only rather than on every push.
+    # See docs (e2e runner runbook) for why: proven on ubuntu-latest (run 29019535180),
+    # but the datapath spec itself has never executed end to end yet.
+    if: github.event_name == 'workflow_dispatch'
+    runs-on: ubuntu-latest
+    needs: [lint, build]
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
 
-      - name: Create Talos E2E cluster
-        run: talosctl cluster create --provisioner docker --name imp-e2e
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: ${{ env.GO_VERSION }}
 
-      - name: Apply CRDs
-        run: kubectl apply -f config/crd/bases/
+      - name: Allow unprivileged access to /dev/kvm
+        run: |
+          echo 'KERNEL=="kvm", GROUP="kvm", MODE="0666"' | sudo tee /etc/udev/rules.d/99-kvm4all.rules
+          sudo udevadm control --reload-rules
+          sudo udevadm trigger --name-match=kvm
+          ls -l /dev/kvm
 
-      - name: Run E2E suite
-        run: go test -v -tags e2e ./test/e2e/...
+      - name: Create Kind cluster
+        uses: helm/kind-action@v1
+        with:
+          cluster_name: imp-e2e-datapath
+          config: test/e2e/kind-datapath.yaml
+
+      - name: Install Helm CLI
+        uses: azure/setup-helm@v4
 
-      - name: Destroy Talos E2E cluster
-        if: always()
-        run: talosctl cluster destroy --name imp-e2e
+      - name: Build local E2E images
+        run: |
+          docker build -f Dockerfile.operator -t local/imp-operator:e2e .
+          docker build -f Dockerfile.agent -t local/imp-agent:e2e .
+
+      - name: Load local E2E images into Kind
+        run: |
+          kind load docker-image local/imp-operator:e2e --name imp-e2e-datapath
+          kind load docker-image local/imp-agent:e2e --name imp-e2e-datapath
+
+      - name: Stage Firecracker binary + guest kernel on the Kind node
+        # No imp-specific guest kernel exists yet (planned "imp-guest-kernel" repo,
+        # not built) — every UNVALIDATED-marked path in this repo uses the public
+        # Firecracker quickstart kernel as a stopgap; same here.
+        run: |
+          node=$(kind get nodes --name imp-e2e-datapath | head -1)
+          fc_ver=v1.15.0
+          curl -fsSL "https://github.com/firecracker-microvm/firecracker/releases/download/${fc_ver}/firecracker-${fc_ver}-x86_64.tgz" -o /tmp/firecracker.tgz
+          tar -xzf /tmp/firecracker.tgz -C /tmp
+          docker cp "/tmp/release-${fc_ver}-x86_64/firecracker-${fc_ver}-x86_64" "${node}:/usr/local/bin/firecracker"
+          docker exec "${node}" chmod +x /usr/local/bin/firecracker
+          curl -fsSL "https://s3.amazonaws.com/spec.ccfc.min/img/quickstart_guide/x86_64/kernels/vmlinux.bin" -o /tmp/vmlinux
+          docker exec "${node}" mkdir -p /var/lib/imp
+          docker cp /tmp/vmlinux "${node}:/var/lib/imp/vmlinux"
+
+      - name: Run datapath E2E suite
+        run: |
+          IMP_E2E_REAL_AGENT=true go test -tags e2e ./test/e2e/... -v -timeout 30m -ginkgo.label-filter="datapath"
+
+      - name: Dump cluster diagnostics on failure
+        if: failure()
+        run: |
+          kubectl get pods -A -o wide || true
+          kubectl get impvm,impnetwork,impvmclass -A -o wide || true
+          kubectl describe pods -n imp-system || true
+          kubectl describe impvm -n default || true
+          echo "--- operator logs ---"
+          kubectl logs -n imp-system -l app.kubernetes.io/component=operator --tail=200 || true
+          echo "--- agent logs ---"
+          kubectl logs -n imp-system -l app.kubernetes.io/component=agent --tail=200 || true
+          echo "--- events ---"
+          kubectl get events -A --sort-by=.lastTimestamp || true
 
   e2e-kind:
     name: E2E Smoke (Kind)

charts/imp/templates/agent/daemonset.yaml

@@ -68,6 +68,9 @@ spec:
         - name: firecracker-bin
           mountPath: {{ .Values.agent.env.fcBinPath | quote }}
           readOnly: true
+        - name: guest-kernel
+          mountPath: {{ .Values.agent.env.kernelPath | quote }}
+          readOnly: true
         - name: socket-dir
           mountPath: /run/imp/sockets
         - name: image-cache
@@ -81,6 +84,10 @@ spec:
         hostPath:
           path: {{ .Values.agent.env.fcBinPath | quote }}
           type: File
+      - name: guest-kernel
+        hostPath:
+          path: {{ .Values.agent.env.kernelPath | quote }}
+          type: File
       - name: socket-dir
         {{- if .Values.agent.hostPaths.socketDir.enabled }}
         hostPath:

charts/imp/tests/agent-daemonset_test.yaml

@@ -50,6 +50,25 @@ tests:
               path: /usr/local/bin/firecracker
               type: File
 
+  - it: mounts the guest kernel hostPath at kernelPath
+    template: templates/agent/daemonset.yaml
+    set:
+      agent.env.kernelPath: /var/lib/imp/vmlinux
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: guest-kernel
+            mountPath: /var/lib/imp/vmlinux
+            readOnly: true
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: guest-kernel
+            hostPath:
+              path: /var/lib/imp/vmlinux
+              type: File
+
   - it: uses emptyDir for socketDir when hostPaths disabled
     template: templates/agent/daemonset.yaml
     set:

internal/agent/network/vxlan.go

@@ -95,13 +95,24 @@ func (m *LinuxNetManager) SyncFDB(_ context.Context, ifaceName string, entries [
 		return fmt.Errorf("list FDB entries for %s: %w", ifaceName, err)
 	}
 
-	// Remove stale entries — skip the all-zeros broadcast entry.
+	// Remove stale entries — skip the all-zeros broadcast entry, and skip any
+	// entry with no resolvable destination IP. When this VXLAN interface is
+	// attached to a bridge, the kernel auto-creates a "self permanent" FDB
+	// entry for the port's own MAC with no tunnel destination; NeighDel on
+	// that entry serializes an empty NDA_DST attribute, which the kernel's
+	// VXLAN driver rejects with EAFNOSUPPORT ("address family not supported by
+	// protocol") — deterministically, every time, since it's not something
+	// SyncFDB itself ever added. Confirmed via CI: this deletion failure
+	// aborted the whole sync before any real VTEP entry was ever added.
 	allZeros := "00:00:00:00:00:00"
 	for _, n := range current {
 		mac := n.HardwareAddr.String()
 		if mac == allZeros {
 			continue
 		}
+		if n.IP.To4() == nil && n.IP.To16() == nil {
+			continue
+		}
 		if _, ok := desired[mac]; !ok {
 			del := &netlink.Neigh{
 				LinkIndex:    link.Attrs().Index,

internal/agent/scaletozero.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1"
 )
@@ -76,25 +77,39 @@ func (w *wakeRegistry) register(ip string, vm client.Object) {
 	w.keyByIP[ip] = key
 	w.ipByKey[key] = ip
 	w.objByKey[key] = vm
+	logf.Log.Info("wake registry: VM registered for wake-on-traffic", "vm", key, "ip", ip)
 }
 
 // onDstIP is the PacketSource callback: a frame arrived for ip. If ip belongs to
 // a registered VM not already signalled, enqueue a reconcile for it. The
 // signalled flag is set only when the event is actually enqueued, so a full
 // channel never silently loses a wake — the next packet retries.
+//
+// Diagnostic logging: this is the one hop the AF_PACKET PacketSource is
+// UNVALIDATED to reach (see scaletozero_linux.go), so every branch here is
+// logged at Info to make a validation run's outcome legible from cluster logs
+// without needing a debugger. Not gated behind a verbosity flag because a
+// suspended VM should see near-zero unmatched traffic — only VMs actually
+// awaiting wake generate log volume here.
 func (w *wakeRegistry) onDstIP(ip string) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	key, ok := w.keyByIP[ip]
-	if !ok || w.signalled[key] {
+	if !ok {
+		return
+	}
+	if w.signalled[key] {
+		logf.Log.V(1).Info("wake registry: frame matched an already-signalled VM", "vm", key, "ip", ip)
 		return
 	}
 	obj := w.objByKey[key]
 	select {
 	case w.events <- event.GenericEvent{Object: obj}:
 		w.signalled[key] = true
+		logf.Log.Info("wake registry: frame matched, wake reconcile enqueued", "vm", key, "ip", ip)
 	default:
 		// Channel full; leave unsignalled so a later packet retries.
+		logf.Log.Info("wake registry: frame matched but event channel full, will retry", "vm", key, "ip", ip)
 	}
 }
 

internal/agent/scaletozero_linux.go

@@ -10,6 +10,7 @@ import (
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
 	"k8s.io/apimachinery/pkg/types"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 // resetIdle forgets any idle sample for key (called on suspend/resume so the VM
@@ -52,24 +53,50 @@ func netlinkLinkStats(iface string) (uint64, error) {
 // raw socket (unbound, so it sees every overlay bridge) and reports each frame's
 // destination IP. One socket serves all suspended VMs on the node.
 //
-// UNVALIDATED (see scaletozero.go): not yet confirmed to observe the first frame
-// destined to a TAP-less suspended VM. Swap for a tc-BPF PacketSource if the
-// cluster spike shows the frame is dropped before this hook.
+// VALIDATED-then-fixed (2026-07-29 real-KVM CI run, see scaletozero.go): the
+// first cluster spike proved this hook never observed overlay traffic when
+// opened with a specific protocol (ETH_P_IP). Root cause: passing a specific
+// EtherType at socket() time registers the handler in the kernel's per-protocol
+// ptype_base hash, which __netif_receive_skb_core only invokes if the receiving
+// device's rx_handler returns RX_HANDLER_PASS. A bridge port's rx_handler
+// (br_handle_frame) returns RX_HANDLER_CONSUMED for every switched/flooded
+// frame — i.e. every ordinary VM-to-VM frame on an impbr-* bridge — so
+// ptype_base delivery never happens for exactly the traffic this hook exists
+// to see. Only ptype_all (registered by an unbound/protocol-0 socket, same as
+// tcpdump's default) is invoked before rx_handler and sees the frame
+// regardless of what the bridge later does with it. Hence: bind with
+// ETH_P_ALL and filter IPv4 in userspace instead of filtering at socket()
+// time. Do not "simplify" this back to ETH_P_IP — it silently stops seeing
+// all bridged/switched traffic while still working for non-bridged devices
+// (lo, an unbridged pod veth), which is exactly what made the original bug
+// hard to notice.
 type afpacketSource struct{}
 
 func htons(v uint16) uint16 { return v<<8 | v>>8 }
 
+const ethTypeIPv4 = 0x0800
+
 func (afpacketSource) Run(ctx context.Context, onDstIP func(string)) error {
-	fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons(unix.ETH_P_IP)))
+	fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, int(htons(unix.ETH_P_ALL)))
 	if err != nil {
+		logf.Log.Error(err, "afpacketSource: failed to open AF_PACKET socket")
 		return err
 	}
+	logf.Log.Info("afpacketSource: AF_PACKET socket open, capturing inbound IPv4 frames")
 	// Unblock the blocking Recvfrom and release the fd when the manager stops.
 	go func() {
 		<-ctx.Done()
 		_ = unix.Close(fd)
 	}()
 
+	// Diagnostic-only liveness counter (see scaletozero.go: this hook is
+	// UNVALIDATED). Proves whether the socket sees ANY IPv4 traffic at all,
+	// independent of whether it matches a registered wake IP — onDstIP() only
+	// logs on a match, so a validation run with zero matches is otherwise
+	// indistinguishable from a socket that receives nothing.
+	var frameCount uint64
+	lastLog := time.Now()
+
 	buf := make([]byte, 65536)
 	for {
 		n, _, err := unix.Recvfrom(fd, buf, 0)
@@ -80,11 +107,22 @@ func (afpacketSource) Run(ctx context.Context, onDstIP func(string)) error {
 			time.Sleep(10 * time.Millisecond) // avoid a tight spin on a persistent recv error
 			continue
 		}
-		// AF_PACKET/SOCK_RAW frames include the 14-byte Ethernet header; the IPv4
-		// destination address sits at bytes 30..34 (eth[14] + ipv4[16..20]).
+		// AF_PACKET/SOCK_RAW frames include the 14-byte Ethernet header. With
+		// ETH_P_ALL the socket now also receives ARP, IPv6, STP, LLDP, etc., so
+		// filter to IPv4 in userspace via the EtherType at bytes 12..14 before
+		// treating bytes 30..34 (eth[14] + ipv4[16..20]) as an IPv4 dest addr.
 		if n < 34 {
 			continue
 		}
+		if ethType := uint16(buf[12])<<8 | uint16(buf[13]); ethType != ethTypeIPv4 {
+			continue
+		}
+		frameCount++
+		if frameCount == 1 || time.Since(lastLog) >= 10*time.Second {
+			logf.Log.Info("afpacketSource: capturing IPv4 traffic", "framesSeen", frameCount,
+				"lastDstIP", net.IP(buf[30:34]).String())
+			lastLog = time.Now()
+		}
 		onDstIP(net.IP(buf[30:34]).String())
 	}
 }

internal/capability/probe.go

@@ -29,9 +29,10 @@ func (r Result) OK() bool {
 func Check(kvmPath, binPath string) Result {
 	var r Result
 
-	if _, err := os.Stat(kvmPath); err != nil {
+	if f, err := os.OpenFile(kvmPath, os.O_RDWR, 0); err != nil { //nolint:gosec // G304: kvmPath is an operator-supplied device path, not user input
 		r.KVMError = fmt.Sprintf("%s not available: %v", kvmPath, err)
 	} else {
+		_ = f.Close()
 		r.KVMAvailable = true
 	}
 

internal/capability/probe_test.go

@@ -53,6 +53,25 @@ func TestCheck_MissingDevice(t *testing.T) {
 	}
 }
 
+func TestCheck_UnopenableDevice(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("running as root: permission bits don't block access")
+	}
+	dir := t.TempDir()
+	kvmPath := filepath.Join(dir, "kvm")
+	binPath := filepath.Join(dir, "firecracker")
+	if err := os.WriteFile(kvmPath, nil, 0o000); err != nil {
+		t.Fatalf("write unopenable kvm device stub: %v", err)
+	}
+	writeExecutable(t, binPath)
+
+	got := Check(kvmPath, binPath)
+
+	if got.KVMAvailable || got.KVMError == "" {
+		t.Errorf("expected KVM unavailable with an error, got %+v", got)
+	}
+}
+
 func TestCheck_MissingBinary(t *testing.T) {
 	dir := t.TempDir()
 	kvmPath := filepath.Join(dir, "kvm")