From f73e151c7718271e42ac1e19a1fa0ce5918a916f Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Wed, 22 Apr 2026 16:45:30 +1000 Subject: [PATCH] feat: in-process DHCP client and Manager in vpcd (mulga-990) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces the spinifex/services/vpcd/dhcp package with a Client interface, typed Lease struct, nclient4-backed implementation, and a Fake for unit tests. Adds DHCPManager in the vpcd package: subscribes to vpc.dhcp.acquire / vpc.dhcp.release NATS requests, runs a renewal goroutine per lease with ±10% jitter on T1/T2, persists leases in the new spinifex-dhcp-leases KV bucket, and fans out vpc.dhcp.lease-expired when a lease permanently fails to renew. Wires the Manager into launchService so it starts on vpcd boot and rehydrates persisted leases via Bootstrap. Handler-side wiring (ExternalIPAM → NATS) and the systemd capability move land separately in mulga-991. See docs/development/bugs/dhcp-allocate-error.md. --- go.mod | 6 + go.sum | 18 +- spinifex/services/vpcd/dhcp/client.go | 88 +++ spinifex/services/vpcd/dhcp/client_test.go | 113 ++++ spinifex/services/vpcd/dhcp/fake.go | 177 +++++ spinifex/services/vpcd/dhcp/nats.go | 73 +++ spinifex/services/vpcd/dhcp/nclient4.go | 173 +++++ spinifex/services/vpcd/dhcp_manager.go | 685 ++++++++++++++++++++ spinifex/services/vpcd/dhcp_manager_test.go | 354 ++++++++++ spinifex/services/vpcd/vpcd.go | 37 +- 10 files changed, 1721 insertions(+), 3 deletions(-) create mode 100644 spinifex/services/vpcd/dhcp/client.go create mode 100644 spinifex/services/vpcd/dhcp/client_test.go create mode 100644 spinifex/services/vpcd/dhcp/fake.go create mode 100644 spinifex/services/vpcd/dhcp/nats.go create mode 100644 spinifex/services/vpcd/dhcp/nclient4.go create mode 100644 spinifex/services/vpcd/dhcp_manager.go create mode 100644 spinifex/services/vpcd/dhcp_manager_test.go diff --git a/go.mod b/go.mod index b46ea0d38..9b6850891 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( github.com/charmbracelet/lipgloss v1.1.0 github.com/go-chi/chi/v5 v5.2.5 github.com/google/uuid v1.6.0 + github.com/insomniacslk/dhcp v0.0.0-20260407060928-11b94ed970f2 github.com/kdomanski/iso9660 v0.4.0 github.com/klauspost/cpuid/v2 v2.3.0 github.com/mulgadc/predastore v1.1.6 @@ -74,6 +75,7 @@ require ( github.com/hashicorp/raft-boltdb/v2 v2.3.1 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jmespath/go-jmespath v0.4.0 // indirect + github.com/josharian/native v1.1.0 // indirect github.com/klauspost/compress v1.18.5 // indirect github.com/klauspost/reedsolomon v1.13.3 // indirect github.com/leodido/go-urn v1.4.0 // indirect @@ -83,6 +85,8 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-localereader v0.0.1 // indirect github.com/mattn/go-runewidth v0.0.21 // indirect + github.com/mdlayher/packet v1.1.2 // indirect + github.com/mdlayher/socket v0.4.1 // indirect github.com/minio/crc64nvme v1.1.1 // indirect github.com/minio/highwayhash v1.0.4 // indirect github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 // indirect @@ -92,6 +96,7 @@ require ( github.com/nats-io/jwt/v2 v2.8.1 // indirect github.com/nats-io/nkeys v0.4.15 // indirect github.com/nats-io/nuid v1.0.1 // indirect + github.com/pierrec/lz4/v4 v4.1.14 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_golang v1.23.2 // indirect github.com/prometheus/client_model v0.6.2 // indirect @@ -104,6 +109,7 @@ require ( github.com/spf13/cast v1.10.0 // indirect github.com/spf13/pflag v1.0.10 // indirect github.com/subosito/gotenv v1.6.0 // indirect + github.com/u-root/uio v0.0.0-20230220225925-ffce2a382923 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect go.etcd.io/bbolt v1.4.3 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect diff --git a/go.sum b/go.sum index b4c15fbed..8b6a301a0 100644 --- a/go.sum +++ b/go.sum @@ -67,8 +67,6 @@ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.10 h1:GtsxyiF3Nd3JahRBJbxLCCd github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.10/go.mod h1:/j67Z5XBVDx8nZVp9EuFM9/BS5dvBznbqILGuu73hug= github.com/aws/aws-sdk-go-v2/service/sts v1.41.2 h1:a5UTtD4mHBU3t0o6aHQZFJTNKVfxFWfPX7J0Lr7G+uY= github.com/aws/aws-sdk-go-v2/service/sts v1.41.2/go.mod h1:6TxbXoDSgBQ225Qd8Q+MbxUxUh6TtNKwbRt/EPS9xso= -github.com/aws/smithy-go v1.23.2 h1:Crv0eatJUQhaManss33hS5r40CG3ZFH+21XSkqMrIUM= -github.com/aws/smithy-go v1.23.2/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0= github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng= github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k= @@ -223,12 +221,19 @@ github.com/hashicorp/raft-boltdb v0.0.0-20230125174641-2a8082862702 h1:RLKEcCuKc github.com/hashicorp/raft-boltdb v0.0.0-20230125174641-2a8082862702/go.mod h1:nTakvJ4XYq45UXtn0DbwR4aU9ZdjlnIenpbs6Cd+FM0= github.com/hashicorp/raft-boltdb/v2 v2.3.1 h1:ackhdCNPKblmOhjEU9+4lHSJYFkJd6Jqyvj6eW9pwkc= github.com/hashicorp/raft-boltdb/v2 v2.3.1/go.mod h1:n4S+g43dXF1tqDT+yzcXHhXM6y7MrlUd3TTwGRcUvQE= +github.com/hugelgupf/socketpair v0.0.0-20190730060125-05d35a94e714 h1:/jC7qQFrv8CrSJVmaolDVOxTfS9kc36uB6H40kdbQq8= +github.com/hugelgupf/socketpair v0.0.0-20190730060125-05d35a94e714/go.mod h1:2Goc3h8EklBH5mspfHFxBnEoURQCGzQQH1ga9Myjvis= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/insomniacslk/dhcp v0.0.0-20260407060928-11b94ed970f2 h1:G3irkWwmpl0vH/nn83K2AHqLUZweC7XAONuwXy/w9Co= +github.com/insomniacslk/dhcp v0.0.0-20260407060928-11b94ed970f2/go.mod h1:qfvBmyDNp+/liLEYWRvqny/PEz9hGe2Dz833eXILSmo= github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg= github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo= github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8= github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U= +github.com/josharian/native v1.0.1-0.20221213033349-c1e37c09b531/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2Cu7qoaN2w= +github.com/josharian/native v1.1.0 h1:uuaP0hAbW7Y4l0ZRQ6C9zfb7Mg1mbFKry/xzDAfmtLA= +github.com/josharian/native v1.1.0/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2Cu7qoaN2w= github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= @@ -277,6 +282,10 @@ github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh github.com/mattn/go-runewidth v0.0.21 h1:jJKAZiQH+2mIinzCJIaIG9Be1+0NR+5sz/lYEEjdM8w= github.com/mattn/go-runewidth v0.0.21/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= +github.com/mdlayher/packet v1.1.2 h1:3Up1NG6LZrsgDVn6X4L9Ge/iyRyxFEFD9o6Pr3Q1nQY= +github.com/mdlayher/packet v1.1.2/go.mod h1:GEu1+n9sG5VtiRE4SydOmX5GTwyyYlteZiFU+x0kew4= +github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U= +github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA= github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI= github.com/minio/crc64nvme v1.1.1/go.mod h1:eVfm2fAzLlxMdUGc0EEBGSMmPwmXD5XiNRpnu9J3bvg= github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= @@ -315,6 +324,8 @@ github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0Mw github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pelletier/go-toml/v2 v2.3.0 h1:k59bC/lIZREW0/iVaQR8nDHxVq8OVlIzYCOJf421CaM= github.com/pelletier/go-toml/v2 v2.3.0/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pierrec/lz4/v4 v4.1.14 h1:+fL8AQEZtz/ijeNnpduH0bROTu0O3NZAlPjQxGn8LwE= +github.com/pierrec/lz4/v4 v4.1.14/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -399,6 +410,8 @@ github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM= +github.com/u-root/uio v0.0.0-20230220225925-ffce2a382923 h1:tHNk7XK9GkmKUR6Gh8gVBKXc2MVSZ4G/NnWLtzw4gNA= +github.com/u-root/uio v0.0.0-20230220225925-ffce2a382923/go.mod h1:eLL9Nub3yfAho7qB0MzZizFhTU2QkLeoVsWdHtDW264= github.com/xo/terminfo v0.0.0-20210125001918-ca9a967f8778/go.mod h1:2MuV+tbUrU1zIOPMxZ5EncGwgmMJsa+9ucAQZXxsObs= github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no= github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM= @@ -479,6 +492,7 @@ golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/spinifex/services/vpcd/dhcp/client.go b/spinifex/services/vpcd/dhcp/client.go new file mode 100644 index 000000000..f0d3559ee --- /dev/null +++ b/spinifex/services/vpcd/dhcp/client.go @@ -0,0 +1,88 @@ +// Package dhcp provides an in-process DHCPv4 client used by spinifex-vpcd +// to obtain, renew and release real server-bound leases on behalf of the +// external IPAM pool. It replaces the historical "shell out to sudo dhcpcd" +// path which had no renewal, phantom leases under -T, and a release channel +// that collided with a host-side dhcpcd daemon managing the same bridge. +// +// The real implementation (NewNClient4) opens an AF_PACKET socket per DORA +// on the target bridge via github.com/insomniacslk/dhcp/dhcpv4/nclient4. +// Tests inject Fake instead. +package dhcp + +import ( + "context" + "net" + "time" +) + +// Client is the per-handshake DHCP surface. Backed by NewNClient4 in +// production; Fake satisfies this for unit tests. +type Client interface { + // Acquire performs a full DORA on req.Bridge and returns the + // server-bound lease. Blocks until ACK or ctx cancellation. + Acquire(ctx context.Context, req AcquireRequest) (*Lease, error) + // Renew refreshes an existing lease with its recorded server-id. + // Returns a new Lease reflecting the refreshed timers. + Renew(ctx context.Context, lease *Lease) (*Lease, error) + // Release sends DHCPRELEASE to the server-id in the lease. + Release(ctx context.Context, lease *Lease) error +} + +// AcquireRequest identifies a new DHCP lease. Fields map 1:1 to DHCP +// options and the payload chaddr. +type AcquireRequest struct { + Bridge string // OVS bridge carrying the AF_PACKET socket + ClientID string // option 61 — client-identifier + Hostname string // option 12 — host-name + VendorClass string // option 60 — vendor-class-identifier + HWAddr net.HardwareAddr // chaddr in the DHCP payload +} + +// Lease is a server-bound DHCPv4 lease. Carries enough state for Renew +// and Release, plus the raw Offer/ACK packets so that the Manager can +// resume after a daemon restart. +type Lease struct { + // Request context — echoed for audit; also used to rebuild the + // underlying nclient4 client on Renew/Release. + Bridge string + ClientID string + Hostname string + VendorClass string + HWAddr net.HardwareAddr + + // Wire state extracted from the server's ACK. + IP net.IP + SubnetMask net.IPMask + Routers []net.IP + DNS []net.IP + ServerID net.IP + + AcquiredAt time.Time + LeaseDuration time.Duration + T1 time.Duration // renew timer (option 58, or lease/2) + T2 time.Duration // rebind timer (option 59, or lease*7/8) + + // Raw wire packets. Opaque to callers; nclient4 needs them to build + // RENEW / RELEASE messages that hit the same server binding. Persisted + // in KV for crash recovery. + RawOffer []byte + RawACK []byte +} + +// ExpiresAt returns the absolute deadline at which the upstream server is +// free to reassign this IP. +func (l *Lease) ExpiresAt() time.Time { + return l.AcquiredAt.Add(l.LeaseDuration) +} + +// RenewAt returns the absolute time the Manager should attempt a RENEW +// (option 58 / T1). +func (l *Lease) RenewAt() time.Time { + return l.AcquiredAt.Add(l.T1) +} + +// RebindAt returns the absolute time the Manager should attempt a REBIND +// (option 59 / T2). +func (l *Lease) RebindAt() time.Time { + return l.AcquiredAt.Add(l.T2) +} diff --git a/spinifex/services/vpcd/dhcp/client_test.go b/spinifex/services/vpcd/dhcp/client_test.go new file mode 100644 index 000000000..05dd9689c --- /dev/null +++ b/spinifex/services/vpcd/dhcp/client_test.go @@ -0,0 +1,113 @@ +package dhcp_test + +import ( + "context" + "errors" + "net" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" +) + +func TestLeaseTimers(t *testing.T) { + acq := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + l := &dhcp.Lease{ + AcquiredAt: acq, + LeaseDuration: time.Hour, + T1: 30 * time.Minute, + T2: 52*time.Minute + 30*time.Second, + } + + if got := l.RenewAt(); !got.Equal(acq.Add(30 * time.Minute)) { + t.Errorf("RenewAt: got %v, want %v", got, acq.Add(30*time.Minute)) + } + if got := l.RebindAt(); !got.Equal(acq.Add(52*time.Minute + 30*time.Second)) { + t.Errorf("RebindAt: got %v, want %v", got, acq.Add(52*time.Minute+30*time.Second)) + } + if got := l.ExpiresAt(); !got.Equal(acq.Add(time.Hour)) { + t.Errorf("ExpiresAt: got %v, want %v", got, acq.Add(time.Hour)) + } +} + +func TestFakeAcquireReturnsLeaseAndTracksCount(t *testing.T) { + f := dhcp.NewFake() + mac, _ := net.ParseMAC("02:00:00:aa:bb:cc") + + lease, err := f.Acquire(context.Background(), dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-1", HWAddr: mac, + }) + if err != nil { + t.Fatalf("Acquire: %v", err) + } + if lease == nil || lease.IP == nil { + t.Fatalf("expected lease with IP, got %+v", lease) + } + if f.AcquireCount() != 1 { + t.Errorf("acquire count = %d, want 1", f.AcquireCount()) + } + if held, ok := f.HeldLease("eni-1"); !ok || held.IP.String() != lease.IP.String() { + t.Errorf("held lease = %v/%v, want match", held, ok) + } +} + +func TestFakeAcquireHookOverridesDefault(t *testing.T) { + f := dhcp.NewFake() + f.AcquireHook = func(req dhcp.AcquireRequest) (*dhcp.Lease, error) { + return nil, errors.New("injected") + } + _, err := f.Acquire(context.Background(), dhcp.AcquireRequest{Bridge: "br-wan", ClientID: "x"}) + if err == nil || err.Error() != "injected" { + t.Fatalf("expected injected error, got %v", err) + } +} + +func TestFakeRenewRefreshesAcquiredAt(t *testing.T) { + f := dhcp.NewFake() + lease, err := f.Acquire(context.Background(), dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-2", + }) + if err != nil { + t.Fatalf("Acquire: %v", err) + } + firstAcq := lease.AcquiredAt + + time.Sleep(10 * time.Millisecond) + renewed, err := f.Renew(context.Background(), lease) + if err != nil { + t.Fatalf("Renew: %v", err) + } + if !renewed.AcquiredAt.After(firstAcq) { + t.Errorf("renewed AcquiredAt %v not after original %v", renewed.AcquiredAt, firstAcq) + } + if f.RenewCount() != 1 { + t.Errorf("renew count = %d, want 1", f.RenewCount()) + } +} + +func TestFakeReleaseClearsTrackedLease(t *testing.T) { + f := dhcp.NewFake() + lease, _ := f.Acquire(context.Background(), dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-3", + }) + if err := f.Release(context.Background(), lease); err != nil { + t.Fatalf("Release: %v", err) + } + if _, ok := f.HeldLease("eni-3"); ok { + t.Errorf("expected lease removed after Release") + } + if f.ReleaseCount() != 1 { + t.Errorf("release count = %d, want 1", f.ReleaseCount()) + } +} + +func TestFakeRenewHookSurfacesError(t *testing.T) { + f := dhcp.NewFake() + f.RenewHook = func(l *dhcp.Lease) (*dhcp.Lease, error) { + return nil, errors.New("server NAK") + } + lease, _ := f.Acquire(context.Background(), dhcp.AcquireRequest{Bridge: "br-wan", ClientID: "eni-4"}) + if _, err := f.Renew(context.Background(), lease); err == nil { + t.Fatal("expected hook error") + } +} diff --git a/spinifex/services/vpcd/dhcp/fake.go b/spinifex/services/vpcd/dhcp/fake.go new file mode 100644 index 000000000..8050a892d --- /dev/null +++ b/spinifex/services/vpcd/dhcp/fake.go @@ -0,0 +1,177 @@ +package dhcp + +import ( + "context" + "fmt" + "net" + "sync" + "time" +) + +// Fake is a configurable in-memory Client for unit tests. It mimics the +// DORA / RENEW / RELEASE lifecycle well enough to exercise the Manager +// without opening real sockets. Hooks let tests inject errors or custom +// responses at each step. +type Fake struct { + mu sync.Mutex + + // Default template for synthesised leases. Tests can override via + // SetDefaultLease. + defaultLease LeaseTemplate + + // Optional hooks — if non-nil the fake calls the hook instead of + // synthesising a response. Useful for fault injection. + AcquireHook func(AcquireRequest) (*Lease, error) + RenewHook func(*Lease) (*Lease, error) + ReleaseHook func(*Lease) error + + // Tracked state. + leases map[string]*Lease // keyed by ClientID + acquireCount int + renewCount int + releaseCount int +} + +// LeaseTemplate configures the synthetic lease Fake returns from Acquire +// when no AcquireHook is set. +type LeaseTemplate struct { + IP net.IP + SubnetMask net.IPMask + Routers []net.IP + DNS []net.IP + ServerID net.IP + LeaseDuration time.Duration +} + +// DefaultLeaseTemplate returns a reasonable template for tests (RFC 5737 +// TEST-NET-1, 1 h lease). +func DefaultLeaseTemplate() LeaseTemplate { + return LeaseTemplate{ + IP: net.IPv4(192, 0, 2, 100), + SubnetMask: net.CIDRMask(24, 32), + Routers: []net.IP{net.IPv4(192, 0, 2, 1)}, + DNS: []net.IP{net.IPv4(192, 0, 2, 1)}, + ServerID: net.IPv4(192, 0, 2, 1), + LeaseDuration: time.Hour, + } +} + +// NewFake creates a Fake pre-loaded with DefaultLeaseTemplate. +func NewFake() *Fake { + return &Fake{ + defaultLease: DefaultLeaseTemplate(), + leases: map[string]*Lease{}, + } +} + +// SetDefaultLease overrides the template used for subsequent Acquire calls. +func (f *Fake) SetDefaultLease(tmpl LeaseTemplate) { + f.mu.Lock() + defer f.mu.Unlock() + f.defaultLease = tmpl +} + +// AcquireCount returns the number of Acquire calls observed. +func (f *Fake) AcquireCount() int { f.mu.Lock(); defer f.mu.Unlock(); return f.acquireCount } + +// RenewCount returns the number of Renew calls observed. +func (f *Fake) RenewCount() int { f.mu.Lock(); defer f.mu.Unlock(); return f.renewCount } + +// ReleaseCount returns the number of Release calls observed. +func (f *Fake) ReleaseCount() int { f.mu.Lock(); defer f.mu.Unlock(); return f.releaseCount } + +// HeldLease returns the last lease Fake synthesised for the given ClientID. +func (f *Fake) HeldLease(clientID string) (*Lease, bool) { + f.mu.Lock() + defer f.mu.Unlock() + l, ok := f.leases[clientID] + return l, ok +} + +func (f *Fake) Acquire(_ context.Context, req AcquireRequest) (*Lease, error) { + f.mu.Lock() + f.acquireCount++ + hook := f.AcquireHook + tmpl := f.defaultLease + f.mu.Unlock() + + if hook != nil { + lease, err := hook(req) + if err == nil && lease != nil { + f.mu.Lock() + f.leases[req.ClientID] = lease + f.mu.Unlock() + } + return lease, err + } + + lease := &Lease{ + Bridge: req.Bridge, + ClientID: req.ClientID, + Hostname: req.Hostname, + VendorClass: req.VendorClass, + HWAddr: append(net.HardwareAddr(nil), req.HWAddr...), + IP: tmpl.IP, + SubnetMask: tmpl.SubnetMask, + Routers: tmpl.Routers, + DNS: tmpl.DNS, + ServerID: tmpl.ServerID, + AcquiredAt: time.Now(), + LeaseDuration: tmpl.LeaseDuration, + T1: tmpl.LeaseDuration / 2, + T2: tmpl.LeaseDuration * 7 / 8, + // Synthetic raw packets — opaque marker bytes so Renew/Release + // distinguish a fake lease from a nil-bytes one. + RawOffer: []byte("fake-offer-" + req.ClientID), + RawACK: []byte("fake-ack-" + req.ClientID), + } + + f.mu.Lock() + f.leases[req.ClientID] = lease + f.mu.Unlock() + return lease, nil +} + +func (f *Fake) Renew(_ context.Context, lease *Lease) (*Lease, error) { + if lease == nil { + return nil, fmt.Errorf("fake renew: lease is nil") + } + f.mu.Lock() + f.renewCount++ + hook := f.RenewHook + f.mu.Unlock() + + if hook != nil { + renewed, err := hook(lease) + if err == nil && renewed != nil { + f.mu.Lock() + f.leases[lease.ClientID] = renewed + f.mu.Unlock() + } + return renewed, err + } + + renewed := *lease + renewed.AcquiredAt = time.Now() + f.mu.Lock() + f.leases[lease.ClientID] = &renewed + f.mu.Unlock() + return &renewed, nil +} + +func (f *Fake) Release(_ context.Context, lease *Lease) error { + if lease == nil { + return nil + } + f.mu.Lock() + f.releaseCount++ + hook := f.ReleaseHook + f.mu.Unlock() + if hook != nil { + return hook(lease) + } + f.mu.Lock() + delete(f.leases, lease.ClientID) + f.mu.Unlock() + return nil +} diff --git a/spinifex/services/vpcd/dhcp/nats.go b/spinifex/services/vpcd/dhcp/nats.go new file mode 100644 index 000000000..758a1d39e --- /dev/null +++ b/spinifex/services/vpcd/dhcp/nats.go @@ -0,0 +1,73 @@ +package dhcp + +// NATS subjects that spinifex-vpcd's DHCP manager subscribes to and +// publishes on. The daemon-side handlers in handlers/ec2/vpc send +// requests against the request/reply subjects. +const ( + // TopicAcquire is the request/reply subject for obtaining a lease. + // Request: AcquireRequestMsg. Reply: AcquireReplyMsg. + TopicAcquire = "vpc.dhcp.acquire" + // TopicRelease is the request/reply subject for releasing a lease. + // Request: ReleaseRequestMsg. Reply: ReleaseReplyMsg. + TopicRelease = "vpc.dhcp.release" + // TopicLeaseExpired is a fan-out event published by vpcd when a + // renewal permanently fails and the lease has passed its expiry. + // Consumers (EIP / instance / VPC reconcilers) decide whether to + // re-request a lease, reschedule, or fail the pending API call. + TopicLeaseExpired = "vpc.dhcp.lease-expired" +) + +// KV bucket for persisted lease state. Keyed by ClientID. Separate from +// the external IPAM allocation ledger by design: IPAM records who owns +// which IP, this bucket records when a lease expires and how to renew it. +const ( + KVBucketDHCPLeases = "spinifex-dhcp-leases" + KVBucketDHCPLeasesVersion = 1 +) + +// AcquireRequestMsg is the NATS payload for TopicAcquire. +type AcquireRequestMsg struct { + Bridge string `json:"bridge"` + ClientID string `json:"client_id"` + Hostname string `json:"hostname"` + VendorClass string `json:"vendor_class"` + // HWAddr is the MAC to put in the DHCP payload chaddr, formatted as + // 02:aa:bb:cc:dd:ee. Empty means the vpcd Manager derives a synthetic + // MAC from ClientID. + HWAddr string `json:"hw_addr,omitempty"` + PoolName string `json:"pool_name,omitempty"` +} + +// AcquireReplyMsg is the NATS reply for TopicAcquire. Error is the only +// field set on failure. +type AcquireReplyMsg struct { + IP string `json:"ip,omitempty"` + SubnetMask string `json:"subnet_mask,omitempty"` + Routers []string `json:"routers,omitempty"` + DNS []string `json:"dns,omitempty"` + ServerID string `json:"server_id,omitempty"` + HWAddr string `json:"hw_addr,omitempty"` + ExpiresUnix int64 `json:"expires_unix,omitempty"` + Error string `json:"error,omitempty"` +} + +// ReleaseRequestMsg is the NATS payload for TopicRelease. ClientID must +// match the identifier the lease was acquired with. +type ReleaseRequestMsg struct { + ClientID string `json:"client_id"` +} + +// ReleaseReplyMsg is the NATS reply for TopicRelease. +type ReleaseReplyMsg struct { + Error string `json:"error,omitempty"` +} + +// LeaseExpiredEvent is the payload fan-out on TopicLeaseExpired. +type LeaseExpiredEvent struct { + ClientID string `json:"client_id"` + PoolName string `json:"pool_name,omitempty"` + IP string `json:"ip,omitempty"` + // Reason is a short machine-readable code (e.g. "renew_failed", + // "nak", "server_unreachable"). + Reason string `json:"reason"` +} diff --git a/spinifex/services/vpcd/dhcp/nclient4.go b/spinifex/services/vpcd/dhcp/nclient4.go new file mode 100644 index 000000000..a9c5dd170 --- /dev/null +++ b/spinifex/services/vpcd/dhcp/nclient4.go @@ -0,0 +1,173 @@ +package dhcp + +import ( + "context" + "fmt" + "time" + + "github.com/insomniacslk/dhcp/dhcpv4" + "github.com/insomniacslk/dhcp/dhcpv4/nclient4" +) + +// NClient4Client is the production DHCP client backed by +// github.com/insomniacslk/dhcp/dhcpv4/nclient4. Each Acquire/Renew/Release +// opens an AF_PACKET socket on the target bridge for the duration of the +// handshake and closes it when done — no long-lived per-lease process. +type NClient4Client struct { + timeout time.Duration + retry int +} + +// NewNClient4 creates an NClient4Client with sensible DORA defaults. +func NewNClient4(timeout time.Duration, retry int) *NClient4Client { + if timeout <= 0 { + timeout = 15 * time.Second + } + if retry <= 0 { + retry = 3 + } + return &NClient4Client{timeout: timeout, retry: retry} +} + +func (c *NClient4Client) Acquire(ctx context.Context, req AcquireRequest) (*Lease, error) { + if req.Bridge == "" { + return nil, fmt.Errorf("dhcp acquire: bridge is required") + } + if len(req.HWAddr) == 0 { + return nil, fmt.Errorf("dhcp acquire: hw_addr is required") + } + + client, err := nclient4.New(req.Bridge, + nclient4.WithHWAddr(req.HWAddr), + nclient4.WithTimeout(c.timeout), + nclient4.WithRetry(c.retry), + ) + if err != nil { + return nil, fmt.Errorf("open nclient4 on %s: %w", req.Bridge, err) + } + defer func() { _ = client.Close() }() + + lease, err := client.Request(ctx, identityModifiers(req.ClientID, req.Hostname, req.VendorClass)...) + if err != nil { + return nil, fmt.Errorf("dhcp DORA on %s (client=%s): %w", req.Bridge, req.ClientID, err) + } + return leaseFromNClient4(req, lease), nil +} + +func (c *NClient4Client) Renew(ctx context.Context, lease *Lease) (*Lease, error) { + if lease == nil { + return nil, fmt.Errorf("dhcp renew: lease is nil") + } + nclient4Lease, err := reconstructNClient4Lease(lease) + if err != nil { + return nil, fmt.Errorf("dhcp renew: %w", err) + } + + client, err := nclient4.New(lease.Bridge, + nclient4.WithHWAddr(lease.HWAddr), + nclient4.WithTimeout(c.timeout), + nclient4.WithRetry(c.retry), + ) + if err != nil { + return nil, fmt.Errorf("open nclient4 on %s for renew: %w", lease.Bridge, err) + } + defer func() { _ = client.Close() }() + + renewed, err := client.Renew(ctx, nclient4Lease, + identityModifiers(lease.ClientID, lease.Hostname, lease.VendorClass)...) + if err != nil { + return nil, fmt.Errorf("dhcp renew on %s (client=%s): %w", lease.Bridge, lease.ClientID, err) + } + + return leaseFromNClient4(AcquireRequest{ + Bridge: lease.Bridge, + ClientID: lease.ClientID, + Hostname: lease.Hostname, + VendorClass: lease.VendorClass, + HWAddr: lease.HWAddr, + }, renewed), nil +} + +func (c *NClient4Client) Release(ctx context.Context, lease *Lease) error { + if lease == nil { + return nil + } + nclient4Lease, err := reconstructNClient4Lease(lease) + if err != nil { + return fmt.Errorf("dhcp release: %w", err) + } + + client, err := nclient4.New(lease.Bridge, + nclient4.WithHWAddr(lease.HWAddr), + nclient4.WithTimeout(c.timeout), + ) + if err != nil { + return fmt.Errorf("open nclient4 on %s for release: %w", lease.Bridge, err) + } + defer func() { _ = client.Close() }() + + if err := client.Release(nclient4Lease, + dhcpv4.WithOption(dhcpv4.OptClientIdentifier([]byte(lease.ClientID)))); err != nil { + return fmt.Errorf("dhcp release on %s (client=%s): %w", lease.Bridge, lease.ClientID, err) + } + return nil +} + +// identityModifiers builds the three identifying DHCP options we set on +// every outbound message: option 61 (client-id), option 12 (hostname), +// option 60 (vendor class). +func identityModifiers(clientID, hostname, vendorClass string) []dhcpv4.Modifier { + var mods []dhcpv4.Modifier + if clientID != "" { + mods = append(mods, dhcpv4.WithOption(dhcpv4.OptClientIdentifier([]byte(clientID)))) + } + if hostname != "" { + mods = append(mods, dhcpv4.WithOption(dhcpv4.OptHostName(hostname))) + } + if vendorClass != "" { + mods = append(mods, dhcpv4.WithOption(dhcpv4.OptClassIdentifier(vendorClass))) + } + return mods +} + +func leaseFromNClient4(req AcquireRequest, in *nclient4.Lease) *Lease { + ack := in.ACK + leaseTime := ack.IPAddressLeaseTime(24 * time.Hour) + return &Lease{ + Bridge: req.Bridge, + ClientID: req.ClientID, + Hostname: req.Hostname, + VendorClass: req.VendorClass, + HWAddr: req.HWAddr, + IP: ack.YourIPAddr, + SubnetMask: ack.SubnetMask(), + Routers: ack.Router(), + DNS: ack.DNS(), + ServerID: ack.ServerIdentifier(), + AcquiredAt: in.CreationTime, + LeaseDuration: leaseTime, + T1: ack.IPAddressRenewalTime(leaseTime / 2), + T2: ack.IPAddressRebindingTime(leaseTime * 7 / 8), + RawOffer: in.Offer.ToBytes(), + RawACK: ack.ToBytes(), + } +} + +func reconstructNClient4Lease(l *Lease) (*nclient4.Lease, error) { + if len(l.RawOffer) == 0 || len(l.RawACK) == 0 { + return nil, fmt.Errorf("lease is missing raw offer/ack bytes; renewal/release not possible") + } + offer, err := dhcpv4.FromBytes(l.RawOffer) + if err != nil { + return nil, fmt.Errorf("parse stored offer: %w", err) + } + ack, err := dhcpv4.FromBytes(l.RawACK) + if err != nil { + return nil, fmt.Errorf("parse stored ack: %w", err) + } + return &nclient4.Lease{ + Offer: offer, + ACK: ack, + CreationTime: l.AcquiredAt, + }, nil +} diff --git a/spinifex/services/vpcd/dhcp_manager.go b/spinifex/services/vpcd/dhcp_manager.go new file mode 100644 index 000000000..ecdae547b --- /dev/null +++ b/spinifex/services/vpcd/dhcp_manager.go @@ -0,0 +1,685 @@ +package vpcd + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "math/rand/v2" + "net" + "strings" + "sync" + "time" + + "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// DHCPManager owns the active DHCP leases held by spinifex-vpcd. It +// services vpc.dhcp.acquire / vpc.dhcp.release NATS requests from the +// daemon-side ExternalIPAM handlers, runs a per-lease renewal goroutine, +// and persists each lease in the spinifex-dhcp-leases KV bucket so that +// a vpcd restart does not forget what has already been allocated upstream. +type DHCPManager struct { + client dhcp.Client + kv nats.KeyValue + nc *nats.Conn + + // macForClientID produces the chaddr for a request that came in with + // HWAddr unset. Defaults to generateMAC; tests override. + macForClientID func(clientID string) net.HardwareAddr + + // acquireTimeout bounds a single DORA (Acquire) / Renew / Release + // handshake. Independent of the RPC reply timeout on the handler side. + acquireTimeout time.Duration + + // jitterFraction is the ± range applied to renewal sleeps, expressed + // as a fraction of T1 (0.1 = ±10%). + jitterFraction float64 + rng *rand.Rand + + mu sync.Mutex + leases map[string]*managedLease // keyed by ClientID + + stopCtx context.Context + stopCancel context.CancelFunc + wg sync.WaitGroup +} + +type managedLease struct { + lease *dhcp.Lease + renewStop context.CancelFunc + renewDone chan struct{} +} + +// DHCPManagerOption configures DHCPManager at construction. +type DHCPManagerOption func(*DHCPManager) + +// WithDHCPMACFunc overrides the default MAC derivation (generateMAC). +func WithDHCPMACFunc(fn func(clientID string) net.HardwareAddr) DHCPManagerOption { + return func(m *DHCPManager) { m.macForClientID = fn } +} + +// WithDHCPAcquireTimeout overrides the per-handshake DORA timeout. +func WithDHCPAcquireTimeout(d time.Duration) DHCPManagerOption { + return func(m *DHCPManager) { m.acquireTimeout = d } +} + +// WithDHCPJitterFraction overrides the ±fraction applied to renewal sleeps. +func WithDHCPJitterFraction(f float64) DHCPManagerOption { + return func(m *DHCPManager) { m.jitterFraction = f } +} + +// WithDHCPRand overrides the random source used for jitter (tests only). +func WithDHCPRand(r *rand.Rand) DHCPManagerOption { + return func(m *DHCPManager) { m.rng = r } +} + +// NewDHCPManager creates a Manager wired against js for KV persistence and +// nc for NATS request/reply. The caller still needs to call Bootstrap (to +// rehydrate persisted leases) and Subscribe (to start servicing requests). +func NewDHCPManager(nc *nats.Conn, js nats.JetStreamContext, client dhcp.Client, opts ...DHCPManagerOption) (*DHCPManager, error) { + if nc == nil { + return nil, fmt.Errorf("DHCPManager: nats.Conn is required") + } + if js == nil { + return nil, fmt.Errorf("DHCPManager: JetStreamContext is required") + } + if client == nil { + return nil, fmt.Errorf("DHCPManager: dhcp.Client is required") + } + + kv, err := utils.GetOrCreateKVBucket(js, dhcp.KVBucketDHCPLeases, 3) + if err != nil { + return nil, fmt.Errorf("create DHCP lease KV bucket: %w", err) + } + if err := migrate.DefaultRegistry.RunKV(dhcp.KVBucketDHCPLeases, kv, dhcp.KVBucketDHCPLeasesVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", dhcp.KVBucketDHCPLeases, err) + } + + ctx, cancel := context.WithCancel(context.Background()) + m := &DHCPManager{ + client: client, + kv: kv, + nc: nc, + acquireTimeout: 15 * time.Second, + jitterFraction: 0.1, + rng: rand.New(rand.NewPCG(uint64(time.Now().UnixNano()), 0xdeadbeef)), //nolint:gosec // non-cryptographic jitter + leases: map[string]*managedLease{}, + stopCtx: ctx, + stopCancel: cancel, + macForClientID: func(id string) net.HardwareAddr { + mac, err := net.ParseMAC(generateMAC(id)) + if err != nil { + return nil + } + return mac + }, + } + for _, opt := range opts { + opt(m) + } + return m, nil +} + +// Bootstrap reads every persisted lease from KV and schedules the +// renewal goroutine. Leases already past their expiry are removed and +// a TopicLeaseExpired event is published so the reconcilers upstream +// know they need to re-request. +func (m *DHCPManager) Bootstrap(ctx context.Context) error { + keys, err := m.kv.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + slog.Info("DHCP Manager bootstrapped", "leases", 0) + return nil + } + return fmt.Errorf("list DHCP lease keys: %w", err) + } + + now := time.Now() + resumed, expired := 0, 0 + for _, key := range keys { + if key == "_version" { + continue + } + entry, err := m.kv.Get(key) + if err != nil { + slog.Warn("DHCP Manager: failed to load lease", "client_id", key, "err", err) + continue + } + lease, err := decodeLease(entry.Value()) + if err != nil { + slog.Warn("DHCP Manager: failed to decode lease", "client_id", key, "err", err) + continue + } + + if now.After(lease.ExpiresAt()) { + _ = m.kv.Delete(key) + m.publishExpired(lease, "", "bootstrap_past_expiry") + expired++ + continue + } + + m.trackAndRenew(lease) + resumed++ + } + slog.Info("DHCP Manager bootstrapped", "leases", resumed, "expired", expired) + return nil +} + +// Subscribe registers queue-subscribed handlers on vpc.dhcp.acquire and +// vpc.dhcp.release. Queue group "vpcd-dhcp-workers" matches the existing +// vpcd-workers convention. +func (m *DHCPManager) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) { + var subs []*nats.Subscription + + acquire, err := nc.QueueSubscribe(dhcp.TopicAcquire, "vpcd-dhcp-workers", m.handleAcquire) + if err != nil { + return nil, fmt.Errorf("subscribe %s: %w", dhcp.TopicAcquire, err) + } + subs = append(subs, acquire) + slog.Info("DHCP Manager subscribed", "topic", dhcp.TopicAcquire) + + release, err := nc.QueueSubscribe(dhcp.TopicRelease, "vpcd-dhcp-workers", m.handleRelease) + if err != nil { + _ = acquire.Unsubscribe() + return nil, fmt.Errorf("subscribe %s: %w", dhcp.TopicRelease, err) + } + subs = append(subs, release) + slog.Info("DHCP Manager subscribed", "topic", dhcp.TopicRelease) + + return subs, nil +} + +// Close stops all renewal goroutines and waits for them to exit. +func (m *DHCPManager) Close() { + m.stopCancel() + m.wg.Wait() +} + +// handleAcquire processes a vpc.dhcp.acquire request. Idempotent: if we +// already hold a valid lease for the requested ClientID we return it +// without a fresh DORA. Lets the daemon-side CAS retry loop be safe. +func (m *DHCPManager) handleAcquire(msg *nats.Msg) { + var req dhcp.AcquireRequestMsg + if err := json.Unmarshal(msg.Data, &req); err != nil { + m.replyAcquireError(msg, fmt.Sprintf("invalid acquire request: %v", err)) + return + } + if req.Bridge == "" || req.ClientID == "" { + m.replyAcquireError(msg, "acquire: bridge and client_id are required") + return + } + + if existing, ok := m.get(req.ClientID); ok { + if time.Now().Before(existing.ExpiresAt()) { + m.replyAcquire(msg, existing) + return + } + m.forget(req.ClientID, "stale_existing") + } + + hwAddr, err := resolveHWAddr(req.HWAddr, req.ClientID, m.macForClientID) + if err != nil { + m.replyAcquireError(msg, err.Error()) + return + } + + ctx, cancel := context.WithTimeout(m.stopCtx, m.acquireTimeout) + defer cancel() + + lease, err := m.client.Acquire(ctx, dhcp.AcquireRequest{ + Bridge: req.Bridge, + ClientID: req.ClientID, + Hostname: req.Hostname, + VendorClass: req.VendorClass, + HWAddr: hwAddr, + }) + if err != nil { + m.replyAcquireError(msg, err.Error()) + return + } + + if err := m.persistLease(lease); err != nil { + slog.Warn("DHCP Manager: persist lease failed; lease will not survive restart", "client_id", req.ClientID, "err", err) + } + m.trackAndRenew(lease) + + slog.Info("DHCP Manager acquired lease", + "client_id", lease.ClientID, "ip", lease.IP.String(), + "bridge", lease.Bridge, "server_id", lease.ServerID.String(), + "expires_in", time.Until(lease.ExpiresAt())) + + m.replyAcquire(msg, lease) +} + +// handleRelease processes a vpc.dhcp.release request. +func (m *DHCPManager) handleRelease(msg *nats.Msg) { + var req dhcp.ReleaseRequestMsg + if err := json.Unmarshal(msg.Data, &req); err != nil { + m.replyRelease(msg, fmt.Sprintf("invalid release request: %v", err)) + return + } + if req.ClientID == "" { + m.replyRelease(msg, "release: client_id is required") + return + } + + lease, ok := m.get(req.ClientID) + if !ok { + // No active lease; treat as a no-op so the daemon can call release + // in clean-up paths without worrying about order. + m.replyRelease(msg, "") + return + } + + ctx, cancel := context.WithTimeout(m.stopCtx, m.acquireTimeout) + defer cancel() + if err := m.client.Release(ctx, lease); err != nil { + slog.Warn("DHCP Manager: release failed", "client_id", req.ClientID, "err", err) + // Still forget locally — upstream server may or may not have seen + // the RELEASE, but holding on to a lease we've been told to drop + // serves no purpose. + } + m.forget(req.ClientID, "released") + slog.Info("DHCP Manager released lease", "client_id", req.ClientID) + m.replyRelease(msg, "") +} + +// trackAndRenew registers the lease in the in-memory map and starts the +// per-lease renewal goroutine. Replaces any prior tracker for the same +// ClientID. +func (m *DHCPManager) trackAndRenew(lease *dhcp.Lease) { + m.mu.Lock() + if prev, ok := m.leases[lease.ClientID]; ok && prev.renewStop != nil { + prev.renewStop() + m.mu.Unlock() + <-prev.renewDone + m.mu.Lock() + } + + // cancel is stored in tracker.renewStop and fired from forget() or Close(). + ctx, cancel := context.WithCancel(m.stopCtx) //nolint:gosec // G118: cancel stored in tracker.renewStop, fired via forget()/Close() + tracker := &managedLease{ + lease: lease, + renewStop: cancel, + renewDone: make(chan struct{}), + } + m.leases[lease.ClientID] = tracker + m.mu.Unlock() + + m.wg.Add(1) + go m.renewLoop(ctx, tracker) +} + +// renewLoop drives one lease's renewal lifecycle. Sleeps until T1 (with +// ±jitter), attempts Renew; on failure retries at T2; on terminal failure +// past the expiry, publishes TopicLeaseExpired and forgets the lease. +func (m *DHCPManager) renewLoop(ctx context.Context, tracker *managedLease) { + defer m.wg.Done() + defer close(tracker.renewDone) + + for { + lease := m.snapshotLease(tracker) + if lease == nil { + return + } + + // Phase 1 — wait for T1 (± jitter) and try RENEW. + wait := m.jitter(time.Until(lease.RenewAt())) + if !sleep(ctx, wait) { + return + } + + renewed, err := m.tryRenew(ctx, lease) + if err == nil { + tracker.lease = renewed + if perr := m.persistLease(renewed); perr != nil { + slog.Warn("DHCP Manager: persist renewed lease failed", "client_id", renewed.ClientID, "err", perr) + } + continue + } + slog.Warn("DHCP Manager: renew failed, will retry at T2", "client_id", lease.ClientID, "err", err) + + // Phase 2 — retry until T2. + wait = m.jitter(time.Until(lease.RebindAt())) + if !sleep(ctx, wait) { + return + } + renewed, err = m.tryRenew(ctx, lease) + if err == nil { + tracker.lease = renewed + if perr := m.persistLease(renewed); perr != nil { + slog.Warn("DHCP Manager: persist rebound lease failed", "client_id", renewed.ClientID, "err", perr) + } + continue + } + + // Phase 3 — wait out the rest of the lease; on expiry, emit + // lease-expired and drop. + wait = time.Until(lease.ExpiresAt()) + if !sleep(ctx, wait) { + return + } + slog.Warn("DHCP Manager: lease expired without renewal", "client_id", lease.ClientID, "ip", lease.IP.String(), "err", err) + m.publishExpired(lease, "", classifyRenewErr(err)) + m.forget(lease.ClientID, "expired") + return + } +} + +// tryRenew calls the backend Renew under a bounded context. +func (m *DHCPManager) tryRenew(parent context.Context, lease *dhcp.Lease) (*dhcp.Lease, error) { + ctx, cancel := context.WithTimeout(parent, m.acquireTimeout) + defer cancel() + return m.client.Renew(ctx, lease) +} + +// snapshotLease returns the tracker's current lease under the lock. +func (m *DHCPManager) snapshotLease(tracker *managedLease) *dhcp.Lease { + m.mu.Lock() + defer m.mu.Unlock() + return tracker.lease +} + +// get returns the current lease for a ClientID, if tracked. +func (m *DHCPManager) get(clientID string) (*dhcp.Lease, bool) { + m.mu.Lock() + defer m.mu.Unlock() + t, ok := m.leases[clientID] + if !ok { + return nil, false + } + return t.lease, true +} + +// forget removes a lease from the tracking map, KV and stops its renewal +// goroutine. +func (m *DHCPManager) forget(clientID, reason string) { + m.mu.Lock() + t, ok := m.leases[clientID] + if ok { + delete(m.leases, clientID) + } + m.mu.Unlock() + + if ok && t.renewStop != nil { + t.renewStop() + } + if err := m.kv.Delete(clientID); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + slog.Debug("DHCP Manager: delete KV key failed", "client_id", clientID, "reason", reason, "err", err) + } +} + +// persistLease writes the lease state to the KV bucket under ClientID. +func (m *DHCPManager) persistLease(lease *dhcp.Lease) error { + data, err := encodeLease(lease) + if err != nil { + return err + } + if _, err := m.kv.Put(lease.ClientID, data); err != nil { + return fmt.Errorf("put lease %s: %w", lease.ClientID, err) + } + return nil +} + +func (m *DHCPManager) publishExpired(lease *dhcp.Lease, poolName, reason string) { + evt := dhcp.LeaseExpiredEvent{ + ClientID: lease.ClientID, + PoolName: poolName, + IP: lease.IP.String(), + Reason: reason, + } + data, err := json.Marshal(evt) + if err != nil { + slog.Warn("DHCP Manager: marshal lease-expired event failed", "client_id", lease.ClientID, "err", err) + return + } + if err := m.nc.Publish(dhcp.TopicLeaseExpired, data); err != nil { + slog.Warn("DHCP Manager: publish lease-expired failed", "client_id", lease.ClientID, "err", err) + } +} + +func (m *DHCPManager) replyAcquire(msg *nats.Msg, lease *dhcp.Lease) { + reply := dhcp.AcquireReplyMsg{ + IP: lease.IP.String(), + SubnetMask: net.IP(lease.SubnetMask).String(), + Routers: ipsToStrings(lease.Routers), + DNS: ipsToStrings(lease.DNS), + ServerID: lease.ServerID.String(), + HWAddr: lease.HWAddr.String(), + ExpiresUnix: lease.ExpiresAt().Unix(), + } + respondJSON(msg, reply) +} + +func (m *DHCPManager) replyAcquireError(msg *nats.Msg, errMsg string) { + respondJSON(msg, dhcp.AcquireReplyMsg{Error: errMsg}) +} + +func (m *DHCPManager) replyRelease(msg *nats.Msg, errMsg string) { + respondJSON(msg, dhcp.ReleaseReplyMsg{Error: errMsg}) +} + +// jitter applies ±jitterFraction of d to produce an actual sleep duration. +// Negative d is clamped to 0. +func (m *DHCPManager) jitter(d time.Duration) time.Duration { + if d <= 0 { + return 0 + } + if m.jitterFraction <= 0 { + return d + } + spread := float64(d) * m.jitterFraction + offset := (m.rng.Float64()*2 - 1) * spread + out := time.Duration(float64(d) + offset) + if out < 0 { + return 0 + } + return out +} + +// resolveHWAddr parses the hw_addr string supplied in the NATS request, +// falling back to a derived MAC when the request carries no value. +func resolveHWAddr(raw, clientID string, derive func(string) net.HardwareAddr) (net.HardwareAddr, error) { + if raw != "" { + mac, err := net.ParseMAC(raw) + if err != nil { + return nil, fmt.Errorf("parse hw_addr %q: %w", raw, err) + } + return mac, nil + } + if derive == nil { + return nil, fmt.Errorf("hw_addr is empty and no derivation function configured") + } + mac := derive(clientID) + if len(mac) == 0 { + return nil, fmt.Errorf("derive hw_addr for client %q: empty result", clientID) + } + return mac, nil +} + +// classifyRenewErr maps a renew error to a short reason code on the +// lease-expired event. Kept coarse; callers that care about specifics +// can re-log from their side. +func classifyRenewErr(err error) string { + if err == nil { + return "unknown" + } + // nclient4 returns *nclient4.ErrNak on NAK. Rather than taking a + // dependency on its private type here, match the known message. + msg := err.Error() + if strings.Contains(msg, "NAK") || strings.Contains(msg, "Nak") { + return "nak" + } + if strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline exceeded") || strings.Contains(msg, "i/o timeout") { + return "server_unreachable" + } + return "renew_failed" +} + +func ipsToStrings(ips []net.IP) []string { + if len(ips) == 0 { + return nil + } + out := make([]string, 0, len(ips)) + for _, ip := range ips { + out = append(out, ip.String()) + } + return out +} + +// respondJSON marshals v and sends it as the reply. If the message has +// no Reply subject (fire-and-forget), drops the reply silently. +func respondJSON(msg *nats.Msg, v any) { + if msg.Reply == "" { + return + } + data, err := json.Marshal(v) + if err != nil { + slog.Warn("DHCP Manager: marshal reply failed", "err", err) + return + } + if err := msg.Respond(data); err != nil { + slog.Warn("DHCP Manager: respond failed", "err", err) + } +} + +// sleep blocks until d elapses or ctx is done. Returns false if ctx was +// cancelled (caller should exit). +func sleep(ctx context.Context, d time.Duration) bool { + if d <= 0 { + select { + case <-ctx.Done(): + return false + default: + return true + } + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} + +// --- KV encoding --- + +// leaseRecord is the JSON representation persisted in KV and used to +// rehydrate on vpcd startup. Explicit format rather than gob-encoding +// *dhcp.Lease so a future Go type change doesn't invalidate every +// in-flight lease. +type leaseRecord struct { + Version int `json:"version"` + Bridge string `json:"bridge"` + ClientID string `json:"client_id"` + Hostname string `json:"hostname,omitempty"` + VendorClass string `json:"vendor_class,omitempty"` + HWAddr string `json:"hw_addr"` + IP string `json:"ip"` + SubnetMask string `json:"subnet_mask,omitempty"` + Routers []string `json:"routers,omitempty"` + DNS []string `json:"dns,omitempty"` + ServerID string `json:"server_id,omitempty"` + AcquiredUnix int64 `json:"acquired_unix"` + LeaseSeconds int64 `json:"lease_seconds"` + T1Seconds int64 `json:"t1_seconds"` + T2Seconds int64 `json:"t2_seconds"` + RawOfferB64 []byte `json:"raw_offer,omitempty"` + RawACKB64 []byte `json:"raw_ack,omitempty"` +} + +const leaseRecordVersion = 1 + +func encodeLease(l *dhcp.Lease) ([]byte, error) { + rec := leaseRecord{ + Version: leaseRecordVersion, + Bridge: l.Bridge, + ClientID: l.ClientID, + Hostname: l.Hostname, + VendorClass: l.VendorClass, + HWAddr: l.HWAddr.String(), + IP: l.IP.String(), + SubnetMask: maskString(l.SubnetMask), + Routers: ipsToStrings(l.Routers), + DNS: ipsToStrings(l.DNS), + ServerID: l.ServerID.String(), + AcquiredUnix: l.AcquiredAt.Unix(), + LeaseSeconds: int64(l.LeaseDuration / time.Second), + T1Seconds: int64(l.T1 / time.Second), + T2Seconds: int64(l.T2 / time.Second), + RawOfferB64: l.RawOffer, + RawACKB64: l.RawACK, + } + return json.Marshal(rec) +} + +func decodeLease(data []byte) (*dhcp.Lease, error) { + var rec leaseRecord + if err := json.Unmarshal(data, &rec); err != nil { + return nil, fmt.Errorf("unmarshal lease: %w", err) + } + if rec.Version != leaseRecordVersion { + return nil, fmt.Errorf("unsupported lease record version %d", rec.Version) + } + mac, err := net.ParseMAC(rec.HWAddr) + if err != nil { + return nil, fmt.Errorf("parse hw_addr %q: %w", rec.HWAddr, err) + } + return &dhcp.Lease{ + Bridge: rec.Bridge, + ClientID: rec.ClientID, + Hostname: rec.Hostname, + VendorClass: rec.VendorClass, + HWAddr: mac, + IP: net.ParseIP(rec.IP), + SubnetMask: parseMask(rec.SubnetMask), + Routers: parseIPs(rec.Routers), + DNS: parseIPs(rec.DNS), + ServerID: net.ParseIP(rec.ServerID), + AcquiredAt: time.Unix(rec.AcquiredUnix, 0), + LeaseDuration: time.Duration(rec.LeaseSeconds) * time.Second, + T1: time.Duration(rec.T1Seconds) * time.Second, + T2: time.Duration(rec.T2Seconds) * time.Second, + RawOffer: rec.RawOfferB64, + RawACK: rec.RawACKB64, + }, nil +} + +func maskString(m net.IPMask) string { + if len(m) == 0 { + return "" + } + return net.IP(m).String() +} + +func parseMask(s string) net.IPMask { + ip := net.ParseIP(s) + if ip == nil { + return nil + } + ip4 := ip.To4() + if ip4 == nil { + return nil + } + return net.IPv4Mask(ip4[0], ip4[1], ip4[2], ip4[3]) +} + +func parseIPs(ss []string) []net.IP { + if len(ss) == 0 { + return nil + } + out := make([]net.IP, 0, len(ss)) + for _, s := range ss { + if ip := net.ParseIP(s); ip != nil { + out = append(out, ip) + } + } + return out +} diff --git a/spinifex/services/vpcd/dhcp_manager_test.go b/spinifex/services/vpcd/dhcp_manager_test.go new file mode 100644 index 000000000..f0b250600 --- /dev/null +++ b/spinifex/services/vpcd/dhcp_manager_test.go @@ -0,0 +1,354 @@ +package vpcd + +import ( + "context" + "encoding/json" + "errors" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/require" +) + +// newTestDHCPManager spins up an embedded JetStream, wires a Fake client, +// and returns the Manager along with the NATS connection. Cleanup of NATS +// is handled by testutil; the Manager is closed via t.Cleanup. +func newTestDHCPManager(t *testing.T, opts ...DHCPManagerOption) (*DHCPManager, *nats.Conn, *dhcp.Fake) { + t.Helper() + _, nc, js := testutil.StartTestJetStream(t) + fake := dhcp.NewFake() + m, err := NewDHCPManager(nc, js, fake, opts...) + require.NoError(t, err) + t.Cleanup(m.Close) + subs, err := m.Subscribe(nc) + require.NoError(t, err) + t.Cleanup(func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }) + return m, nc, fake +} + +func requestAcquire(t *testing.T, nc *nats.Conn, req dhcp.AcquireRequestMsg) dhcp.AcquireReplyMsg { + t.Helper() + data, err := json.Marshal(req) + require.NoError(t, err) + msg, err := nc.Request(dhcp.TopicAcquire, data, 5*time.Second) + require.NoError(t, err) + var reply dhcp.AcquireReplyMsg + require.NoError(t, json.Unmarshal(msg.Data, &reply)) + return reply +} + +func requestRelease(t *testing.T, nc *nats.Conn, clientID string) dhcp.ReleaseReplyMsg { + t.Helper() + data, err := json.Marshal(dhcp.ReleaseRequestMsg{ClientID: clientID}) + require.NoError(t, err) + msg, err := nc.Request(dhcp.TopicRelease, data, 5*time.Second) + require.NoError(t, err) + var reply dhcp.ReleaseReplyMsg + require.NoError(t, json.Unmarshal(msg.Data, &reply)) + return reply +} + +func TestDHCPManager_AcquireReturnsLease(t *testing.T) { + _, nc, fake := newTestDHCPManager(t) + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{ + Bridge: "br-wan", ClientID: "eni-1", Hostname: "eni-1", VendorClass: "i-abc", + }) + require.Empty(t, reply.Error, "unexpected error: %s", reply.Error) + require.Equal(t, "192.0.2.100", reply.IP) + require.NotEmpty(t, reply.ServerID) + require.NotZero(t, reply.ExpiresUnix) + require.NotEmpty(t, reply.HWAddr) + require.Equal(t, 1, fake.AcquireCount()) +} + +func TestDHCPManager_AcquireIdempotentWhileLeaseActive(t *testing.T) { + _, nc, fake := newTestDHCPManager(t) + req := dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-idem"} + + first := requestAcquire(t, nc, req) + require.Empty(t, first.Error) + second := requestAcquire(t, nc, req) + require.Empty(t, second.Error) + + require.Equal(t, first.IP, second.IP, "idempotent acquire should return the same IP") + require.Equal(t, 1, fake.AcquireCount(), "second request must not trigger a fresh DORA") +} + +func TestDHCPManager_AcquireFailureSurfacesError(t *testing.T) { + _, nc, fake := newTestDHCPManager(t) + fake.AcquireHook = func(dhcp.AcquireRequest) (*dhcp.Lease, error) { + return nil, errors.New("server unreachable") + } + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-err"}) + require.Contains(t, reply.Error, "server unreachable") + require.Empty(t, reply.IP) +} + +func TestDHCPManager_ReleaseRemovesLease(t *testing.T) { + m, nc, fake := newTestDHCPManager(t) + req := dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-release"} + require.Empty(t, requestAcquire(t, nc, req).Error) + + reply := requestRelease(t, nc, req.ClientID) + require.Empty(t, reply.Error) + require.Equal(t, 1, fake.ReleaseCount()) + + _, ok := m.get("eni-release") + require.False(t, ok, "lease should be forgotten after Release") +} + +func TestDHCPManager_ReleaseUnknownIsNoop(t *testing.T) { + _, nc, fake := newTestDHCPManager(t) + reply := requestRelease(t, nc, "nonexistent") + require.Empty(t, reply.Error) + require.Equal(t, 0, fake.ReleaseCount()) +} + +func TestDHCPManager_AcquireRejectsMissingFields(t *testing.T) { + _, nc, _ := newTestDHCPManager(t) + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "", ClientID: "eni-x"}) + require.Contains(t, reply.Error, "bridge and client_id are required") +} + +func TestDHCPManager_HWAddrDerivedFromClientID(t *testing.T) { + _, nc, fake := newTestDHCPManager(t) + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-mac"}) + require.Empty(t, reply.Error) + lease, ok := fake.HeldLease("eni-mac") + require.True(t, ok) + // generateMAC uses the 02:00:00 locally-administered unicast prefix. + require.Equal(t, "02:00:00", lease.HWAddr.String()[:8]) +} + +func TestDHCPManager_HWAddrExplicitlySuppliedIsUsed(t *testing.T) { + _, nc, fake := newTestDHCPManager(t) + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{ + Bridge: "br-wan", ClientID: "eni-mac-explicit", HWAddr: "02:aa:bb:cc:dd:ee", + }) + require.Empty(t, reply.Error) + lease, _ := fake.HeldLease("eni-mac-explicit") + require.Equal(t, "02:aa:bb:cc:dd:ee", lease.HWAddr.String()) +} + +func TestDHCPManager_RenewalFiresAndUpdatesLease(t *testing.T) { + tmpl := dhcp.DefaultLeaseTemplate() + tmpl.LeaseDuration = 40 * time.Millisecond + fake := dhcp.NewFake() + fake.SetDefaultLease(tmpl) + + _, nc, js := testutil.StartTestJetStream(t) + m, err := NewDHCPManager(nc, js, fake, WithDHCPJitterFraction(0)) + require.NoError(t, err) + t.Cleanup(m.Close) + subs, err := m.Subscribe(nc) + require.NoError(t, err) + t.Cleanup(func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }) + + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-renew"}) + require.Empty(t, reply.Error) + + // T1 = lease/2 = 20 ms. Wait a few multiples to catch at least one renew. + require.Eventually(t, func() bool { return fake.RenewCount() >= 1 }, 2*time.Second, 10*time.Millisecond) +} + +func TestDHCPManager_RenewFailureEmitsLeaseExpired(t *testing.T) { + tmpl := dhcp.DefaultLeaseTemplate() + tmpl.LeaseDuration = 60 * time.Millisecond + fake := dhcp.NewFake() + fake.SetDefaultLease(tmpl) + fake.RenewHook = func(*dhcp.Lease) (*dhcp.Lease, error) { + return nil, errors.New("server NAK") + } + + _, nc, js := testutil.StartTestJetStream(t) + m, err := NewDHCPManager(nc, js, fake, WithDHCPJitterFraction(0)) + require.NoError(t, err) + t.Cleanup(m.Close) + + var events atomic.Int32 + expiredMsg := make(chan dhcp.LeaseExpiredEvent, 1) + sub, err := nc.Subscribe(dhcp.TopicLeaseExpired, func(msg *nats.Msg) { + events.Add(1) + var evt dhcp.LeaseExpiredEvent + if err := json.Unmarshal(msg.Data, &evt); err == nil { + select { + case expiredMsg <- evt: + default: + } + } + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + subs, err := m.Subscribe(nc) + require.NoError(t, err) + t.Cleanup(func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }) + + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-expire"}) + require.Empty(t, reply.Error) + + // T1=30ms → try, fail. T2=~52ms → try, fail. Expires at 60ms. + select { + case evt := <-expiredMsg: + require.Equal(t, "eni-expire", evt.ClientID) + require.Equal(t, "nak", evt.Reason) + case <-time.After(2 * time.Second): + t.Fatalf("expected lease-expired event (events=%d)", events.Load()) + } +} + +func TestDHCPManager_BootstrapRehydratesLease(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + fake := dhcp.NewFake() + + m1, err := NewDHCPManager(nc, js, fake) + require.NoError(t, err) + subs, err := m1.Subscribe(nc) + require.NoError(t, err) + + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-boot"}) + require.Empty(t, reply.Error) + + // Close first manager; KV entry should survive. + for _, s := range subs { + _ = s.Unsubscribe() + } + m1.Close() + + // Second manager picks up the lease on Bootstrap. + fake2 := dhcp.NewFake() + m2, err := NewDHCPManager(nc, js, fake2) + require.NoError(t, err) + t.Cleanup(m2.Close) + require.NoError(t, m2.Bootstrap(context.Background())) + + lease, ok := m2.get("eni-boot") + require.True(t, ok, "manager should rehydrate persisted lease") + require.Equal(t, "192.0.2.100", lease.IP.String()) + require.Equal(t, 0, fake2.AcquireCount(), "no fresh DORA on bootstrap") +} + +func TestDHCPManager_BootstrapDropsExpiredLeases(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + fake := dhcp.NewFake() + + m1, err := NewDHCPManager(nc, js, fake) + require.NoError(t, err) + subs, err := m1.Subscribe(nc) + require.NoError(t, err) + + reply := requestAcquire(t, nc, dhcp.AcquireRequestMsg{Bridge: "br-wan", ClientID: "eni-stale"}) + require.Empty(t, reply.Error) + + for _, s := range subs { + _ = s.Unsubscribe() + } + m1.Close() + + // Rewrite the persisted record so AcquiredAt + LeaseDuration is in the past. + kv, err := js.KeyValue(dhcp.KVBucketDHCPLeases) + require.NoError(t, err) + entry, err := kv.Get("eni-stale") + require.NoError(t, err) + var rec map[string]any + require.NoError(t, json.Unmarshal(entry.Value(), &rec)) + rec["acquired_unix"] = time.Now().Add(-2 * time.Hour).Unix() + rec["lease_seconds"] = int64(60) + raw, err := json.Marshal(rec) + require.NoError(t, err) + _, err = kv.Put("eni-stale", raw) + require.NoError(t, err) + + fake2 := dhcp.NewFake() + m2, err := NewDHCPManager(nc, js, fake2) + require.NoError(t, err) + t.Cleanup(m2.Close) + + var expired sync.WaitGroup + expired.Add(1) + sub, err := nc.Subscribe(dhcp.TopicLeaseExpired, func(msg *nats.Msg) { + var evt dhcp.LeaseExpiredEvent + if err := json.Unmarshal(msg.Data, &evt); err == nil && evt.ClientID == "eni-stale" { + expired.Done() + } + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + require.NoError(t, m2.Bootstrap(context.Background())) + + done := make(chan struct{}) + go func() { expired.Wait(); close(done) }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("expected lease-expired event for stale lease") + } + _, err = kv.Get("eni-stale") + require.ErrorIs(t, err, nats.ErrKeyNotFound) +} + +func TestDHCPManager_JitterWithinBounds(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + m, err := NewDHCPManager(nc, js, dhcp.NewFake(), WithDHCPJitterFraction(0.1)) + require.NoError(t, err) + t.Cleanup(m.Close) + + base := 100 * time.Millisecond + minOK := time.Duration(float64(base) * 0.9) + maxOK := time.Duration(float64(base) * 1.1) + for i := range 100 { + got := m.jitter(base) + if got < minOK || got > maxOK { + t.Fatalf("iteration %d: jitter produced %v, outside [%v,%v]", i, got, minOK, maxOK) + } + } +} + +func TestDHCPManager_EncodeDecodeLeaseRoundTrip(t *testing.T) { + mac, _ := net.ParseMAC("02:00:00:de:ad:be") + in := &dhcp.Lease{ + Bridge: "br-wan", ClientID: "eni-rt", Hostname: "eni-rt", VendorClass: "i-xyz", + HWAddr: mac, + IP: net.ParseIP("192.0.2.50"), + SubnetMask: net.CIDRMask(24, 32), + Routers: []net.IP{net.ParseIP("192.0.2.1")}, + DNS: []net.IP{net.ParseIP("1.1.1.1")}, + ServerID: net.ParseIP("192.0.2.1"), + AcquiredAt: time.Unix(1_700_000_000, 0).UTC(), + LeaseDuration: time.Hour, + T1: 30 * time.Minute, + T2: 52 * time.Minute, + RawOffer: []byte{0x01, 0x02}, + RawACK: []byte{0x03, 0x04}, + } + data, err := encodeLease(in) + require.NoError(t, err) + out, err := decodeLease(data) + require.NoError(t, err) + + require.Equal(t, in.ClientID, out.ClientID) + require.Equal(t, in.IP.String(), out.IP.String()) + require.Equal(t, in.LeaseDuration, out.LeaseDuration) + require.Equal(t, in.HWAddr.String(), out.HWAddr.String()) + require.Equal(t, in.RawACK, out.RawACK) + require.Equal(t, in.RawOffer, out.RawOffer) +} diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index 2716aa3cc..24508dcbe 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -10,7 +10,9 @@ import ( "path/filepath" "strings" "syscall" + "time" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -365,7 +367,40 @@ func launchService(cfg *Config) error { // Runs after subscribing so new events are not missed during reconciliation. ReconcileFromKV(ctx, nc, topo) - slog.Info("vpcd service started, waiting for VPC lifecycle events", "subscriptions", len(subs)) + // DHCP manager: services vpc.dhcp.acquire/release requests from the + // daemon-side ExternalIPAM handlers, runs a renewal goroutine per lease, + // and persists leases in spinifex-dhcp-leases KV. Started unconditionally + // — pools with source="static" never issue acquire requests, so the + // Manager sits idle until something actually wants DHCP. + js, err := nc.JetStream() + if err != nil { + slog.Error("Failed to get JetStream context for DHCP manager", "err", err) + return err + } + dhcpManager, err := NewDHCPManager(nc, js, dhcp.NewNClient4(15*time.Second, 3)) + if err != nil { + slog.Error("Failed to create DHCP manager", "err", err) + return err + } + defer dhcpManager.Close() + + if err := dhcpManager.Bootstrap(ctx); err != nil { + slog.Error("DHCP manager bootstrap failed", "err", err) + return err + } + dhcpSubs, err := dhcpManager.Subscribe(nc) + if err != nil { + slog.Error("DHCP manager subscribe failed", "err", err) + return err + } + defer func() { + for _, s := range dhcpSubs { + _ = s.Unsubscribe() + } + }() + + slog.Info("vpcd service started, waiting for VPC lifecycle events", + "subscriptions", len(subs), "dhcp_subscriptions", len(dhcpSubs)) // Wait for shutdown signal sigChan := make(chan os.Signal, 1)