From 69b62f3ca0c0757e7f87521f2bd3a848d5501d0f Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 10 Jun 2026 08:58:58 +1000 Subject: [PATCH] feat(eks): HA control-plane spread placement (mulga-siv-231.7.1) (#332) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(eks): node-targetable control-plane VM (HA spread transport prereq) Groundwork for HA control-plane placement (mulga-siv-231.7.1): make the k3s control-plane VM placeable on a specific Spinifex host. - Add a sys.medium instance type (2 vCPU / 4 GB) and boot the control-plane VM as it instead of t3.medium. A sys.* type keeps the CP VM out of customer DescribeInstanceTypes and makes the daemon register the node-targeted system.LaunchInstance.{type}.{nodeID} subject (only sys.* types get it). - Add K3sServerInput.TargetNodeID and a LaunchSystemInstanceOnNode method on the launcher: empty/local node launches in-process (unchanged single-CP path); any other node dispatches over the node-targeted subject, where the receiving daemon runs the launch and owns the VM. The spread orchestrator (reserve N hosts, launch per host, verify, rollback, <3-host fallback) lands separately; CreateCluster still launches locally (TargetNodeID empty), so behaviour is unchanged. * feat(eks): HA control-plane spread orchestrator Place N=3 k3s server VMs across distinct Spinifex hosts via a per-cluster spread placement group, with all-or-nothing rollback and single-CP fallback. - placeControlPlane: capacity fan-out (HostScheduler) -> >=3-host gate -> ensure spread group (system account) -> ReserveSpreadNodes(Min=Max=3) -> parallel per-node LaunchK3sServerVM(TargetNodeID) -> node-vms placement verify -> FinalizeSpreadInstances. Any launch/verify/finalize failure terminates launched VMs, releases the reservation, drops the group. - <3 schedulable hosts or a lost reservation race falls back to a single local control-plane VM (prior behaviour). - ClusterMeta gains ControlPlaneNodes[] + ControlPlaneSpreadGroup; [0] is mirrored into the scalar ControlPlane* fields for back-compat. DeleteCluster tears down every CP VM and the spread group. - Wire PlacementGroup + Scheduler into EKSServiceDeps + orchestration-deps gate. * fix(eks): size HA control-plane spread on raw host headroom SchedulableHosts matched the CP type by name in node.status's per-type capacity list, but that list omits sys.* types (system types are hidden from customers). The CP VM is sys.medium, so the match never hit and SchedulableHosts returned 0 on every node — placeControlPlane always fell back to a single CP, making the >=3-host spread unreachable in production. System VMs consume host vCPU/memory like any guest, so size the fit against each node's raw schedulable headroom (Total - Reserved - Alloc) and the type's footprint instead. Adds instancetypes.SpecForSystemType for the sys.* vCPU/memory spec and a pure nodeFitsSystemInstance helper. --- spinifex/daemon/daemon_system_dispatch.go | 47 ++ .../daemon/daemon_system_dispatch_test.go | 61 +++ spinifex/daemon/eks_deps.go | 3 + spinifex/handlers/eks/cluster_state.go | 21 + spinifex/handlers/eks/delete_cluster_test.go | 2 + spinifex/handlers/eks/k3s_ha_control_plane.go | 429 ++++++++++++++++ .../handlers/eks/k3s_ha_control_plane_test.go | 476 ++++++++++++++++++ spinifex/handlers/eks/k3s_server_vm.go | 20 +- spinifex/handlers/eks/k3s_server_vm_test.go | 38 ++ spinifex/handlers/eks/service_impl.go | 43 +- spinifex/instancetypes/definitions.go | 3 +- spinifex/instancetypes/generate.go | 19 + spinifex/instancetypes/generate_test.go | 8 +- 13 files changed, 1153 insertions(+), 17 deletions(-) create mode 100644 spinifex/handlers/eks/k3s_ha_control_plane.go create mode 100644 spinifex/handlers/eks/k3s_ha_control_plane_test.go diff --git a/spinifex/daemon/daemon_system_dispatch.go b/spinifex/daemon/daemon_system_dispatch.go index 2208fcdc6..88ff7c6a0 100644 --- a/spinifex/daemon/daemon_system_dispatch.go +++ b/spinifex/daemon/daemon_system_dispatch.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "log/slog" + "time" "github.com/mulgadc/spinifex/spinifex/awserrors" handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2" @@ -11,6 +12,11 @@ import ( "github.com/nats-io/nats.go" ) +// nodeSystemLaunchTimeout caps a node-targeted system.LaunchInstance round +// trip. Sized for a full-AMI control-plane VM boot (clone root + cloud-init +// seed + QEMU start) on a remote host, well above the local microVM path. +const nodeSystemLaunchTimeout = 5 * time.Minute + // systemInstanceLaunchEnvelope is the wire format for // system.LaunchInstance.{type}[.{nodeID}] replies. Either Output or Error // is set. Encoded as JSON so non-Go consumers stay possible later. @@ -58,6 +64,47 @@ func (d *Daemon) handleSystemLaunchInstance(msg *nats.Msg) { respondWithSystemLaunchOutput(msg, output) } +// LaunchSystemInstanceOnNode launches a system VM on a specific Spinifex host. +// An empty nodeID or the local node runs the launch in-process (matching +// LaunchSystemInstance); any other node is reached over the node-targeted +// system.LaunchInstance.{type}.{nodeID} subject, where the receiving daemon +// runs the launch locally and binds the per-instance terminate subscription +// (handleSystemLaunchInstance). The VM stays bound to the node that runs it. +func (d *Daemon) LaunchSystemInstanceOnNode(nodeID string, input *handlers_elbv2.SystemInstanceInput) (*handlers_elbv2.SystemInstanceOutput, error) { + if nodeID == "" || nodeID == d.node { + return d.LaunchSystemInstance(input) + } + if d.natsConn == nil { + return nil, fmt.Errorf("system instance: cannot target node %s without a NATS connection", nodeID) + } + if input == nil || input.InstanceType == "" { + return nil, fmt.Errorf("system instance input missing InstanceType") + } + + payload, err := json.Marshal(input) + if err != nil { + return nil, fmt.Errorf("marshal SystemInstanceInput: %w", err) + } + + subject := fmt.Sprintf("system.LaunchInstance.%s.%s", input.InstanceType, nodeID) + reply, err := d.natsConn.Request(subject, payload, nodeSystemLaunchTimeout) + if err != nil { + return nil, fmt.Errorf("nats request %s: %w", subject, err) + } + + var env systemInstanceLaunchEnvelope + if err := json.Unmarshal(reply.Data, &env); err != nil { + return nil, fmt.Errorf("decode launch reply: %w", err) + } + if env.Error != "" { + return nil, fmt.Errorf("%s", env.Error) + } + if env.Output == nil { + return nil, fmt.Errorf("launch reply missing output payload") + } + return env.Output, nil +} + // subscribeSystemTerminate registers an owning-node subscription for // system.TerminateInstance.{instanceID}. Idempotent — calling for an // already-bound subscription is a no-op. diff --git a/spinifex/daemon/daemon_system_dispatch_test.go b/spinifex/daemon/daemon_system_dispatch_test.go index a98f390f7..b206a8f2f 100644 --- a/spinifex/daemon/daemon_system_dispatch_test.go +++ b/spinifex/daemon/daemon_system_dispatch_test.go @@ -294,6 +294,67 @@ func TestSubscribeSystemTerminate_Idempotent(t *testing.T) { assert.Same(t, first, second, "second call must be a no-op and keep the original subscription") } +func TestLaunchSystemInstanceOnNode_RemoteRoundTrip(t *testing.T) { + nc, err := nats.Connect(sharedNATSURL) + require.NoError(t, err) + t.Cleanup(nc.Close) + + d := &Daemon{natsConn: nc, node: "node-a", natsSubscriptions: make(map[string]*nats.Subscription)} + + // Stand in for the remote node's owning daemon on the node-targeted subject. + gotCh := make(chan *handlers_elbv2.SystemInstanceInput, 1) + sub, err := nc.Subscribe("system.LaunchInstance.sys.medium.node-b", func(msg *nats.Msg) { + var in handlers_elbv2.SystemInstanceInput + _ = json.Unmarshal(msg.Data, &in) + gotCh <- &in + payload, _ := json.Marshal(systemInstanceLaunchEnvelope{ + Output: &handlers_elbv2.SystemInstanceOutput{InstanceID: "i-remote", PrivateIP: "10.0.2.7"}, + }) + _ = msg.Respond(payload) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + out, err := d.LaunchSystemInstanceOnNode("node-b", &handlers_elbv2.SystemInstanceInput{ + InstanceType: "sys.medium", ImageID: "ami-x", + }) + require.NoError(t, err) + require.NotNil(t, out) + assert.Equal(t, "i-remote", out.InstanceID) + + select { + case in := <-gotCh: + assert.Equal(t, "sys.medium", in.InstanceType, "node-targeted launch carries the CP instance type") + case <-time.After(2 * time.Second): + t.Fatal("remote node never received the node-targeted launch") + } +} + +func TestLaunchSystemInstanceOnNode_RemoteErrorPropagated(t *testing.T) { + nc, err := nats.Connect(sharedNATSURL) + require.NoError(t, err) + t.Cleanup(nc.Close) + + d := &Daemon{natsConn: nc, node: "node-a", natsSubscriptions: make(map[string]*nats.Subscription)} + + sub, err := nc.Subscribe("system.LaunchInstance.sys.medium.node-b", func(msg *nats.Msg) { + payload, _ := json.Marshal(systemInstanceLaunchEnvelope{Error: "insufficient capacity"}) + _ = msg.Respond(payload) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + _, err = d.LaunchSystemInstanceOnNode("node-b", &handlers_elbv2.SystemInstanceInput{InstanceType: "sys.medium"}) + require.Error(t, err) + assert.Contains(t, err.Error(), "insufficient capacity") +} + +func TestLaunchSystemInstanceOnNode_RemoteWithoutConn(t *testing.T) { + d := &Daemon{node: "node-a"} + _, err := d.LaunchSystemInstanceOnNode("node-b", &handlers_elbv2.SystemInstanceInput{InstanceType: "sys.medium"}) + require.Error(t, err, "remote placement requires a NATS connection") +} + // msgWithConn returns msg with its internal connection set so msg.Respond // publishes through nc. nats.Msg.Respond requires a backing connection set // either via Subscribe delivery or by hand for synthetic test messages. diff --git a/spinifex/daemon/eks_deps.go b/spinifex/daemon/eks_deps.go index 6284e3150..6a93b8aae 100644 --- a/spinifex/daemon/eks_deps.go +++ b/spinifex/daemon/eks_deps.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" + handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" "log/slog" @@ -62,6 +63,8 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps { Image: d.imageService, EIP: d.eipService, Worker: d, + PlacementGroup: handlers_ec2_placementgroup.NewNATSPlacementGroupService(d.natsConn), + Scheduler: handlers_eks.NewNATSHostScheduler(d.natsConn), } } diff --git a/spinifex/handlers/eks/cluster_state.go b/spinifex/handlers/eks/cluster_state.go index dcea217a6..aa37114c9 100644 --- a/spinifex/handlers/eks/cluster_state.go +++ b/spinifex/handlers/eks/cluster_state.go @@ -65,6 +65,16 @@ type ClusterMeta struct { // resolve or route to the VPC-internal NLB DNS endpoint, so it probes // /healthz on this host-reachable address instead. ControlPlaneMgmtIP string `json:"controlPlaneMgmtIp,omitempty"` + // ControlPlaneNodes lists the placed control-plane server VMs. HA spread + // holds one entry per distinct host; the single-CP path holds one. [0] is the + // primary the NLB target + egress SNAT wire to until per-node NLB + // registration (231.7.3). The scalar ControlPlane* fields above mirror [0] + // for readers that predate this field (reconciler, teardown) and for clusters + // persisted before HA spread existed (empty ControlPlaneNodes). + ControlPlaneNodes []ControlPlaneNode `json:"controlPlaneNodes,omitempty"` + // ControlPlaneSpreadGroup is the spread placement-group name reserving the CP + // hosts; "" for the single-CP fallback. DeleteCluster releases + deletes it. + ControlPlaneSpreadGroup string `json:"controlPlaneSpreadGroup,omitempty"` // EgressEIPAllocationID / EgressEIPPublicIP track the hidden pool address // SNAT'd to the control-plane VM for egress-only internet (image pulls). // Released + the snat removed on DeleteCluster. @@ -93,6 +103,17 @@ type ClusterMeta struct { BuiltinIngress bool `json:"builtinIngress,omitempty"` } +// ControlPlaneNode identifies one placed control-plane server VM and the host +// it landed on. NodeID is the Spinifex host — distinct per entry under HA +// spread, empty for a single control plane launched on the local node. +type ControlPlaneNode struct { + NodeID string `json:"nodeId,omitempty"` + InstanceID string `json:"instanceId"` + ENIID string `json:"eniId,omitempty"` + ENIIP string `json:"eniIp,omitempty"` + MgmtIP string `json:"mgmtIp,omitempty"` +} + // ErrClusterNotFound is returned by GetClusterMeta / SetClusterStatus / // DeleteClusterPrefix when the cluster meta key is absent. Callers translate // to the AWS shape (ResourceNotFoundException) at the service boundary. diff --git a/spinifex/handlers/eks/delete_cluster_test.go b/spinifex/handlers/eks/delete_cluster_test.go index 66b18593f..01af55477 100644 --- a/spinifex/handlers/eks/delete_cluster_test.go +++ b/spinifex/handlers/eks/delete_cluster_test.go @@ -71,6 +71,8 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture { Image: ami, EIP: eip, Worker: worker, + PlacementGroup: &fakePlacer{}, + Scheduler: &fakeHostScheduler{}, }) require.NoError(t, err) t.Cleanup(svc.Shutdown) diff --git a/spinifex/handlers/eks/k3s_ha_control_plane.go b/spinifex/handlers/eks/k3s_ha_control_plane.go new file mode 100644 index 000000000..4cf1160a8 --- /dev/null +++ b/spinifex/handlers/eks/k3s_ha_control_plane.go @@ -0,0 +1,429 @@ +package handlers_eks + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/admin" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" + "github.com/mulgadc/spinifex/spinifex/instancetypes" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/nats-io/nats.go" +) + +// haControlPlaneCount is the control-plane spread width: an odd quorum of three +// so an embedded-etcd control plane tolerates a single host loss. Fewer than +// this many schedulable hosts falls back to a single control-plane VM (today's +// behaviour) instead of failing the create. +const haControlPlaneCount = 3 + +// controlPlanePlacer is the spread-placement surface the HA control-plane +// orchestrator needs. handlers/ec2/placementgroup.PlacementGroupService (wired +// by the daemon as NewNATSPlacementGroupService) satisfies it. +type controlPlanePlacer interface { + CreatePlacementGroup(*ec2.CreatePlacementGroupInput, string) (*ec2.CreatePlacementGroupOutput, error) + DeletePlacementGroup(*ec2.DeletePlacementGroupInput, string) (*ec2.DeletePlacementGroupOutput, error) + ReserveSpreadNodes(*handlers_ec2_placementgroup.ReserveSpreadNodesInput, string) (*handlers_ec2_placementgroup.ReserveSpreadNodesOutput, error) + ReleaseSpreadNodes(*handlers_ec2_placementgroup.ReleaseSpreadNodesInput, string) (*handlers_ec2_placementgroup.ReleaseSpreadNodesOutput, error) + FinalizeSpreadInstances(*handlers_ec2_placementgroup.FinalizeSpreadInstancesInput, string) (*handlers_ec2_placementgroup.FinalizeSpreadInstancesOutput, error) + RemoveInstance(*handlers_ec2_placementgroup.RemoveInstanceInput, string) (*handlers_ec2_placementgroup.RemoveInstanceOutput, error) +} + +var _ controlPlanePlacer = (handlers_ec2_placementgroup.PlacementGroupService)(nil) + +// HostScheduler answers the capacity + placement fan-out questions the HA +// control-plane orchestrator asks. The daemon wires a NATS implementation; unit +// tests fake it so the orchestration logic needs no live cluster. +type HostScheduler interface { + // SchedulableHosts returns the distinct node IDs that can fit at least one + // VM of instanceType. + SchedulableHosts(instanceType string) []string + // InstanceHosts maps each instance ID to the node currently hosting it. + // Instances absent from the result are not (yet) visible to the fan-out. + InstanceHosts(instanceIDs []string) map[string]string +} + +// placeControlPlane places the cluster's control-plane VM(s). With at least +// haControlPlaneCount schedulable hosts it spreads that many k3s server VMs +// across distinct hosts, all-or-nothing; otherwise it launches a single +// control-plane VM on the local node (today's behaviour). Returns the placed +// nodes ([0] is the primary the NLB + egress wire to until 231.7.3) and the +// spread placement-group name ("" for the single-CP fallback) the caller +// persists for DeleteCluster teardown. +func (s *EKSServiceImpl) placeControlPlane(accountID, clusterName string, tmpl K3sServerInput) ([]ControlPlaneNode, string, error) { + instanceType := tmpl.InstanceType + if instanceType == "" { + instanceType = defaultK3sServerInstanceType + } + + hosts := s.deps.Scheduler.SchedulableHosts(instanceType) + if len(hosts) < haControlPlaneCount { + slog.Info("placeControlPlane: insufficient hosts for HA spread, launching single control plane", + "cluster", clusterName, "schedulableHosts", len(hosts), "want", haControlPlaneCount) + return s.launchSingleControlPlane(tmpl) + } + + pgAccount := admin.SystemAccountID() + groupName := haSpreadGroupName(accountID, clusterName) + if err := s.ensureSpreadGroup(groupName, pgAccount); err != nil { + return nil, "", err + } + + reserve, err := s.deps.PlacementGroup.ReserveSpreadNodes(&handlers_ec2_placementgroup.ReserveSpreadNodesInput{ + GroupName: groupName, + EligibleNodes: hosts, + MinCount: haControlPlaneCount, + MaxCount: haControlPlaneCount, + }, pgAccount) + if err != nil { + // Could not reserve the full quorum of distinct hosts (capacity raced + // away between the status fan-out and the CAS reservation). Drop the + // group and fall back to a single control plane rather than failing the + // whole create. + slog.Warn("placeControlPlane: spread reservation failed, falling back to single control plane", + "cluster", clusterName, "group", groupName, "err", err) + s.deleteSpreadGroup(groupName, pgAccount) + return s.launchSingleControlPlane(tmpl) + } + reserved := reserve.ReservedNodes + + results := s.launchControlPlaneSpread(tmpl, reserved) + + var launched []ControlPlaneNode + var launchErrs []error + for _, r := range results { + if r.err != nil { + launchErrs = append(launchErrs, fmt.Errorf("node %s: %w", r.node, r.err)) + continue + } + launched = append(launched, controlPlaneNode(r.node, r.out)) + } + + // All-or-nothing: any launch failure rolls back every VM that did come up, + // releases the reservation, drops the group, and fails the create. + if len(launchErrs) > 0 { + slog.Error("placeControlPlane: partial spread launch, rolling back", + "cluster", clusterName, "launched", len(launched), "failed", len(launchErrs)) + s.rollbackControlPlaneSpread(accountID, groupName, pgAccount, launched, reserved) + return nil, "", fmt.Errorf("eks: HA control-plane launch failed: %w", errors.Join(launchErrs...)) + } + + // Confirm each VM actually landed on its reserved host (acceptance: verified + // via the node-VM fan-out). The node-targeted launch subject already pins + // placement, so an instance not yet visible to the fan-out is tolerated; + // only a definitive wrong-host placement triggers rollback. + if err := s.verifyControlPlaneSpread(launched); err != nil { + slog.Error("placeControlPlane: placement verification failed, rolling back", + "cluster", clusterName, "err", err) + s.rollbackControlPlaneSpread(accountID, groupName, pgAccount, launched, reserved) + return nil, "", err + } + + nodeInstances := make(map[string][]string, len(launched)) + for _, n := range launched { + nodeInstances[n.NodeID] = []string{n.InstanceID} + } + if _, err := s.deps.PlacementGroup.FinalizeSpreadInstances(&handlers_ec2_placementgroup.FinalizeSpreadInstancesInput{ + GroupName: groupName, + NodeInstances: nodeInstances, + }, pgAccount); err != nil { + slog.Error("placeControlPlane: finalize failed, rolling back", "cluster", clusterName, "err", err) + s.rollbackControlPlaneSpread(accountID, groupName, pgAccount, launched, reserved) + return nil, "", fmt.Errorf("eks: finalize HA control-plane placement: %w", err) + } + + slog.Info("placeControlPlane: HA control plane placed", + "cluster", clusterName, "group", groupName, "nodes", reserved) + return launched, groupName, nil +} + +// launchSingleControlPlane launches one control-plane VM on the local node, the +// pre-HA behaviour. NodeID is empty: the local launch is node-blind. +func (s *EKSServiceImpl) launchSingleControlPlane(tmpl K3sServerInput) ([]ControlPlaneNode, string, error) { + in := tmpl + in.TargetNodeID = "" + out, err := LaunchK3sServerVM(s.deps.VPCK3s, s.deps.Instance, s.deps.Image, in) + if err != nil { + return nil, "", err + } + return []ControlPlaneNode{controlPlaneNode("", out)}, "", nil +} + +type cpLaunchResult struct { + node string + out *K3sServerOutput + err error +} + +// launchControlPlaneSpread launches one server VM per reserved node in parallel +// (each remote launch blocks on a full-AMI boot round trip, so sequential would +// be N×). Results stay index-aligned with nodes. +func (s *EKSServiceImpl) launchControlPlaneSpread(tmpl K3sServerInput, nodes []string) []cpLaunchResult { + results := make([]cpLaunchResult, len(nodes)) + var wg sync.WaitGroup + for i, node := range nodes { + wg.Add(1) + go func(idx int, nodeID string) { + defer wg.Done() + in := tmpl + in.TargetNodeID = nodeID + out, err := LaunchK3sServerVM(s.deps.VPCK3s, s.deps.Instance, s.deps.Image, in) + results[idx] = cpLaunchResult{node: nodeID, out: out, err: err} + }(i, node) + } + wg.Wait() + return results +} + +// verifyControlPlaneSpread confirms every launched VM sits on its reserved host +// and that no two share a host. A VM not yet visible to the fan-out is tolerated +// (listing lag — the node-targeted launch subject already guarantees the host). +func (s *EKSServiceImpl) verifyControlPlaneSpread(nodes []ControlPlaneNode) error { + ids := make([]string, 0, len(nodes)) + for _, n := range nodes { + ids = append(ids, n.InstanceID) + } + hosts := s.deps.Scheduler.InstanceHosts(ids) + + seen := make(map[string]string, len(nodes)) // host -> instanceID + for _, n := range nodes { + actual, ok := hosts[n.InstanceID] + if !ok { + slog.Warn("verifyControlPlaneSpread: instance not yet visible in node fan-out", + "instanceId", n.InstanceID, "expectedNode", n.NodeID) + continue + } + if actual != n.NodeID { + return fmt.Errorf("eks: control-plane VM %s landed on %s, expected %s", n.InstanceID, actual, n.NodeID) + } + if other, dup := seen[actual]; dup { + return fmt.Errorf("eks: control-plane VMs %s and %s share host %s", other, n.InstanceID, actual) + } + seen[actual] = n.InstanceID + } + return nil +} + +// rollbackControlPlaneSpread tears down a partial HA placement: terminate every +// VM that launched, release the reservation (clears the empty placeholders — +// the record was never finalized on any rollback path), and drop the group. +// Best-effort: a stranded internal placement group is harmless and re-creatable. +func (s *EKSServiceImpl) rollbackControlPlaneSpread(accountID, groupName, pgAccount string, launched []ControlPlaneNode, reserved []string) { + for _, n := range launched { + if err := TerminateK3sServerVM(s.deps.VPCK3s, s.deps.Instance, accountID, n.InstanceID, n.ENIID); err != nil { + slog.Warn("rollbackControlPlaneSpread: terminate failed", "instanceId", n.InstanceID, "err", err) + } + } + if _, err := s.deps.PlacementGroup.ReleaseSpreadNodes(&handlers_ec2_placementgroup.ReleaseSpreadNodesInput{ + GroupName: groupName, + Nodes: reserved, + }, pgAccount); err != nil { + slog.Warn("rollbackControlPlaneSpread: release nodes failed", "group", groupName, "err", err) + } + s.deleteSpreadGroup(groupName, pgAccount) +} + +// ensureSpreadGroup creates the per-cluster spread placement group. A duplicate +// (a prior attempt left it behind) is reused — ReserveSpreadNodes filters +// already-occupied hosts, so reserving against a stale group is safe. +func (s *EKSServiceImpl) ensureSpreadGroup(groupName, pgAccount string) error { + _, err := s.deps.PlacementGroup.CreatePlacementGroup(&ec2.CreatePlacementGroupInput{ + GroupName: aws.String(groupName), + Strategy: aws.String(ec2.PlacementStrategySpread), + }, pgAccount) + if err == nil { + return nil + } + if awserrors.IsErrorCode(err, awserrors.ErrorInvalidPlacementGroupDuplicate) { + slog.Info("ensureSpreadGroup: spread group exists, reusing", "group", groupName) + return nil + } + return fmt.Errorf("eks: create spread placement group %s: %w", groupName, err) +} + +func (s *EKSServiceImpl) deleteSpreadGroup(groupName, pgAccount string) { + if _, err := s.deps.PlacementGroup.DeletePlacementGroup(&ec2.DeletePlacementGroupInput{ + GroupName: aws.String(groupName), + }, pgAccount); err != nil { + slog.Warn("deleteSpreadGroup: delete failed", "group", groupName, "err", err) + } +} + +// teardownSpreadGroup removes the cluster's CP instances from the spread group +// and deletes it. Best-effort: a leaked internal group strands nothing billable. +// Runs only for HA clusters (ControlPlaneSpreadGroup set). +func (s *EKSServiceImpl) teardownSpreadGroup(meta *ClusterMeta) { + if meta.ControlPlaneSpreadGroup == "" { + return + } + pgAccount := admin.SystemAccountID() + for _, cp := range meta.ControlPlaneNodes { + if cp.NodeID == "" || cp.InstanceID == "" { + continue + } + if _, err := s.deps.PlacementGroup.RemoveInstance(&handlers_ec2_placementgroup.RemoveInstanceInput{ + GroupName: meta.ControlPlaneSpreadGroup, + NodeName: cp.NodeID, + InstanceID: cp.InstanceID, + }, pgAccount); err != nil { + slog.Warn("teardownSpreadGroup: remove instance failed", + "group", meta.ControlPlaneSpreadGroup, "instanceId", cp.InstanceID, "err", err) + } + } + s.deleteSpreadGroup(meta.ControlPlaneSpreadGroup, pgAccount) +} + +// controlPlaneTeardownNodes returns the control-plane VMs DeleteCluster must +// tear down. It prefers the multi-node list; for clusters persisted before that +// field existed it synthesizes a single entry from the scalar fields. +func controlPlaneTeardownNodes(meta *ClusterMeta) []ControlPlaneNode { + if len(meta.ControlPlaneNodes) > 0 { + return meta.ControlPlaneNodes + } + if meta.ControlPlaneInstanceID == "" && meta.ControlPlaneENIID == "" { + return nil + } + return []ControlPlaneNode{{ + InstanceID: meta.ControlPlaneInstanceID, + ENIID: meta.ControlPlaneENIID, + ENIIP: meta.ControlPlaneENIIP, + MgmtIP: meta.ControlPlaneMgmtIP, + }} +} + +// haSpreadGroupName namespaces the spread group by account + cluster. The group +// lives under the system account (admin.SystemAccountID) so it never surfaces in +// a customer's DescribePlacementGroups; the account therefore cannot +// disambiguate same-named clusters across tenants, so it is folded into the name. +func haSpreadGroupName(accountID, clusterName string) string { + return "eks-cp-" + accountID + "-" + clusterName +} + +func controlPlaneNode(nodeID string, out *K3sServerOutput) ControlPlaneNode { + return ControlPlaneNode{ + NodeID: nodeID, + InstanceID: out.InstanceID, + ENIID: out.ENIID, + ENIIP: out.ENIIP, + MgmtIP: out.MgmtIP, + } +} + +// --- NATS host scheduler --- + +var _ HostScheduler = (*natsHostScheduler)(nil) + +// hostFanoutTimeout bounds a node status / node VMs fan-out. Sized for a LAN +// round trip to every daemon; CreateCluster is already multi-second so a fixed +// short collection window trades a little latency for reliably seeing every host +// (a missed host would wrongly drop the cluster to single-CP). +const hostFanoutTimeout = time.Second + +type natsHostScheduler struct { + nc *nats.Conn +} + +// NewNATSHostScheduler builds the NATS fan-out HostScheduler the daemon wires +// into EKSServiceDeps. +func NewNATSHostScheduler(nc *nats.Conn) HostScheduler { + return &natsHostScheduler{nc: nc} +} + +func (h *natsHostScheduler) SchedulableHosts(instanceType string) []string { + // The control-plane VM is a system type (sys.*), which node.status omits from + // its per-type capacity list (system types are hidden from customers). A + // system VM still consumes host vCPU/memory like any guest, so size the fit + // against each node's raw schedulable headroom and the type's footprint. + vcpu, memGB, ok := instancetypes.SpecForSystemType(instanceType) + if !ok { + slog.Warn("SchedulableHosts: unknown system instance type, no schedulable hosts", + "instanceType", instanceType) + return nil + } + + var hosts []string + seen := make(map[string]bool) + h.fanout("spinifex.node.status", func(data []byte) { + var st types.NodeStatusResponse + if json.Unmarshal(data, &st) != nil || st.Node == "" || seen[st.Node] { + return + } + if nodeFitsSystemInstance(st, vcpu, memGB) { + seen[st.Node] = true + hosts = append(hosts, st.Node) + } + }) + return hosts +} + +// nodeFitsSystemInstance reports whether a node's schedulable headroom +// (Total - Reserved - Alloc, the same arithmetic node.status documents for guest +// scheduling) admits at least one VM of the given vCPU/memory footprint. +func nodeFitsSystemInstance(st types.NodeStatusResponse, vcpu int, memGB float64) bool { + remainVCPU := st.TotalVCPU - st.ReservedVCPU - st.AllocVCPU + remainMem := st.TotalMemGB - st.ReservedMemGB - st.AllocMemGB + return remainVCPU >= vcpu && remainMem >= memGB +} + +func (h *natsHostScheduler) InstanceHosts(instanceIDs []string) map[string]string { + want := make(map[string]bool, len(instanceIDs)) + for _, id := range instanceIDs { + want[id] = true + } + out := make(map[string]string) + h.fanout("spinifex.node.vms", func(data []byte) { + var resp types.NodeVMsResponse + if json.Unmarshal(data, &resp) != nil || resp.Node == "" { + return + } + for _, vm := range resp.VMs { + if want[vm.InstanceID] { + out[vm.InstanceID] = resp.Node + } + } + }) + return out +} + +// fanout publishes an empty request on subject and invokes handle for every +// reply that arrives within hostFanoutTimeout. +func (h *natsHostScheduler) fanout(subject string, handle func([]byte)) { + if h.nc == nil { + return + } + inbox := nats.NewInbox() + sub, err := h.nc.SubscribeSync(inbox) + if err != nil { + slog.Warn("hostScheduler: subscribe inbox failed", "subject", subject, "err", err) + return + } + defer func() { _ = sub.Unsubscribe() }() + + msg := nats.NewMsg(subject) + msg.Reply = inbox + msg.Data = []byte("{}") + if err := h.nc.PublishMsg(msg); err != nil { + slog.Warn("hostScheduler: publish failed", "subject", subject, "err", err) + return + } + + deadline := time.Now().Add(hostFanoutTimeout) + for { + remaining := time.Until(deadline) + if remaining <= 0 { + return + } + reply, err := sub.NextMsg(remaining) + if err != nil { + return + } + handle(reply.Data) + } +} diff --git a/spinifex/handlers/eks/k3s_ha_control_plane_test.go b/spinifex/handlers/eks/k3s_ha_control_plane_test.go new file mode 100644 index 000000000..76e3475fc --- /dev/null +++ b/spinifex/handlers/eks/k3s_ha_control_plane_test.go @@ -0,0 +1,476 @@ +package handlers_eks + +import ( + "errors" + "fmt" + "strings" + "sync" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" + "github.com/mulgadc/spinifex/spinifex/handlers/sysinstance" + "github.com/mulgadc/spinifex/spinifex/instancetypes" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- HA control-plane orchestrator test doubles --- + +// fakeHostScheduler fakes the capacity + placement fan-out. SchedulableHosts +// returns the configured hosts; InstanceHosts maps each instance ID back to its +// node by the "i-" convention seqK3sInst launches with, unless overridden +// by notVisible (listing lag) or wrongHost (placement mismatch). +type fakeHostScheduler struct { + hosts []string + notVisible map[string]bool + wrongHost map[string]string +} + +var _ HostScheduler = (*fakeHostScheduler)(nil) + +func (f *fakeHostScheduler) SchedulableHosts(string) []string { return f.hosts } + +func (f *fakeHostScheduler) InstanceHosts(ids []string) map[string]string { + out := make(map[string]string) + for _, id := range ids { + if f.notVisible[id] { + continue + } + if h, ok := f.wrongHost[id]; ok { + out[id] = h + continue + } + out[id] = strings.TrimPrefix(id, "i-") + } + return out +} + +// seqK3sVPC issues a distinct ENI per call so a 3-node spread yields 3 ENIs. +type seqK3sVPC struct { + mu sync.Mutex + n int + deleted []string +} + +var _ k3sVPCProvisioner = (*seqK3sVPC)(nil) + +func (v *seqK3sVPC) CreateNetworkInterface(in *ec2.CreateNetworkInterfaceInput, _ string) (*ec2.CreateNetworkInterfaceOutput, error) { + v.mu.Lock() + defer v.mu.Unlock() + v.n++ + return &ec2.CreateNetworkInterfaceOutput{NetworkInterface: &ec2.NetworkInterface{ + NetworkInterfaceId: aws.String(fmt.Sprintf("eni-%03d", v.n)), + PrivateIpAddress: aws.String(fmt.Sprintf("10.0.1.%d", v.n+10)), + SubnetId: in.SubnetId, + }}, nil +} + +func (v *seqK3sVPC) DeleteNetworkInterface(in *ec2.DeleteNetworkInterfaceInput, _ string) (*ec2.DeleteNetworkInterfaceOutput, error) { + v.mu.Lock() + defer v.mu.Unlock() + v.deleted = append(v.deleted, aws.StringValue(in.NetworkInterfaceId)) + return &ec2.DeleteNetworkInterfaceOutput{}, nil +} + +// seqK3sInst returns instance ID "i-" so launches map deterministically +// back to their target host. failNodes makes a node's launch return an error. +type seqK3sInst struct { + mu sync.Mutex + nodes []string + terminated []string + failNodes map[string]bool +} + +var _ k3sInstanceLauncher = (*seqK3sInst)(nil) + +func (i *seqK3sInst) LaunchSystemInstance(in *sysinstance.SystemInstanceInput) (*sysinstance.SystemInstanceOutput, error) { + return i.LaunchSystemInstanceOnNode("", in) +} + +func (i *seqK3sInst) LaunchSystemInstanceOnNode(nodeID string, _ *sysinstance.SystemInstanceInput) (*sysinstance.SystemInstanceOutput, error) { + i.mu.Lock() + defer i.mu.Unlock() + if i.failNodes[nodeID] { + return nil, errors.New("InsufficientInstanceCapacity") + } + i.nodes = append(i.nodes, nodeID) + id := "i-" + nodeID + if nodeID == "" { + id = "i-local" + } + return &sysinstance.SystemInstanceOutput{InstanceID: id, MgmtIP: "10.255.0.9"}, nil +} + +func (i *seqK3sInst) TerminateSystemInstance(instanceID string) error { + i.mu.Lock() + defer i.mu.Unlock() + i.terminated = append(i.terminated, instanceID) + return nil +} + +// fakePlacer records every spread-placement call and returns configurable +// reservation/finalize outcomes. Reserve defaults to the first MaxCount eligible +// nodes when reserved is unset. +type fakePlacer struct { + mu sync.Mutex + + createGroups []string + deleteGroups []string + reserveInputs []*handlers_ec2_placementgroup.ReserveSpreadNodesInput + releaseInputs []*handlers_ec2_placementgroup.ReleaseSpreadNodesInput + finalizeInputs []*handlers_ec2_placementgroup.FinalizeSpreadInstancesInput + removeInputs []*handlers_ec2_placementgroup.RemoveInstanceInput + + reserved []string + createErr error + reserveErr error + finalizeErr error +} + +var _ controlPlanePlacer = (*fakePlacer)(nil) + +func (p *fakePlacer) CreatePlacementGroup(in *ec2.CreatePlacementGroupInput, _ string) (*ec2.CreatePlacementGroupOutput, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.createGroups = append(p.createGroups, aws.StringValue(in.GroupName)) + if p.createErr != nil { + return nil, p.createErr + } + return &ec2.CreatePlacementGroupOutput{}, nil +} + +func (p *fakePlacer) DeletePlacementGroup(in *ec2.DeletePlacementGroupInput, _ string) (*ec2.DeletePlacementGroupOutput, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.deleteGroups = append(p.deleteGroups, aws.StringValue(in.GroupName)) + return &ec2.DeletePlacementGroupOutput{}, nil +} + +func (p *fakePlacer) ReserveSpreadNodes(in *handlers_ec2_placementgroup.ReserveSpreadNodesInput, _ string) (*handlers_ec2_placementgroup.ReserveSpreadNodesOutput, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.reserveInputs = append(p.reserveInputs, in) + if p.reserveErr != nil { + return nil, p.reserveErr + } + reserved := p.reserved + if reserved == nil { + n := min(in.MaxCount, len(in.EligibleNodes)) + reserved = append([]string(nil), in.EligibleNodes[:n]...) + } + return &handlers_ec2_placementgroup.ReserveSpreadNodesOutput{ReservedNodes: reserved}, nil +} + +func (p *fakePlacer) ReleaseSpreadNodes(in *handlers_ec2_placementgroup.ReleaseSpreadNodesInput, _ string) (*handlers_ec2_placementgroup.ReleaseSpreadNodesOutput, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.releaseInputs = append(p.releaseInputs, in) + return &handlers_ec2_placementgroup.ReleaseSpreadNodesOutput{}, nil +} + +func (p *fakePlacer) FinalizeSpreadInstances(in *handlers_ec2_placementgroup.FinalizeSpreadInstancesInput, _ string) (*handlers_ec2_placementgroup.FinalizeSpreadInstancesOutput, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.finalizeInputs = append(p.finalizeInputs, in) + if p.finalizeErr != nil { + return nil, p.finalizeErr + } + return &handlers_ec2_placementgroup.FinalizeSpreadInstancesOutput{}, nil +} + +func (p *fakePlacer) RemoveInstance(in *handlers_ec2_placementgroup.RemoveInstanceInput, _ string) (*handlers_ec2_placementgroup.RemoveInstanceOutput, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.removeInputs = append(p.removeInputs, in) + return &handlers_ec2_placementgroup.RemoveInstanceOutput{}, nil +} + +// seqK3sAMI resolves the eks-server AMI; concurrency-safe so the parallel +// spread launches can each call DescribeImages without racing. +type seqK3sAMI struct { + mu sync.Mutex + inner fakeK3sAMI +} + +var _ k3sAMIResolver = (*seqK3sAMI)(nil) + +func (a *seqK3sAMI) DescribeImages(in *ec2.DescribeImagesInput, accountID string) (*ec2.DescribeImagesOutput, error) { + a.mu.Lock() + defer a.mu.Unlock() + return a.inner.DescribeImages(in, accountID) +} + +func newPlacerService(sched HostScheduler, placer controlPlanePlacer, vpc k3sVPCProvisioner, inst k3sInstanceLauncher) *EKSServiceImpl { + return &EKSServiceImpl{deps: EKSServiceDeps{ + VPCK3s: vpc, + Instance: inst, + Image: &seqK3sAMI{}, + Scheduler: sched, + PlacementGroup: placer, + }} +} + +func nodeIDs(nodes []ControlPlaneNode) []string { + out := make([]string, 0, len(nodes)) + for _, n := range nodes { + out = append(out, n.NodeID) + } + return out +} + +const testHAAccountID = "111122223333" + +func TestPlaceControlPlane_SpreadHappyPath(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b", "node-c", "node-d"}} + placer := &fakePlacer{reserved: []string{"node-a", "node-b", "node-c"}} + vpc := &seqK3sVPC{} + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, vpc, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.NoError(t, err) + assert.Equal(t, "eks-cp-"+testHAAccountID+"-alpha", group) + require.Len(t, nodes, 3) + + assert.ElementsMatch(t, []string{"node-a", "node-b", "node-c"}, nodeIDs(nodes)) + enis := map[string]bool{} + for _, n := range nodes { + assert.Equal(t, "i-"+n.NodeID, n.InstanceID) + assert.NotEmpty(t, n.ENIID) + enis[n.ENIID] = true + } + assert.Len(t, enis, 3, "each spread VM gets a distinct ENI") + + assert.Equal(t, []string{group}, placer.createGroups) + require.Len(t, placer.reserveInputs, 1) + assert.Equal(t, haControlPlaneCount, placer.reserveInputs[0].MinCount) + assert.Equal(t, haControlPlaneCount, placer.reserveInputs[0].MaxCount) + require.Len(t, placer.finalizeInputs, 1) + assert.Len(t, placer.finalizeInputs[0].NodeInstances, 3) + assert.Empty(t, placer.releaseInputs) + assert.Empty(t, placer.deleteGroups) + + assert.ElementsMatch(t, []string{"node-a", "node-b", "node-c"}, inst.nodes) + assert.Empty(t, inst.terminated) +} + +func TestPlaceControlPlane_FallbackUnderThreeHosts(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b"}} + placer := &fakePlacer{} + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.NoError(t, err) + assert.Equal(t, "", group) + require.Len(t, nodes, 1) + assert.Equal(t, "", nodes[0].NodeID) + assert.Equal(t, "i-local", nodes[0].InstanceID) + + assert.Empty(t, placer.createGroups) + assert.Empty(t, placer.reserveInputs) + assert.Equal(t, []string{""}, inst.nodes) +} + +func TestPlaceControlPlane_BoundaryExactlyThree(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b", "node-c"}} + placer := &fakePlacer{} + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.NoError(t, err) + assert.Equal(t, "eks-cp-"+testHAAccountID+"-alpha", group) + require.Len(t, nodes, 3) + require.Len(t, placer.finalizeInputs, 1) + assert.ElementsMatch(t, []string{"node-a", "node-b", "node-c"}, inst.nodes) +} + +func TestPlaceControlPlane_PartialLaunchRollsBack(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b", "node-c"}} + placer := &fakePlacer{reserved: []string{"node-a", "node-b", "node-c"}} + inst := &seqK3sInst{failNodes: map[string]bool{"node-c": true}} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.Error(t, err) + assert.Nil(t, nodes) + assert.Equal(t, "", group) + + assert.Len(t, inst.terminated, 2, "the two VMs that launched are terminated") + require.Len(t, placer.releaseInputs, 1) + assert.ElementsMatch(t, []string{"node-a", "node-b", "node-c"}, placer.releaseInputs[0].Nodes) + assert.Empty(t, placer.finalizeInputs) + assert.Len(t, placer.deleteGroups, 1) +} + +func TestPlaceControlPlane_ReserveFailureFallsBackToSingle(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b", "node-c"}} + placer := &fakePlacer{reserveErr: errors.New(awserrors.ErrorInsufficientInstanceCapacity)} + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.NoError(t, err) + assert.Equal(t, "", group) + require.Len(t, nodes, 1) + assert.Equal(t, "", nodes[0].NodeID) + + assert.Len(t, placer.createGroups, 1) + require.Len(t, placer.reserveInputs, 1) + assert.Len(t, placer.deleteGroups, 1, "the reserved-but-empty group is dropped") + assert.Equal(t, []string{""}, inst.nodes) +} + +func TestPlaceControlPlane_VerifyMismatchRollsBack(t *testing.T) { + sched := &fakeHostScheduler{ + hosts: []string{"node-a", "node-b", "node-c"}, + wrongHost: map[string]string{"i-node-c": "node-x"}, + } + placer := &fakePlacer{reserved: []string{"node-a", "node-b", "node-c"}} + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.Error(t, err) + assert.Nil(t, nodes) + assert.Equal(t, "", group) + + assert.Len(t, inst.terminated, 3) + require.Len(t, placer.releaseInputs, 1) + assert.Empty(t, placer.finalizeInputs) + assert.Len(t, placer.deleteGroups, 1) +} + +func TestPlaceControlPlane_VerifyToleratesNotYetVisible(t *testing.T) { + sched := &fakeHostScheduler{ + hosts: []string{"node-a", "node-b", "node-c"}, + notVisible: map[string]bool{"i-node-b": true}, + } + placer := &fakePlacer{reserved: []string{"node-a", "node-b", "node-c"}} + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.NoError(t, err) + assert.Equal(t, "eks-cp-"+testHAAccountID+"-alpha", group) + require.Len(t, nodes, 3) + require.Len(t, placer.finalizeInputs, 1) + assert.Empty(t, placer.releaseInputs) + assert.Empty(t, inst.terminated) +} + +func TestPlaceControlPlane_FinalizeFailureRollsBack(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b", "node-c"}} + placer := &fakePlacer{ + reserved: []string{"node-a", "node-b", "node-c"}, + finalizeErr: errors.New("kv write failed"), + } + inst := &seqK3sInst{} + svc := newPlacerService(sched, placer, &seqK3sVPC{}, inst) + + nodes, group, err := svc.placeControlPlane(testHAAccountID, "alpha", validK3sInput()) + require.Error(t, err) + assert.Nil(t, nodes) + assert.Equal(t, "", group) + + assert.Len(t, inst.terminated, 3) + require.Len(t, placer.finalizeInputs, 1) + require.Len(t, placer.releaseInputs, 1) + assert.Len(t, placer.deleteGroups, 1) +} + +func TestControlPlaneTeardownNodes(t *testing.T) { + multi := &ClusterMeta{ControlPlaneNodes: []ControlPlaneNode{ + {NodeID: "n1", InstanceID: "i-1", ENIID: "eni-1"}, + {NodeID: "n2", InstanceID: "i-2", ENIID: "eni-2"}, + }} + assert.Len(t, controlPlaneTeardownNodes(multi), 2) + + scalar := &ClusterMeta{ + ControlPlaneInstanceID: "i-old", + ControlPlaneENIID: "eni-old", + ControlPlaneENIIP: "10.0.0.5", + ControlPlaneMgmtIP: "10.255.0.5", + } + got := controlPlaneTeardownNodes(scalar) + require.Len(t, got, 1) + assert.Equal(t, "i-old", got[0].InstanceID) + assert.Equal(t, "eni-old", got[0].ENIID) + assert.Equal(t, "10.0.0.5", got[0].ENIIP) + + assert.Nil(t, controlPlaneTeardownNodes(&ClusterMeta{})) +} + +func TestTeardownSpreadGroup(t *testing.T) { + placer := &fakePlacer{} + svc := &EKSServiceImpl{deps: EKSServiceDeps{PlacementGroup: placer}} + meta := &ClusterMeta{ + ControlPlaneSpreadGroup: "eks-cp-" + testHAAccountID + "-alpha", + ControlPlaneNodes: []ControlPlaneNode{ + {NodeID: "n1", InstanceID: "i-1"}, + {NodeID: "n2", InstanceID: "i-2"}, + {NodeID: "n3", InstanceID: "i-3"}, + }, + } + svc.teardownSpreadGroup(meta) + assert.Len(t, placer.removeInputs, 3) + assert.Equal(t, []string{meta.ControlPlaneSpreadGroup}, placer.deleteGroups) + + noGroup := &fakePlacer{} + svc2 := &EKSServiceImpl{deps: EKSServiceDeps{PlacementGroup: noGroup}} + svc2.teardownSpreadGroup(&ClusterMeta{}) + assert.Empty(t, noGroup.removeInputs) + assert.Empty(t, noGroup.deleteGroups) +} + +func TestNodeFitsSystemInstance(t *testing.T) { + // sys.medium = 2 vCPU / 4 GB. + vcpu, memGB, ok := instancetypes.SpecForSystemType("sys.medium") + require.True(t, ok) + require.Equal(t, 2, vcpu) + require.Equal(t, 4.0, memGB) + + tests := []struct { + name string + st types.NodeStatusResponse + want bool + }{ + { + name: "ample headroom", + st: types.NodeStatusResponse{TotalVCPU: 16, TotalMemGB: 64}, + want: true, + }, + { + name: "exact fit after reserve+alloc", + st: types.NodeStatusResponse{TotalVCPU: 8, ReservedVCPU: 2, AllocVCPU: 4, TotalMemGB: 12, ReservedMemGB: 4, AllocMemGB: 4}, + want: true, // remain 2 vCPU / 4 GB == footprint + }, + { + name: "vcpu exhausted", + st: types.NodeStatusResponse{TotalVCPU: 8, AllocVCPU: 7, TotalMemGB: 64}, + want: false, // remain 1 vCPU < 2 + }, + { + name: "memory exhausted", + st: types.NodeStatusResponse{TotalVCPU: 16, TotalMemGB: 8, AllocMemGB: 5}, + want: false, // remain 3 GB < 4 + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, nodeFitsSystemInstance(tt.st, vcpu, memGB)) + }) + } +} + +func TestSpecForSystemType_RejectsNonSystem(t *testing.T) { + _, _, ok := instancetypes.SpecForSystemType("t3.medium") + assert.False(t, ok, "customer types are not system types") + _, _, ok = instancetypes.SpecForSystemType("sys.nonexistent") + assert.False(t, ok, "unknown system size has no spec") +} diff --git a/spinifex/handlers/eks/k3s_server_vm.go b/spinifex/handlers/eks/k3s_server_vm.go index 870a413d9..3139a2641 100644 --- a/spinifex/handlers/eks/k3s_server_vm.go +++ b/spinifex/handlers/eks/k3s_server_vm.go @@ -41,6 +41,10 @@ type k3sVPCProvisioner interface { // *daemon.Daemon satisfies this interface. type k3sInstanceLauncher interface { LaunchSystemInstance(input *sysinstance.SystemInstanceInput) (*sysinstance.SystemInstanceOutput, error) + // LaunchSystemInstanceOnNode places the VM on a specific Spinifex host for + // HA control-plane spread. An empty nodeID (or the local node) launches + // in-process, matching LaunchSystemInstance. + LaunchSystemInstanceOnNode(nodeID string, input *sysinstance.SystemInstanceInput) (*sysinstance.SystemInstanceOutput, error) TerminateSystemInstance(instanceID string) error } @@ -51,10 +55,12 @@ type k3sAMIResolver interface { } const ( - // defaultK3sServerInstanceType is the spinifex instance type closest to - // the AWS EKS minimum control-plane footprint (2 vCPU / 8 GB / 40 GB). - // Callers can override via K3sServerInput.InstanceType. - defaultK3sServerInstanceType = "t3.medium" + // defaultK3sServerInstanceType is the internal system type the k3s + // control-plane VM boots as (2 vCPU / 4 GB / 40 GB root). A sys.* type + // keeps the CP VM out of customer DescribeInstanceTypes and lets the daemon + // register the node-targeted system.LaunchInstance.{type}.{nodeID} subject + // the HA spread path uses. Callers can override via K3sServerInput.InstanceType. + defaultK3sServerInstanceType = "sys.medium" // k3sOIDCSigningKeyPath is the on-VM path where cloud-init drops the // per-cluster OIDC private key PEM. K3s reads it via the @@ -137,6 +143,10 @@ type K3sServerInput struct { SecretKey string GatewayCACert string InstanceType string + // TargetNodeID pins the control-plane VM to a specific Spinifex host for HA + // spread across distinct failure domains. Empty launches on the local + // daemon node (the single-CP default). + TargetNodeID string // BuiltinIngress keeps K3s' bundled traefik + servicelb enabled (interim // in-VPC app exposure). When false the config disables them for AWS parity, // where Service type=LoadBalancer / Ingress are satisfied by the AWS Load @@ -208,7 +218,7 @@ func LaunchK3sServerVM( userData := buildK3sUserData(in) - sysOut, err := instSvc.LaunchSystemInstance(&sysinstance.SystemInstanceInput{ + sysOut, err := instSvc.LaunchSystemInstanceOnNode(in.TargetNodeID, &sysinstance.SystemInstanceInput{ BootMode: sysinstance.BootAMI, ManagedBy: tags.ManagedByEKS, InstanceType: instanceType, diff --git a/spinifex/handlers/eks/k3s_server_vm_test.go b/spinifex/handlers/eks/k3s_server_vm_test.go index 77f7bb407..7ed7af3de 100644 --- a/spinifex/handlers/eks/k3s_server_vm_test.go +++ b/spinifex/handlers/eks/k3s_server_vm_test.go @@ -53,6 +53,7 @@ func (f *fakeK3sVPC) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInp type fakeK3sInst struct { launchCalls []*sysinstance.SystemInstanceInput + launchNodes []string // TargetNodeID per launch (parallel to launchCalls) terminateCalls []string launchOut *sysinstance.SystemInstanceOutput @@ -63,7 +64,12 @@ type fakeK3sInst struct { var _ k3sInstanceLauncher = (*fakeK3sInst)(nil) func (f *fakeK3sInst) LaunchSystemInstance(input *sysinstance.SystemInstanceInput) (*sysinstance.SystemInstanceOutput, error) { + return f.LaunchSystemInstanceOnNode("", input) +} + +func (f *fakeK3sInst) LaunchSystemInstanceOnNode(nodeID string, input *sysinstance.SystemInstanceInput) (*sysinstance.SystemInstanceOutput, error) { f.launchCalls = append(f.launchCalls, input) + f.launchNodes = append(f.launchNodes, nodeID) if f.launchErr != nil { return nil, f.launchErr } @@ -246,6 +252,38 @@ func TestLaunchK3sServerVM_HonorsCustomInstanceType(t *testing.T) { assert.Equal(t, "t3.large", inst.launchCalls[0].InstanceType) } +func TestLaunchK3sServerVM_DefaultsToSystemInstanceType(t *testing.T) { + // The control-plane VM defaults to a sys.* type so the daemon registers the + // node-targeted system.LaunchInstance subject the HA spread path depends on. + assert.Equal(t, "sys.medium", defaultK3sServerInstanceType) + + vpc, inst, ami := &fakeK3sVPC{}, &fakeK3sInst{}, &fakeK3sAMI{} + _, err := LaunchK3sServerVM(vpc, inst, ami, validK3sInput()) + require.NoError(t, err) + require.Len(t, inst.launchCalls, 1) + assert.Equal(t, "sys.medium", inst.launchCalls[0].InstanceType) +} + +func TestLaunchK3sServerVM_NoTargetNodeLaunchesLocal(t *testing.T) { + vpc, inst, ami := &fakeK3sVPC{}, &fakeK3sInst{}, &fakeK3sAMI{} + + _, err := LaunchK3sServerVM(vpc, inst, ami, validK3sInput()) + require.NoError(t, err) + require.Len(t, inst.launchNodes, 1) + assert.Empty(t, inst.launchNodes[0], "empty TargetNodeID launches on the local node") +} + +func TestLaunchK3sServerVM_TargetNodeIDRoutedToLauncher(t *testing.T) { + vpc, inst, ami := &fakeK3sVPC{}, &fakeK3sInst{}, &fakeK3sAMI{} + in := validK3sInput() + in.TargetNodeID = "node-c" + + _, err := LaunchK3sServerVM(vpc, inst, ami, in) + require.NoError(t, err) + require.Len(t, inst.launchNodes, 1) + assert.Equal(t, "node-c", inst.launchNodes[0], "TargetNodeID pins placement to a specific host") +} + func TestLaunchK3sServerVM_UserDataContainsAllArtifacts(t *testing.T) { vpc, inst, ami := &fakeK3sVPC{}, &fakeK3sInst{}, &fakeK3sAMI{} diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index 899463b94..a9f153559 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -64,6 +64,12 @@ type EKSServiceDeps struct { EIP eipProvisioner Worker WorkerLauncher + // PlacementGroup reserves distinct hosts for HA control-plane spread; + // Scheduler answers the capacity + placement fan-out the spread path needs. + // The daemon wires the NATS implementations. + PlacementGroup controlPlanePlacer + Scheduler HostScheduler + // AddonInstaller delivers managed-addon manifests to clusters. Optional: // when nil the service defaults to the KV staging installer // (newStagingInstaller) bound to NATSConn. @@ -254,6 +260,12 @@ func (s *EKSServiceImpl) missingOrchestrationDeps() []string { if s.deps.Worker == nil { missing = append(missing, "Worker") } + if s.deps.PlacementGroup == nil { + missing = append(missing, "PlacementGroup") + } + if s.deps.Scheduler == nil { + missing = append(missing, "Scheduler") + } if len(s.deps.MasterKey) == 0 { missing = append(missing, "MasterKey") } @@ -438,7 +450,7 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, return nil, logCreateErr(name, accountID, "derive OIDC public key", err) } - k3sOut, err := LaunchK3sServerVM(s.deps.VPCK3s, s.deps.Instance, s.deps.Image, K3sServerInput{ + cpNodes, spreadGroup, err := s.placeControlPlane(accountID, name, K3sServerInput{ AccountID: accountID, ClusterName: name, Region: region, @@ -466,10 +478,16 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, "cluster", name, "accountID", accountID, "err", err) return nil, fmt.Errorf("launch K3s VM: %w", err) } - meta.ControlPlaneInstanceID = k3sOut.InstanceID - meta.ControlPlaneENIID = k3sOut.ENIID - meta.ControlPlaneENIIP = k3sOut.ENIIP - meta.ControlPlaneMgmtIP = k3sOut.MgmtIP + // Mirror the primary ([0]) into the scalar fields the reconciler, NLB + // registration, egress wiring, and teardown read today. Until per-node NLB + // registration (231.7.3) only the primary is target-registered + egress-wired. + primary := cpNodes[0] + meta.ControlPlaneNodes = cpNodes + meta.ControlPlaneSpreadGroup = spreadGroup + meta.ControlPlaneInstanceID = primary.InstanceID + meta.ControlPlaneENIID = primary.ENIID + meta.ControlPlaneENIIP = primary.ENIIP + meta.ControlPlaneMgmtIP = primary.MgmtIP // Egress: the control-plane VM must pull container images. Allocate a // hidden pool address and wire an egress-only SNAT for its /32 (no DNAT — @@ -484,7 +502,7 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, utils.PublishEvent(s.deps.NATSConn, "vpc.add-system-egress", systemEgressEvent{ VpcId: vpcID, SubnetId: subnetIDs[0], - InstanceIp: k3sOut.ENIIP, + InstanceIp: primary.ENIIP, ExternalIp: egressIP, }) @@ -495,7 +513,7 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, return nil, logCreateErr(name, accountID, "persist control-plane ids", err) } - if err := RegisterClusterTarget(s.deps.NLB, accountID, nlb.TargetGroupArn, k3sOut.ENIIP); err != nil { + if err := RegisterClusterTarget(s.deps.NLB, accountID, nlb.TargetGroupArn, primary.ENIIP); err != nil { s.markFailed(acctKV, name) return nil, logCreateErr(name, accountID, "register NLB target", err) } @@ -697,10 +715,10 @@ func (s *EKSServiceImpl) purgeClusterInfra(accountID, name string, meta *Cluster } } - if meta.ControlPlaneInstanceID != "" || meta.ControlPlaneENIID != "" { + for _, cp := range controlPlaneTeardownNodes(meta) { if err := TerminateK3sServerVM(s.deps.VPCK3s, s.deps.Instance, - accountID, meta.ControlPlaneInstanceID, meta.ControlPlaneENIID); err != nil { - teardownErrs = append(teardownErrs, fmt.Errorf("terminate K3s VM: %w", err)) + accountID, cp.InstanceID, cp.ENIID); err != nil { + teardownErrs = append(teardownErrs, fmt.Errorf("terminate K3s VM %s: %w", cp.InstanceID, err)) } } @@ -740,6 +758,11 @@ func (s *EKSServiceImpl) purgeClusterInfra(accountID, name string, meta *Cluster } } + // Release + delete the HA spread placement group (no-op for single-CP + // clusters). Best-effort: the VMs are already gone, so a leaked internal + // group strands nothing. + s.teardownSpreadGroup(meta) + // Only now, with NLB + VM confirmed gone, erase the meta + per-cluster KV. if err := DeleteClusterPrefix(acctKV, name); err != nil { return fmt.Errorf("delete cluster prefix: %w", err) diff --git a/spinifex/instancetypes/definitions.go b/spinifex/instancetypes/definitions.go index 164a266c1..4657d3c91 100644 --- a/spinifex/instancetypes/definitions.go +++ b/spinifex/instancetypes/definitions.go @@ -295,7 +295,8 @@ var p5eSizes = []instanceSize{ // systemSizes defines internal-only instance types for system VMs (LB, NAT GW, etc.). // These are registered in the type map for allocation but excluded from DescribeInstanceTypes. var systemSizes = []instanceSize{ - {"micro", 1, 0.125}, // 1 vCPU, 128 MB + {"micro", 1, 0.125}, // 1 vCPU, 128 MB — ELBv2 LB microVMs + {"medium", 2, 4}, // 2 vCPU, 4 GB — EKS k3s control-plane VMs } // instanceFamilyDefs defines all supported instance families with their vendor and sizes. diff --git a/spinifex/instancetypes/generate.go b/spinifex/instancetypes/generate.go index 48cbdf480..c49ecc563 100644 --- a/spinifex/instancetypes/generate.go +++ b/spinifex/instancetypes/generate.go @@ -17,6 +17,25 @@ func IsSystemType(name string) bool { return strings.HasPrefix(name, "sys.") } +// SpecForSystemType returns the vCPU and memory (GB) footprint of a system +// instance type (sys.*). System VMs consume host capacity like any guest, so +// the EKS HA control-plane scheduler sizes spread placement against these +// figures rather than the customer-facing node.status type list (which omits +// sys.* types). ok is false for an unknown name or a non-system type. The +// footprint is arch-independent, so the host GOARCH is used for the lookup. +func SpecForSystemType(name string) (vcpu int, memGB float64, ok bool) { + if !IsSystemType(name) { + return 0, 0, false + } + it, found := generateSystemTypes(runtime.GOARCH)[name] + if !found { + return 0, 0, false + } + return int(aws.Int64Value(it.VCpuInfo.DefaultVCpus)), + float64(aws.Int64Value(it.MemoryInfo.SizeInMiB)) / 1024, + true +} + // generateForGeneration creates the instance type map for the given CPU generation. // It generates all instance families matching the generation's family list across // burstable, general purpose, compute optimized, and memory optimized categories. diff --git a/spinifex/instancetypes/generate_test.go b/spinifex/instancetypes/generate_test.go index ddcbc0c16..8bcd22ed6 100644 --- a/spinifex/instancetypes/generate_test.go +++ b/spinifex/instancetypes/generate_test.go @@ -41,13 +41,19 @@ func TestIsSystemType(t *testing.T) { func TestGenerateSystemTypes(t *testing.T) { types := generateSystemTypes("x86_64") - require.Len(t, types, 1, "should have exactly 1 system type (sys.micro)") + require.Len(t, types, 2, "should have 2 system types (sys.micro, sys.medium)") sysMicro, ok := types["sys.micro"] require.True(t, ok, "sys.micro must exist") assert.Equal(t, int64(1), *sysMicro.VCpuInfo.DefaultVCpus, "sys.micro should have 1 vCPU") assert.Equal(t, int64(128), *sysMicro.MemoryInfo.SizeInMiB, "sys.micro should have 128 MiB") assert.False(t, *sysMicro.BurstablePerformanceSupported, "sys.micro should not be burstable") + + sysMedium, ok := types["sys.medium"] + require.True(t, ok, "sys.medium (EKS control-plane VM) must exist") + assert.Equal(t, int64(2), *sysMedium.VCpuInfo.DefaultVCpus, "sys.medium should have 2 vCPU") + assert.Equal(t, int64(4096), *sysMedium.MemoryInfo.SizeInMiB, "sys.medium should have 4 GiB") + assert.False(t, *sysMedium.BurstablePerformanceSupported, "sys.medium should not be burstable") } func TestDetectAndGenerate_IncludesSystemTypes(t *testing.T) {