From a22bc2f716fd2ead8c0f825ff6d0194d26a2cb3b Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Sat, 6 Jun 2026 00:52:18 +1000 Subject: [PATCH] feat(eks): 6d nodegroup provisioning + NATS-gated cluster health (#300) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(eks): nodegroup provisioning + K3s agent join (6d) CreateNodegroup launches N customer-owned worker EC2 from the unified eks-node AMI in agent role; each joins the cluster via K3S_URL + K3S_TOKEN baked into cloud-init (first-boot.env SPINIFEX_K3S_ROLE=agent + agent.env). Workers are customer-owned and DescribeInstances-visible (no ManagedBy, no mgmt NIC) via the normal RunInstances path, matching real EKS. K3S_URL targets the control-plane ENI's in-VPC private IP on :6443 — the cluster NLB DNS endpoint is not resolvable in-VPC yet; switch to the NLB endpoint on :443 once authoritative in-VPC DNS lands. K3S_TOKEN is the decrypted cluster node-token persisted at server bootstrap. Describe/List/Update(scaling)/Delete round out CRUD. Cluster<->nodegroup SG ingress rules (apiserver 6443, flannel VXLAN 8472/udp, kubelet 10250, intra-nodegroup) authorized idempotently. UpdateNodegroupVersion deferred. * feat: gate EKS cluster health via NATS state reports The K3s apiserver binds the VPC node-ip, unreachable from the host-side spinifex daemon, so the reconciler cannot probe /healthz directly. Add a control-plane state reporter that publishes {healthz, node_count} over the management NATS bus the daemon already shares. - eks-node server role runs mulga-eks-state-report as an OpenRC service, publishing eks.state.{account}.{cluster}.server every 30s. - ClusterReconciler subscribes to that subject, gates CREATING -> ACTIVE on a healthy report, and records Ready node_count into ClusterMeta. - node_count counts Ready nodes only; k3s retains terminated workers as NotReady until pruned, so a raw count never drops on scale-down. --- scripts/images/eks-node/eks-node-role.sh | 2 + scripts/images/eks-node/manifest.conf | 9 +- .../eks-node/mulga-eks-state-report.initd | 29 + .../images/eks-node/mulga-eks-state-report.sh | 65 +- scripts/images/eks-node/setup.sh | 6 +- spinifex/daemon/daemon_handlers_eks_test.go | 15 +- spinifex/daemon/eks_deps.go | 1 + spinifex/daemon/eks_worker_launch.go | 67 ++ spinifex/daemon/eks_worker_launch_test.go | 37 ++ spinifex/handlers/eks/cluster_reconciler.go | 129 +++- .../eks/cluster_reconciler_state_test.go | 154 +++++ spinifex/handlers/eks/cluster_state.go | 25 + spinifex/handlers/eks/delete_cluster_test.go | 23 +- spinifex/handlers/eks/k3s_server_vm.go | 5 + spinifex/handlers/eks/nodegroup.go | 605 ++++++++++++++++++ spinifex/handlers/eks/nodegroup_test.go | 321 ++++++++++ spinifex/handlers/eks/nodegroup_userdata.go | 90 +++ spinifex/handlers/eks/security_groups.go | 66 ++ spinifex/handlers/eks/security_groups_test.go | 22 +- spinifex/handlers/eks/service_impl.go | 98 ++- spinifex/handlers/eks/service_impl_test.go | 16 +- spinifex/handlers/eks/state_report.go | 37 ++ spinifex/handlers/eks/store.go | 8 +- spinifex/handlers/eks/types.go | 29 +- 24 files changed, 1743 insertions(+), 116 deletions(-) create mode 100644 scripts/images/eks-node/mulga-eks-state-report.initd create mode 100644 spinifex/daemon/eks_worker_launch.go create mode 100644 spinifex/daemon/eks_worker_launch_test.go create mode 100644 spinifex/handlers/eks/cluster_reconciler_state_test.go create mode 100644 spinifex/handlers/eks/nodegroup.go create mode 100644 spinifex/handlers/eks/nodegroup_test.go create mode 100644 spinifex/handlers/eks/nodegroup_userdata.go create mode 100644 spinifex/handlers/eks/state_report.go diff --git a/scripts/images/eks-node/eks-node-role.sh b/scripts/images/eks-node/eks-node-role.sh index 2d3663a77..22948623f 100755 --- a/scripts/images/eks-node/eks-node-role.sh +++ b/scripts/images/eks-node/eks-node-role.sh @@ -49,9 +49,11 @@ case "${ROLE}" in rc-update add eks-token-webhook default rc-update add k3s default rc-update add k3s-first-boot default + rc-update add mulga-eks-state-report default rc-service eks-token-webhook start rc-service k3s start rc-service k3s-first-boot start + rc-service mulga-eks-state-report start ;; agent) log "configuring agent role" diff --git a/scripts/images/eks-node/manifest.conf b/scripts/images/eks-node/manifest.conf index a4c7aaab7..10fc2713e 100644 --- a/scripts/images/eks-node/manifest.conf +++ b/scripts/images/eks-node/manifest.conf @@ -30,12 +30,13 @@ scripts/images/eks-node/k3s-agent.initd:/etc/init.d/k3s-agent \ scripts/images/eks-node/eks-token-webhook.initd:/etc/init.d/eks-token-webhook \ scripts/images/eks-node/k3s-first-boot.initd:/etc/init.d/k3s-first-boot \ scripts/images/eks-node/k3s-first-boot.sh:/usr/local/sbin/k3s-first-boot \ -scripts/images/eks-node/mulga-eks-state-report.sh:/etc/periodic/15min/mulga-eks-state-report \ +scripts/images/eks-node/mulga-eks-state-report.initd:/etc/init.d/mulga-eks-state-report \ +scripts/images/eks-node/mulga-eks-state-report.sh:/usr/local/sbin/mulga-eks-state-report \ scripts/images/eks-node/mulga-eks-etcd-snapshot.sh:/etc/periodic/daily/mulga-eks-etcd-snapshot" -# Only crond (state-report) + the role selector at bake. The selector adds the -# role-specific services (k3s/eks-token-webhook/k3s-first-boot or k3s-agent) to -# the default runlevel on first boot. +# Only crond (etcd snapshot) + the role selector at bake. The selector adds the +# role-specific services (k3s/eks-token-webhook/k3s-first-boot/state-report or +# k3s-agent) to the default runlevel on first boot. ENABLE_SERVICES="crond eks-node-role" SETUP_SCRIPT="scripts/images/eks-node/setup.sh" diff --git a/scripts/images/eks-node/mulga-eks-state-report.initd b/scripts/images/eks-node/mulga-eks-state-report.initd new file mode 100644 index 000000000..34b26fb4c --- /dev/null +++ b/scripts/images/eks-node/mulga-eks-state-report.initd @@ -0,0 +1,29 @@ +#!/sbin/openrc-run +# mulga-eks-state-report — server-role service that publishes K3s health + +# node_count to the management NATS bus every STATE_REPORT_INTERVAL seconds. The +# spinifex cluster reconciler consumes this to gate CREATING → ACTIVE and to +# track nodegroup join/scale, since the apiserver (bound to the VPC node-ip) is +# unreachable from the host. Enabled on the server role by eks-node-role.sh. + +description="Mulga EKS control-plane state reporter" +command="/usr/local/sbin/mulga-eks-state-report" +command_background="yes" +pidfile="/run/mulga-eks-state-report.pid" +output_log="/var/log/mulga-eks-state-report.log" +error_log="/var/log/mulga-eks-state-report.log" + +# Cadence the report loop sleeps between publishes. 30s is well under the +# reconciler's 90s stale window, so a missed tick does not flap health. +export STATE_REPORT_INTERVAL="${STATE_REPORT_INTERVAL:-30}" + +depend() { + need net + after k3s +} + +start_pre() { + if [ ! -f /etc/spinifex-eks/first-boot.env ]; then + eerror "/etc/spinifex-eks/first-boot.env not present — cloud-init did not seed NATS creds" + return 1 + fi +} diff --git a/scripts/images/eks-node/mulga-eks-state-report.sh b/scripts/images/eks-node/mulga-eks-state-report.sh index 1427e7e43..5339aa738 100755 --- a/scripts/images/eks-node/mulga-eks-state-report.sh +++ b/scripts/images/eks-node/mulga-eks-state-report.sh @@ -1,15 +1,20 @@ #!/bin/sh set -eu -# mulga-eks-state-report — periodic NATS publish of K3s server health. -# Invoked every 15 minutes via /etc/periodic/15min (Alpine crond). One-shot -# best-effort; failures log but do not retry — the spinifex reconciler also -# does its own NLB healthcheck so this is supplementary signal. +# mulga-eks-state-report — NATS publish of K3s server health for the spinifex +# cluster reconciler. The apiserver binds the VPC node-ip, unreachable from the +# host-side daemon, so the daemon cannot probe /healthz directly — this report +# carries {healthz, node_count} over the management NATS bus the daemon shares. # # Subject: eks.state.{accountID}.{clusterName}.server # Payload: JSON { healthz: "ok"|"fail", node_count: N, ts: } +# +# Reads NATS creds + cluster identity from the cloud-init-seeded first-boot env +# (same vars k3s-first-boot.sh publishes its bootstrap artifacts with). Run as a +# loop by the mulga-eks-state-report OpenRC service when STATE_REPORT_INTERVAL is +# set; one-shot otherwise (e.g. a manual invocation). -ENVFILE=/etc/spinifex-eks/state-report.env +ENVFILE=/etc/spinifex-eks/first-boot.env [ -f "${ENVFILE}" ] || { logger -t mulga-eks-state-report "${ENVFILE} missing — exiting"; exit 0; } # shellcheck disable=SC1090 . "${ENVFILE}" @@ -21,27 +26,39 @@ ENVFILE=/etc/spinifex-eks/state-report.env KUBECONFIG=${KUBECONFIG:-/etc/rancher/k3s/k3s.yaml} export KUBECONFIG -if curl -sk --max-time 3 https://127.0.0.1:6443/healthz | grep -q '^ok$'; then - HEALTH=ok -else - HEALTH=fail -fi - -if [ "${HEALTH}" = "ok" ]; then - NODE_COUNT=$(kubectl get nodes --no-headers 2>/dev/null | wc -l | tr -d ' ') -else - NODE_COUNT=0 -fi - -TS=$(date +%s) -PAYLOAD=$(printf '{"healthz":"%s","node_count":%s,"ts":%s}' "${HEALTH}" "${NODE_COUNT}" "${TS}") SUBJ="eks.state.${EKS_ACCOUNT_ID}.${EKS_CLUSTER_NAME}.server" +# NATS args mirror k3s-first-boot.sh (token + TLS CA) — the proven publish path. NATS_ARGS="-s ${SPINIFEX_NATS_URL}" -if [ -n "${SPINIFEX_NATS_CREDS_FILE:-}" ] && [ -f "${SPINIFEX_NATS_CREDS_FILE}" ]; then - NATS_ARGS="${NATS_ARGS} --creds ${SPINIFEX_NATS_CREDS_FILE}" +if [ -n "${SPINIFEX_NATS_TOKEN:-}" ]; then + NATS_ARGS="${NATS_ARGS} --token ${SPINIFEX_NATS_TOKEN}" +fi +if [ -n "${SPINIFEX_NATS_CA:-}" ] && [ -f "${SPINIFEX_NATS_CA}" ]; then + NATS_ARGS="${NATS_ARGS} --tlsca ${SPINIFEX_NATS_CA}" fi -# shellcheck disable=SC2086 -printf '%s' "${PAYLOAD}" | nats ${NATS_ARGS} pub "${SUBJ}" --stdin 2>&1 \ - | logger -t mulga-eks-state-report +publish_report() { + if curl -sk --max-time 3 https://127.0.0.1:6443/healthz | grep -q '^ok$'; then + health=ok + # Count Ready nodes only — k3s retains terminated workers as NotReady + # until manually pruned, so a raw node count never drops on scale-down. + node_count=$(kubectl get nodes --no-headers 2>/dev/null | awk '$2=="Ready"' | wc -l | tr -d ' ') + else + health=fail + node_count=0 + fi + payload=$(printf '{"healthz":"%s","node_count":%s,"ts":%s}' "${health}" "${node_count}" "$(date +%s)") + # shellcheck disable=SC2086 + printf '%s' "${payload}" | nats ${NATS_ARGS} pub "${SUBJ}" --force-stdin 2>&1 \ + | logger -t mulga-eks-state-report +} + +# STATE_REPORT_INTERVAL set → run forever (service mode); unset → one-shot. +if [ -n "${STATE_REPORT_INTERVAL:-}" ] && [ "${STATE_REPORT_INTERVAL}" -gt 0 ] 2>/dev/null; then + while true; do + publish_report + sleep "${STATE_REPORT_INTERVAL}" + done +else + publish_report +fi diff --git a/scripts/images/eks-node/setup.sh b/scripts/images/eks-node/setup.sh index 9fe3e3390..a81db515c 100755 --- a/scripts/images/eks-node/setup.sh +++ b/scripts/images/eks-node/setup.sh @@ -55,9 +55,9 @@ apk del --no-cache unzip # Init scripts ship as 0644 from INSTALL_FILES; OpenRC requires 0755. Every # role's services are baked; the selector enables the right ones at first boot. chmod 0755 /etc/init.d/eks-node-role /etc/init.d/k3s /etc/init.d/k3s-agent \ - /etc/init.d/eks-token-webhook /etc/init.d/k3s-first-boot -chmod 0755 /usr/local/sbin/eks-node-role /usr/local/sbin/k3s-first-boot -chmod 0755 /etc/periodic/15min/mulga-eks-state-report + /etc/init.d/eks-token-webhook /etc/init.d/k3s-first-boot /etc/init.d/mulga-eks-state-report +chmod 0755 /usr/local/sbin/eks-node-role /usr/local/sbin/k3s-first-boot \ + /usr/local/sbin/mulga-eks-state-report chmod 0755 /etc/periodic/daily/mulga-eks-etcd-snapshot # K3s server config — skeleton; cloud-init / first-boot fills in the diff --git a/spinifex/daemon/daemon_handlers_eks_test.go b/spinifex/daemon/daemon_handlers_eks_test.go index 9c9623aa3..166ae00d8 100644 --- a/spinifex/daemon/daemon_handlers_eks_test.go +++ b/spinifex/daemon/daemon_handlers_eks_test.go @@ -61,6 +61,11 @@ func TestDaemonHandleEKS_AllHandlersDispatchToService(t *testing.T) { // no error, so wantCode is "" (success — assert a non-error payload). notImpl := awserrors.ErrorNotImplemented invalid := awserrors.ErrorInvalidParameterValue + // Nodegroup mutators gate on orchestration deps, which the shim service + // lacks → ServiceUnavailable. The read paths reach input validation and an + // empty body fails with InvalidParameterValue. UpdateNodegroupVersion stays + // NotImplemented (v1 doesn't do AMI upgrades). + unavailable := awserrors.ErrorServiceUnavailable cases := []struct { subject string handler nats.MsgHandler @@ -68,12 +73,12 @@ func TestDaemonHandleEKS_AllHandlersDispatchToService(t *testing.T) { }{ {"eks.UpdateClusterConfig", d.handleEKSUpdateClusterConfig, notImpl}, {"eks.UpdateClusterVersion", d.handleEKSUpdateClusterVersion, notImpl}, - {"eks.CreateNodegroup", d.handleEKSCreateNodegroup, notImpl}, - {"eks.DescribeNodegroup", d.handleEKSDescribeNodegroup, notImpl}, - {"eks.ListNodegroups", d.handleEKSListNodegroups, notImpl}, - {"eks.UpdateNodegroupConfig", d.handleEKSUpdateNodegroupConfig, notImpl}, + {"eks.CreateNodegroup", d.handleEKSCreateNodegroup, unavailable}, + {"eks.DescribeNodegroup", d.handleEKSDescribeNodegroup, invalid}, + {"eks.ListNodegroups", d.handleEKSListNodegroups, invalid}, + {"eks.UpdateNodegroupConfig", d.handleEKSUpdateNodegroupConfig, unavailable}, {"eks.UpdateNodegroupVersion", d.handleEKSUpdateNodegroupVersion, notImpl}, - {"eks.DeleteNodegroup", d.handleEKSDeleteNodegroup, notImpl}, + {"eks.DeleteNodegroup", d.handleEKSDeleteNodegroup, unavailable}, {"eks.CreateAccessEntry", d.handleEKSCreateAccessEntry, invalid}, {"eks.DescribeAccessEntry", d.handleEKSDescribeAccessEntry, invalid}, {"eks.ListAccessEntries", d.handleEKSListAccessEntries, invalid}, diff --git a/spinifex/daemon/eks_deps.go b/spinifex/daemon/eks_deps.go index 2a0e63ef3..8a3f51f7a 100644 --- a/spinifex/daemon/eks_deps.go +++ b/spinifex/daemon/eks_deps.go @@ -59,6 +59,7 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps { Instance: d, Image: d.imageService, EIP: d.eipService, + Worker: d, } } diff --git a/spinifex/daemon/eks_worker_launch.go b/spinifex/daemon/eks_worker_launch.go new file mode 100644 index 000000000..82e2e1893 --- /dev/null +++ b/spinifex/daemon/eks_worker_launch.go @@ -0,0 +1,67 @@ +package daemon + +import ( + "encoding/base64" + "errors" + "fmt" + "log/slog" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + spxtypes "github.com/mulgadc/spinifex/spinifex/types" +) + +// Compile-time check that Daemon satisfies the EKS worker-launch surface. +var _ handlers_eks.WorkerLauncher = (*Daemon)(nil) + +// RunWorkerInstance launches a nodegroup worker through the normal customer +// RunInstances path: customer-owned, primary ENI auto-created in input.SubnetId +// with input.SecurityGroupIds, no ManagedBy tag, no mgmt NIC — so the worker is +// visible in DescribeInstances and reclaimed by TerminateInstances like any +// other customer EC2. The EKS service hands us raw cloud-config YAML in +// input.UserData; RunInstances expects it base64-encoded, so wrap it here. +func (d *Daemon) RunWorkerInstance(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) { + if d.instanceService == nil { + return nil, errors.New("eks worker: instance service not initialized") + } + if input == nil { + return nil, errors.New("eks worker: nil RunInstancesInput") + } + if input.UserData != nil && *input.UserData != "" { + input.UserData = aws.String(base64.StdEncoding.EncodeToString([]byte(*input.UserData))) + } + return d.instanceService.RunInstances(input, accountID) +} + +// TerminateWorkerInstances terminates each nodegroup worker via the same path +// the EC2 NATS terminate handler uses. A worker no longer tracked by the local +// vmMgr is treated as already-gone (idempotent success) so a retried +// DeleteNodegroup / scale-down does not wedge on instances that already drained. +func (d *Daemon) TerminateWorkerInstances(instanceIDs []string, accountID string) error { + if d.instanceService == nil { + return errors.New("eks worker: instance service not initialized") + } + var errs []error + for _, id := range instanceIDs { + if id == "" { + continue + } + instance, ok := d.vmMgr.Get(id) + if !ok { + slog.Debug("TerminateWorkerInstances: instance already gone", "instanceId", id) + continue + } + cmd := spxtypes.EC2InstanceCommand{ + ID: id, + Attributes: spxtypes.EC2CommandAttributes{TerminateInstance: true}, + } + if err := d.instanceService.StopOrTerminateInstance(instance, cmd); err != nil { + errs = append(errs, fmt.Errorf("terminate worker %s: %w", id, err)) + } + } + if len(errs) > 0 { + return errors.Join(errs...) + } + return nil +} diff --git a/spinifex/daemon/eks_worker_launch_test.go b/spinifex/daemon/eks_worker_launch_test.go new file mode 100644 index 000000000..33b4ebb58 --- /dev/null +++ b/spinifex/daemon/eks_worker_launch_test.go @@ -0,0 +1,37 @@ +package daemon + +import ( + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" + "github.com/stretchr/testify/require" +) + +// A worker whose ID is not tracked by the local vmMgr is already-gone, so +// terminating it is an idempotent no-op success (a retried DeleteNodegroup / +// scale-down must not wedge on drained instances). +func TestTerminateWorkerInstances_NotFoundIsIdempotent(t *testing.T) { + d := newDaemonWithVMs() + // Non-nil service passes the init guard; no VM matches the IDs so + // StopOrTerminateInstance is never reached — the not-found branch returns + // success. + d.instanceService = &handlers_ec2_instance.InstanceServiceImpl{} + + require.NoError(t, d.TerminateWorkerInstances([]string{"i-gone1", "i-gone2"}, "111122223333")) + // Empty / blank IDs are skipped. + require.NoError(t, d.TerminateWorkerInstances([]string{"", ""}, "111122223333")) + require.NoError(t, d.TerminateWorkerInstances(nil, "111122223333")) +} + +// Both worker methods must guard against an uninitialized instance service +// rather than nil-panic. +func TestWorkerLauncher_NilInstanceService(t *testing.T) { + d := &Daemon{} + + _, err := d.RunWorkerInstance(&ec2.RunInstancesInput{ImageId: aws.String("ami-1")}, "111122223333") + require.Error(t, err) + + require.Error(t, d.TerminateWorkerInstances([]string{"i-1"}, "111122223333")) +} diff --git a/spinifex/handlers/eks/cluster_reconciler.go b/spinifex/handlers/eks/cluster_reconciler.go index fc4cde12d..3f0d8edd2 100644 --- a/spinifex/handlers/eks/cluster_reconciler.go +++ b/spinifex/handlers/eks/cluster_reconciler.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "net/http" + "sync/atomic" "time" "github.com/nats-io/nats.go" @@ -21,8 +22,18 @@ const ( // (absolute), so a VM that never boots reaches a terminal state even across // daemon restarts. Plan Risk #1. defaultCreateTimeout = 15 * time.Minute + // defaultStateStaleAfter bounds how old the control plane's last NATS state + // report may be before health is treated as unknown/failed. The CP publishes + // every ~30s; 3x that tolerates a missed tick without flapping. + defaultStateStaleAfter = 90 * time.Second ) +// natsSubscriber is the minimum surface the reconciler needs to consume the +// control-plane state report; *nats.Conn satisfies it, tests stub it. +type natsSubscriber interface { + Subscribe(subject string, cb nats.MsgHandler) (*nats.Subscription, error) +} + // ErrReconcilerLeaseLost is returned by Run when the leader lease was lost // mid-loop. Callers can re-Acquire and re-Run if they want to keep driving // reconcile from the same node. @@ -58,6 +69,14 @@ type ClusterReconciler struct { healthTimeout time.Duration createTimeout time.Duration httpClient HTTPDoer + + // State-report health source. When stateSub is non-nil the reconciler gates + // on the CP's NATS self-report (latest) rather than the HTTP /healthz probe — + // the apiserver is bound to the VPC node-ip and unreachable from the host. + stateSub natsSubscriber + stateSubject string + stateStaleAfter time.Duration + latest atomic.Pointer[ServerStateReport] } // ReconcilerOption tunes a ClusterReconciler. Tests use the With* helpers to @@ -91,6 +110,22 @@ func WithHTTPClient(c HTTPDoer) ReconcilerOption { return func(r *ClusterReconciler) { r.httpClient = c } } +// WithStateSource configures the reconciler to gate health on the control +// plane's NATS state report (subject) consumed via sub, instead of the HTTP +// /healthz probe. The subscription is opened in Run and closed when it returns. +func WithStateSource(sub natsSubscriber, subject string) ReconcilerOption { + return func(r *ClusterReconciler) { + r.stateSub = sub + r.stateSubject = subject + } +} + +// WithStateStaleAfter overrides how old the last state report may be before +// health is treated as failing. +func WithStateStaleAfter(d time.Duration) ReconcilerOption { + return func(r *ClusterReconciler) { r.stateStaleAfter = d } +} + // NewClusterReconciler constructs a ClusterReconciler. healthURL is the // fully-qualified probe target (e.g. https://nlb-dns:443/healthz) or empty // to skip /healthz probing (CREATING transitions on bootstrap state alone). @@ -111,16 +146,17 @@ func NewClusterReconciler(leaderKV, acctKV nats.KeyValue, accountID, clusterName return nil, errors.New("eks: NewClusterReconciler empty holderID") } r := &ClusterReconciler{ - leaderKV: leaderKV, - acctKV: acctKV, - accountID: accountID, - clusterName: clusterName, - holderID: holderID, - healthURL: healthURL, - leaseRefresh: defaultReconcileLeaseRefresh, - interval: defaultReconcileInterval, - healthTimeout: defaultHealthzTimeout, - createTimeout: defaultCreateTimeout, + leaderKV: leaderKV, + acctKV: acctKV, + accountID: accountID, + clusterName: clusterName, + holderID: holderID, + healthURL: healthURL, + leaseRefresh: defaultReconcileLeaseRefresh, + interval: defaultReconcileInterval, + healthTimeout: defaultHealthzTimeout, + createTimeout: defaultCreateTimeout, + stateStaleAfter: defaultStateStaleAfter, httpClient: &http.Client{ Timeout: defaultHealthzTimeout, Transport: &http.Transport{ @@ -186,6 +222,22 @@ func (r *ClusterReconciler) RefreshLease() bool { // cluster reaches a terminal state. Caller is expected to AcquireLease first; // Run does not re-acquire. func (r *ClusterReconciler) Run(ctx context.Context) error { + if r.stateSub != nil && r.stateSubject != "" { + sub, err := r.stateSub.Subscribe(r.stateSubject, func(m *nats.Msg) { + report, perr := unmarshalServerStateReport(m.Data) + if perr != nil { + slog.Warn("ClusterReconciler: bad state report", + "cluster", r.clusterName, "subject", m.Subject, "err", perr) + return + } + r.latest.Store(&report) + }) + if err != nil { + return fmt.Errorf("subscribe state report %s: %w", r.stateSubject, err) + } + defer func() { _ = sub.Unsubscribe() }() + } + refreshT := time.NewTicker(r.leaseRefresh) defer refreshT.Stop() reconcileT := time.NewTicker(r.interval) @@ -238,24 +290,30 @@ func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error { "cluster", r.clusterName, "reason", reason) return r.failIfCreateTimedOut(meta, "bootstrap not ready: "+reason) } - if err := r.probeHealthz(ctx); err != nil { - slog.Debug("ClusterReconciler: /healthz still failing", - "cluster", r.clusterName, "err", err) - return r.failIfCreateTimedOut(meta, "healthz not ready: "+err.Error()) + issue, nodeCount := r.observe(ctx) + if issue != "" { + slog.Debug("ClusterReconciler: health not ready", + "cluster", r.clusterName, "issue", issue) + return r.failIfCreateTimedOut(meta, "healthz not ready: "+issue) } if err := SetClusterStatus(r.acctKV, r.clusterName, ClusterStatusActive); err != nil { return err } - slog.Info("ClusterReconciler: cluster transitioned to ACTIVE", - "cluster", r.clusterName) - case ClusterStatusActive: - issue := "" - if err := r.probeHealthz(ctx); err != nil { - slog.Warn("ClusterReconciler: /healthz failing for ACTIVE cluster", - "cluster", r.clusterName, "err", err) - issue = err.Error() + if err := SetClusterHealthState(r.acctKV, r.clusterName, "", nodeCount); err != nil { + if errors.Is(err, ErrClusterNotFound) { + return ErrClusterNotFound + } + return fmt.Errorf("record cluster health: %w", err) } - if err := SetClusterHealth(r.acctKV, r.clusterName, issue); err != nil { + slog.Info("ClusterReconciler: cluster transitioned to ACTIVE", + "cluster", r.clusterName, "nodes", nodeCount) + case ClusterStatusActive: + issue, nodeCount := r.observe(ctx) + if issue != "" { + slog.Warn("ClusterReconciler: health failing for ACTIVE cluster", + "cluster", r.clusterName, "issue", issue) + } + if err := SetClusterHealthState(r.acctKV, r.clusterName, issue, nodeCount); err != nil { if errors.Is(err, ErrClusterNotFound) { return ErrClusterNotFound } @@ -269,6 +327,31 @@ func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error { return nil } +// observe returns the current health issue ("" when healthy) and the reported +// node count. With a NATS state source it reads the control plane's latest +// self-report (rejecting a stale or unhealthy one); otherwise it falls back to +// the HTTP /healthz probe (node count is then 0, the probe carries no count). +func (r *ClusterReconciler) observe(ctx context.Context) (issue string, nodeCount int) { + if r.stateSub != nil && r.stateSubject != "" { + report := r.latest.Load() + if report == nil { + return "no control-plane state report received yet", 0 + } + age := time.Since(time.Unix(report.TS, 0)) + if age > r.stateStaleAfter { + return fmt.Sprintf("control-plane state report stale (%s old)", age.Round(time.Second)), report.NodeCount + } + if !report.Healthy() { + return fmt.Sprintf("apiserver healthz=%q", report.Healthz), report.NodeCount + } + return "", report.NodeCount + } + if err := r.probeHealthz(ctx); err != nil { + return err.Error(), 0 + } + return "", 0 +} + // failIfCreateTimedOut transitions a still-CREATING cluster to FAILED once it // has exceeded createTimeout since meta.CreatedAt, and returns // ErrReconcilerClusterFailed so Run exits the loop. While still within the diff --git a/spinifex/handlers/eks/cluster_reconciler_state_test.go b/spinifex/handlers/eks/cluster_reconciler_state_test.go new file mode 100644 index 000000000..49335d17a --- /dev/null +++ b/spinifex/handlers/eks/cluster_reconciler_state_test.go @@ -0,0 +1,154 @@ +package handlers_eks + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newStateReconcilerHarness builds a reconciler wired to a real NATS conn as its +// state source (so observe() reads r.latest) and returns the conn + account KV. +// It does NOT start a subscription — tests drive r.latest directly for +// determinism; the integration test below exercises the real Subscribe path. +func newStateReconcilerHarness(t *testing.T, opts ...ReconcilerOption) (*ClusterReconciler, *nats.Conn, nats.KeyValue) { + t.Helper() + _, nc, js := testutil.StartTestJetStream(t) + leaderKV, err := InitLeaderBucket(js) + require.NoError(t, err) + acctKV, err := GetOrCreateAccountBucket(js, testAccountID) + require.NoError(t, err) + require.NoError(t, PutClusterMeta(acctKV, sampleClusterMeta("alpha"))) + + base := []ReconcilerOption{WithStateSource(nc, StateSubject(testAccountID, "alpha"))} + r, err := NewClusterReconciler(leaderKV, acctKV, testAccountID, "alpha", "holder-1", "", append(base, opts...)...) + require.NoError(t, err) + return r, nc, acctKV +} + +func freshReport(healthz string, nodes int) *ServerStateReport { + return &ServerStateReport{Healthz: healthz, NodeCount: nodes, TS: time.Now().Unix()} +} + +func TestClusterReconciler_CreatingTransitionsToActiveOnStateReport(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t) + freshenClusterCreatedAt(t, acctKV) + seedBootstrapState(t, acctKV) + r.latest.Store(freshReport("ok", 3)) + + require.NoError(t, r.reconcileOnce(context.Background())) + + meta, err := GetClusterMeta(acctKV, "alpha") + require.NoError(t, err) + assert.Equal(t, ClusterStatusActive, meta.Status) + assert.Equal(t, 3, meta.NodeCount) + assert.Empty(t, meta.HealthIssue) +} + +func TestClusterReconciler_CreatingStaysWithoutStateReport(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t) + freshenClusterCreatedAt(t, acctKV) + seedBootstrapState(t, acctKV) + // No report stored. + + require.NoError(t, r.reconcileOnce(context.Background())) + + meta, err := GetClusterMeta(acctKV, "alpha") + require.NoError(t, err) + assert.Equal(t, ClusterStatusCreating, meta.Status, "no state report yet → stays CREATING") +} + +func TestClusterReconciler_CreatingStaysOnUnhealthyReport(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t) + freshenClusterCreatedAt(t, acctKV) + seedBootstrapState(t, acctKV) + r.latest.Store(freshReport("fail", 0)) + + require.NoError(t, r.reconcileOnce(context.Background())) + + meta, err := GetClusterMeta(acctKV, "alpha") + require.NoError(t, err) + assert.Equal(t, ClusterStatusCreating, meta.Status, "unhealthy apiserver → stays CREATING") +} + +func TestClusterReconciler_ActiveRecordsNodeCountAndClearsIssue(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t) + require.NoError(t, SetClusterStatus(acctKV, "alpha", ClusterStatusActive)) + require.NoError(t, SetClusterHealthState(acctKV, "alpha", "stale", 0)) + r.latest.Store(freshReport("ok", 4)) + + require.NoError(t, r.reconcileOnce(context.Background())) + + meta, err := GetClusterMeta(acctKV, "alpha") + require.NoError(t, err) + assert.Equal(t, ClusterStatusActive, meta.Status) + assert.Equal(t, 4, meta.NodeCount) + assert.Empty(t, meta.HealthIssue, "healthy report clears the prior issue") +} + +func TestClusterReconciler_ActiveFlagsStaleReport(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t, WithStateStaleAfter(90*time.Second)) + require.NoError(t, SetClusterStatus(acctKV, "alpha", ClusterStatusActive)) + r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()}) + + require.NoError(t, r.reconcileOnce(context.Background())) + + meta, err := GetClusterMeta(acctKV, "alpha") + require.NoError(t, err) + assert.Contains(t, meta.HealthIssue, "stale", "report older than stale window flags an issue") + assert.Equal(t, 2, meta.NodeCount, "last known node count still recorded") +} + +func TestClusterReconciler_StateReportSubscriptionDrivesActive(t *testing.T) { + r, nc, acctKV := newStateReconcilerHarness(t, + WithReconcileInterval(10*time.Millisecond), + WithLeaseRefresh(10*time.Second), + ) + freshenClusterCreatedAt(t, acctKV) + seedBootstrapState(t, acctKV) + + release, ok := r.AcquireLease() + require.True(t, ok) + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + runErr := make(chan error, 1) + go func() { runErr <- r.Run(ctx) }() + + // Publish healthy reports until the cluster goes ACTIVE (core NATS only + // delivers post-subscribe, so keep emitting while Run's subscription opens). + stop := make(chan struct{}) + defer close(stop) + go func() { + payload, _ := json.Marshal(ServerStateReport{Healthz: "ok", NodeCount: 3, TS: time.Now().Unix()}) + tick := time.NewTicker(20 * time.Millisecond) + defer tick.Stop() + for { + select { + case <-stop: + return + case <-tick.C: + _ = nc.Publish(StateSubject(testAccountID, "alpha"), payload) + } + } + }() + + require.Eventually(t, func() bool { + meta, err := GetClusterMeta(acctKV, "alpha") + return err == nil && meta.Status == ClusterStatusActive && meta.NodeCount == 3 + }, 1500*time.Millisecond, 10*time.Millisecond, "state reports should drive CREATING → ACTIVE") + + cancel() + select { + case err := <-runErr: + assert.ErrorIs(t, err, context.Canceled) + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } +} diff --git a/spinifex/handlers/eks/cluster_state.go b/spinifex/handlers/eks/cluster_state.go index fafb7690f..89740a6a8 100644 --- a/spinifex/handlers/eks/cluster_state.go +++ b/spinifex/handlers/eks/cluster_state.go @@ -69,6 +69,10 @@ type ClusterMeta struct { // only) and never treats a time.Time struct as empty, so this field always // serializes. The tag states that honestly rather than implying it elides. LastHealthProbe time.Time `json:"lastHealthProbe"` + // NodeCount is the node total from the control plane's last NATS state + // report (server + joined agents). Surfaces nodegroup join/scale progress + // without the host needing apiserver reachability. + NodeCount int `json:"nodeCount,omitempty"` } // ErrClusterNotFound is returned by GetClusterMeta / SetClusterStatus / @@ -231,6 +235,27 @@ func SetClusterHealth(kv nats.KeyValue, name, issue string) error { }) } +// SetClusterHealthState records the latest health outcome together with the +// node count from the control-plane state report. issue is the failure reason +// (empty clears it / marks healthy). LastHealthProbe is stamped whenever health +// or node count changes; a steady-state healthy cluster with an unchanging node +// count produces no KV write. Returns ErrClusterNotFound if the meta record was +// deleted underneath us. +func SetClusterHealthState(kv nats.KeyValue, name, issue string, nodeCount int) error { + if name == "" { + return errors.New("eks: SetClusterHealthState empty name") + } + return casUpdateMeta(kv, name, func(m *ClusterMeta) bool { + if m.HealthIssue == issue && m.NodeCount == nodeCount { + return false + } + m.HealthIssue = issue + m.NodeCount = nodeCount + m.LastHealthProbe = time.Now().UTC() + return true + }) +} + // DeleteClusterPrefix removes every KV key under clusters/{name}/. Called // from DeleteCluster after the OIDC private key has been zeroized. // Returns the first delete error encountered but continues sweeping so diff --git a/spinifex/handlers/eks/delete_cluster_test.go b/spinifex/handlers/eks/delete_cluster_test.go index 2088881c0..6d89e82b2 100644 --- a/spinifex/handlers/eks/delete_cluster_test.go +++ b/spinifex/handlers/eks/delete_cluster_test.go @@ -27,13 +27,15 @@ func deleteInput(name string) *eks.DeleteClusterInput { // and an embedded JetStream KV. Tests poke the fakes' *Err fields to force // failures at specific lifecycle steps. type eksServiceFixture struct { - svc *EKSServiceImpl - kv nats.KeyValue - nlb *fakeNLBProvisioner - inst *fakeK3sInst - vpc *fakeK3sVPC - ami *fakeK3sAMI - eip *fakeEIPProvisioner + svc *EKSServiceImpl + kv nats.KeyValue + nlb *fakeNLBProvisioner + inst *fakeK3sInst + vpc *fakeK3sVPC + ami *fakeK3sAMI + eip *fakeEIPProvisioner + sg *fakeSGProvisioner + worker *fakeWorkerLauncher } func newEKSServiceFixture(t *testing.T) *eksServiceFixture { @@ -47,6 +49,8 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture { vpc := &fakeK3sVPC{} ami := &fakeK3sAMI{} eip := newFakeEIPProvisioner() + sg := newFakeSGProvisioner() + worker := newFakeWorkerLauncher() svc, err := NewEKSServiceImpl(EKSServiceDeps{ NATSConn: nc, @@ -57,18 +61,19 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture { NATSURL: "nats://gw.local:4222", NATSToken: "s3cr3t-token", NATSCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", - VPCSG: newFakeSGProvisioner(), + VPCSG: sg, VPCK3s: vpc, VPCSubnet: fakeSubnetResolver{}, NLB: nlb, Instance: inst, Image: ami, EIP: eip, + Worker: worker, }) require.NoError(t, err) t.Cleanup(svc.Shutdown) - return &eksServiceFixture{svc: svc, kv: kv, nlb: nlb, inst: inst, vpc: vpc, ami: ami, eip: eip} + return &eksServiceFixture{svc: svc, kv: kv, nlb: nlb, inst: inst, vpc: vpc, ami: ami, eip: eip, sg: sg, worker: worker} } // deleteClusterFixture is an eksServiceFixture pre-seeded with a CREATING diff --git a/spinifex/handlers/eks/k3s_server_vm.go b/spinifex/handlers/eks/k3s_server_vm.go index 05f0a7ae1..b92f01c2b 100644 --- a/spinifex/handlers/eks/k3s_server_vm.go +++ b/spinifex/handlers/eks/k3s_server_vm.go @@ -71,6 +71,11 @@ const ( // (see scripts/images/eks-server/k3s-first-boot.sh ENVFILE). k3sFirstBootEnvPath = "/etc/spinifex-eks/first-boot.env" + // agentEnvPath is the env file the k3s-agent OpenRC service sources for its + // K3S_URL/K3S_TOKEN/K3S_NODE_NAME/K3S_NODE_LABEL. Path matches the AGENT_ENVFILE + // in scripts/images/eks-node/eks-node-role.sh and k3s-agent.initd. + agentEnvPath = "/etc/spinifex-eks/agent.env" + // k3sNATSCAPath is the on-VM destination for the NATS CA cert PEM. The K3s // VM uses it to verify the daemon's NATS TLS when publishing bootstrap // messages. Path matches k3s-first-boot.sh SPINIFEX_NATS_CA. diff --git a/spinifex/handlers/eks/nodegroup.go b/spinifex/handlers/eks/nodegroup.go new file mode 100644 index 000000000..38e4cf081 --- /dev/null +++ b/spinifex/handlers/eks/nodegroup.go @@ -0,0 +1,605 @@ +package handlers_eks + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "maps" + "sort" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/eks" + "github.com/google/uuid" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/nats-io/nats.go" +) + +// defaultNodegroupInstanceType mirrors the AWS EKS managed-nodegroup default +// when the caller omits instanceTypes. +const defaultNodegroupInstanceType = "t3.medium" + +// ErrNodegroupNotFound is returned by GetNodegroupRecord when the record key is +// absent. Callers translate to the AWS shape (ResourceNotFoundException) at the +// service boundary. +var ErrNodegroupNotFound = errors.New("eks: nodegroup not found") + +// NodegroupARN composes a nodegroup ARN matching the AWS shape +// (.../nodegroup/{cluster}/{ng}/{uuid}). The trailing UUID is the per-nodegroup +// discriminator AWS appends; it is generated once at create time and persisted. +func NodegroupARN(region, accountID, cluster, ng, id string) string { + return fmt.Sprintf("arn:aws:eks:%s:%s:nodegroup/%s/%s/%s", region, accountID, cluster, ng, id) +} + +// PutNodegroupRecord writes the record unconditionally. +func PutNodegroupRecord(kv nats.KeyValue, rec *NodegroupRecord) error { + if rec == nil { + return errors.New("eks: PutNodegroupRecord nil record") + } + if rec.ClusterName == "" || rec.Name == "" { + return errors.New("eks: PutNodegroupRecord missing cluster or name") + } + data, err := json.Marshal(rec) + if err != nil { + return fmt.Errorf("marshal nodegroup %s: %w", rec.Name, err) + } + key := NodegroupKey(rec.ClusterName, rec.Name) + if _, err := kv.Put(key, data); err != nil { + return fmt.Errorf("kv put %s: %w", key, err) + } + return nil +} + +// GetNodegroupRecord reads one record. Returns ErrNodegroupNotFound if absent. +func GetNodegroupRecord(kv nats.KeyValue, cluster, ng string) (*NodegroupRecord, error) { + if cluster == "" || ng == "" { + return nil, errors.New("eks: GetNodegroupRecord empty cluster or name") + } + entry, err := kv.Get(NodegroupKey(cluster, ng)) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, ErrNodegroupNotFound + } + return nil, fmt.Errorf("kv get nodegroup: %w", err) + } + var rec NodegroupRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return nil, fmt.Errorf("unmarshal nodegroup %s: %w", ng, err) + } + return &rec, nil +} + +// ListNodegroupRecords returns every nodegroup record under a cluster, sorted +// by name for stable output. +func ListNodegroupRecords(kv nats.KeyValue, cluster string) ([]*NodegroupRecord, error) { + if cluster == "" { + return nil, errors.New("eks: ListNodegroupRecords empty cluster") + } + keys, err := kv.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + return nil, fmt.Errorf("kv keys: %w", err) + } + prefix := NodegroupsPrefix(cluster) + out := make([]*NodegroupRecord, 0) + for _, k := range keys { + if !strings.HasPrefix(k, prefix) { + continue + } + entry, err := kv.Get(k) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + continue + } + return nil, fmt.Errorf("kv get %s: %w", k, err) + } + var rec NodegroupRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return nil, fmt.Errorf("unmarshal nodegroup %s: %w", k, err) + } + out = append(out, &rec) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out, nil +} + +// DeleteNodegroupRecord removes one record. A missing key is a no-op so +// DeleteNodegroup stays idempotent. +func DeleteNodegroupRecord(kv nats.KeyValue, cluster, ng string) error { + key := NodegroupKey(cluster, ng) + if err := kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + return fmt.Errorf("kv delete %s: %w", key, err) + } + return nil +} + +// nodegroupAcctKV opens the per-account bucket for nodegroup handlers. +func (s *EKSServiceImpl) nodegroupAcctKV(accountID string) (nats.KeyValue, error) { + js, err := s.deps.NATSConn.JetStream() + if err != nil { + return nil, fmt.Errorf("jetstream: %w", err) + } + acctKV, err := GetOrCreateAccountBucket(js, accountID) + if err != nil { + return nil, fmt.Errorf("get account bucket: %w", err) + } + return acctKV, nil +} + +func (s *EKSServiceImpl) createNodegroup(acctKV nats.KeyValue, input *eks.CreateNodegroupInput, accountID string) (*eks.CreateNodegroupOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + ng := aws.StringValue(input.NodegroupName) + if cluster == "" || ng == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + subnets := aws.StringValueSlice(input.Subnets) + if len(subnets) == 0 { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + meta, err := GetClusterMeta(acctKV, cluster) + if err != nil { + if errors.Is(err, ErrClusterNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + if meta.Status != ClusterStatusActive { + slog.Warn("createNodegroup: cluster not ACTIVE", "cluster", cluster, "status", meta.Status) + return nil, errors.New(awserrors.ErrorInvalidRequest) + } + if meta.ControlPlaneENIIP == "" { + slog.Error("createNodegroup: cluster has no control-plane ENI IP", "cluster", cluster) + return nil, errors.New(awserrors.ErrorInvalidRequest) + } + if meta.ResourcesVpcConfig == nil || meta.ResourcesVpcConfig.VpcId == "" { + slog.Error("createNodegroup: cluster has no VPC", "cluster", cluster) + return nil, errors.New(awserrors.ErrorInvalidRequest) + } + + if _, err := GetNodegroupRecord(acctKV, cluster, ng); err == nil { + return nil, errors.New(awserrors.ErrorEKSResourceInUse) + } else if !errors.Is(err, ErrNodegroupNotFound) { + return nil, err + } + + instanceTypes := aws.StringValueSlice(input.InstanceTypes) + if len(instanceTypes) == 0 { + instanceTypes = []string{defaultNodegroupInstanceType} + } + minSize, maxSize, desired := scalingFromInput(input.ScalingConfig) + + amiType := aws.StringValue(input.AmiType) + if amiType == "" { + amiType = eks.AMITypesAl2X8664 + } + version := aws.StringValue(input.Version) + if version == "" { + version = meta.Version + } + + now := time.Now().UTC() + rec := &NodegroupRecord{ + ClusterName: cluster, + Name: ng, + Arn: NodegroupARN(s.deps.Region, accountID, cluster, ng, uuid.NewString()), + Status: eks.NodegroupStatusCreating, + Subnets: subnets, + InstanceTypes: instanceTypes, + AMIType: amiType, + DiskSize: aws.Int64Value(input.DiskSize), + ScalingMin: minSize, + ScalingMax: maxSize, + ScalingDesired: desired, + Version: version, + NodeRole: aws.StringValue(input.NodeRole), + Labels: aws.StringValueMap(input.Labels), + CreatedAt: now, + ModifiedAt: now, + } + + cpSGID, ngSGID, err := EnsureClusterSGs(s.deps.VPCSG, accountID, cluster, meta.ResourcesVpcConfig.VpcId) + if err != nil { + return nil, fmt.Errorf("ensure cluster SGs: %w", err) + } + if err := EnsureNodegroupSGRules(s.deps.VPCSG, accountID, cluster, cpSGID, ngSGID); err != nil { + return nil, fmt.Errorf("ensure nodegroup SG rules: %w", err) + } + + if err := PutNodegroupRecord(acctKV, rec); err != nil { + return nil, err + } + + token, err := s.decryptNodeToken(acctKV, cluster) + if err != nil { + s.markNodegroupFailed(acctKV, cluster, ng, "decrypt node token: "+err.Error()) + return nil, fmt.Errorf("decrypt node token: %w", err) + } + + amiID, err := lookupEKSServerAMI(s.deps.Image, accountID) + if err != nil { + s.markNodegroupFailed(acctKV, cluster, ng, "resolve eks-node AMI: "+err.Error()) + if errors.Is(err, ErrEKSServerAMINotFound) { + return nil, errors.New(awserrors.ErrorServiceUnavailable) + } + return nil, fmt.Errorf("resolve eks-node AMI: %w", err) + } + + instanceIDs, err := s.launchWorkers(accountID, rec, meta, ngSGID, amiID, token, int(desired)) + rec.InstanceIDs = instanceIDs + if err != nil { + // Persist whatever launched so DeleteNodegroup can reclaim it, then fail. + rec.Status = eks.NodegroupStatusCreateFailed + rec.StatusReason = "launch workers: " + err.Error() + rec.ModifiedAt = time.Now().UTC() + if perr := PutNodegroupRecord(acctKV, rec); perr != nil { + slog.Error("createNodegroup: persist CREATE_FAILED record", "cluster", cluster, "nodegroup", ng, "err", perr) + } + return nil, fmt.Errorf("launch workers: %w", err) + } + + // v1 marks ACTIVE on successful RunInstances (instance-level), not on + // k8s node-Ready — node-Ready gating needs a kubectl probe from the + // reconciler and lands in a follow-up. + rec.Status = eks.NodegroupStatusActive + rec.ModifiedAt = time.Now().UTC() + if err := PutNodegroupRecord(acctKV, rec); err != nil { + return nil, err + } + + return &eks.CreateNodegroupOutput{Nodegroup: nodegroupRecordToAWS(rec)}, nil +} + +// launchWorkers issues count customer-owned RunInstances calls (one per worker +// so each gets a distinct node name + node label in its user-data) and returns +// the launched instance IDs. On a partial failure it returns the IDs that did +// launch plus the error so the caller can persist them for teardown. +func (s *EKSServiceImpl) launchWorkers(accountID string, rec *NodegroupRecord, meta *ClusterMeta, ngSGID, amiID, token string, count int) ([]string, error) { + instanceType := defaultNodegroupInstanceType + if len(rec.InstanceTypes) > 0 { + instanceType = rec.InstanceTypes[0] + } + subnet := rec.Subnets[0] + + ids := make([]string, 0, count) + for i := range count { + shortID := uuid.NewString()[:8] + userData := buildAgentUserData(agentUserDataInput{ + ClusterName: rec.ClusterName, + NodegroupName: rec.Name, + ControlPlaneENIIP: meta.ControlPlaneENIIP, + JoinToken: token, + NodeName: fmt.Sprintf("%s-%s-%s", rec.ClusterName, rec.Name, shortID), + }) + runInput := &ec2.RunInstancesInput{ + ImageId: aws.String(amiID), + InstanceType: aws.String(instanceType), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + SubnetId: aws.String(subnet), + SecurityGroupIds: aws.StringSlice([]string{ngSGID}), + UserData: aws.String(userData), + TagSpecifications: []*ec2.TagSpecification{{ + ResourceType: aws.String("instance"), + Tags: []*ec2.Tag{ + {Key: aws.String(clusterEKSClusterTagKey), Value: aws.String(rec.ClusterName)}, + {Key: aws.String(clusterEKSNodegroupTagKey), Value: aws.String(rec.Name)}, + }, + }}, + } + res, err := s.deps.Worker.RunWorkerInstance(runInput, accountID) + if err != nil { + return ids, fmt.Errorf("run worker %d/%d: %w", i+1, count, err) + } + for _, inst := range res.Instances { + if id := aws.StringValue(inst.InstanceId); id != "" { + ids = append(ids, id) + } + } + } + return ids, nil +} + +func (s *EKSServiceImpl) describeNodegroup(acctKV nats.KeyValue, input *eks.DescribeNodegroupInput) (*eks.DescribeNodegroupOutput, error) { + cluster := aws.StringValue(input.ClusterName) + ng := aws.StringValue(input.NodegroupName) + if cluster == "" || ng == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + rec, err := GetNodegroupRecord(acctKV, cluster, ng) + if err != nil { + if errors.Is(err, ErrNodegroupNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return &eks.DescribeNodegroupOutput{Nodegroup: nodegroupRecordToAWS(rec)}, nil +} + +func (s *EKSServiceImpl) listNodegroups(acctKV nats.KeyValue, input *eks.ListNodegroupsInput) (*eks.ListNodegroupsOutput, error) { + cluster := aws.StringValue(input.ClusterName) + if cluster == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + if _, err := GetClusterMeta(acctKV, cluster); err != nil { + if errors.Is(err, ErrClusterNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + recs, err := ListNodegroupRecords(acctKV, cluster) + if err != nil { + return nil, err + } + names := make([]string, 0, len(recs)) + for _, rec := range recs { + names = append(names, rec.Name) + } + return &eks.ListNodegroupsOutput{Nodegroups: aws.StringSlice(names)}, nil +} + +func (s *EKSServiceImpl) updateNodegroupConfig(acctKV nats.KeyValue, input *eks.UpdateNodegroupConfigInput, accountID string) (*eks.UpdateNodegroupConfigOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + ng := aws.StringValue(input.NodegroupName) + if cluster == "" || ng == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + meta, err := GetClusterMeta(acctKV, cluster) + if err != nil { + if errors.Is(err, ErrClusterNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + rec, err := GetNodegroupRecord(acctKV, cluster, ng) + if err != nil { + if errors.Is(err, ErrNodegroupNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + + if input.ScalingConfig != nil { + if v := input.ScalingConfig.MinSize; v != nil { + rec.ScalingMin = *v + } + if v := input.ScalingConfig.MaxSize; v != nil { + rec.ScalingMax = *v + } + if v := input.ScalingConfig.DesiredSize; v != nil { + rec.ScalingDesired = *v + } + } + if input.Labels != nil { + rec.Labels = applyLabelUpdate(rec.Labels, input.Labels) + } + + if err := s.reconcileWorkerCount(acctKV, accountID, rec, meta); err != nil { + // reconcileWorkerCount may have launched some workers before failing; their + // IDs are already on rec. Persist so DeleteNodegroup can reclaim them rather + // than leaking the partial scale-up. + rec.ModifiedAt = time.Now().UTC() + if perr := PutNodegroupRecord(acctKV, rec); perr != nil { + slog.Error("updateNodegroupConfig: persist after reconcile failure", + "cluster", cluster, "nodegroup", ng, "err", perr) + } + return nil, err + } + + rec.ModifiedAt = time.Now().UTC() + if err := PutNodegroupRecord(acctKV, rec); err != nil { + return nil, err + } + + return &eks.UpdateNodegroupConfigOutput{Update: &eks.Update{ + Id: aws.String(uuid.NewString()), + Status: aws.String(eks.UpdateStatusSuccessful), + Type: aws.String(eks.UpdateTypeConfigUpdate), + CreatedAt: aws.Time(rec.ModifiedAt), + }}, nil +} + +// reconcileWorkerCount launches or terminates workers so len(InstanceIDs) +// matches ScalingDesired. Surplus removal terminates the last (highest-index) +// instance IDs so scale-down is deterministic. +func (s *EKSServiceImpl) reconcileWorkerCount(acctKV nats.KeyValue, accountID string, rec *NodegroupRecord, meta *ClusterMeta) error { + current := len(rec.InstanceIDs) + desired := int(rec.ScalingDesired) + + switch { + case desired > current: + token, err := s.decryptNodeToken(acctKV, rec.ClusterName) + if err != nil { + return fmt.Errorf("decrypt node token: %w", err) + } + amiID, err := lookupEKSServerAMI(s.deps.Image, accountID) + if err != nil { + if errors.Is(err, ErrEKSServerAMINotFound) { + return errors.New(awserrors.ErrorServiceUnavailable) + } + return fmt.Errorf("resolve eks-node AMI: %w", err) + } + _, ngSGID, err := EnsureClusterSGs(s.deps.VPCSG, accountID, rec.ClusterName, meta.ResourcesVpcConfig.VpcId) + if err != nil { + return fmt.Errorf("ensure cluster SGs: %w", err) + } + newIDs, err := s.launchWorkers(accountID, rec, meta, ngSGID, amiID, token, desired-current) + rec.InstanceIDs = append(rec.InstanceIDs, newIDs...) + if err != nil { + return fmt.Errorf("launch workers: %w", err) + } + case desired < current: + surplus := rec.InstanceIDs[desired:] + if err := s.deps.Worker.TerminateWorkerInstances(surplus, accountID); err != nil { + return fmt.Errorf("terminate surplus workers: %w", err) + } + rec.InstanceIDs = rec.InstanceIDs[:desired] + } + return nil +} + +func (s *EKSServiceImpl) updateNodegroupVersion(input *eks.UpdateNodegroupVersionInput) (*eks.UpdateNodegroupVersionOutput, error) { + // Worker AMI version upgrades (drain + replace) land in a follow-up; v1 + // nodegroups are fixed at their create-time version. + if input != nil { + slog.Info("UpdateNodegroupVersion not implemented in v1", + "cluster", aws.StringValue(input.ClusterName), "nodegroup", aws.StringValue(input.NodegroupName)) + } + return nil, notImpl() +} + +func (s *EKSServiceImpl) deleteNodegroup(acctKV nats.KeyValue, input *eks.DeleteNodegroupInput, accountID string) (*eks.DeleteNodegroupOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + ng := aws.StringValue(input.NodegroupName) + if cluster == "" || ng == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + rec, err := GetNodegroupRecord(acctKV, cluster, ng) + if err != nil { + if errors.Is(err, ErrNodegroupNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + + if len(rec.InstanceIDs) > 0 { + if err := s.deps.Worker.TerminateWorkerInstances(rec.InstanceIDs, accountID); err != nil { + return nil, fmt.Errorf("terminate workers: %w", err) + } + } + if err := DeleteNodegroupRecord(acctKV, cluster, ng); err != nil { + return nil, err + } + + rec.Status = eks.NodegroupStatusDeleting + return &eks.DeleteNodegroupOutput{Nodegroup: nodegroupRecordToAWS(rec)}, nil +} + +// decryptNodeToken reads + decrypts the cluster's K3s join token from KV. +func (s *EKSServiceImpl) decryptNodeToken(kv nats.KeyValue, cluster string) (string, error) { + if kv == nil { + return "", errors.New("eks: nil KV for node token") + } + entry, err := kv.Get(NodeTokenKey(cluster)) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return "", errors.New("eks: cluster join token not yet provisioned") + } + return "", fmt.Errorf("kv get node token: %w", err) + } + token, err := handlers_iam.DecryptSecret(string(entry.Value()), s.deps.MasterKey) + if err != nil { + return "", fmt.Errorf("decrypt node token: %w", err) + } + if token == "" { + return "", errors.New("eks: decrypted node token is empty") + } + return token, nil +} + +func (s *EKSServiceImpl) markNodegroupFailed(kv nats.KeyValue, cluster, ng, reason string) { + rec, err := GetNodegroupRecord(kv, cluster, ng) + if err != nil { + return + } + rec.Status = eks.NodegroupStatusCreateFailed + rec.StatusReason = reason + rec.ModifiedAt = time.Now().UTC() + if err := PutNodegroupRecord(kv, rec); err != nil { + slog.Warn("markNodegroupFailed: persist failed", "cluster", cluster, "nodegroup", ng, "err", err) + } +} + +// scalingFromInput derives min/max/desired from a NodegroupScalingConfig, +// defaulting to a single node when the caller omits the config or fields. +func scalingFromInput(cfg *eks.NodegroupScalingConfig) (minSize, maxSize, desired int64) { + minSize, maxSize, desired = 1, 1, 1 + if cfg == nil { + return minSize, maxSize, desired + } + if cfg.MinSize != nil { + minSize = *cfg.MinSize + } + if cfg.DesiredSize != nil { + desired = *cfg.DesiredSize + } else { + desired = minSize + } + if cfg.MaxSize != nil { + maxSize = *cfg.MaxSize + } else if desired > maxSize { + maxSize = desired + } + return minSize, maxSize, desired +} + +// applyLabelUpdate applies an UpdateLabelsPayload (addOrUpdate + removeLabels) +// onto the record's label map. +func applyLabelUpdate(existing map[string]string, payload *eks.UpdateLabelsPayload) map[string]string { + if payload == nil { + return existing + } + out := map[string]string{} + maps.Copy(out, existing) + for k, v := range payload.AddOrUpdateLabels { + out[k] = aws.StringValue(v) + } + for _, k := range payload.RemoveLabels { + delete(out, aws.StringValue(k)) + } + return out +} + +func nodegroupRecordToAWS(rec *NodegroupRecord) *eks.Nodegroup { + if rec == nil { + return nil + } + out := &eks.Nodegroup{ + ClusterName: aws.String(rec.ClusterName), + NodegroupName: aws.String(rec.Name), + NodegroupArn: aws.String(rec.Arn), + Status: aws.String(rec.Status), + Subnets: aws.StringSlice(rec.Subnets), + InstanceTypes: aws.StringSlice(rec.InstanceTypes), + AmiType: aws.String(rec.AMIType), + NodeRole: aws.String(rec.NodeRole), + Version: aws.String(rec.Version), + CreatedAt: aws.Time(rec.CreatedAt), + ModifiedAt: aws.Time(rec.ModifiedAt), + ScalingConfig: &eks.NodegroupScalingConfig{ + MinSize: aws.Int64(rec.ScalingMin), + MaxSize: aws.Int64(rec.ScalingMax), + DesiredSize: aws.Int64(rec.ScalingDesired), + }, + Resources: &eks.NodegroupResources{}, + } + if rec.DiskSize > 0 { + out.DiskSize = aws.Int64(rec.DiskSize) + } + if len(rec.Labels) > 0 { + out.Labels = aws.StringMap(rec.Labels) + } + if rec.StatusReason != "" { + out.Health = &eks.NodegroupHealth{Issues: []*eks.Issue{{ + Code: aws.String(eks.NodegroupIssueCodeNodeCreationFailure), + Message: aws.String(rec.StatusReason), + ResourceIds: aws.StringSlice(rec.InstanceIDs), + }}} + } + return out +} diff --git a/spinifex/handlers/eks/nodegroup_test.go b/spinifex/handlers/eks/nodegroup_test.go new file mode 100644 index 000000000..d3432f9f3 --- /dev/null +++ b/spinifex/handlers/eks/nodegroup_test.go @@ -0,0 +1,321 @@ +package handlers_eks + +import ( + "errors" + "strconv" + "strings" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/eks" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeWorkerLauncher records RunWorkerInstance / TerminateWorkerInstances calls +// and hands back sequential instance IDs so tests can assert on the IDs a +// nodegroup tracks. +type fakeWorkerLauncher struct { + runCalls []*ec2.RunInstancesInput + terminateCalls [][]string + + nextID int + runErr error + termErr error +} + +var _ WorkerLauncher = (*fakeWorkerLauncher)(nil) + +func newFakeWorkerLauncher() *fakeWorkerLauncher { + return &fakeWorkerLauncher{} +} + +func (f *fakeWorkerLauncher) RunWorkerInstance(input *ec2.RunInstancesInput, _ string) (*ec2.Reservation, error) { + f.runCalls = append(f.runCalls, input) + if f.runErr != nil { + return nil, f.runErr + } + f.nextID++ + id := "i-worker" + strconv.Itoa(f.nextID) + return &ec2.Reservation{Instances: []*ec2.Instance{{InstanceId: aws.String(id)}}}, nil +} + +func (f *fakeWorkerLauncher) TerminateWorkerInstances(ids []string, _ string) error { + f.terminateCalls = append(f.terminateCalls, ids) + return f.termErr +} + +// seedActiveClusterWithToken lays down an ACTIVE cluster meta with the control- +// plane ENI IP + VPC populated and an encrypted K3s join token, so the +// nodegroup path can run end-to-end. +func seedActiveClusterWithToken(t *testing.T, f *eksServiceFixture, cluster string) { + t.Helper() + meta := &ClusterMeta{ + Name: cluster, + Status: ClusterStatusActive, + Version: "1.32", + ControlPlaneENIIP: "10.0.1.42", + ResourcesVpcConfig: &ClusterVpcConfig{ + SubnetIds: []string{"subnet-aaa"}, + VpcId: "vpc-aaa", + }, + } + require.NoError(t, PutClusterMeta(f.kv, meta)) + + ct, err := handlers_iam.EncryptSecret("K10node-join-token::server:abc", bootstrapTestMasterKey) + require.NoError(t, err) + _, err = f.kv.Put(NodeTokenKey(cluster), []byte(ct)) + require.NoError(t, err) +} + +func createNGInput(cluster, ng string, desired int64) *eks.CreateNodegroupInput { + return &eks.CreateNodegroupInput{ + ClusterName: aws.String(cluster), + NodegroupName: aws.String(ng), + NodeRole: aws.String("arn:aws:iam::111122223333:role/eks-node"), + Subnets: aws.StringSlice([]string{"subnet-aaa"}), + ScalingConfig: &eks.NodegroupScalingConfig{ + MinSize: aws.Int64(1), + MaxSize: aws.Int64(3), + DesiredSize: aws.Int64(desired), + }, + } +} + +func TestNodegroupRecord_CRUDRoundTrip(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + kv, err := GetOrCreateAccountBucket(js, testAccountID) + require.NoError(t, err) + + rec := &NodegroupRecord{ + ClusterName: "c1", + Name: "ng1", + Arn: "arn:aws:eks:us-east-1:111122223333:nodegroup/c1/ng1/abc", + Status: eks.NodegroupStatusActive, + Subnets: []string{"subnet-aaa"}, + InstanceTypes: []string{"t3.medium"}, + ScalingMin: 1, + ScalingMax: 3, + ScalingDesired: 2, + InstanceIDs: []string{"i-1", "i-2"}, + } + require.NoError(t, PutNodegroupRecord(kv, rec)) + + got, err := GetNodegroupRecord(kv, "c1", "ng1") + require.NoError(t, err) + assert.Equal(t, rec.InstanceIDs, got.InstanceIDs) + assert.Equal(t, int64(2), got.ScalingDesired) + + list, err := ListNodegroupRecords(kv, "c1") + require.NoError(t, err) + require.Len(t, list, 1) + assert.Equal(t, "ng1", list[0].Name) + + require.NoError(t, DeleteNodegroupRecord(kv, "c1", "ng1")) + _, err = GetNodegroupRecord(kv, "c1", "ng1") + assert.ErrorIs(t, err, ErrNodegroupNotFound) + + // Delete of an absent record is idempotent. + require.NoError(t, DeleteNodegroupRecord(kv, "c1", "ng1")) +} + +func TestBuildAgentUserData_Shape(t *testing.T) { + ud := buildAgentUserData(agentUserDataInput{ + ClusterName: "c1", + NodegroupName: "ng1", + ControlPlaneENIIP: "10.0.1.42", + JoinToken: "K10secret::server:xyz", + NodeName: "c1-ng1-abc123de", + }) + + assert.True(t, strings.HasPrefix(ud, "#cloud-config\n")) + assert.Contains(t, ud, "SPINIFEX_K3S_ROLE=agent") + assert.Contains(t, ud, "K3S_URL=https://10.0.1.42:6443") + assert.Contains(t, ud, "K3S_TOKEN=K10secret::server:xyz") + assert.Contains(t, ud, "K3S_NODE_NAME=c1-ng1-abc123de") + assert.Contains(t, ud, "eks.amazonaws.com/nodegroup=ng1") + assert.Contains(t, ud, agentEnvPath) + // IMDS on-link route line present. + assert.Contains(t, ud, "ip route replace "+imdsServerIP+"/32") + // Exactly one write_files block (single "write_files:" key). + assert.Equal(t, 1, strings.Count(ud, "write_files:")) +} + +func TestEnsureNodegroupSGRules_AuthorizesExpectedRules(t *testing.T) { + sg := newFakeSGProvisioner() + + require.NoError(t, EnsureNodegroupSGRules(sg, testAccountID, "c1", "sg-cp", "sg-ng")) + + require.Len(t, sg.authorizeCalls, 6) + + // Verify the apiserver rule: 6443/tcp on cpSG sourced from ngSG. + var found6443 bool + for _, call := range sg.authorizeCalls { + require.Len(t, call.IpPermissions, 1) + perm := call.IpPermissions[0] + if aws.StringValue(call.GroupId) == "sg-cp" && + aws.StringValue(perm.IpProtocol) == "tcp" && + aws.Int64Value(perm.FromPort) == 6443 { + require.Len(t, perm.UserIdGroupPairs, 1) + assert.Equal(t, "sg-ng", aws.StringValue(perm.UserIdGroupPairs[0].GroupId)) + assert.Empty(t, perm.IpRanges, "rules must be group-to-group, not CIDR") + found6443 = true + } + } + assert.True(t, found6443, "expected agent→apiserver 6443/tcp rule") +} + +func TestEnsureNodegroupSGRules_DuplicateRuleTolerated(t *testing.T) { + sg := newFakeSGProvisioner() + sg.authorizeErr = errors.New(awserrors.ErrorInvalidPermissionDuplicate) + + require.NoError(t, EnsureNodegroupSGRules(sg, testAccountID, "c1", "sg-cp", "sg-ng"), + "duplicate-rule error must be treated as success") +} + +func TestCreateNodegroup_HappyPath(t *testing.T) { + f := newEKSServiceFixture(t) + seedActiveClusterWithToken(t, f, "c1") + + out, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 2), testAccountID) + require.NoError(t, err) + require.NotNil(t, out.Nodegroup) + assert.Equal(t, eks.NodegroupStatusActive, aws.StringValue(out.Nodegroup.Status)) + assert.Equal(t, int64(2), aws.Int64Value(out.Nodegroup.ScalingConfig.DesiredSize)) + assert.Contains(t, aws.StringValue(out.Nodegroup.NodegroupArn), ":nodegroup/c1/ng1/") + + // Two workers launched and tracked. + assert.Len(t, f.worker.runCalls, 2) + rec, err := GetNodegroupRecord(f.kv, "c1", "ng1") + require.NoError(t, err) + assert.Len(t, rec.InstanceIDs, 2) + + // Workers go through the customer path: no ManagedBy tag, SG = nodegroup SG. + for _, call := range f.worker.runCalls { + require.NotEmpty(t, call.SecurityGroupIds) + assert.Equal(t, "subnet-aaa", aws.StringValue(call.SubnetId)) + for _, ts := range call.TagSpecifications { + for _, tg := range ts.Tags { + assert.NotEqual(t, "spinifex:managed-by", aws.StringValue(tg.Key)) + } + } + } + + // SG ingress rules were authorized. + assert.NotEmpty(t, f.sg.authorizeCalls) + + // Duplicate create → ResourceInUseException. + _, err = f.svc.CreateNodegroup(createNGInput("c1", "ng1", 2), testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceInUse) +} + +func TestCreateNodegroup_ClusterNotActive(t *testing.T) { + f := newEKSServiceFixture(t) + require.NoError(t, PutClusterMeta(f.kv, &ClusterMeta{ + Name: "c1", + Status: ClusterStatusCreating, + ControlPlaneENIIP: "10.0.1.42", + ResourcesVpcConfig: &ClusterVpcConfig{VpcId: "vpc-aaa"}, + })) + + _, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID) + require.EqualError(t, err, awserrors.ErrorInvalidRequest) +} + +func TestCreateNodegroup_UnknownCluster(t *testing.T) { + f := newEKSServiceFixture(t) + _, err := f.svc.CreateNodegroup(createNGInput("ghost", "ng1", 1), testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) +} + +func TestDescribeAndListNodegroups(t *testing.T) { + f := newEKSServiceFixture(t) + seedActiveClusterWithToken(t, f, "c1") + _, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID) + require.NoError(t, err) + + desc, err := f.svc.DescribeNodegroup(&eks.DescribeNodegroupInput{ + ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, "ng1", aws.StringValue(desc.Nodegroup.NodegroupName)) + assert.Equal(t, "c1", aws.StringValue(desc.Nodegroup.ClusterName)) + + list, err := f.svc.ListNodegroups(&eks.ListNodegroupsInput{ClusterName: aws.String("c1")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, []string{"ng1"}, aws.StringValueSlice(list.Nodegroups)) + + _, err = f.svc.DescribeNodegroup(&eks.DescribeNodegroupInput{ + ClusterName: aws.String("c1"), NodegroupName: aws.String("ghost"), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) +} + +func TestUpdateNodegroupConfig_ScaleUpAndDown(t *testing.T) { + f := newEKSServiceFixture(t) + seedActiveClusterWithToken(t, f, "c1") + _, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 1), testAccountID) + require.NoError(t, err) + require.Len(t, f.worker.runCalls, 1) + + // Scale up 1 → 3: two new workers launched. + upOut, err := f.svc.UpdateNodegroupConfig(&eks.UpdateNodegroupConfigInput{ + ClusterName: aws.String("c1"), + NodegroupName: aws.String("ng1"), + ScalingConfig: &eks.NodegroupScalingConfig{DesiredSize: aws.Int64(3)}, + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, eks.UpdateStatusSuccessful, aws.StringValue(upOut.Update.Status)) + assert.Len(t, f.worker.runCalls, 3) + rec, err := GetNodegroupRecord(f.kv, "c1", "ng1") + require.NoError(t, err) + assert.Len(t, rec.InstanceIDs, 3) + + // Scale down 3 → 1: surplus (last two) terminated. + _, err = f.svc.UpdateNodegroupConfig(&eks.UpdateNodegroupConfigInput{ + ClusterName: aws.String("c1"), + NodegroupName: aws.String("ng1"), + ScalingConfig: &eks.NodegroupScalingConfig{DesiredSize: aws.Int64(1)}, + }, testAccountID) + require.NoError(t, err) + require.Len(t, f.worker.terminateCalls, 1) + assert.Len(t, f.worker.terminateCalls[0], 2) + rec, err = GetNodegroupRecord(f.kv, "c1", "ng1") + require.NoError(t, err) + assert.Len(t, rec.InstanceIDs, 1) +} + +func TestDeleteNodegroup_TerminatesAndIsIdempotent(t *testing.T) { + f := newEKSServiceFixture(t) + seedActiveClusterWithToken(t, f, "c1") + _, err := f.svc.CreateNodegroup(createNGInput("c1", "ng1", 2), testAccountID) + require.NoError(t, err) + + _, err = f.svc.DeleteNodegroup(&eks.DeleteNodegroupInput{ + ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"), + }, testAccountID) + require.NoError(t, err) + require.Len(t, f.worker.terminateCalls, 1) + assert.Len(t, f.worker.terminateCalls[0], 2) + + _, err = GetNodegroupRecord(f.kv, "c1", "ng1") + assert.ErrorIs(t, err, ErrNodegroupNotFound) + + // Second delete → ResourceNotFoundException (record already gone). + _, err = f.svc.DeleteNodegroup(&eks.DeleteNodegroupInput{ + ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) +} + +func TestUpdateNodegroupVersion_NotImplemented(t *testing.T) { + f := newEKSServiceFixture(t) + _, err := f.svc.UpdateNodegroupVersion(&eks.UpdateNodegroupVersionInput{ + ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1"), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorNotImplemented) +} diff --git a/spinifex/handlers/eks/nodegroup_userdata.go b/spinifex/handlers/eks/nodegroup_userdata.go new file mode 100644 index 000000000..40329d2bd --- /dev/null +++ b/spinifex/handlers/eks/nodegroup_userdata.go @@ -0,0 +1,90 @@ +package handlers_eks + +import ( + "fmt" + "net" + "strings" +) + +// agentUserDataInput is the input shape for buildAgentUserData. ControlPlaneENIIP +// is the cluster's in-VPC control-plane ENI private IP (the K3s join target); +// JoinToken is the decrypted K3s node token. +type agentUserDataInput struct { + ClusterName string + NodegroupName string + ControlPlaneENIIP string + JoinToken string + NodeName string +} + +// buildAgentUserData renders the cloud-config YAML a nodegroup worker consumes +// at first boot. It mirrors buildK3sUserData: a bootcmd resolver fix, a single +// write_files block (first-boot.env + agent.env + the IMDS on-link route +// script), and a runcmd that enables the OpenRC `local` service. Output is +// unencoded YAML; the caller base64-wraps it for the RunInstances UserData +// field. The eks-node-role first-boot selector reads SPINIFEX_K3S_ROLE=agent +// (and the presence of agent.env) to enable the k3s-agent service, which sources +// agent.env for K3S_URL/K3S_TOKEN. +func buildAgentUserData(in agentUserDataInput) string { + // K3S_URL targets the control-plane ENI's in-VPC private IP on :6443 + // because the cluster NLB DNS endpoint is not resolvable in-VPC yet. + // Workers share the VPC with the server, so this is directly reachable and + // TLS-pinned by the join token's CA hash (SAN/hostname irrelevant for the + // k3s agent join). Switch to the NLB endpoint on :443 once authoritative + // in-VPC DNS is available. + k3sURL := "https://" + net.JoinHostPort(in.ControlPlaneENIIP, "6443") + nodeLabel := "eks.amazonaws.com/nodegroup=" + in.NodegroupName + + envBody := strings.Join([]string{ + "SPINIFEX_K3S_ROLE=agent", + }, "\n") + + agentBody := strings.Join([]string{ + "K3S_URL=" + k3sURL, + "K3S_TOKEN=" + in.JoinToken, + "K3S_NODE_NAME=" + in.NodeName, + "K3S_NODE_LABEL=" + nodeLabel, + }, "\n") + + files := []userDataFile{ + {Path: k3sFirstBootEnvPath, Perms: "0600", Body: envBody}, + {Path: agentEnvPath, Perms: "0600", Body: agentBody}, + { + Path: "/etc/local.d/imds-onlink-route.start", + Perms: "0755", + Body: "#!/bin/sh\n" + + "dev=$(ip route show default | awk '{print $5; exit}')\n" + + "[ -n \"$dev\" ] && ip route replace " + imdsServerIP + "/32 dev \"$dev\" scope link", + }, + } + + var buf strings.Builder + buf.WriteString("#cloud-config\n") + + // Resolver fix via bootcmd (not write_files) for the same reason as the + // server path: /etc/resolv.conf is a dangling symlink on the Alpine AMI, so + // write_files would follow the dead link and abort the whole block. + // Containerd needs a working resolver to pull the worker's system images. + buf.WriteString("bootcmd:\n") + buf.WriteString(" - rm -f " + k3sResolvConfPath + "\n") + fmt.Fprintf(&buf, " - printf '%s\\n' > %s\n", + strings.ReplaceAll(k3sResolvConf, "\n", "\\n"), k3sResolvConfPath) + + buf.WriteString("write_files:\n") + for _, f := range files { + fmt.Fprintf(&buf, " - path: %s\n", f.Path) + fmt.Fprintf(&buf, " permissions: '%s'\n", f.Perms) + buf.WriteString(" content: |\n") + for line := range strings.SplitSeq(f.Body, "\n") { + buf.WriteString(" ") + buf.WriteString(line) + buf.WriteString("\n") + } + } + + buf.WriteString("runcmd:\n") + buf.WriteString(" - [ rc-update, add, local, default ]\n") + buf.WriteString(" - [ rc-service, local, start ]\n") + + return buf.String() +} diff --git a/spinifex/handlers/eks/security_groups.go b/spinifex/handlers/eks/security_groups.go index 33cf083dc..126f00416 100644 --- a/spinifex/handlers/eks/security_groups.go +++ b/spinifex/handlers/eks/security_groups.go @@ -7,6 +7,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/tags" ) @@ -17,6 +18,7 @@ type sgProvisioner interface { CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error) DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error) DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error) + AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) } // clusterEKSClusterTagKey is stamped on every cluster-scoped resource so the @@ -27,6 +29,10 @@ const clusterEKSClusterTagKey = "spinifex:eks-cluster" // SG when both live in the same VPC. const clusterEKSRoleTagKey = "spinifex:eks-role" +// clusterEKSNodegroupTagKey is stamped on worker instances so the nodegroup +// they belong to is recoverable from EC2 tags alone. +const clusterEKSNodegroupTagKey = "spinifex:eks-nodegroup" + const ( clusterEKSRoleControlPlane = "control-plane" clusterEKSRoleNodegroup = "nodegroup" @@ -142,6 +148,66 @@ func ensureClusterSG(sgp sgProvisioner, accountID, vpcID, clusterName, name, des return *out.GroupId, nil } +// EnsureNodegroupSGRules authorizes the ingress rules a nodegroup's workers +// need to join the K3s control plane and form the flannel overlay. All rules +// are group-to-group (UserIdGroupPairs) so they stay correct as instances come +// and go. Idempotent: the InvalidPermission.Duplicate error a re-run triggers +// for an already-authorized rule is treated as success. +// +// Rules: +// - agent → server apiserver/supervisor: 6443/tcp (ngSG → cpSG) +// - flannel VXLAN: 8472/udp both directions between cpSG and ngSG, and +// ngSG ↔ ngSG +// - kubelet: 10250/tcp (cpSG → ngSG) +// - intra-nodegroup all traffic: ngSG self-reference +func EnsureNodegroupSGRules(sgp sgProvisioner, accountID, clusterName, cpSGID, ngSGID string) error { + if cpSGID == "" || ngSGID == "" { + return errors.New("eks: EnsureNodegroupSGRules empty SG id") + } + + type rule struct { + targetSG string + sourceSG string + proto string + from int64 + to int64 + desc string + } + rules := []rule{ + {cpSGID, ngSGID, "tcp", 6443, 6443, "EKS agent to apiserver/supervisor"}, + {cpSGID, ngSGID, "udp", 8472, 8472, "EKS flannel VXLAN node to control-plane"}, + {ngSGID, cpSGID, "udp", 8472, 8472, "EKS flannel VXLAN control-plane to node"}, + {ngSGID, ngSGID, "udp", 8472, 8472, "EKS flannel VXLAN node to node"}, + {ngSGID, cpSGID, "tcp", 10250, 10250, "EKS kubelet control-plane to node"}, + {ngSGID, ngSGID, "-1", -1, -1, "EKS intra-nodegroup all traffic"}, + } + + for _, r := range rules { + perm := &ec2.IpPermission{ + IpProtocol: aws.String(r.proto), + UserIdGroupPairs: []*ec2.UserIdGroupPair{{ + GroupId: aws.String(r.sourceSG), + Description: aws.String(r.desc), + }}, + } + if r.proto != "-1" { + perm.FromPort = aws.Int64(r.from) + perm.ToPort = aws.Int64(r.to) + } + _, err := sgp.AuthorizeSecurityGroupIngress(&ec2.AuthorizeSecurityGroupIngressInput{ + GroupId: aws.String(r.targetSG), + IpPermissions: []*ec2.IpPermission{perm}, + }, accountID) + if err != nil { + if awserrors.IsErrorCode(err, awserrors.ErrorInvalidPermissionDuplicate) { + continue + } + return fmt.Errorf("authorize %s ingress on %s: %w", r.desc, r.targetSG, err) + } + } + return nil +} + func lookupSGByName(sgp sgProvisioner, accountID, vpcID, name string) (string, error) { out, err := sgp.DescribeSecurityGroups(&ec2.DescribeSecurityGroupsInput{ Filters: []*ec2.Filter{ diff --git a/spinifex/handlers/eks/security_groups_test.go b/spinifex/handlers/eks/security_groups_test.go index b84a0e34a..75b002f42 100644 --- a/spinifex/handlers/eks/security_groups_test.go +++ b/spinifex/handlers/eks/security_groups_test.go @@ -12,9 +12,10 @@ import ( ) type fakeSGProvisioner struct { - createCalls []*ec2.CreateSecurityGroupInput - describeCalls []*ec2.DescribeSecurityGroupsInput - deleteCalls []*ec2.DeleteSecurityGroupInput + createCalls []*ec2.CreateSecurityGroupInput + describeCalls []*ec2.DescribeSecurityGroupsInput + deleteCalls []*ec2.DeleteSecurityGroupInput + authorizeCalls []*ec2.AuthorizeSecurityGroupIngressInput // existing maps "name|vpcId" → groupId for DescribeSecurityGroups lookup. existing map[string]string @@ -24,9 +25,10 @@ type fakeSGProvisioner struct { // nodegroup SG IDs. createIDs []string - createErr error - describeErr error - deleteErr error + createErr error + describeErr error + deleteErr error + authorizeErr error } var _ sgProvisioner = (*fakeSGProvisioner)(nil) @@ -84,6 +86,14 @@ func (f *fakeSGProvisioner) DeleteSecurityGroup(input *ec2.DeleteSecurityGroupIn return &ec2.DeleteSecurityGroupOutput{}, nil } +func (f *fakeSGProvisioner) AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, _ string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) { + f.authorizeCalls = append(f.authorizeCalls, input) + if f.authorizeErr != nil { + return nil, f.authorizeErr + } + return &ec2.AuthorizeSecurityGroupIngressOutput{}, nil +} + func TestEnsureClusterSGs_EmptyInputsRejected(t *testing.T) { sgp := newFakeSGProvisioner() diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index 0c35e70a1..5b0cad159 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -55,6 +55,19 @@ type EKSServiceDeps struct { Instance k3sInstanceLauncher Image k3sAMIResolver EIP eipProvisioner + Worker WorkerLauncher +} + +// WorkerLauncher is the narrow EC2 surface CreateNodegroup needs to launch and +// terminate nodegroup worker instances. Workers go through the normal +// customer-owned RunInstances path (no ManagedBy tag, no mgmt NIC) so they are +// visible in DescribeInstances and reclaimed by TerminateInstances like any +// other customer EC2. The daemon adapts its concrete instance service onto this. +// Exported so the daemon can assert *Daemon satisfies it with a compile-time +// check. +type WorkerLauncher interface { + RunWorkerInstance(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) + TerminateWorkerInstances(instanceIDs []string, accountID string) error } // eipProvisioner is the narrow EIP surface CreateCluster needs to give the @@ -226,6 +239,9 @@ func (s *EKSServiceImpl) missingOrchestrationDeps() []string { if s.deps.EIP == nil { missing = append(missing, "EIP") } + if s.deps.Worker == nil { + missing = append(missing, "Worker") + } if len(s.deps.MasterKey) == 0 { missing = append(missing, "MasterKey") } @@ -682,28 +698,63 @@ func (s *EKSServiceImpl) UpdateClusterVersion(_ *eks.UpdateClusterVersionInput, // --- Nodegroup --- -func (s *EKSServiceImpl) CreateNodegroup(_ *eks.CreateNodegroupInput, _ string) (*eks.CreateNodegroupOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) CreateNodegroup(input *eks.CreateNodegroupInput, accountID string) (*eks.CreateNodegroupOutput, error) { + if err := s.requireOrchestrationDeps("CreateNodegroup"); err != nil { + return nil, err + } + acctKV, err := s.nodegroupAcctKV(accountID) + if err != nil { + return nil, err + } + return s.createNodegroup(acctKV, input, accountID) } -func (s *EKSServiceImpl) DescribeNodegroup(_ *eks.DescribeNodegroupInput, _ string) (*eks.DescribeNodegroupOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) DescribeNodegroup(input *eks.DescribeNodegroupInput, accountID string) (*eks.DescribeNodegroupOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + acctKV, err := s.nodegroupAcctKV(accountID) + if err != nil { + return nil, err + } + return s.describeNodegroup(acctKV, input) } -func (s *EKSServiceImpl) ListNodegroups(_ *eks.ListNodegroupsInput, _ string) (*eks.ListNodegroupsOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) ListNodegroups(input *eks.ListNodegroupsInput, accountID string) (*eks.ListNodegroupsOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + acctKV, err := s.nodegroupAcctKV(accountID) + if err != nil { + return nil, err + } + return s.listNodegroups(acctKV, input) } -func (s *EKSServiceImpl) UpdateNodegroupConfig(_ *eks.UpdateNodegroupConfigInput, _ string) (*eks.UpdateNodegroupConfigOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) UpdateNodegroupConfig(input *eks.UpdateNodegroupConfigInput, accountID string) (*eks.UpdateNodegroupConfigOutput, error) { + if err := s.requireOrchestrationDeps("UpdateNodegroupConfig"); err != nil { + return nil, err + } + acctKV, err := s.nodegroupAcctKV(accountID) + if err != nil { + return nil, err + } + return s.updateNodegroupConfig(acctKV, input, accountID) } -func (s *EKSServiceImpl) UpdateNodegroupVersion(_ *eks.UpdateNodegroupVersionInput, _ string) (*eks.UpdateNodegroupVersionOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) UpdateNodegroupVersion(input *eks.UpdateNodegroupVersionInput, accountID string) (*eks.UpdateNodegroupVersionOutput, error) { + return s.updateNodegroupVersion(input) } -func (s *EKSServiceImpl) DeleteNodegroup(_ *eks.DeleteNodegroupInput, _ string) (*eks.DeleteNodegroupOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) DeleteNodegroup(input *eks.DeleteNodegroupInput, accountID string) (*eks.DeleteNodegroupOutput, error) { + if err := s.requireOrchestrationDeps("DeleteNodegroup"); err != nil { + return nil, err + } + acctKV, err := s.nodegroupAcctKV(accountID) + if err != nil { + return nil, err + } + return s.deleteNodegroup(acctKV, input, accountID) } // --- AccessEntry + AccessPolicy --- @@ -1120,21 +1171,7 @@ func (s *EKSServiceImpl) spawnBootstrap(accountID, clusterName string, kv nats.K }() } -func (s *EKSServiceImpl) spawnReconciler(accountID, clusterName string, meta *ClusterMeta) { - healthURL := "" - switch { - case meta.ControlPlaneMgmtIP != "": - // Workaround until authoritative DNS (Eclipso/Route53) lands: the - // reconciler runs on the host, outside the VPC overlay, and can neither - // resolve the NLB DNS name nor route to the VPC-internal NLB front-end. - // The control-plane VM's br-mgmt NIC is the one apiserver address - // reachable from the daemon, so probe :6443/healthz there directly. The - // probe client sets InsecureSkipVerify, so the mgmt IP need not be in the - // apiserver cert SAN. - healthURL = "https://" + net.JoinHostPort(meta.ControlPlaneMgmtIP, "6443") + "/healthz" - case meta.Endpoint != "": - healthURL = strings.TrimRight(meta.Endpoint, "/") + "/healthz" - } +func (s *EKSServiceImpl) spawnReconciler(accountID, clusterName string, _ *ClusterMeta) { js, err := s.deps.NATSConn.JetStream() if err != nil { slog.Error("spawnReconciler: jetstream", "err", err) @@ -1145,8 +1182,13 @@ func (s *EKSServiceImpl) spawnReconciler(accountID, clusterName string, meta *Cl slog.Error("spawnReconciler: account bucket", "err", err) return } + // Health gates on the control plane's NATS self-report, not an HTTP probe: + // k3s binds the apiserver to the VPC node-ip, unreachable from the host. The + // CP publishes {healthz,node_count} on the mgmt bus the daemon already shares. + stateSubject := StateSubject(accountID, clusterName) spawn := func(ctx context.Context, _, _ string) (func(), error) { - return RunClusterReconciler(ctx, s.leaderKV, acctKV, accountID, clusterName, s.deps.HolderID, healthURL) + return RunClusterReconciler(ctx, s.leaderKV, acctKV, accountID, clusterName, s.deps.HolderID, "", + WithStateSource(s.deps.NATSConn, stateSubject)) } if err := s.registry.Spawn(s.bgCtx, accountID, clusterName, spawn); err != nil { slog.Error("spawnReconciler: registry spawn", "cluster", clusterName, "err", err) diff --git a/spinifex/handlers/eks/service_impl_test.go b/spinifex/handlers/eks/service_impl_test.go index 99d990b41..f89a1214b 100644 --- a/spinifex/handlers/eks/service_impl_test.go +++ b/spinifex/handlers/eks/service_impl_test.go @@ -132,26 +132,30 @@ func TestDeleteCluster_ZeroizesOIDCKeyBeforeTeardown(t *testing.T) { assert.Equal(t, ClusterStatusDeleting, meta.Status) } -func TestEKSServiceImpl_NodegroupMethodsReturnNotImplemented(t *testing.T) { +// In shim mode (orchestration deps absent) the mutating nodegroup methods +// short-circuit to ServiceUnavailable, the read methods reach an empty +// per-account bucket and surface ResourceNotFoundException, and +// UpdateNodegroupVersion stays NotImplemented (v1 doesn't do AMI upgrades). +func TestEKSServiceImpl_NodegroupMethodsShimMode(t *testing.T) { svc := setupTestService(t) _, err := svc.CreateNodegroup(&eks.CreateNodegroupInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + require.EqualError(t, err, awserrors.ErrorServiceUnavailable) _, err = svc.DescribeNodegroup(&eks.DescribeNodegroupInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) _, err = svc.ListNodegroups(&eks.ListNodegroupsInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) _, err = svc.UpdateNodegroupConfig(&eks.UpdateNodegroupConfigInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + require.EqualError(t, err, awserrors.ErrorServiceUnavailable) _, err = svc.UpdateNodegroupVersion(&eks.UpdateNodegroupVersionInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID) require.EqualError(t, err, awserrors.ErrorNotImplemented) _, err = svc.DeleteNodegroup(&eks.DeleteNodegroupInput{ClusterName: aws.String("c1"), NodegroupName: aws.String("ng1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + require.EqualError(t, err, awserrors.ErrorServiceUnavailable) } // seedTestCluster lays down a minimal ACTIVE cluster meta in the per-account diff --git a/spinifex/handlers/eks/state_report.go b/spinifex/handlers/eks/state_report.go new file mode 100644 index 000000000..ccaa91c73 --- /dev/null +++ b/spinifex/handlers/eks/state_report.go @@ -0,0 +1,37 @@ +package handlers_eks + +import ( + "encoding/json" + "fmt" +) + +// StateSubject returns the NATS subject the control-plane VM publishes its +// periodic health self-report to: "eks.state.{accountID}.{clusterName}.server". +// The CP VM is on the management NATS bus, so this report reaches the host-side +// reconciler without any route into the VPC overlay — unlike an HTTP /healthz +// probe to the apiserver, which k3s binds to the unreachable VPC node-ip. +func StateSubject(accountID, clusterName string) string { + return fmt.Sprintf("eks.state.%s.%s.server", accountID, clusterName) +} + +// ServerStateReport is the JSON payload of a control-plane state report, emitted +// by scripts/images/eks-node/mulga-eks-state-report.sh. Healthz mirrors the +// apiserver's /healthz body ("ok" when serving); NodeCount is the number of +// nodes the apiserver reports (server + joined agents); TS is the publish time +// in unix seconds, used to reject stale reports from a CP that stopped emitting. +type ServerStateReport struct { + Healthz string `json:"healthz"` + NodeCount int `json:"node_count"` + TS int64 `json:"ts"` +} + +// Healthy reports whether the apiserver was serving at publish time. +func (s ServerStateReport) Healthy() bool { return s.Healthz == "ok" } + +func unmarshalServerStateReport(data []byte) (ServerStateReport, error) { + var r ServerStateReport + if err := json.Unmarshal(data, &r); err != nil { + return ServerStateReport{}, fmt.Errorf("eks: unmarshal server state report: %w", err) + } + return r, nil +} diff --git a/spinifex/handlers/eks/store.go b/spinifex/handlers/eks/store.go index 072091a38..5a46d4ca4 100644 --- a/spinifex/handlers/eks/store.go +++ b/spinifex/handlers/eks/store.go @@ -56,9 +56,15 @@ func ClusterMetaKey(name string) string { return fmt.Sprintf("clusters/%s/meta", name) } +// NodegroupsPrefix returns the KV key prefix under which all of a cluster's +// nodegroup records live. Used by ListNodegroups to enumerate. +func NodegroupsPrefix(cluster string) string { + return fmt.Sprintf("clusters/%s/nodegroups/", cluster) +} + // NodegroupKey returns the KV key for a nodegroup record under a cluster. func NodegroupKey(cluster, ng string) string { - return fmt.Sprintf("clusters/%s/nodegroups/%s", cluster, ng) + return NodegroupsPrefix(cluster) + ng } // AccessEntriesPrefix returns the KV key prefix under which all of a cluster's diff --git a/spinifex/handlers/eks/types.go b/spinifex/handlers/eks/types.go index 3e91d889d..cf17dd393 100644 --- a/spinifex/handlers/eks/types.go +++ b/spinifex/handlers/eks/types.go @@ -27,14 +27,29 @@ type ClusterRecord struct { } // NodegroupRecord is the persisted-state envelope for an EKS managed -// nodegroup. +// nodegroup. It is the source of truth for Describe/ListNodegroups and tracks +// the worker EC2 instance IDs the nodegroup owns so Update/DeleteNodegroup can +// scale and tear them down. type NodegroupRecord struct { - ClusterName string `json:"clusterName"` - Name string `json:"name"` - Status string `json:"status"` - DesiredSize int64 `json:"desiredSize"` - InstanceTypes []string `json:"instanceTypes,omitempty"` - CreatedAt time.Time `json:"createdAt"` + ClusterName string `json:"clusterName"` + Name string `json:"name"` + Arn string `json:"arn"` + Status string `json:"status"` + StatusReason string `json:"statusReason,omitempty"` + Subnets []string `json:"subnets,omitempty"` + InstanceTypes []string `json:"instanceTypes,omitempty"` + AMIType string `json:"amiType,omitempty"` + DiskSize int64 `json:"diskSize,omitempty"` + ScalingMin int64 `json:"scalingMin"` + ScalingMax int64 `json:"scalingMax"` + ScalingDesired int64 `json:"scalingDesired"` + Version string `json:"version,omitempty"` + NodeRole string `json:"nodeRole,omitempty"` + Labels map[string]string `json:"labels,omitempty"` + InstanceIDs []string `json:"instanceIds,omitempty"` + Health string `json:"health,omitempty"` + CreatedAt time.Time `json:"createdAt"` + ModifiedAt time.Time `json:"modifiedAt"` } // AccessEntryRecord is the persisted-state envelope for an EKS API-mode