diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 42ae467..4199f9d 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -6,6 +6,7 @@ import ( "context" "flag" "os" + "time" _ "k8s.io/client-go/plugin/pkg/client/auth" @@ -111,6 +112,14 @@ func main() { alloc = fcDrv.Alloc } + // Scale-to-zero (wake-on-traffic) is experimental and its capture hook is not + // yet hardware-validated — opt in explicitly via IMP_SCALE_TO_ZERO=true. + var sz *agent.ScaleToZero + if os.Getenv("IMP_SCALE_TO_ZERO") == "true" { + sz = agent.NewLinuxScaleToZero(1024, 15*time.Second) + log.Info("scale-to-zero enabled (experimental; wake path not hardware-validated)") + } + if err := (&agent.ImpVMReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), @@ -120,6 +129,7 @@ func main() { Metrics: mc, Net: prodNet, Alloc: alloc, + SZ: sz, Recorder: mgr.GetEventRecorderFor("imp-agent"), //nolint:staticcheck // controller-runtime returns legacy recorder type expected by reconciler }).SetupWithManager(mgr); err != nil { log.Error(err, "Unable to set up ImpVMReconciler") diff --git a/internal/agent/reconciler.go b/internal/agent/reconciler.go index 09fd303..6cb9e25 100644 --- a/internal/agent/reconciler.go +++ b/internal/agent/reconciler.go @@ -22,7 +22,9 @@ import ( "k8s.io/client-go/util/retry" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" logf "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/source" impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1" "github.com/syscode-labs/imp/internal/agent/network" @@ -56,6 +58,9 @@ type ImpVMReconciler struct { // SuspendDir is the node-local base directory under which suspend snapshots // are written (one subdir per VM). Defaults to /var/lib/imp/suspend when empty. SuspendDir string + // SZ is optional. When non-nil it enables scale-to-zero: ScaleToZero VMs + // auto-suspend on traffic idle and auto-resume on the first inbound packet. + SZ *ScaleToZero } // suspendBaseDir returns the configured suspend snapshot base directory, @@ -256,6 +261,19 @@ func (r *ImpVMReconciler) handleRunning(ctx context.Context, vm *impdevv1alpha1. return ctrl.Result{}, nil } + // ScaleToZero: poll TAP traffic and auto-suspend after idleTimeout of silence. + if vm.Spec.DesiredState == impdevv1alpha1.VMDesiredStateScaleToZero && r.SZ != nil { + suspended, err := r.maybeSuspendIdle(ctx, vm) + if err != nil { + log.Error(err, "ScaleToZero idle probe failed; will retry") + return ctrl.Result{RequeueAfter: r.SZ.interval}, nil + } + if suspended { + return ctrl.Result{}, nil + } + // Not idle — fall through to the liveness check; runningResult requeues. + } + state, err := r.Driver.Inspect(ctx, vm) if err != nil { log.Error(err, "Driver Inspect failed") @@ -263,7 +281,7 @@ func (r *ImpVMReconciler) handleRunning(ctx context.Context, vm *impdevv1alpha1. } if state.Running { - return ctrl.Result{}, nil // watch-driven steady state + return r.runningResult(vm), nil // watch-driven steady state (ScaleToZero: polled) } // Inspect returned Running=false. Before declaring the VM dead, check whether @@ -333,6 +351,46 @@ func (r *ImpVMReconciler) handleRunning(ctx context.Context, vm *impdevv1alpha1. return r.finishFailed(ctx, vm) } +// runningResult is the steady-state reconcile result for a Running VM: empty +// (watch-driven) normally, or a poll interval for ScaleToZero VMs whose idle +// detector needs periodic wake-ups. +func (r *ImpVMReconciler) runningResult(vm *impdevv1alpha1.ImpVM) ctrl.Result { + if r.SZ != nil && vm.Spec.DesiredState == impdevv1alpha1.VMDesiredStateScaleToZero { + return ctrl.Result{RequeueAfter: r.SZ.interval} + } + return ctrl.Result{} +} + +// maybeSuspendIdle probes vm's TAP traffic. If the VM has been idle for its +// idleTimeout it stamps lastActivityTime and transitions to Suspending, returning +// true; otherwise it returns false without touching status. We deliberately do NOT +// write lastActivityTime on every busy probe — a genuinely active VM changes bytes +// each interval, so that would be a status write every interval per VM (churn at the +// density this feature targets). The field records when the VM last had traffic +// before suspending; live activity is observable via /metrics. +func (r *ImpVMReconciler) maybeSuspendIdle(ctx context.Context, vm *impdevv1alpha1.ImpVM) (bool, error) { + key := client.ObjectKeyFromObject(vm) + tap := network.TAPName(vmKey(vm)) + idle, lastActivity, err := r.SZ.observe(key, tap, idleTimeoutOrDefault(vm), time.Now()) + if err != nil { + return false, err + } + if !idle { + return false, nil + } + base := vm.DeepCopy() + t := metav1.NewTime(lastActivity) + vm.Status.LastActivityTime = &t + vm.Status.Phase = impdevv1alpha1.VMPhaseSuspending + if err := r.Status().Patch(ctx, vm, client.MergeFrom(base)); err != nil { + if apierrors.IsConflict(err) { + return false, nil // requeue re-evaluates + } + return false, err + } + return true, nil +} + func (r *ImpVMReconciler) handleTerminating(ctx context.Context, vm *impdevv1alpha1.ImpVM) (result ctrl.Result, err error) { log := logf.FromContext(ctx) @@ -401,8 +459,11 @@ func (r *ImpVMReconciler) handleSuspending(ctx context.Context, vm *impdevv1alph return ctrl.Result{RequeueAfter: r.retryInterval()}, err } - // Deregister VTEP so other nodes stop routing to the VM about to be stopped. - if vm.Spec.NetworkRef != nil && vm.Status.IP != "" { + // Deregister VTEP so other nodes stop routing to the VM about to be stopped — + // EXCEPT for ScaleToZero, which keeps the VTEP so the overlay still delivers + // the wake packet to this node while the VM is suspended. + if vm.Spec.NetworkRef != nil && vm.Status.IP != "" && + vm.Spec.DesiredState != impdevv1alpha1.VMDesiredStateScaleToZero { if derr := r.deregisterVTEP(ctx, vm); derr != nil { log.Error(derr, "Failed to deregister VTEP during suspend") } @@ -431,6 +492,14 @@ func (r *ImpVMReconciler) handleSuspending(ctx context.Context, vm *impdevv1alph if r.Metrics != nil { r.Metrics.SetVMState(vm.Namespace+"/"+vm.Name, "Suspended", r.NodeName) } + + // ScaleToZero: register the VM's IP so the activator wakes it on the first + // inbound packet. Reset any idle sample so a fresh window applies on resume. + if vm.Spec.DesiredState == impdevv1alpha1.VMDesiredStateScaleToZero && r.SZ != nil && vm.Status.IP != "" { + r.SZ.reg.register(vm.Status.IP, vm) + r.SZ.resetIdle(client.ObjectKeyFromObject(vm)) + } + log.Info("VM suspended", "snapshotPath", destDir) return ctrl.Result{}, nil } @@ -438,9 +507,17 @@ func (r *ImpVMReconciler) handleSuspending(ctx context.Context, vm *impdevv1alph // handleSuspended holds the VM in the Suspended state until desiredState flips // back to Running, at which point it advances to Resuming. func (r *ImpVMReconciler) handleSuspended(ctx context.Context, vm *impdevv1alpha1.ImpVM) (ctrl.Result, error) { - if vm.Spec.DesiredState == impdevv1alpha1.VMDesiredStateSuspended { + switch vm.Spec.DesiredState { + case impdevv1alpha1.VMDesiredStateSuspended: return ctrl.Result{}, nil // steady state + case impdevv1alpha1.VMDesiredStateScaleToZero: + // Stay suspended until the activator observes an inbound packet. + if r.SZ == nil || !r.SZ.reg.pending(client.ObjectKeyFromObject(vm)) { + return ctrl.Result{}, nil + } + // Wake packet observed — fall through to resume. } + // desiredState=Running (explicit resume) or ScaleToZero with a pending wake. base := vm.DeepCopy() vm.Status.Phase = impdevv1alpha1.VMPhaseResuming if err := r.Status().Patch(ctx, vm, client.MergeFrom(base)); err != nil { @@ -501,6 +578,13 @@ func (r *ImpVMReconciler) handleResuming(ctx context.Context, vm *impdevv1alpha1 r.Metrics.SetVMState(vm.Namespace+"/"+vm.Name, "Running", r.NodeName) } + // Clear scale-to-zero wake state now that the VM is running again. + if r.SZ != nil { + key := client.ObjectKeyFromObject(vm) + r.SZ.reg.clear(key) + r.SZ.resetIdle(key) + } + // Re-register VTEP + sync FDB so other nodes route to the resumed VM. r.ensureVTEPAndFDB(ctx, vm, state.IP) @@ -597,10 +681,22 @@ func (r *ImpVMReconciler) SetupWithManager(mgr ctrl.Manager) error { // Detect and patch CPU model onto ClusterImpNodeProfile at startup (best-effort). go detectAndPatchCPUModel(context.Background(), r.Client, r.NodeName) - return ctrl.NewControllerManagedBy(mgr). + b := ctrl.NewControllerManagedBy(mgr). For(&impdevv1alpha1.ImpVM{}). - Named("agent-impvm"). - Complete(r) + Named("agent-impvm") + + // Scale-to-zero: let the activator fire reconciles via a channel, and run the + // packet source as a per-node runnable. + if r.SZ != nil { + b = b.WatchesRawSource(source.Channel(r.SZ.reg.events, &handler.EnqueueRequestForObject{})) + if r.SZ.src != nil { + if err := mgr.Add(&activator{src: r.SZ.src, reg: r.SZ.reg}); err != nil { + return err + } + } + } + + return b.Complete(r) } // registerVTEP adds or updates the VTEPEntry for vm in ImpNetwork.status.vtepTable. diff --git a/internal/agent/scaletozero.go b/internal/agent/scaletozero.go new file mode 100644 index 0000000..6757fe2 --- /dev/null +++ b/internal/agent/scaletozero.go @@ -0,0 +1,180 @@ +package agent + +// Scale-to-zero wake-on-traffic support (Phase 3). +// +// This file holds the platform-neutral, fully unit-tested core: the wake +// registry (which suspended VMs are awaiting a packet), the traffic-idle +// detector, and the reconcile-triggering plumbing. The two pieces that need a +// real host — reading NIC byte counters and capturing packets — are injected as +// a linkStatsFunc and a PacketSource, faked in tests and implemented for real in +// scaletozero_linux.go. +// +// UNVALIDATED: the real PacketSource (AF_PACKET on the overlay) has NOT been +// confirmed to observe the first frame destined to a TAP-less (suspended) VM. +// The first cluster spike (Phase 1 of the wake-on-traffic plan) must validate +// hook placement. The PacketSource interface exists precisely so the hook can be +// swapped to tc-BPF later without touching any of the logic below. + +import ( + "context" + "sync" + "time" + + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + + impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1" +) + +// defaultIdleTimeout is used when a ScaleToZero VM leaves spec.idleTimeout unset. +const defaultIdleTimeout = 5 * time.Minute + +// PacketSource delivers the destination IP of each inbound overlay frame to +// onDstIP until ctx is cancelled. Implementations are host-specific. +type PacketSource interface { + Run(ctx context.Context, onDstIP func(ip string)) error +} + +// linkStatsFunc returns cumulative rx+tx bytes for a host interface (a VM TAP). +type linkStatsFunc func(iface string) (bytes uint64, err error) + +// wakeRegistry tracks suspended ScaleToZero VMs by IP and fires a reconcile when +// a packet arrives for one. Safe for concurrent use: the activator goroutine +// calls onDstIP while the reconcile loop calls register/clear/pending. +type wakeRegistry struct { + mu sync.Mutex + keyByIP map[string]types.NamespacedName + ipByKey map[types.NamespacedName]string + objByKey map[types.NamespacedName]client.Object + signalled map[types.NamespacedName]bool + events chan event.GenericEvent +} + +func newWakeRegistry(bufSize int) *wakeRegistry { + if bufSize <= 0 { + bufSize = 1024 + } + return &wakeRegistry{ + keyByIP: map[string]types.NamespacedName{}, + ipByKey: map[types.NamespacedName]string{}, + objByKey: map[types.NamespacedName]client.Object{}, + signalled: map[types.NamespacedName]bool{}, + events: make(chan event.GenericEvent, bufSize), + } +} + +// register marks vm as suspended and awaiting a wake packet at ip. +func (w *wakeRegistry) register(ip string, vm client.Object) { + key := client.ObjectKeyFromObject(vm) + w.mu.Lock() + defer w.mu.Unlock() + // Drop any stale IP mapping for this VM before recording the new one. + if old, ok := w.ipByKey[key]; ok { + delete(w.keyByIP, old) + } + w.keyByIP[ip] = key + w.ipByKey[key] = ip + w.objByKey[key] = vm +} + +// 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. +func (w *wakeRegistry) onDstIP(ip string) { + w.mu.Lock() + defer w.mu.Unlock() + key, ok := w.keyByIP[ip] + if !ok || w.signalled[key] { + return + } + obj := w.objByKey[key] + select { + case w.events <- event.GenericEvent{Object: obj}: + w.signalled[key] = true + default: + // Channel full; leave unsignalled so a later packet retries. + } +} + +// pending reports whether a wake packet has been observed for key. +func (w *wakeRegistry) pending(key types.NamespacedName) bool { + w.mu.Lock() + defer w.mu.Unlock() + return w.signalled[key] +} + +// clear drops all state for key (called once the VM has resumed). +func (w *wakeRegistry) clear(key types.NamespacedName) { + w.mu.Lock() + defer w.mu.Unlock() + if ip, ok := w.ipByKey[key]; ok { + delete(w.keyByIP, ip) + } + delete(w.ipByKey, key) + delete(w.objByKey, key) + delete(w.signalled, key) +} + +// ScaleToZero bundles the wake registry, the idle detector, and the packet +// source into the optional feature attached to the reconciler as SZ. +type ScaleToZero struct { + reg *wakeRegistry + stats linkStatsFunc + src PacketSource + interval time.Duration + + mu sync.Mutex + samples map[types.NamespacedName]idleSample +} + +type idleSample struct { + bytes uint64 + since time.Time +} + +func newScaleToZero(stats linkStatsFunc, src PacketSource, interval time.Duration, bufSize int) *ScaleToZero { + if interval <= 0 { + interval = 15 * time.Second + } + return &ScaleToZero{ + reg: newWakeRegistry(bufSize), + stats: stats, + src: src, + interval: interval, + samples: map[types.NamespacedName]idleSample{}, + } +} + +// observe samples iface's byte counter and reports whether the VM has seen no +// traffic for at least idleTimeout. Any change resets the idle clock, so a +// freshly-resumed VM (no prior sample) always gets a full idleTimeout of grace — +// this is the anti-thrash hysteresis. +// +// ASSUMPTION (validate on cluster): byte-idle suspends a VM holding an idle-but- +// open connection (long-poll, pooled DB conn), which the resume must transparently +// re-establish or the connection breaks. Combining with the guest CPU-idle signal +// is an open design question; not built here. +func (s *ScaleToZero) observe(key types.NamespacedName, iface string, idleTimeout time.Duration, now time.Time) (idle bool, lastActivity time.Time, err error) { + b, err := s.stats(iface) + if err != nil { + return false, time.Time{}, err + } + s.mu.Lock() + defer s.mu.Unlock() + prev, ok := s.samples[key] + if !ok || b != prev.bytes { + s.samples[key] = idleSample{bytes: b, since: now} + return false, now, nil + } + return now.Sub(prev.since) >= idleTimeout, prev.since, nil +} + +// idleTimeoutOrDefault resolves the effective idle window for a VM. +func idleTimeoutOrDefault(vm *impdevv1alpha1.ImpVM) time.Duration { + if vm.Spec.IdleTimeout != nil && vm.Spec.IdleTimeout.Duration > 0 { + return vm.Spec.IdleTimeout.Duration + } + return defaultIdleTimeout +} diff --git a/internal/agent/scaletozero_linux.go b/internal/agent/scaletozero_linux.go new file mode 100644 index 0000000..d042fab --- /dev/null +++ b/internal/agent/scaletozero_linux.go @@ -0,0 +1,95 @@ +//go:build linux + +package agent + +import ( + "context" + "net" + "time" + + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" + "k8s.io/apimachinery/pkg/types" +) + +// resetIdle forgets any idle sample for key (called on suspend/resume so the VM +// starts a fresh idle window next time it runs). Only the linux reconciler uses it. +func (s *ScaleToZero) resetIdle(key types.NamespacedName) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.samples, key) +} + +// activator runs the packet source, feeding matches into the wake registry. It +// runs on every node (no leader election) since each agent owns its own VMs. +type activator struct { + src PacketSource + reg *wakeRegistry +} + +func (a *activator) Start(ctx context.Context) error { + return a.src.Run(ctx, a.reg.onDstIP) +} + +// NeedLeaderElection marks the activator as a per-node runnable. +func (a *activator) NeedLeaderElection() bool { return false } + +// netlinkLinkStats returns cumulative rx+tx bytes for iface (a VM TAP), used by +// the idle detector to decide whether a ScaleToZero VM has gone quiet. +func netlinkLinkStats(iface string) (uint64, error) { + link, err := netlink.LinkByName(iface) + if err != nil { + return 0, err + } + st := link.Attrs().Statistics + if st == nil { + return 0, nil + } + return st.RxBytes + st.TxBytes, nil +} + +// afpacketSource captures inbound IPv4 frames on the node via a single AF_PACKET +// 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. +type afpacketSource struct{} + +func htons(v uint16) uint16 { return v<<8 | v>>8 } + +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))) + if err != nil { + return err + } + // Unblock the blocking Recvfrom and release the fd when the manager stops. + go func() { + <-ctx.Done() + _ = unix.Close(fd) + }() + + buf := make([]byte, 65536) + for { + n, _, err := unix.Recvfrom(fd, buf, 0) + if err != nil { + if ctx.Err() != nil { + return nil // closed on shutdown + } + 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]). + if n < 34 { + continue + } + onDstIP(net.IP(buf[30:34]).String()) + } +} + +// NewLinuxScaleToZero wires the real host implementations into the neutral core. +func NewLinuxScaleToZero(bufSize int, interval time.Duration) *ScaleToZero { + return newScaleToZero(netlinkLinkStats, afpacketSource{}, interval, bufSize) +} diff --git a/internal/agent/scaletozero_reconciler_test.go b/internal/agent/scaletozero_reconciler_test.go new file mode 100644 index 0000000..7932ba9 --- /dev/null +++ b/internal/agent/scaletozero_reconciler_test.go @@ -0,0 +1,143 @@ +//go:build linux + +package agent + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1" +) + +var _ = Describe("ImpVM Agent: scale-to-zero", func() { + ctx := context.Background() + + // szWith builds a ScaleToZero whose idle detector reads *bytes (so a test can + // simulate traffic) and has no real packet source. + szWith := func(bytes *uint64) *ScaleToZero { + return newScaleToZero(func(string) (uint64, error) { return *bytes, nil }, nil, time.Second, 16) + } + + It("auto-suspends a Running ScaleToZero VM after it goes idle", func() { + driver := NewStubDriver() + var bytes uint64 = 100 + sz := szWith(&bytes) + r := &ImpVMReconciler{Client: k8sClient, NodeName: testNode, Driver: driver, SuspendDir: GinkgoT().TempDir(), SZ: sz} + + vm := &impdevv1alpha1.ImpVM{ + ObjectMeta: metav1.ObjectMeta{Name: "tc-sz-idle", Namespace: "default", Finalizers: []string{"imp/finalizer"}}, + Spec: impdevv1alpha1.ImpVMSpec{NodeName: testNode, DesiredState: impdevv1alpha1.VMDesiredStateScaleToZero}, + } + Expect(k8sClient.Create(ctx, vm)).To(Succeed()) + DeferCleanup(func() { k8sClient.Delete(ctx, vm) }) //nolint:errcheck + + pid, err := driver.Start(ctx, vm) + Expect(err).NotTo(HaveOccurred()) + base := vm.DeepCopy() + vm.Status.Phase = impdevv1alpha1.VMPhaseRunning + vm.Status.RuntimePID = pid + vm.Status.IP = "192.168.100.60" + Expect(k8sClient.Status().Patch(ctx, vm, client.MergeFrom(base))).To(Succeed()) + + key := types.NamespacedName{Name: "tc-sz-idle", Namespace: "default"} + req := reconcile.Request{NamespacedName: key} + + // Seed the idle sample as if traffic last moved an hour ago, so the probe + // (default 5m idleTimeout) sees the VM as idle immediately. + sz.samples[key] = idleSample{bytes: bytes, since: time.Now().Add(-time.Hour)} + + _, err = r.Reconcile(ctx, req) + Expect(err).NotTo(HaveOccurred()) + updated := &impdevv1alpha1.ImpVM{} + Expect(k8sClient.Get(ctx, key, updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(impdevv1alpha1.VMPhaseSuspending)) + }) + + It("keeps a Running ScaleToZero VM running while it has traffic", func() { + driver := NewStubDriver() + var bytes uint64 = 100 + sz := szWith(&bytes) + r := &ImpVMReconciler{Client: k8sClient, NodeName: testNode, Driver: driver, SZ: sz} + + vm := &impdevv1alpha1.ImpVM{ + ObjectMeta: metav1.ObjectMeta{Name: "tc-sz-busy", Namespace: "default", Finalizers: []string{"imp/finalizer"}}, + Spec: impdevv1alpha1.ImpVMSpec{NodeName: testNode, DesiredState: impdevv1alpha1.VMDesiredStateScaleToZero}, + } + Expect(k8sClient.Create(ctx, vm)).To(Succeed()) + DeferCleanup(func() { k8sClient.Delete(ctx, vm) }) //nolint:errcheck + + pid, err := driver.Start(ctx, vm) + Expect(err).NotTo(HaveOccurred()) + base := vm.DeepCopy() + vm.Status.Phase = impdevv1alpha1.VMPhaseRunning + vm.Status.RuntimePID = pid + vm.Status.IP = "192.168.100.61" + Expect(k8sClient.Status().Patch(ctx, vm, client.MergeFrom(base))).To(Succeed()) + + key := types.NamespacedName{Name: "tc-sz-busy", Namespace: "default"} + req := reconcile.Request{NamespacedName: key} + + // First reconcile establishes a baseline (never idle). Result requeues. + res, err := r.Reconcile(ctx, req) + Expect(err).NotTo(HaveOccurred()) + Expect(res.RequeueAfter).To(Equal(sz.interval)) + updated := &impdevv1alpha1.ImpVM{} + Expect(k8sClient.Get(ctx, key, updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(impdevv1alpha1.VMPhaseRunning)) + }) + + It("resumes a Suspended ScaleToZero VM once a wake packet is observed", func() { + driver := NewStubDriver() + var bytes uint64 = 100 + sz := szWith(&bytes) + suspendDir := GinkgoT().TempDir() + r := &ImpVMReconciler{Client: k8sClient, NodeName: testNode, Driver: driver, SuspendDir: suspendDir, SZ: sz} + + vm := &impdevv1alpha1.ImpVM{ + ObjectMeta: metav1.ObjectMeta{Name: "tc-sz-wake", Namespace: "default", Finalizers: []string{"imp/finalizer"}}, + Spec: impdevv1alpha1.ImpVMSpec{NodeName: testNode, DesiredState: impdevv1alpha1.VMDesiredStateScaleToZero}, + } + Expect(k8sClient.Create(ctx, vm)).To(Succeed()) + DeferCleanup(func() { k8sClient.Delete(ctx, vm) }) //nolint:errcheck + + base := vm.DeepCopy() + vm.Status.Phase = impdevv1alpha1.VMPhaseSuspended + vm.Status.IP = "192.168.100.62" + Expect(k8sClient.Status().Patch(ctx, vm, client.MergeFrom(base))).To(Succeed()) + + key := types.NamespacedName{Name: "tc-sz-wake", Namespace: "default"} + req := reconcile.Request{NamespacedName: key} + + // No wake yet → stays Suspended. + _, err := r.Reconcile(ctx, req) + Expect(err).NotTo(HaveOccurred()) + updated := &impdevv1alpha1.ImpVM{} + Expect(k8sClient.Get(ctx, key, updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(impdevv1alpha1.VMPhaseSuspended)) + + // A packet arrives for the VM → registry marks it pending. + sz.reg.register("192.168.100.62", vm) + sz.reg.onDstIP("192.168.100.62") + Expect(sz.reg.pending(key)).To(BeTrue()) + + // Suspended + pending wake → Resuming. + _, err = r.Reconcile(ctx, req) + Expect(err).NotTo(HaveOccurred()) + Expect(k8sClient.Get(ctx, key, updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(impdevv1alpha1.VMPhaseResuming)) + + // Resuming → Running; wake state cleared. + _, err = r.Reconcile(ctx, req) + Expect(err).NotTo(HaveOccurred()) + Expect(k8sClient.Get(ctx, key, updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(impdevv1alpha1.VMPhaseRunning)) + Expect(sz.reg.pending(key)).To(BeFalse()) + }) +}) diff --git a/internal/agent/scaletozero_test.go b/internal/agent/scaletozero_test.go new file mode 100644 index 0000000..f7016f6 --- /dev/null +++ b/internal/agent/scaletozero_test.go @@ -0,0 +1,185 @@ +package agent + +import ( + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + impdevv1alpha1 "github.com/syscode-labs/imp/api/v1alpha1" +) + +func sztVM(ns, name string) *impdevv1alpha1.ImpVM { + return &impdevv1alpha1.ImpVM{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name}} +} + +func TestWakeRegistry_MatchAndDedup(t *testing.T) { + reg := newWakeRegistry(8) + vm := sztVM("ns", "web") + key := client.ObjectKeyFromObject(vm) + reg.register("10.0.0.5", vm) + + reg.onDstIP("10.0.0.5") + select { + case ev := <-reg.events: + if ev.Object.GetName() != "web" { + t.Fatalf("event for wrong object: %s", ev.Object.GetName()) + } + default: + t.Fatal("expected a wake event, got none") + } + if !reg.pending(key) { + t.Fatal("expected pending after wake") + } + + // Second packet for the same VM must not enqueue a duplicate. + reg.onDstIP("10.0.0.5") + select { + case <-reg.events: + t.Fatal("expected dedup, got a second event") + default: + } +} + +func TestWakeRegistry_UnknownIP(t *testing.T) { + reg := newWakeRegistry(8) + reg.register("10.0.0.5", sztVM("ns", "web")) + reg.onDstIP("10.0.0.9") // not registered + select { + case <-reg.events: + t.Fatal("unexpected event for unregistered IP") + default: + } +} + +func TestWakeRegistry_Clear(t *testing.T) { + reg := newWakeRegistry(8) + vm := sztVM("ns", "web") + key := client.ObjectKeyFromObject(vm) + reg.register("10.0.0.5", vm) + reg.onDstIP("10.0.0.5") + <-reg.events + + reg.clear(key) + if reg.pending(key) { + t.Fatal("pending should be false after clear") + } + // IP mapping is gone: a later packet must not wake. + reg.onDstIP("10.0.0.5") + select { + case <-reg.events: + t.Fatal("cleared IP should not wake") + default: + } +} + +// A full channel must not lose a wake: signalled stays false so the next packet +// retries once there is room. +func TestWakeRegistry_FullChannelRetries(t *testing.T) { + reg := newWakeRegistry(1) + a, b := sztVM("ns", "a"), sztVM("ns", "b") + reg.register("10.0.0.1", a) + reg.register("10.0.0.2", b) + + reg.onDstIP("10.0.0.1") // fills the buffer of 1 + reg.onDstIP("10.0.0.2") // channel full → dropped, not signalled + if reg.pending(client.ObjectKeyFromObject(b)) { + t.Fatal("b must not be signalled while channel is full") + } + + <-reg.events // drain a's event + reg.onDstIP("10.0.0.2") // retry now succeeds + if !reg.pending(client.ObjectKeyFromObject(b)) { + t.Fatal("b should be signalled after retry") + } +} + +func TestWakeRegistry_ReregisterDropsOldIP(t *testing.T) { + reg := newWakeRegistry(8) + vm := sztVM("ns", "web") + reg.register("10.0.0.5", vm) + reg.register("10.0.0.6", vm) // VM came back with a new IP + + reg.onDstIP("10.0.0.5") // stale IP must not wake + select { + case <-reg.events: + t.Fatal("stale IP should have been dropped on re-register") + default: + } + reg.onDstIP("10.0.0.6") + if !reg.pending(client.ObjectKeyFromObject(vm)) { + t.Fatal("new IP should wake") + } +} + +func TestObserve_IdleAfterTimeout(t *testing.T) { + var bytes uint64 = 100 + sz := newScaleToZero(func(string) (uint64, error) { return bytes, nil }, nil, time.Second, 8) + key := client.ObjectKeyFromObject(sztVM("ns", "web")) + t0 := time.Now() + + if idle, _, _ := sz.observe(key, "tap0", time.Minute, t0); idle { + t.Fatal("first sample must never be idle (grace window)") + } + // Same byte count, within timeout → still not idle. + if idle, _, _ := sz.observe(key, "tap0", time.Minute, t0.Add(30*time.Second)); idle { + t.Fatal("within idleTimeout should not be idle") + } + // Same byte count, past timeout → idle. + if idle, _, _ := sz.observe(key, "tap0", time.Minute, t0.Add(90*time.Second)); !idle { + t.Fatal("past idleTimeout with no traffic should be idle") + } +} + +func TestObserve_TrafficResetsClock(t *testing.T) { + var bytes uint64 = 100 + sz := newScaleToZero(func(string) (uint64, error) { return bytes, nil }, nil, time.Second, 8) + key := client.ObjectKeyFromObject(sztVM("ns", "web")) + t0 := time.Now() + + sz.observe(key, "tap0", time.Minute, t0) + bytes = 500 // traffic happened + idle, lastActivity, _ := sz.observe(key, "tap0", time.Minute, t0.Add(90*time.Second)) + if idle { + t.Fatal("traffic must reset the idle clock") + } + if !lastActivity.Equal(t0.Add(90 * time.Second)) { + t.Fatalf("lastActivity should be the traffic time, got %v", lastActivity) + } + // Now idle from the reset point. + if idle, _, _ := sz.observe(key, "tap0", time.Minute, t0.Add(90*time.Second).Add(2*time.Minute)); !idle { + t.Fatal("should be idle a full timeout after the reset") + } +} + +func TestIdleTimeoutOrDefault(t *testing.T) { + if got := idleTimeoutOrDefault(sztVM("ns", "a")); got != defaultIdleTimeout { + t.Errorf("unset: got %v, want %v", got, defaultIdleTimeout) + } + vm := sztVM("ns", "a") + vm.Spec.IdleTimeout = &metav1.Duration{Duration: 2 * time.Minute} + if got := idleTimeoutOrDefault(vm); got != 2*time.Minute { + t.Errorf("set: got %v, want 2m", got) + } +} + +// Run with -race: concurrent activator (onDstIP) vs reconcile (register/clear/pending). +func TestWakeRegistry_ConcurrentAccess(t *testing.T) { + reg := newWakeRegistry(1024) + vm := sztVM("ns", "web") + key := client.ObjectKeyFromObject(vm) + done := make(chan struct{}) + go func() { + for i := 0; i < 2000; i++ { + reg.onDstIP("10.0.0.5") + } + close(done) + }() + for i := 0; i < 2000; i++ { + reg.register("10.0.0.5", vm) + reg.pending(key) + reg.clear(key) + } + <-done +} diff --git a/internal/webhook/v1alpha1/impvm_webhook.go b/internal/webhook/v1alpha1/impvm_webhook.go index 8428195..34db164 100644 --- a/internal/webhook/v1alpha1/impvm_webhook.go +++ b/internal/webhook/v1alpha1/impvm_webhook.go @@ -114,7 +114,7 @@ func (w *ImpVMWebhook) mergeRestartPolicy(ctx context.Context, vm *impdevv1alpha // ValidateCreate implements admission.Validator[*impdevv1alpha1.ImpVM]. func (w *ImpVMWebhook) ValidateCreate(_ context.Context, vm *impdevv1alpha1.ImpVM) (admission.Warnings, error) { - return nil, validateImpVM(vm).ToAggregate() + return scaleToZeroWarnings(vm), validateImpVM(vm).ToAggregate() } // ValidateUpdate implements admission.Validator[*impdevv1alpha1.ImpVM]. @@ -128,7 +128,18 @@ func (w *ImpVMWebhook) ValidateUpdate(_ context.Context, oldVM, newVM *impdevv1a )) } - return nil, errs.ToAggregate() + return scaleToZeroWarnings(newVM), errs.ToAggregate() +} + +// scaleToZeroWarnings warns that ScaleToZero is experimental: its wake-on-traffic +// path has not yet been validated on real hardware. +func scaleToZeroWarnings(vm *impdevv1alpha1.ImpVM) admission.Warnings { + if vm.Spec.DesiredState == impdevv1alpha1.VMDesiredStateScaleToZero { + return admission.Warnings{ + "desiredState=ScaleToZero is experimental: the wake-on-traffic path is not yet hardware-validated and requires the agent's IMP_SCALE_TO_ZERO opt-in", + } + } + return nil } // ValidateDelete implements admission.Validator[*impdevv1alpha1.ImpVM]. diff --git a/internal/webhook/v1alpha1/impvm_webhook_test.go b/internal/webhook/v1alpha1/impvm_webhook_test.go index ddaa64a..6d7f455 100644 --- a/internal/webhook/v1alpha1/impvm_webhook_test.go +++ b/internal/webhook/v1alpha1/impvm_webhook_test.go @@ -216,10 +216,26 @@ func TestImpVMWebhook_ValidateCreate_IdleTimeoutValid(t *testing.T) { vm.Spec.DesiredState = impdevv1alpha1.VMDesiredStateScaleToZero vm.Spec.IdleTimeout = &metav1.Duration{Duration: 2 * time.Minute} - _, err := wh.ValidateCreate(context.Background(), vm) + warns, err := wh.ValidateCreate(context.Background(), vm) if err != nil { t.Errorf("expected no error for valid ScaleToZero+idleTimeout, got: %v", err) } + if len(warns) == 0 { + t.Error("expected an experimental warning for ScaleToZero, got none") + } +} + +func TestImpVMWebhook_ValidateCreate_NoWarningWithoutScaleToZero(t *testing.T) { + wh := &ImpVMWebhook{} + vm := newVM("", "my-class", "my-image") // desiredState unset (Running) + + warns, err := wh.ValidateCreate(context.Background(), vm) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if len(warns) != 0 { + t.Errorf("expected no warnings for non-ScaleToZero VM, got: %v", warns) + } } func TestImpVMWebhook_ValidateCreate_Valid_TemplateRef(t *testing.T) {