P3: add mesh networking and pod IP routing

cmd/macvz-kubelet/main.go

@@ -102,9 +102,14 @@ func run(ctx context.Context, configPath, runtimeBinary string) error {
 		return fmt.Errorf("build kubernetes client: %w", err)
 	}
 
-	// Construct the provider over the runtime driver. Pod lifecycle methods are
-	// stubbed until #16; this issue only wires the node controller loop.
-	p := provider.New(cfg.NodeName, driver)
+	internalIP := cfg.Node.InternalIP
+	if internalIP == "" {
+		internalIP = detectInternalIP()
+	}
+
+	// Construct the provider over the runtime driver. The node's reachable
+	// address is reported as each Pod's HostIP so Services/`-o wide` resolve it.
+	p := provider.New(cfg.NodeName, driver, provider.WithHostIP(internalIP))
 
 	// Resolve the configured node shape (capacity/taints validated at load).
 	capacity, err := cfg.Capacity()
@@ -115,10 +120,6 @@ func run(ctx context.Context, configPath, runtimeBinary string) error {
 	if err != nil {
 		return fmt.Errorf("node taints: %w", err)
 	}
-	internalIP := cfg.Node.InternalIP
-	if internalIP == "" {
-		internalIP = detectInternalIP()
-	}
 	nodeSpec := provider.NodeSpec{
 		KubeletVersion: version.Version,
 		OS:             cfg.Node.OS,
@@ -197,6 +198,28 @@ func run(ctx context.Context, configPath, runtimeBinary string) error {
 		return nil
 	}
 
+	// Enable coordinated Pod IPAM now that Kubernetes has assigned this node a
+	// Pod CIDR, recovering any existing allocations before Pods are reconciled.
+	if err := setupIPAM(ctx, cfg, clientset, p); err != nil {
+		return fmt.Errorf("setup pod IPAM: %w", err)
+	}
+
+	// Bring up the cross-host WireGuard mesh (if enabled) so remote Pod CIDRs are
+	// routed through encrypted tunnels before Pods start landing on this node.
+	stopMesh, err := setupMesh(ctx, cfg)
+	if err != nil {
+		return fmt.Errorf("setup mesh: %w", err)
+	}
+	defer stopMesh()
+
+	// Start the host Pod network path so each micro-VM is reachable at its Pod IP
+	// across the mesh, and attach it to the provider before Pods are reconciled.
+	stopPodNet, err := setupPodNetwork(ctx, cfg, p)
+	if err != nil {
+		return fmt.Errorf("setup pod network: %w", err)
+	}
+	defer stopPodNet()
+
 	// Start the Pod lifecycle controller so scheduled Pods become micro-VMs.
 	stopPods, err := startPodController(ctx, cfg, clientset, p, runtime.NumCPU())
 	if err != nil {

cmd/macvz-kubelet/serve.go

@@ -9,11 +9,16 @@ import (
 	"time"
 
 	"github.com/chimerakang/macvz/pkg/config"
+	"github.com/chimerakang/macvz/pkg/network"
+	"github.com/chimerakang/macvz/pkg/network/podnet"
+	"github.com/chimerakang/macvz/pkg/network/wireguard"
 	"github.com/chimerakang/macvz/pkg/provider"
 	vknode "github.com/virtual-kubelet/virtual-kubelet/node"
 	vkapi "github.com/virtual-kubelet/virtual-kubelet/node/api"
 	"github.com/virtual-kubelet/virtual-kubelet/node/nodeutil"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -25,6 +30,75 @@ import (
 // informerResync is how often the shared informers do a full relist.
 const informerResync = time.Minute
 
+// podCIDRWaitTimeout bounds how long startup waits for Kubernetes to assign this
+// node a Pod CIDR before continuing without coordinated IPAM.
+const podCIDRWaitTimeout = 30 * time.Second
+
+// setupIPAM enables coordinated Pod IPAM for this node. The address range is the
+// node's Kubernetes-assigned Spec.PodCIDR (or cfg.Node.PodCIDR when set as an
+// override). It then recovers existing allocations from the API server so a
+// restart neither leaks addresses nor reassigns a live Pod's IP.
+//
+// IPAM is best-effort: on a cluster that assigns no Pod CIDR (and with no
+// override configured), it logs and returns nil so Pods still run with the Pod
+// IP derived from the runtime-reported address.
+func setupIPAM(ctx context.Context, cfg config.Config, clientset *kubernetes.Clientset, p *provider.Provider) error {
+	cidr := cfg.Node.PodCIDR
+	if cidr == "" {
+		var err error
+		cidr, err = waitForPodCIDR(ctx, clientset, cfg.NodeName)
+		if err != nil {
+			klog.ErrorS(err, "coordinated Pod IPAM disabled; Pod IPs will come from the runtime",
+				"hint", "run kube-controller-manager with --allocate-node-cidrs or set node.podCIDR")
+			return nil
+		}
+	}
+
+	ipam, err := network.NewPodIPAM(cidr)
+	if err != nil {
+		return fmt.Errorf("build pod IPAM for %q: %w", cidr, err)
+	}
+	p.SetIPAM(ipam)
+	klog.InfoS("coordinated Pod IPAM enabled", "node", cfg.NodeName, "podCIDR", ipam.CIDR())
+
+	// Rebuild allocations from Kubernetes state before the Pod controller runs.
+	podList, err := clientset.CoreV1().Pods(corev1.NamespaceAll).List(ctx, metav1.ListOptions{
+		FieldSelector: fields.OneTermEqualSelector("spec.nodeName", cfg.NodeName).String(),
+	})
+	if err != nil {
+		// A failed recovery list is non-fatal: the allocator starts empty and
+		// re-derives IPs as Pods are (re)created.
+		klog.ErrorS(err, "could not list existing Pods for IPAM recovery", "node", cfg.NodeName)
+		return nil
+	}
+	pods := make([]*corev1.Pod, 0, len(podList.Items))
+	for i := range podList.Items {
+		pods = append(pods, &podList.Items[i])
+	}
+	p.RecoverAllocations(pods)
+	return nil
+}
+
+// waitForPodCIDR polls the node until Kubernetes assigns its Spec.PodCIDR, which
+// happens shortly after registration on clusters with node-CIDR allocation.
+func waitForPodCIDR(ctx context.Context, clientset *kubernetes.Clientset, nodeName string) (string, error) {
+	deadlineCtx, cancel := context.WithTimeout(ctx, podCIDRWaitTimeout)
+	defer cancel()
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		node, err := clientset.CoreV1().Nodes().Get(deadlineCtx, nodeName, metav1.GetOptions{})
+		if err == nil && node.Spec.PodCIDR != "" {
+			return node.Spec.PodCIDR, nil
+		}
+		select {
+		case <-deadlineCtx.Done():
+			return "", fmt.Errorf("node %q has no Spec.PodCIDR after %s", nodeName, podCIDRWaitTimeout)
+		case <-ticker.C:
+		}
+	}
+}
+
 // startPodController wires the Virtual Kubelet pod controller: pod/configmap/
 // secret/service informers (pods filtered to this node), an event recorder, and
 // the controller itself driving the provider's Pod lifecycle. It returns a
@@ -77,6 +151,64 @@ func startPodController(ctx context.Context, cfg config.Config, clientset *kuber
 	return eb.Shutdown, nil
 }
 
+// setupMesh brings up this node's WireGuard mesh when enabled, returning a
+// cleanup func that tears it back down. The mesh encrypts and routes cross-host
+// Pod traffic (issue #21); each peer's Pod CIDR becomes a route through the
+// tunnel. When the mesh is disabled it is a no-op returning a no-op cleanup.
+func setupMesh(ctx context.Context, cfg config.Config) (func(), error) {
+	if !cfg.Mesh.Enabled {
+		klog.InfoS("WireGuard mesh disabled; Pods are reachable only on their local node")
+		return func() {}, nil
+	}
+
+	ifc, err := cfg.MeshInterfaceConfig()
+	if err != nil {
+		return nil, fmt.Errorf("resolve mesh config: %w", err)
+	}
+	mesh := wireguard.New(ifc)
+	if err := mesh.Up(ctx); err != nil {
+		return nil, fmt.Errorf("bring up mesh interface %q: %w", ifc.Name, err)
+	}
+	klog.InfoS("WireGuard mesh up",
+		"interface", mesh.InterfaceName(),
+		"publicKey", ifc.PrivateKey.PublicKey().String(),
+		"peers", mesh.Peers(),
+		"routes", mesh.InstalledRoutes(),
+	)
+
+	return func() {
+		// Tear down with a fresh context: the root ctx is already cancelled by
+		// the time cleanup runs during shutdown.
+		if err := mesh.Down(context.Background()); err != nil {
+			klog.ErrorS(err, "failed to tear down mesh", "interface", ifc.Name)
+		}
+	}, nil
+}
+
+// setupPodNetwork starts the host Pod network path (when enabled) and attaches
+// it to the provider so each micro-VM is reachable at its Pod IP across the mesh
+// (#22). It returns a cleanup func that flushes the host rules. When disabled it
+// is a no-op returning a no-op cleanup.
+func setupPodNetwork(ctx context.Context, cfg config.Config, p *provider.Provider) (func(), error) {
+	if !cfg.PodNetwork.Enabled {
+		klog.InfoS("Pod network path disabled; Pods keep the runtime host-only address")
+		return func() {}, nil
+	}
+
+	router := podnet.New(cfg.PodNetworkRouterConfig())
+	if err := router.Start(ctx); err != nil {
+		return nil, fmt.Errorf("start pod network path: %w", err)
+	}
+	p.SetPodNetwork(router)
+	klog.InfoS("Pod network path started", "interface", cfg.PodNetwork.Interface)
+
+	return func() {
+		if err := router.Stop(context.Background()); err != nil {
+			klog.ErrorS(err, "failed to stop pod network path")
+		}
+	}, nil
+}
+
 // startKubeletServer starts the HTTPS kubelet API used by `kubectl logs`/`exec`,
 // routing to the provider. It is a no-op (returning a no-op cleanup) when no
 // serving certificate is configured, mirroring upstream Virtual Kubelet: Pods
@@ -95,6 +227,7 @@ func startKubeletServer(ctx context.Context, cfg config.Config, p *provider.Prov
 	handler := vkapi.PodHandler(vkapi.PodHandlerConfig{
 		RunInContainer:        p.RunInContainer,
 		GetContainerLogs:      p.GetContainerLogs,
+		PortForward:           p.PortForward,
 		GetPods:               p.GetPods,
 		GetPodsFromKubernetes: p.GetPods,
 	}, false)

config.example.yaml

@@ -35,6 +35,11 @@ node:
   # Node InternalIP. Leave empty to auto-detect the host's primary outbound IPv4.
   internalIP: ""
 
+  # Pod CIDR override for coordinated Pod IPAM. Leave empty to use the
+  # Kubernetes-assigned Node.Spec.PodCIDR. Only set this on clusters that do not
+  # allocate node CIDRs automatically, and give every MacVz node a disjoint CIDR.
+  podCIDR: ""
+
   # Extra labels/annotations merged onto the node (override built-ins on clash).
   labels: {}
   annotations: {}
@@ -61,3 +66,28 @@ node:
   kubeletPort: 10250
   servingTLSCertFile: ""     # e.g. /etc/macvz/pki/kubelet.crt
   servingTLSKeyFile: ""      # e.g. /etc/macvz/pki/kubelet.key
+
+# Cross-host WireGuard mesh. Disabled by default; see docs/NETWORKING.md for a
+# complete two-node setup and pf anchor requirements.
+mesh:
+  enabled: false
+  interface: utun7
+  privateKeyFile: /etc/macvz/wireguard.key
+  address: 10.99.0.1/32
+  listenPort: 51820
+  mtu: 0
+  peers:
+    - name: mac-mini-02
+      publicKey: ""
+      endpoint: 192.168.1.20:51820
+      podCIDR: 10.244.2.0/24
+      address: 10.99.0.2/32
+      persistentKeepalive: 25
+
+# Host Pod network path. Disabled by default; enable with mesh once the
+# apple/container vmnet interface is known on the host.
+podNetwork:
+  enabled: false
+  interface: bridge100
+  anchor: macvz/pods
+  enableForwarding: true

docs/MASTER_TASKS.md

@@ -44,11 +44,11 @@ Source of truth for the phased roadmap. Phases map to GitHub **Milestones**; tas
 | #17 | Translate Kubernetes Pod specs into runtime workload specs | P2 | in progress |
 | #18 | Wire kubectl logs and exec through the runtime | P2 | in progress |
 | #19 | Add RBAC, manifests, and P2 MVP smoke test docs | P2 | in progress |
-| #20 | Implement Kubernetes-coordinated Pod IPAM | P3 | open |
-| #21 | Bring up WireGuard mesh between MacVz nodes | P3 | open |
-| #22 | Connect micro-VM networking to the controllable Pod network path | P3 | open |
-| #23 | Report Pod IPs and readiness so Services resolve across MacVz nodes | P3 | open |
-| #24 | Implement kubectl port-forward for MacVz-backed Pods | P3 | open |
+| #20 | Implement Kubernetes-coordinated Pod IPAM | P3 | closed |
+| #21 | Bring up WireGuard mesh between MacVz nodes | P3 | closed |
+| #22 | Connect micro-VM networking to the controllable Pod network path | P3 | closed |
+| #23 | Report Pod IPs and readiness so Services resolve across MacVz nodes | P3 | closed |
+| #24 | Implement kubectl port-forward for MacVz-backed Pods | P3 | closed |
 | #25 | Implement node and pod metrics reporting | P4 | open |
 | #26 | Support VirtioFS-backed volumes for MacVz Pods | P4 | open |
 | #27 | Handle image architecture and Rosetta-for-Linux behavior | P4 | open |