mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat: in-process DHCP client and Manager in vpcd (mulga-990)
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.
This commit is contained in:
Ben Duncan
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user