diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e14bac..2502074 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -5,6 +5,7 @@ on: branches: [main] pull_request: branches: [main] + workflow_dispatch: env: GO_VERSION: "1.26" @@ -124,28 +125,6 @@ jobs: with: files: coverage.out - e2e: - name: E2E (Talos) - if: vars.E2E_RUNNER_LABEL != '' - runs-on: ${{ vars.E2E_RUNNER_LABEL }} - needs: test - steps: - - name: Checkout repository - uses: actions/checkout@v6 - - - name: Create Talos E2E cluster - run: talosctl cluster create --provisioner docker --name imp-e2e - - - name: Apply CRDs - run: kubectl apply -f config/crd/bases/ - - - name: Run E2E suite - run: go test -v -tags e2e ./test/e2e/... - - - name: Destroy Talos E2E cluster - if: always() - run: talosctl cluster destroy --name imp-e2e - e2e-kind: name: E2E Smoke (Kind) runs-on: ubuntu-latest diff --git a/.github/workflows/nightly-e2e-datapath.yml b/.github/workflows/nightly-e2e-datapath.yml new file mode 100644 index 0000000..8c52eb7 --- /dev/null +++ b/.github/workflows/nightly-e2e-datapath.yml @@ -0,0 +1,130 @@ +name: Nightly E2E Datapath (Kind, real agent) + +on: + schedule: + # 04:00 UTC daily (staggered after the 02:00/03:00 nightly smoke/cilium jobs) + - cron: "0 4 * * *" + pull_request: + branches: [main] + paths: + - "internal/agent/scaletozero*.go" + - "internal/agent/network/**" + - "internal/agent/reconciler.go" + - "internal/capability/**" + - "charts/imp/templates/agent/**" + - "test/e2e/scaletozero_datapath_test.go" + - "test/e2e/kind-datapath.yaml" + - ".github/workflows/nightly-e2e-datapath.yml" + workflow_dispatch: + +concurrency: + group: nightly-e2e-datapath + cancel-in-progress: false + +env: + GO_VERSION: "1.26" + +jobs: + 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 runs nightly + on PRs touching the + # datapath's own fragile paths, not on every push. See docs (e2e runner + # runbook) for provenance: proven on ubuntu-latest (run 29019535180); wake + # hook + FDB sync validated end-to-end in imp#35 (2026-07-29). + runs-on: ubuntu-latest + timeout-minutes: 45 + steps: + - name: Checkout repository + uses: actions/checkout@v6 + + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version: ${{ env.GO_VERSION }} + + - 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: 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: 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 (retry once) + run: | + set -euo pipefail + if ! IMP_E2E_REAL_AGENT=true go test -tags e2e ./test/e2e/... -v -timeout 30m -ginkgo.label-filter="datapath"; then + echo "First run failed; retrying once..." + IMP_E2E_REAL_AGENT=true go test -tags e2e ./test/e2e/... -v -timeout 30m -ginkgo.label-filter="datapath" + fi + + - name: Dump cluster diagnostics on failure + if: failure() + run: | + mkdir -p artifacts + kubectl get pods -A -o wide > artifacts/pods-all.txt || true + kubectl get impvm,impnetwork,impvmclass -A -o wide > artifacts/impvm-all.txt || true + kubectl describe pods -n imp-system > artifacts/imp-system-pods-describe.txt || true + kubectl describe impvm -n default > artifacts/impvm-describe.txt || true + kubectl logs -n imp-system -l app.kubernetes.io/component=operator --tail=500 > artifacts/operator-logs.txt || true + kubectl logs -n imp-system -l app.kubernetes.io/component=agent --tail=500 > artifacts/agent-logs.txt || true + kubectl get events -A --sort-by=.lastTimestamp > artifacts/events.txt || true + + - name: Upload diagnostics + if: failure() + uses: actions/upload-artifact@v7 + with: + name: nightly-e2e-datapath-diagnostics + path: artifacts/ + + - name: Notify Telegram on failure + # Reuses the org's existing bot pattern (see oci-free-tier-monitor). + # Skips cleanly (not a failed step) until TELEGRAM_BOT_TOKEN and + # TELEGRAM_CHAT_ID are added as repo secrets. + if: failure() && github.event_name != 'pull_request' + env: + TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }} + TELEGRAM_CHAT_ID: ${{ secrets.TELEGRAM_CHAT_ID }} + run: | + if [ -z "${TELEGRAM_BOT_TOKEN:-}" ] || [ -z "${TELEGRAM_CHAT_ID:-}" ]; then + echo "TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID not set, skipping notification" + exit 0 + fi + run_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" + curl -fsS -X POST "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/sendMessage" \ + --data-urlencode "chat_id=${TELEGRAM_CHAT_ID}" \ + --data-urlencode "text=imp: Nightly E2E Datapath failed on ${GITHUB_REF_NAME} (${GITHUB_SHA:0:7}) -- ${run_url}" diff --git a/charts/imp/templates/agent/daemonset.yaml b/charts/imp/templates/agent/daemonset.yaml index 714da80..a2f51b8 100644 --- a/charts/imp/templates/agent/daemonset.yaml +++ b/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: diff --git a/charts/imp/tests/agent-daemonset_test.yaml b/charts/imp/tests/agent-daemonset_test.yaml index 378960e..a99946c 100644 --- a/charts/imp/tests/agent-daemonset_test.yaml +++ b/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: diff --git a/internal/agent/network/vxlan.go b/internal/agent/network/vxlan.go index 495e38b..6f6d00b 100644 --- a/internal/agent/network/vxlan.go +++ b/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, diff --git a/internal/agent/scaletozero.go b/internal/agent/scaletozero.go index 6757fe2..ee37b4f 100644 --- a/internal/agent/scaletozero.go +++ b/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) } } diff --git a/internal/agent/scaletozero_linux.go b/internal/agent/scaletozero_linux.go index d042fab..61dc2f5 100644 --- a/internal/agent/scaletozero_linux.go +++ b/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()) } } diff --git a/internal/capability/probe.go b/internal/capability/probe.go index 8639636..2598350 100644 --- a/internal/capability/probe.go +++ b/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 } diff --git a/internal/capability/probe_test.go b/internal/capability/probe_test.go index 84c06c2..2440d6d 100644 --- a/internal/capability/probe_test.go +++ b/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") diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go index e941189..297cb9d 100644 --- a/test/e2e/e2e_suite_test.go +++ b/test/e2e/e2e_suite_test.go @@ -20,6 +20,7 @@ limitations under the License. package e2e import ( + "fmt" "os" "os/exec" "testing" @@ -87,22 +88,61 @@ var _ = BeforeSuite(func() { agentRepo := getenvOrDefault("IMP_E2E_AGENT_IMAGE_REPOSITORY", "local/imp-agent") agentTag := getenvOrDefault("IMP_E2E_AGENT_IMAGE_TAG", "e2e") - impCmd := exec.Command("helm", "install", helmRelease, "charts/imp", + impArgs := []string{"install", helmRelease, "charts/imp", "--namespace", namespace, - "--set", "operator.image.repository="+operatorRepo, - "--set", "operator.image.tag="+operatorTag, - "--set", "agent.image.repository="+agentRepo, - "--set", "agent.image.tag="+agentTag, + "--set", "operator.image.repository=" + operatorRepo, + "--set", "operator.image.tag=" + operatorTag, + "--set", "agent.image.repository=" + agentRepo, + "--set", "agent.image.tag=" + agentTag, "--set", "agent.env.kernelPath=/var/lib/imp/vmlinux", - "--set-string", "agent.nodeSelector.imp\\.dev/no-agent=true", "--set", "metrics.serviceMonitor.enabled=false", "--set", "metrics.podMonitor.enabled=false", - "--wait", "--timeout", "10m") + } + if os.Getenv("IMP_E2E_REAL_AGENT") == "true" { + // Real KVM runner: let the agent schedule (no no-agent nodeSelector) + // and enable the scale-to-zero datapath under test. + impArgs = append(impArgs, + "--set", "agent.extraEnv[0].name=IMP_SCALE_TO_ZERO", + "--set-string", "agent.extraEnv[0].value=true", + // Firecracker runs as a child process of the agent, so guest RAM counts + // against the agent container's own memory cgroup. The chart default + // (128Mi) assumes no hosted VMs and OOM-kills almost immediately once a + // real microVM boots — silently, since a SIGKILL leaves no panic/log line. + "--set", "agent.resources.limits.memory=1Gi", + ) + } else { + impArgs = append(impArgs, "--set-string", "agent.nodeSelector.imp\\.dev/no-agent=true") + } + impArgs = append(impArgs, "--wait", "--timeout", "10m") + + impCmd := exec.Command("helm", impArgs...) _, err = utils.Run(impCmd) Expect(err).NotTo(HaveOccurred(), "helm install imp failed") }) var _ = AfterSuite(func() { + if os.Getenv("IMP_E2E_REAL_AGENT") == "true" { + By("dumping pod status + agent/operator logs before teardown") + podStatus := exec.Command("kubectl", "get", "pods", "-n", namespace, "-o", "wide") + out, _ := utils.Run(podStatus) + fmt.Println("--- pod status (check RESTARTS) ---\n" + out) + + agentLogsPrev := exec.Command("kubectl", "logs", "-n", namespace, + "-l", "app.kubernetes.io/component=agent", "--tail=500", "--prefix", "--previous") + out, _ = utils.Run(agentLogsPrev) + fmt.Println("--- agent logs (previous container, if it restarted) ---\n" + out) + + agentLogs := exec.Command("kubectl", "logs", "-n", namespace, + "-l", "app.kubernetes.io/component=agent", "--tail=500", "--prefix") + out, _ = utils.Run(agentLogs) + fmt.Println("--- agent logs ---\n" + out) + + operatorLogs := exec.Command("kubectl", "logs", "-n", namespace, + "-l", "app.kubernetes.io/component=operator", "--tail=500", "--prefix") + out, _ = utils.Run(operatorLogs) + fmt.Println("--- operator logs ---\n" + out) + } + By("uninstalling imp chart") unimpCmd := exec.Command("helm", "uninstall", helmRelease, "--namespace", namespace) _, _ = utils.Run(unimpCmd) diff --git a/test/e2e/kind-datapath.yaml b/test/e2e/kind-datapath.yaml new file mode 100644 index 0000000..7e0f1b2 --- /dev/null +++ b/test/e2e/kind-datapath.yaml @@ -0,0 +1,11 @@ +# Kind cluster configuration for the real-agent ScaleToZero datapath e2e. +# Mounts host /dev/kvm into the control-plane node so a real Firecracker agent +# can boot microVMs. See docs (e2e runner runbook) for the proven nested-KVM path. +# Usage: kind create cluster --name imp-e2e-datapath --config test/e2e/kind-datapath.yaml +kind: Cluster +apiVersion: kind.x-k8s.io/v1alpha4 +nodes: + - role: control-plane + extraMounts: + - hostPath: /dev/kvm + containerPath: /dev/kvm diff --git a/test/e2e/scaletozero_datapath_test.go b/test/e2e/scaletozero_datapath_test.go index 43bcb71..bbd3675 100644 --- a/test/e2e/scaletozero_datapath_test.go +++ b/test/e2e/scaletozero_datapath_test.go @@ -20,40 +20,293 @@ limitations under the License. package e2e import ( + "encoding/json" + "fmt" + "net" + "net/http" + "os" + "os/exec" + "strings" + "time" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/syscode-labs/imp/test/utils" ) -// Wake-on-traffic datapath e2e — DELIBERATELY Pending (PIt). +// Wake-on-traffic datapath e2e. // // This validates the one thing the host-only unit + envtest coverage cannot: that // the agent's AF_PACKET hook actually observes the first frame destined to a // TAP-less (suspended) ScaleToZero VM on imp's VXLAN overlay, and that a real -// Firecracker suspend/resume round-trips. It is UNVALIDATED and cannot run yet — -// see docs (runner runbook) for why the harness must change before it can. -// -// Prerequisites that do NOT exist in the current e2e harness (all are runbook -// decisions, which is why this spec stays Pending until the runbook lands them): -// -// 1. A KVM-capable node. BeforeSuite currently pins the agent to -// nodeSelector imp.dev/no-agent=true (agent off) and the Kind smoke runner has -// no /dev/kvm. A real Firecracker agent needs nested virt + the Firecracker -// Talos system extension + a guest kernel at agent.env.kernelPath. -// 2. The agent enabled with IMP_SCALE_TO_ZERO=true. BeforeSuite must conditionally -// drop the no-agent selector and set the env when targeting the KVM runner -// (e.g. gate on an IMP_E2E_REAL_AGENT env var). -// 3. A traffic source on the same ImpNetwork. OPEN QUESTION for first run: how does -// a frame reach a suspended VM's overlay IP? The realistic source is a second -// always-on VM on the same ImpNetwork pinging the suspended VM — but that pulls -// in guest exec / vsock. This is the crux to resolve when the runner is live. +// Firecracker suspend/resume round-trips. // -// Intended flow once the harness supports it: -// - Create an ImpNetwork + a ScaleToZero ImpVM with a short idleTimeout (e.g. 15s). -// - Wait for Running, then (no traffic) wait for the agent's idle detector to -// auto-suspend it → status.phase == Suspended, VTEP retained. -// - Send a frame to the VM's overlay IP from the traffic source. -// - Assert the VM returns to Running (Suspended → Resuming → Running). +// UNVALIDATED: this spec has never run against a live KVM node — no CI job wires +// IMP_E2E_REAL_AGENT=true onto a KVM runner yet. Skips itself unless that env var +// is set, so it stays inert everywhere else. Traffic source: a second always-on +// VM on the same ImpNetwork execs `ping` into the suspended VM's overlay IP via +// the agent's vsock guest-exec API (internal/agent/api/exec.go) — guest agent +// injection defaults to enabled, so no extra fixture wiring needed for that part. var _ = Describe("Imp ScaleToZero datapath", Label("datapath"), func() { - PIt("wakes a suspended ScaleToZero VM when a frame arrives for its overlay IP", func() { - Skip("requires a KVM node + real Firecracker agent; see the e2e runner runbook") + const ( + networkName = "e2e-sz-datapath-net" + className = "e2e-sz-datapath" + pingerName = "e2e-sz-datapath-pinger" + targetName = "e2e-sz-datapath-target" + ) + + It("wakes a suspended ScaleToZero VM when a frame arrives for its overlay IP", func() { + if os.Getenv("IMP_E2E_REAL_AGENT") != "true" { + Skip("requires a KVM node + real Firecracker agent (IMP_E2E_REAL_AGENT=true); see the e2e runner runbook") + } + + DeferCleanup(func() { + _, _ = utils.Run(exec.Command("kubectl", "delete", "impvm", targetName, "-n", "default", "--ignore-not-found")) + _, _ = utils.Run(exec.Command("kubectl", "delete", "impvm", pingerName, "-n", "default", "--ignore-not-found")) + _, _ = utils.Run(exec.Command("kubectl", "delete", "impnetwork", networkName, "-n", "default", "--ignore-not-found")) + _, _ = utils.Run(exec.Command("kubectl", "delete", "impvmclass", className, "--ignore-not-found")) + }) + + By("getting the Kind node name") + nodeOut, err := utils.Run(exec.Command("kubectl", "get", "nodes", "-o", "jsonpath={.items[0].metadata.name}")) + Expect(err).NotTo(HaveOccurred()) + nodeName := strings.TrimSpace(nodeOut) + Expect(nodeName).NotTo(BeEmpty()) + + By("removing control-plane taint and labeling the node imp/enabled=true") + _, _ = utils.Run(exec.Command("kubectl", "taint", "nodes", nodeName, + "node-role.kubernetes.io/control-plane:NoSchedule-")) + _, err = utils.Run(exec.Command("kubectl", "label", "node", nodeName, "imp/enabled=true", "--overwrite")) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { + _, _ = utils.Run(exec.Command("kubectl", "label", "node", nodeName, "imp/enabled-")) + }) + + By("creating the ImpVMClass") + classManifest := fmt.Sprintf(` +apiVersion: imp.dev/v1alpha1 +kind: ImpVMClass +metadata: + name: %s +spec: + vcpu: 1 + memoryMiB: 256 + diskGiB: 1 +`, className) + applyClass := exec.Command("kubectl", "apply", "-f", "-") + applyClass.Stdin = strings.NewReader(classManifest) + _, err = utils.Run(applyClass) + Expect(err).NotTo(HaveOccurred()) + + By("creating the ImpNetwork") + netManifest := fmt.Sprintf(` +apiVersion: imp.dev/v1alpha1 +kind: ImpNetwork +metadata: + name: %s + namespace: default +spec: + subnet: 10.45.0.0/24 +`, networkName) + applyNet := exec.Command("kubectl", "apply", "-f", "-") + applyNet.Stdin = strings.NewReader(netManifest) + _, err = utils.Run(applyNet) + Expect(err).NotTo(HaveOccurred()) + + By("creating the always-on pinger VM") + pingerManifest := fmt.Sprintf(` +apiVersion: imp.dev/v1alpha1 +kind: ImpVM +metadata: + name: %s + namespace: default +spec: + classRef: + name: %s + networkRef: + name: %s + image: docker.io/library/nginx:alpine +`, pingerName, className, networkName) + applyPinger := exec.Command("kubectl", "apply", "-f", "-") + applyPinger.Stdin = strings.NewReader(pingerManifest) + _, err = utils.Run(applyPinger) + Expect(err).NotTo(HaveOccurred()) + + By("creating the ScaleToZero target VM with a short idleTimeout") + targetManifest := fmt.Sprintf(` +apiVersion: imp.dev/v1alpha1 +kind: ImpVM +metadata: + name: %s + namespace: default +spec: + classRef: + name: %s + networkRef: + name: %s + image: docker.io/library/nginx:alpine + desiredState: ScaleToZero + idleTimeout: 15s +`, targetName, className, networkName) + applyTarget := exec.Command("kubectl", "apply", "-f", "-") + applyTarget.Stdin = strings.NewReader(targetManifest) + _, err = utils.Run(applyTarget) + Expect(err).NotTo(HaveOccurred()) + + By("waiting for both VMs to reach Running") + Eventually(func(g Gomega) { + phase, _ := vmPhaseAndIP(g, pingerName) + g.Expect(phase).To(Equal("Running")) + }, "5m", "5s").Should(Succeed()) + + var targetIP string + Eventually(func(g Gomega) { + phase, ip := vmPhaseAndIP(g, targetName) + g.Expect(phase).To(Equal("Running")) + g.Expect(ip).NotTo(BeEmpty()) + targetIP = ip + }, "5m", "5s").Should(Succeed()) + + By("finding the agent pod colocated with the pinger VM") + pingerNode := vmNodeName(pingerName) + Expect(pingerNode).NotTo(BeEmpty()) + agentPod := agentPodOnNode(pingerNode) + Expect(agentPod).NotTo(BeEmpty()) + + By("port-forwarding to the agent's guest-exec API") + pf := exec.Command("kubectl", "port-forward", "pod/"+agentPod, "19091:9091", "-n", namespace) + Expect(pf.Start()).To(Succeed()) + DeferCleanup(func() { + if pf.Process != nil { + _ = pf.Process.Kill() + } + }) + Eventually(func(g Gomega) { + conn, dialErr := net.DialTimeout("tcp", "localhost:19091", 2*time.Second) + g.Expect(dialErr).NotTo(HaveOccurred()) + _ = conn.Close() + }, "30s", "1s").Should(Succeed()) + + By("warming the pinger's ARP cache for the target before it suspends") + // The target's TAP is torn down on suspend (frees memory) — only the VTEP + // is kept, so cross-node overlay routing still works, but the pinger's own + // ARP resolution for the target's IP can only happen while the target's TAP + // exists to answer it. A sender with a warm ARP entry sends the IP frame + // directly on wake with no fresh ARP round-trip needed; a cold sender never + // gets an IP frame onto the wire in the first place, so the AF_PACKET wake + // hook (bound to ETH_P_IP, not ARP) never fires. This mirrors the real + // intended use case: traffic to an already-known, now-idle destination. + // Retry: status.phase==Running reflects the host starting Firecracker, not + // that the guest kernel has finished booting and brought its network up — + // an immediate single-shot ping can race a guest that isn't ready yet. + var warmOut string + Eventually(func(g Gomega) { + warmExit, out, execErr := execInVM(pingerName, "/bin/ping", "-c", "1", "-W", "5", targetIP) + warmOut = out + g.Expect(execErr).NotTo(HaveOccurred()) + g.Expect(warmExit).To(Equal(int32(0))) + }, "1m", "3s").Should(Succeed(), func() string { + return "ARP warm-up ping never succeeded while target was still Running:\n" + warmOut + }) + + By("waiting for the target to auto-suspend after going idle") + Eventually(func(g Gomega) { + phase, _ := vmPhaseAndIP(g, targetName) + g.Expect(phase).To(Equal("Suspended")) + }, "2m", "5s").Should(Succeed()) + + By("pinging the suspended VM's overlay IP from the pinger VM's guest agent") + wakeExit, wakeOut, err := execInVM(pingerName, "/bin/ping", "-c", "3", "-W", "2", targetIP) + Expect(err).NotTo(HaveOccurred()) + GinkgoWriter.Printf("wake ping exit=%d output:\n%s\n", wakeExit, wakeOut) + + By("asserting the target wakes: Suspended -> Resuming -> Running") + Eventually(func(g Gomega) { + phase, _ := vmPhaseAndIP(g, targetName) + g.Expect(phase).To(Equal("Running")) + }, "2m", "5s").Should(Succeed()) }) }) + +// execInVM runs command inside vmName's guest via the agent's vsock guest-exec +// API (assumes a port-forward to localhost:19091 is already active) and returns +// the process exit code plus the combined stdout+stderr NDJSON stream decoded to +// plain text. +func execInVM(vmName string, command ...string) (int32, string, error) { + body, err := json.Marshal(map[string][]string{"command": command}) + if err != nil { + return 0, "", err + } + resp, err := http.Post( //nolint:noctx + fmt.Sprintf("http://localhost:19091/v1/exec/default/%s", vmName), + "application/json", strings.NewReader(string(body))) + if err != nil { + return 0, "", err + } + defer resp.Body.Close() //nolint:errcheck + if resp.StatusCode != http.StatusOK { + return 0, "", fmt.Errorf("exec %s: unexpected status %d", vmName, resp.StatusCode) + } + + var out strings.Builder + var exitCode int32 + dec := json.NewDecoder(resp.Body) + for dec.More() { + var line struct { + Stream string `json:"stream"` + Line string `json:"line,omitempty"` + Code *int32 `json:"code,omitempty"` + } + if err := dec.Decode(&line); err != nil { + return 0, out.String(), fmt.Errorf("decode exec response: %w", err) + } + switch line.Stream { + case "exit": + if line.Code != nil { + exitCode = *line.Code + } + default: + out.WriteString(line.Line) + out.WriteString("\n") + } + } + return exitCode, out.String(), nil +} + +func vmPhaseAndIP(g Gomega, name string) (string, string) { + out, err := utils.Run(exec.Command("kubectl", "get", "impvm", name, "-n", "default", + "-o", "jsonpath={.status.phase} {.status.ip}")) + g.Expect(err).NotTo(HaveOccurred()) + parts := strings.Fields(strings.TrimSpace(out)) + switch len(parts) { + case 0: + return "", "" + case 1: + return parts[0], "" + default: + return parts[0], parts[1] + } +} + +func vmNodeName(name string) string { + // status.nodeName exists on the CRD but nothing in the codebase ever writes it + // (only spec.nodeName is set, by the scheduler) — read spec instead. + out, err := utils.Run(exec.Command("kubectl", "get", "impvm", name, "-n", "default", + "-o", "jsonpath={.spec.nodeName}")) + if err != nil { + return "" + } + return strings.TrimSpace(out) +} + +func agentPodOnNode(nodeName string) string { + out, err := utils.Run(exec.Command("kubectl", "get", "pods", "-n", namespace, + "-l", "app.kubernetes.io/component=agent", + "--field-selector", "spec.nodeName="+nodeName, + "-o", "jsonpath={.items[0].metadata.name}")) + if err != nil { + return "" + } + return strings.TrimSpace(out) +}