feat(eks): HA control-plane spread placement (mulga-siv-231.7.1) (#332)

* 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.
This commit is contained in:
Josh Siviour
2026-06-10 08:58:58 +10:00
committed by GitHub
parent 686e374758
commit 69b62f3ca0
13 changed files with 1153 additions and 17 deletions
+47
View File
@@ -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.
@@ -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.
+3
View File
@@ -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),
}
}
+21
View File
@@ -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.
@@ -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)
@@ -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)
}
}
@@ -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-<node>" 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-<nodeID>" 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")
}
+15 -5
View File
@@ -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,
@@ -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{}
+33 -10
View File
@@ -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)
+2 -1
View File
@@ -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.
+19
View File
@@ -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.
+7 -1
View File
@@ -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) {