From c7f337229f27757630b0d61ebddd19d49ab695b4 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 11 Jun 2026 10:11:23 +1000 Subject: [PATCH] feat(vpcd): drain held DHCP leases on teardown (#346) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DHCP-source external-pool leases are DHCPRELEASEd only on explicit instance/EIP terminate. Manager.Stop() preserves leases for adopt-on-reboot, so env teardown/reset kills vpcd and destroys the per-AZ lease KV without releasing — stranding every live lease (instance ENIs, EIP allocations, per-VPC LRP gateways) on the upstream DHCP server until TTL. Across CI churn with fresh client-ids the upstream scope fills and new launches get no public IP. Add Manager.DrainAll + a vpc.dhcp.drain RPC and a standalone 'spx admin cluster drain-dhcp' command that releases every held lease. Wired into both teardown paths (single-node teardown role and multi-node cluster-teardown.yml) before services stop. Stop() is unchanged, so normal restarts still adopt leases on reboot. --- cmd/spinifex/cmd/admin.go | 13 ++ cmd/spinifex/cmd/cluster.go | 74 +++++++++++ cmd/spinifex/cmd/cluster_test.go | 58 +++++++++ docs/COMMANDS.md | 1 + spinifex/network/external/dhcp/manager.go | 46 +++++++ .../network/external/dhcp/manager_test.go | 118 ++++++++++++++++++ spinifex/network/external/dhcp/wire.go | 12 ++ 7 files changed, 322 insertions(+) create mode 100644 cmd/spinifex/cmd/cluster_test.go diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 655e5ded0..aa8c21c67 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -115,6 +115,16 @@ Each phase waits for all nodes to ACK before proceeding to the next.`, Run: runClusterShutdown, } +var clusterDrainDHCPCmd = &cobra.Command{ + Use: "drain-dhcp", + Short: "Release all upstream DHCP leases held by vpcd", + Long: `Ask each vpcd to DHCPRELEASE every external-pool DHCP lease it currently +holds, returning them to the upstream DHCP server. Run this on teardown before +stopping services: an env reset otherwise strands held leases on the upstream +server until their TTL expires, eventually exhausting the upstream scope.`, + Run: runClusterDrainDHCP, +} + var adminInitCmd = &cobra.Command{ Use: "init", Short: "Initialize Spinifex platform configuration", @@ -245,6 +255,9 @@ func init() { clusterShutdownCmd.Flags().Duration("timeout", 120*time.Second, "Maximum time to wait per phase") clusterShutdownCmd.Flags().Bool("dry-run", false, "Print phase plan without executing") + clusterCmd.AddCommand(clusterDrainDHCPCmd) + clusterDrainDHCPCmd.Flags().Duration("timeout", 30*time.Second, "Reply-collection window for vpcd drain responders") + adminCmd.AddCommand(imagesCmd) imagesCmd.AddCommand(imagesImportCmd) imagesCmd.AddCommand(imagesListCmd) diff --git a/cmd/spinifex/cmd/cluster.go b/cmd/spinifex/cmd/cluster.go index 2a08c1b99..5185343f2 100644 --- a/cmd/spinifex/cmd/cluster.go +++ b/cmd/spinifex/cmd/cluster.go @@ -8,6 +8,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/daemon" + "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" "github.com/nats-io/nats.go" "github.com/spf13/cobra" ) @@ -160,6 +161,79 @@ func runClusterShutdown(cmd *cobra.Command, args []string) { fmt.Printf("Cluster shutdown complete (%s)\n", time.Since(start).Round(time.Millisecond)) } +// runClusterDrainDHCP asks every vpcd to DHCPRELEASE all external-pool leases +// it currently holds, returning them to the upstream DHCP server. Intended for +// the teardown path: an env reset otherwise strands held leases until TTL +// because vpcd's Stop() preserves them for adopt-on-reboot. Best-effort — if +// the cluster is already partly down it warns and exits 0 so teardown proceeds. +func runClusterDrainDHCP(cmd *cobra.Command, args []string) { + timeout, _ := cmd.Flags().GetDuration("timeout") + if timeout <= 0 { + timeout = 30 * time.Second + } + + _, nc, err := loadConfigAndConnect() + if err != nil { + fmt.Fprintf(os.Stderr, "Warning: drain-dhcp could not connect (cluster may be down): %v\n", err) + return + } + defer nc.Close() + + total, responders := drainDHCPLeases(nc, timeout) + fmt.Printf("DHCP drain complete: released %d lease(s) from %d vpcd responder(s)\n", total, responders) +} + +// drainDHCPLeases publishes a drain request to vpc.dhcp.drain and collects +// replies until timeout — one responder per AZ (vpcd uses a per-AZ queue +// group). Returns the summed released count and the number of responders. +func drainDHCPLeases(nc *nats.Conn, timeout time.Duration) (released, responders int) { + reqData, err := json.Marshal(struct{}{}) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + return 0, 0 + } + + inbox := nats.NewInbox() + sub, err := nc.SubscribeSync(inbox) + if err != nil { + fmt.Fprintf(os.Stderr, "Warning: drain-dhcp subscribe failed: %v\n", err) + return 0, 0 + } + defer sub.Unsubscribe() + + if err := nc.PublishRequest(dhcp.TopicDrain, inbox, reqData); err != nil { + fmt.Fprintf(os.Stderr, "Warning: drain-dhcp publish failed: %v\n", err) + return 0, 0 + } + nc.Flush() + + deadline := time.Now().Add(timeout) + for { + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + msg, err := sub.NextMsg(remaining) + if err != nil { + break + } + var reply struct { + Released int `json:"released"` + Error string `json:"error,omitempty"` + } + if err := json.Unmarshal(msg.Data, &reply); err != nil { + continue + } + responders++ + if reply.Error != "" { + fmt.Fprintf(os.Stderr, " vpcd drain error: %s\n", reply.Error) + continue + } + released += reply.Released + } + return released, responders +} + // collectShutdownACKs publishes a shutdown request and collects ACKs from nodes. func collectShutdownACKs(nc *nats.Conn, topic string, reqData []byte, nodeCount int, timeout time.Duration) ([]daemon.ShutdownACK, error) { inbox := nats.NewInbox() diff --git a/cmd/spinifex/cmd/cluster_test.go b/cmd/spinifex/cmd/cluster_test.go new file mode 100644 index 000000000..2c4034fd6 --- /dev/null +++ b/cmd/spinifex/cmd/cluster_test.go @@ -0,0 +1,58 @@ +package cmd + +import ( + "encoding/json" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeDrainResponder subscribes a stand-in vpcd to vpc.dhcp.drain that +// replies with the given released count (or error), mirroring +// Manager.handleDrainMsg without a real lease store. +func fakeDrainResponder(t *testing.T, nc *nats.Conn, released int, errMsg string) { + t.Helper() + sub, err := nc.Subscribe(dhcp.TopicDrain, func(msg *nats.Msg) { + body, _ := json.Marshal(struct { + Released int `json:"released"` + Error string `json:"error,omitempty"` + }{Released: released, Error: errMsg}) + _ = msg.Respond(body) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +func TestDrainDHCPLeasesSumsReleased(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + fakeDrainResponder(t, nc, 7, "") + + released, responders := drainDHCPLeases(nc, 2*time.Second) + assert.Equal(t, 7, released) + assert.Equal(t, 1, responders) +} + +func TestDrainDHCPLeasesReportsResponderError(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + fakeDrainResponder(t, nc, 0, "list leases for drain: boom") + + // An erroring responder still counts as a responder but contributes + // zero released leases. + released, responders := drainDHCPLeases(nc, 2*time.Second) + assert.Equal(t, 0, released) + assert.Equal(t, 1, responders) +} + +func TestDrainDHCPLeasesNoResponders(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + + // No vpcd subscribed: the collection window elapses with zero replies. + released, responders := drainDHCPLeases(nc, 200*time.Millisecond) + assert.Equal(t, 0, released) + assert.Equal(t, 0, responders) +} diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index fb6bbbddc..a53ba7b3f 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -65,6 +65,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t | Command | Flags | Description | |---------|-------|-------------| | `spx admin cluster shutdown` | `--force` (shutdown even if nodes don't respond), `--timeout` (max wait per phase, default 120s), `--dry-run` (print phase plan without executing) | Performs coordinated, phased shutdown of entire cluster. Phases execute in order: GATE (stop API/UI) → DRAIN (stop VMs) → STORAGE (stop viperblock) → PERSIST (stop predastore) → INFRA (stop NATS/daemon). Each phase waits for all nodes to ACK before proceeding. Uses JetStream state tracking. | +| `spx admin cluster drain-dhcp` | `--timeout` (reply-collection window, default 30s) | Asks each vpcd to DHCPRELEASE every external-pool DHCP lease it currently holds, returning them to the upstream DHCP server. Run on teardown before stopping services — an env reset otherwise strands held leases upstream until their TTL expires, eventually exhausting the upstream scope. Best-effort: warns and exits 0 if the cluster is already down. | ### Certificate Management diff --git a/spinifex/network/external/dhcp/manager.go b/spinifex/network/external/dhcp/manager.go index e0fff6f7b..875171024 100644 --- a/spinifex/network/external/dhcp/manager.go +++ b/spinifex/network/external/dhcp/manager.go @@ -186,6 +186,7 @@ func (m *Manager) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) { subs := []sub{ {TopicAcquire, m.handleAcquireMsg}, {TopicRelease, m.handleReleaseMsg}, + {TopicDrain, m.handleDrainMsg}, } var out []*nats.Subscription for _, s := range subs { @@ -486,6 +487,51 @@ func (m *Manager) acquireLocked(ctx context.Context, req acquireWireRequest) (*E return &entry, nil } +// DrainAll DHCPRELEASEs every lease currently held in this vpcd's per-AZ +// store and deletes the KV records. Best-effort: a per-lease release +// failure is logged and skipped so one stuck lease can't block the rest. +// Returns the number of leases successfully released. Used by the teardown +// path so an env reset returns its leases to the upstream pool rather than +// stranding them until TTL (Stop() deliberately preserves leases for +// adopt-on-reboot, which is wrong on a destructive teardown). +func (m *Manager) DrainAll(ctx context.Context) (int, error) { + entries, err := m.store.List() + if err != nil { + return 0, fmt.Errorf("list leases for drain: %w", err) + } + released := 0 + for _, e := range entries { + if e.Lease == nil { + continue + } + if relErr := m.handleRelease(ctx, e.Lease.ClientID); relErr != nil { + slog.Warn("dhcp manager: drain release failed", "client_id", e.Lease.ClientID, "err", relErr) + continue + } + released++ + } + slog.Info("dhcp manager: drained leases", "released", released, "total", len(entries)) + return released, nil +} + +func (m *Manager) handleDrainMsg(msg *nats.Msg) { + if msg.Reply == "" { + slog.Warn("dhcp manager: drain request missing reply subject; dropping") + return + } + n, err := m.DrainAll(context.Background()) + reply := drainWireReply{Released: n} + if err != nil { + reply.Error = err.Error() + } + body, mErr := json.Marshal(reply) + if mErr != nil { + slog.Warn("dhcp manager: encode drain reply failed", "err", mErr) + return + } + _ = msg.Respond(body) +} + func (m *Manager) handleRelease(ctx context.Context, clientID string) error { if clientID == "" { return errors.New("client_id required") diff --git a/spinifex/network/external/dhcp/manager_test.go b/spinifex/network/external/dhcp/manager_test.go index bac9df909..17efcd473 100644 --- a/spinifex/network/external/dhcp/manager_test.go +++ b/spinifex/network/external/dhcp/manager_test.go @@ -2,6 +2,7 @@ package dhcp_test import ( "context" + "encoding/json" "errors" "fmt" "net" @@ -124,6 +125,123 @@ func TestNATSClientAcquireReleaseRoundtrip(t *testing.T) { assert.Equal(t, 1, fake.ReleaseCount()) } +func seedLeases(t *testing.T, store *dhcp.Store, ids []string) { + t.Helper() + hw, _ := net.ParseMAC("02:00:00:aa:bb:cc") + for i, id := range ids { + lease := &dhcp.Lease{ + Bridge: "br-wan", + ClientID: id, + HWAddr: hw, + IP: net.IPv4(192, 0, 2, byte(10+i)), + AcquiredAt: time.Now(), + LeaseDuration: time.Hour, + } + require.NoError(t, store.Put(dhcp.Entry{Purpose: "eip", PoolName: "default", Lease: lease})) + } +} + +func TestManagerDrainAll(t *testing.T) { + fake := dhcp.NewFake() + mgr, store, _ := newTestManager(t, "az1", fake, time.Now) + + // Seed before Start so the manager adopts them and spawns a renewal + // loop per lease — drain must release every lease AND cancel its loop. + ids := []string{"eipalloc-1", "eni-2", "i-3", "gw-vpc-4"} + seedLeases(t, store, ids) + require.NoError(t, mgr.Start(context.Background())) + require.Eventually(t, func() bool { return mgr.LoopCount() == len(ids) }, time.Second, 10*time.Millisecond) + + n, err := mgr.DrainAll(context.Background()) + require.NoError(t, err) + assert.Equal(t, len(ids), n) + assert.Equal(t, len(ids), fake.ReleaseCount()) + + entries, err := store.List() + require.NoError(t, err) + assert.Empty(t, entries) + assert.Equal(t, 0, mgr.LoopCount()) +} + +func TestManagerDrainAllBestEffortOnReleaseError(t *testing.T) { + fake := dhcp.NewFake() + // A stuck upstream release on one lease must not block draining the rest. + fake.ReleaseHook = func(l *dhcp.Lease) error { + if l.ClientID == "eni-2" { + return errors.New("upstream unreachable") + } + return nil + } + mgr, store, _ := newTestManager(t, "az1", fake, time.Now) + + ids := []string{"eipalloc-1", "eni-2", "i-3"} + seedLeases(t, store, ids) + + n, err := mgr.DrainAll(context.Background()) + require.NoError(t, err) + assert.Equal(t, len(ids), n) + assert.Equal(t, len(ids), fake.ReleaseCount()) + + entries, err := store.List() + require.NoError(t, err) + assert.Empty(t, entries) +} + +func TestManagerDrainMsgWithoutReplyIsDropped(t *testing.T) { + fake := dhcp.NewFake() + mgr, store, nc := newTestManager(t, "az1", fake, time.Now) + require.NoError(t, mgr.Start(context.Background())) + subs, err := mgr.Subscribe(nc) + require.NoError(t, err) + t.Cleanup(func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }) + + seedLeases(t, store, []string{"eipalloc-x"}) + + // A drain publish with no reply subject must be dropped before draining — + // the lease stays put and nothing is released. + require.NoError(t, nc.Publish(dhcp.TopicDrain, []byte("{}"))) + require.NoError(t, nc.Flush()) + time.Sleep(100 * time.Millisecond) + + entries, err := store.List() + require.NoError(t, err) + assert.Len(t, entries, 1) + assert.Equal(t, 0, fake.ReleaseCount()) +} + +func TestManagerDrainRPCRoundtrip(t *testing.T) { + fake := dhcp.NewFake() + mgr, store, nc := newTestManager(t, "az1", fake, time.Now) + require.NoError(t, mgr.Start(context.Background())) + subs, err := mgr.Subscribe(nc) + require.NoError(t, err) + t.Cleanup(func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }) + + seedLeases(t, store, []string{"eipalloc-A", "eipalloc-B"}) + + msg, err := nc.Request(dhcp.TopicDrain, []byte("{}"), 2*time.Second) + require.NoError(t, err) + var reply struct { + Released int `json:"released"` + Error string `json:"error,omitempty"` + } + require.NoError(t, json.Unmarshal(msg.Data, &reply)) + assert.Empty(t, reply.Error) + assert.Equal(t, 2, reply.Released) + + entries, err := store.List() + require.NoError(t, err) + assert.Empty(t, entries) +} + func TestNATSClientAcquireIdempotent(t *testing.T) { fake := dhcp.NewFake() mgr, _, nc := newTestManager(t, "az1", fake, time.Now) diff --git a/spinifex/network/external/dhcp/wire.go b/spinifex/network/external/dhcp/wire.go index 93eaa8f25..47c485c36 100644 --- a/spinifex/network/external/dhcp/wire.go +++ b/spinifex/network/external/dhcp/wire.go @@ -14,6 +14,10 @@ import ( const ( TopicAcquire = "vpc.dhcp.acquire" TopicRelease = "vpc.dhcp.release" + // TopicDrain asks the bridge-owning vpcd to DHCPRELEASE every lease it + // currently holds. Invoked on cluster teardown so an env reset returns + // its leases to the upstream pool instead of stranding them until TTL. + TopicDrain = "vpc.dhcp.drain" ) // acquireWireRequest is the JSON payload sent by daemon-side @@ -51,6 +55,14 @@ type releaseWireReply struct { Error string `json:"error,omitempty"` } +// drainWireReply reports how many leases the responding vpcd released. The +// drain request itself carries no body — the manager drains its entire +// per-AZ lease store, so there are no parameters. +type drainWireReply struct { + Released int `json:"released"` + Error string `json:"error,omitempty"` +} + // wireLease is Lease in JSON-friendly form: dotted-quad IPs, MAC string, // durations as nanoseconds. type wireLease struct {