diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index ccc47b264..49596d02a 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -2030,22 +2030,10 @@ func (d *Daemon) startCluster() error { }) } - // DNS record writer: a queue-group consumer of dns.recordset.change. Every - // node subscribes, so the writer locks each zone before its - // read-modify-write. No-op when northstar S3 is not configured. - if sub, err := d.dnsWriter.Subscribe(d.natsConn); err != nil { - return fmt.Errorf("failed to subscribe DNS record writer: %w", err) - } else if sub != nil { - d.mu.Lock() - d.natsSubscriptions[handlers_dns.SubjectRecordsetChange] = sub - d.mu.Unlock() - slog.Info("Subscribed DNS record writer", "subject", handlers_dns.SubjectRecordsetChange, "queue", handlers_dns.QueueGroup) - } - - // DNS drift backstop: periodically rebuild managed records from the live - // cross-tenant inventory and converge the zone. Started on every node but - // gated on a per-cycle leader election, so one node publishes per interval. - // No-op when northstar is not configured. + // DNS: rebuild managed records from the live cross-tenant inventory and + // converge the zone, on a resource change or the interval. Started on every + // node but gated on a per-cycle leader election, so exactly one node writes a + // zone. No-op when northstar is not configured. if d.dnsReconciler.Enabled() { go d.dnsReconciler.Run(d.ctx) slog.Info("Started DNS reconcile backstop", "interval_ms", otelsetup.Millis(handlers_dns.DefaultReconcileInterval)) diff --git a/spinifex/handlers/dns/publish.go b/spinifex/handlers/dns/publish.go deleted file mode 100644 index 9bce1f12c..000000000 --- a/spinifex/handlers/dns/publish.go +++ /dev/null @@ -1,36 +0,0 @@ -package dns - -import ( - "context" - "log/slog" - - "github.com/mulgadc/spinifex/spinifex/utils" - "github.com/nats-io/nats.go" -) - -// PublishChanges sends a batch of record-set changes to the writer via NATS -// request-reply and waits for the ack. It is best-effort from the caller's -// perspective: a nil connection or empty batch is a no-op, and the error is -// returned for the caller to log without failing the resource operation. -func PublishChanges(nc *nats.Conn, accountID string, changes []Change) (*ChangeResult, error) { - if nc == nil || len(changes) == 0 { - return &ChangeResult{}, nil - } - // Detached from any request context: the publish is best-effort and its own - // timeout bounds it, so a cancelled caller ctx must not abort the write. - return utils.NATSRequest[ChangeResult](context.Background(), nc, SubjectRecordsetChange, ChangeBatch{Changes: changes}, requestTimeout, accountID) -} - -// PublishChangesBestEffort publishes a batch and logs the outcome without -// propagating the error, so DNS registration never blocks a resource operation. -// The reconcile loop repairs anything missed by a failed publish. -func PublishChangesBestEffort(nc *nats.Conn, accountID string, changes []Change) { - if len(changes) == 0 { - return - } - if res, err := PublishChanges(nc, accountID, changes); err != nil { - slog.Warn("dns: publish changes failed (continuing)", "count", len(changes), "error", err) - } else { - slog.Info("dns: published changes", "count", len(changes), "zones", res.Zones) - } -} diff --git a/spinifex/handlers/dns/reconcile_election_test.go b/spinifex/handlers/dns/reconcile_election_test.go new file mode 100644 index 000000000..5fa483658 --- /dev/null +++ b/spinifex/handlers/dns/reconcile_election_test.go @@ -0,0 +1,63 @@ +//test:in-package — the election gate lives inside reconcileOnce, so driving it +//needs the unexported Reconciler fields and the writer's resolved S3 config. + +package dns + +import ( + "testing" + + reconcilelock "github.com/mulgadc/spinifex/spinifex/network/reconcile" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// electingReconcilerFor is reconcilerFor with a live connection, so a pass runs +// the election instead of skipping it. +func electingReconcilerFor(w *Writer, nc *nats.Conn, holder string, desired func() DesiredSet) *Reconciler { + r := reconcilerFor(w, desired) + r.nc = nc + r.holder = holder + return r +} + +// The zone lock is gone, and what replaces it is that a zone has one writer: +// the node holding the reconcile lease. A node that loses the election must +// therefore not touch the object at all — if it did, two nodes could be +// read-modify-writing the same zone with nothing serialising them. +func TestReconcileOnce_ALoserOfTheElectionDoesNotWriteTheZone(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + w, objects := newTestWriter(t) + before := objects[testBase+".toml"] + require.NotEmpty(t, before) + + // A peer holds the lease for the whole pass, so this node cannot win it. + release, elected := reconcilelock.AcquireLeader(t.Context(), nc, KVBucketDNSReconcile, "peer") + require.True(t, elected, "the peer has to hold the lease or this test proves nothing") + + r := electingReconcilerFor(w, nc, "node1", func() DesiredSet { + return DesiredSet{Changes: []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}} + }) + r.reconcileOnce(t.Context()) + + assert.Equal(t, before, objects[testBase+".toml"], + "a node that did not win the election must leave the zone object exactly as it found it") + + release() +} + +// The counterpart, so the test above is not passing because the pass was broken +// for some other reason: with the lease free, the same reconciler writes. +func TestReconcileOnce_TheWinnerOfTheElectionWritesTheZone(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + w, objects := newTestWriter(t) + + r := electingReconcilerFor(w, nc, "node1", func() DesiredSet { + return DesiredSet{Changes: []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}} + }) + r.reconcileOnce(t.Context()) + + assert.Contains(t, objects[testBase+".toml"], `address = "10.200.1.9"`, + "an uncontended pass has to reach the zone object") +} diff --git a/spinifex/handlers/dns/types.go b/spinifex/handlers/dns/types.go index fe7b9c567..91bac4ad0 100644 --- a/spinifex/handlers/dns/types.go +++ b/spinifex/handlers/dns/types.go @@ -1,35 +1,23 @@ -// Package dns is the control-plane DNS record writer. Every daemon consumes -// dns.recordset.change from one queue group and owns the read-modify-write of -// zone TOML files in s3://northstar/, using the system predastore credentials. -// The queue group balances load, so the read-modify-write is serialised per zone -// by an explicit lock. Northstar itself stays read-only (N4 intact). +// Package dns is the control-plane DNS record writer. Zone TOML files in +// s3://northstar/ are written by the node that wins the reconcile election, and +// by nothing else: the desired record set is derived from the resource stores +// each pass, so there is one writer per zone and no lock is needed to serialise +// it. Northstar itself stays read-only (N4 intact). package dns -import "time" - -// NATS transport for record-set changes. +// NATS transport for zone changes. const ( - // SubjectRecordsetChange is the request-reply subject lifecycle handlers - // publish DNS changes on. - SubjectRecordsetChange = "dns.recordset.change" // SubjectZoneReload is the fan-out subject the writer publishes after a zone // PUT so every northstar instance reloads just that zone immediately, instead // of waiting for the next S3 sync poll. No queue group: all servers consume. SubjectZoneReload = "dns.zone.reload" - // QueueGroup delivers each message to exactly one writer. It does NOT - // serialise distinct messages: concurrent changes land on different daemons in - // parallel, so the per-zone read-modify-write takes its own lock. - QueueGroup = "spinifex-workers" - // KVBucketDNSReconcile elects the one node that publishes a converging batch - // per drift cycle. Its own bucket, so the election never blocks another - // reconcile loop. + // KVBucketDNSReconcile elects the one node that writes the zones for a drift + // cycle. Its own bucket, so the election never blocks another reconcile loop. KVBucketDNSReconcile = "spinifex-dns-reconcile" // PrivateZone is the fixed AWS-parity private DNS zone (IMDS synthHostname). PrivateZone = "compute.internal" // DefaultTTL is applied when a change omits a TTL. DefaultTTL uint32 = 60 - // requestTimeout bounds a producer's wait for the writer's ack. - requestTimeout = 5 * time.Second ) // Action is the change verb (UPSERT replaces the RRset, DELETE withdraws it). diff --git a/spinifex/handlers/dns/writer.go b/spinifex/handlers/dns/writer.go index ba84524a6..8ac59dc19 100644 --- a/spinifex/handlers/dns/writer.go +++ b/spinifex/handlers/dns/writer.go @@ -1,7 +1,6 @@ package dns import ( - "context" "encoding/json" "errors" "fmt" @@ -12,13 +11,13 @@ import ( nsconfig "github.com/mulgadc/northstar/pkg/config" "github.com/mulgadc/spinifex/spinifex/config" - "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" toml "github.com/pelletier/go-toml/v2" ) // Writer is the control-plane DNS record writer. It owns the read-modify-write // of zone TOML files in s3://northstar/ using the system predastore credentials. +// Only the reconcile's elected node calls it, so a zone has one writer at a time. type Writer struct { enabled bool s3cfg *nsconfig.S3Config @@ -28,7 +27,6 @@ type Writer struct { nc *nats.Conn quotaEnabled bool quotas Quotas - locker *zoneLocker } // NewWriter resolves the northstar S3 endpoint/bucket from the node's @@ -39,9 +37,6 @@ type Writer struct { // events. func NewWriter(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn) *Writer { w := &Writer{ttl: DefaultTTL, nc: nc, quotas: DefaultQuotas()} - if cfg != nil { - w.locker = newZoneLocker(nc, cfg.Node) - } zoneCfg, ok := zoneS3Config(cfg) if !ok { slog.Info("dns writer: northstar S3 not configured, record registration disabled") @@ -65,19 +60,6 @@ func NewWriter(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn) // reconciler asks before it has one, and a missing writer is a disabled one. func (w *Writer) Enabled() bool { return w != nil && w.enabled } -// Subscribe registers the queue-group request-reply consumer. It is a no-op when -// the writer is disabled. Joining the queue group is what exposes this writer to -// its peers, so the zone locker adopts the connection here if it has none yet. -func (w *Writer) Subscribe(nc *nats.Conn) (*nats.Subscription, error) { - if !w.enabled { - return nil, nil - } - w.locker.bindConn(nc) - return nc.QueueSubscribe(SubjectRecordsetChange, QueueGroup, func(msg *nats.Msg) { - utils.ServeNATSRequest(msg, w.ApplyBatch) - }) -} - // ApplyBatch applies a batch of changes, grouped per zone so each zone object is // read-modified-written once. func (w *Writer) ApplyBatch(batch *ChangeBatch) (*ChangeResult, error) { @@ -128,22 +110,10 @@ func (w *Writer) publishReload(zone string) { } // applyZone read-modify-writes a single zone TOML for its changes. It returns -// whether the zone object was rewritten. The read-modify-write holds the -// cluster-wide per-zone lock: a NATS queue group load-balances messages rather -// than serialising them, so without it concurrent daemons lose each other's -// records and can leave the object unparseable. +// whether the zone object was rewritten. Unserialised on purpose: the only +// caller is the reconcile pass on the elected node, so there is no second writer +// for a lock to exclude. func (w *Writer) applyZone(zone string, changes []Change) (bool, error) { - // Bounded by the producer's ack budget: a lock wait outliving the request - // would apply a change nobody is listening for any more. - ctx, cancel := context.WithTimeout(context.Background(), requestTimeout) - defer cancel() - - lock, err := w.locker.lockZone(ctx, zone) - if err != nil { - return false, err - } - defer lock.Release(ctx) - cfg, exists, err := nsconfig.ReadZoneRaw(w.s3cfg, zone) switch { case isCorruptZone(err): @@ -207,11 +177,6 @@ func (w *Writer) applyZone(zone string, changes []Change) (bool, error) { if err != nil { return false, err } - // The lease may have been reaped while the read ran, in which case a peer can - // already hold the zone. Abort rather than complete the write unserialised. - if lock.Expired() { - return false, fmt.Errorf("apply zone %s: lock lease expired before write, retry", zone) - } if err := nsconfig.WriteZoneFile(w.s3cfg, zone, body); err != nil { return false, err } diff --git a/spinifex/handlers/dns/writer_concurrency_test.go b/spinifex/handlers/dns/writer_concurrency_test.go deleted file mode 100644 index 65d116179..000000000 --- a/spinifex/handlers/dns/writer_concurrency_test.go +++ /dev/null @@ -1,220 +0,0 @@ -package dns - -import ( - "fmt" - "os" - "path/filepath" - "sync" - "testing" - "time" - - nsconfig "github.com/mulgadc/northstar/pkg/config" - "github.com/mulgadc/spinifex/spinifex/config" - "github.com/mulgadc/spinifex/spinifex/testutil" - "github.com/nats-io/nats.go" - toml "github.com/pelletier/go-toml/v2" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// parseZone decodes a stored zone object the same way ReadZoneRaw does, so a -// body that fails here is exactly the body that wedges the writer and northstar. -func parseZone(t *testing.T, body string) nsconfig.ConfigArr { - t.Helper() - var cfg nsconfig.ConfigArr - require.NoError(t, toml.Unmarshal([]byte(body), &cfg)) - return cfg -} - -// hasA reports whether the zone holds an A record with the given address. -func hasA(cfg nsconfig.ConfigArr, address string) bool { - for _, rec := range cfg.Records { - if rec.Type == nsconfig.TypeA && rec.Address == address { - return true - } - } - return false -} - -// clusterWriters builds n Writers that share one fakeS3 bucket and one NATS -// connection, standing in for n daemons in a queue group. The base zone is -// pre-seeded so every writer read-modify-writes an existing object, which is the -// racing path. -func clusterWriters(t *testing.T, nc *nats.Conn, n int) ([]*Writer, map[string]string) { - t.Helper() - endpoint, objects := fakeS3(t, "northstar") - objects["spx3.net.toml"] = `version = 1.0 -[domain] -domain = "spx3.net" -active = true -soa = "ns1.spx3.net." -[defaults] -ttl = 300 -type = 1 -class = 1 -[[records]] -domain = "" -type = 2 -address = "ns1.spx3.net." -` - tomlBody := fmt.Sprintf(`listen = "0.0.0.0:5300" -default_domain = "spx3.net" -[s3] -endpoint = %q -bucket = "northstar" -region = "us-east-1" -access_key = "READONLY" -secret_key = "READONLY" -`, endpoint) - configPath := filepath.Join(t.TempDir(), "northstar.toml") - require.NoError(t, os.WriteFile(configPath, []byte(tomlBody), 0o600)) - - writers := make([]*Writer, 0, n) - for i := range n { - cfg := &config.Config{ - Node: fmt.Sprintf("node%d", i+1), - Predastore: config.PredastoreConfig{AccessKey: "SYSTEM", SecretKey: "SYSTEMSECRET"}, - Northstar: config.NorthstarConfig{ConfigPath: configPath}, - } - w := NewWriter(cfg, nil, nc) - require.True(t, w.Enabled()) - writers = append(writers, w) - } - return writers, objects -} - -// TestConcurrentWritersDoNotLoseRecords is the regression test for the zone -// read-modify-write race: a NATS queue group load-balances messages rather than -// serialising them, so before the per-zone lock, concurrent daemons overwrote -// each other's records. -// -// fakeS3 serialises whole requests under a mutex, so this reproduces the -// lost-update half of the bug, not the byte-splicing half that needs a real -// non-atomic multi-shard PUT. -func TestConcurrentWritersDoNotLoseRecords(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, 10*time.Second) - - const nodes = 6 - writers, objects := clusterWriters(t, nc, nodes) - - var wg sync.WaitGroup - errs := make([]error, nodes) - for i, w := range writers { - wg.Go(func() { - _, errs[i] = w.ApplyBatch(&ChangeBatch{Changes: []Change{{ - Action: ActionUpsert, - Zone: "spx3.net", - Name: fmt.Sprintf("lb-%d.elb.spx3.net", i), - Type: "A", - Value: fmt.Sprintf("10.200.1.%d", i+10), - }}}) - }) - } - wg.Wait() - - for i, err := range errs { - require.NoError(t, err, "writer %d", i) - } - - body, ok := objects["spx3.net.toml"] - require.True(t, ok, "zone object must exist") - - // The object must still parse: an interleaved write is what produced - // "toml: unterminated basic string" in production. - cfg := parseZone(t, body) - - // Every writer's record must have survived, not just the last one in. - for i := range nodes { - want := fmt.Sprintf("10.200.1.%d", i+10) - assert.True(t, hasA(cfg, want), "record %s from writer %d was lost", want, i) - } - - // The pre-seeded NS record must not have been dropped along the way. - var foundNS bool - for _, rec := range cfg.Records { - if rec.Type == nsconfig.TypeNS { - foundNS = true - } - } - assert.True(t, foundNS, "structural NS record must survive concurrent upserts") -} - -// TestConcurrentWritersAcrossZonesDoNotBlock proves the lock is per-zone: two -// writers hitting different zones must both complete without contending. -func TestConcurrentWritersAcrossZonesDoNotBlock(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - // A wait budget this short fails if the two writers serialise on one key. - shrinkZoneLockWaits(t, 200*time.Millisecond) - - writers, objects := clusterWriters(t, nc, 2) - - var wg sync.WaitGroup - errs := make([]error, 2) - zones := []string{"spx3.net", "compute.internal"} - for i, w := range writers { - wg.Go(func() { - _, errs[i] = w.ApplyBatch(&ChangeBatch{Changes: []Change{{ - Action: ActionUpsert, - Zone: zones[i], - Name: "host." + zones[i], - Type: "A", - Value: fmt.Sprintf("10.0.0.%d", i+1), - }}}) - }) - } - wg.Wait() - - for i, err := range errs { - require.NoError(t, err, "zone %s", zones[i]) - } - assert.Contains(t, objects, "spx3.net.toml") - assert.Contains(t, objects, "compute.internal.toml") -} - -// TestWriterRebuildsCorruptZone covers the recovery half: a zone whose stored -// bytes do not parse must be replaced, not wedge every future write. Both repair -// paths read the zone first, so without this a corrupt object was permanent. -func TestWriterRebuildsCorruptZone(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, time.Second) - - writers, objects := clusterWriters(t, nc, 1) - objects["spx3.net.toml"] = "version = 1.0\n[domain]\ndomain = \"spx3.ne" - - res, err := writers[0].ApplyBatch(&ChangeBatch{Changes: []Change{{ - Action: ActionUpsert, - Zone: "spx3.net", - Name: "lb-1.elb.spx3.net", - Type: "A", - Value: "10.200.1.9", - }}}) - require.NoError(t, err, "a corrupt zone must be rebuilt, not returned as an error") - assert.Equal(t, []string{"spx3.net"}, res.Zones) - - cfg := parseZone(t, objects["spx3.net.toml"]) - assert.True(t, hasA(cfg, "10.200.1.9"), "the change that hit the corrupt zone must be applied to the rebuild") -} - -// TestWriterCorruptZoneDeleteOnlyIsNoop guards the other branch: a withdrawal -// against a corrupt zone has nothing to materialise, so it must not write a -// stub zone that drops every other record. -func TestWriterCorruptZoneDeleteOnlyIsNoop(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, time.Second) - - writers, objects := clusterWriters(t, nc, 1) - corrupt := "version = 1.0\n[domain]\ndomain = \"spx3.ne" - objects["spx3.net.toml"] = corrupt - - res, err := writers[0].ApplyBatch(&ChangeBatch{Changes: []Change{{ - Action: ActionDelete, - Zone: "spx3.net", - Name: "lb-1.elb.spx3.net", - Type: "A", - Value: "10.200.1.9", - }}}) - require.NoError(t, err) - assert.Empty(t, res.Zones, "a delete-only batch must not materialise a zone") - assert.Equal(t, corrupt, objects["spx3.net.toml"], "the reconciler's UPSERT set owns the rebuild") -} diff --git a/spinifex/handlers/dns/zonelock.go b/spinifex/handlers/dns/zonelock.go deleted file mode 100644 index 912ffe19f..000000000 --- a/spinifex/handlers/dns/zonelock.go +++ /dev/null @@ -1,217 +0,0 @@ -package dns - -import ( - "context" - "errors" - "fmt" - "github.com/mulgadc/spinifex/spinifex/otelsetup" - "log/slog" - "math/rand/v2" - "strings" - "sync" - "time" - - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" -) - -const ( - // KVBucketDNSZoneLock is dedicated to zone write locks: a zone held mid-write - // must never block an unrelated reconcile loop sharing a bucket. - KVBucketDNSZoneLock = "spinifex-dns-zone-lock" - - // zoneLockTTL bounds crash recovery — a holder that dies mid-write parks its - // zone for at most this long. Deliberately above requestTimeout so a producer - // has already abandoned its request before the lock it is waiting on can - // expire under a still-running holder. - zoneLockTTL = 10 * time.Second - - // zoneLockBucketTimeout bounds attaching to the bucket. Kept short because a - // producer is blocked on requestTimeout throughout; JetStream being unready is - // better reported than waited out. - zoneLockBucketTimeout = 2 * time.Second -) - -// Acquire blocks with backoff rather than skipping like a reconcile leader -// election: a writer must apply its change. The total wait stays under -// requestTimeout so the producer's ack is not lost to lock contention. Vars, not -// consts, so tests can shrink them. -var ( - zoneLockWaitFor = 2500 * time.Millisecond - zoneLockStep = 20 * time.Millisecond - zoneLockStepMax = 200 * time.Millisecond -) - -// zoneLocker hands out cluster-wide per-zone write locks over JetStream KV. It -// exists because a NATS queue group load-balances messages, it does not -// serialise them: every daemon in the queue group can otherwise read-modify-write -// the same zone object concurrently, losing records or corrupting the TOML. -type zoneLocker struct { - nc *nats.Conn - holder string - - mu sync.Mutex - kv jetstream.KeyValue -} - -// newZoneLocker builds a locker for one node. A nil conn yields a locker that -// no-ops, which is correct only because a writer without a NATS connection has no -// queue-group peer to race — bindConn upgrades it if a connection appears. -func newZoneLocker(nc *nats.Conn, holder string) *zoneLocker { - if strings.TrimSpace(holder) == "" { - holder = "unknown" - } - return &zoneLocker{nc: nc, holder: holder} -} - -// bindConn adopts a connection when the locker was built without one, so a -// writer that only receives its conn at Subscribe time still locks. -func (l *zoneLocker) bindConn(nc *nats.Conn) { - if l == nil || nc == nil { - return - } - l.mu.Lock() - defer l.mu.Unlock() - if l.nc == nil { - l.nc = nc - } -} - -// zoneLock is a held per-zone lock. Callers must Release it, and should check -// Expired before a write that would land outside the lease. -type zoneLock struct { - locker *zoneLocker - kv jetstream.KeyValue - key string - zone string - acquired time.Time -} - -// lockZone acquires the cluster-wide write lock for one zone, blocking until it -// is free or the wait budget is spent. -func (l *zoneLocker) lockZone(ctx context.Context, zone string) (*zoneLock, error) { - if l == nil { - return nil, nil - } - l.mu.Lock() - nc := l.nc - l.mu.Unlock() - if nc == nil { - // No connection means no queue-group peer, so no cluster-wide race exists - // to guard against. Covers direct in-process ApplyBatch callers and tests. - return nil, nil - } - - kv, err := l.bucket(ctx) - if err != nil { - return nil, err - } - - key := zoneLockKey(zone) - deadline := time.Now().Add(zoneLockWaitFor) - step := zoneLockStep - for { - // Create maps a taken zone to ErrKeyExists on every replica count, - // which is the retryable case; anything else is a real failure. - if _, err := kv.Create(ctx, key, []byte(l.holder)); err == nil { - return &zoneLock{locker: l, kv: kv, key: key, zone: zone, acquired: time.Now()}, nil - } else if !errors.Is(err, jetstream.ErrKeyExists) { - return nil, fmt.Errorf("acquire dns zone lock %s: %w", zone, err) - } - if time.Now().After(deadline) { - return nil, fmt.Errorf("acquire dns zone lock %s: contended for %s", zone, zoneLockWaitFor) - } - // Jitter so several blocked writers do not retry in lockstep and starve - // each other on the same millisecond. - wait := step + rand.N(step) //nolint:gosec // jitter, not cryptographic - select { - case <-ctx.Done(): - return nil, fmt.Errorf("acquire dns zone lock %s: %w", zone, ctx.Err()) - case <-time.After(wait): - } - step = min(step*2, zoneLockStepMax) - } -} - -// Expired reports whether the lease may already have been reaped, meaning -// another writer could now hold the zone. A caller that has not yet written must -// abort rather than write unserialised. -func (z *zoneLock) Expired() bool { - if z == nil { - return false - } - return time.Since(z.acquired) >= zoneLockTTL -} - -// Release frees the zone. It runs on a detached context because the common -// caller is a deferred cleanup whose own context may already be done, and -// skipping the delete would park the zone for the full TTL. -func (z *zoneLock) Release(ctx context.Context) { - if z == nil { - return - } - releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), zoneLockBucketTimeout) - defer cancel() - if err := z.kv.Delete(releaseCtx, z.key); err != nil { - slog.Warn("dns writer: failed to release zone lock (TTL will reap)", - "zone", z.zone, "holder", z.locker.holder, "ttl_ms", otelsetup.Millis(zoneLockTTL), "error", err) - } -} - -// bucket attaches to the lock bucket, creating it when absent, and caches the -// handle. Lazy because the writer is constructed before JetStream is guaranteed -// ready. -func (l *zoneLocker) bucket(ctx context.Context) (jetstream.KeyValue, error) { - l.mu.Lock() - defer l.mu.Unlock() - if l.kv != nil { - return l.kv, nil - } - - js, err := jetstream.New(l.nc) - if err != nil { - return nil, fmt.Errorf("dns zone lock: JetStream unavailable: %w", err) - } - - attachCtx, cancel := context.WithTimeout(ctx, zoneLockBucketTimeout) - defer cancel() - - // Get-or-create: CreateKeyValue reports the stream name as in use when the - // bucket exists, so attach first and create only when genuinely absent. - kv, err := js.KeyValue(attachCtx, KVBucketDNSZoneLock) - if errors.Is(err, jetstream.ErrBucketNotFound) { - kv, err = js.CreateKeyValue(attachCtx, jetstream.KeyValueConfig{ - Bucket: KVBucketDNSZoneLock, - History: 1, - TTL: zoneLockTTL, - }) - // Another node created it in the gap; attach to theirs. - if err != nil { - kv, err = js.KeyValue(attachCtx, KVBucketDNSZoneLock) - } - } - if err != nil { - return nil, fmt.Errorf("dns zone lock: KV bucket %s unavailable: %w", KVBucketDNSZoneLock, err) - } - - l.kv = kv - return kv, nil -} - -// zoneLockKey maps a zone to a NATS KV key. DNS names are already within the -// legal key charset, so this only normalises case and guards odd input rather -// than encoding anything. -func zoneLockKey(zone string) string { - z := strings.ToLower(strings.Trim(strings.TrimSpace(zone), ".")) - if z == "" { - return "_apex" - } - return strings.Map(func(r rune) rune { - switch { - case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '-', r == '.', r == '_': - return r - default: - return '_' - } - }, z) -} diff --git a/spinifex/handlers/dns/zonelock_test.go b/spinifex/handlers/dns/zonelock_test.go deleted file mode 100644 index 10761bc37..000000000 --- a/spinifex/handlers/dns/zonelock_test.go +++ /dev/null @@ -1,159 +0,0 @@ -package dns - -import ( - "sync" - "testing" - "time" - - "github.com/mulgadc/spinifex/spinifex/testutil" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// shrinkZoneLockWaits keeps contention tests fast without changing the code -// path, restoring the production budgets afterwards. -func shrinkZoneLockWaits(t *testing.T, waitFor time.Duration) { - t.Helper() - origWait, origStep, origMax := zoneLockWaitFor, zoneLockStep, zoneLockStepMax - zoneLockWaitFor, zoneLockStep, zoneLockStepMax = waitFor, time.Millisecond, 5*time.Millisecond - t.Cleanup(func() { - zoneLockWaitFor, zoneLockStep, zoneLockStepMax = origWait, origStep, origMax - }) -} - -func TestZoneLockIsMutuallyExclusive(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, 50*time.Millisecond) - - a := newZoneLocker(nc, "node1") - b := newZoneLocker(nc, "node2") - - held, err := a.lockZone(t.Context(), "spx3.net") - require.NoError(t, err) - require.NotNil(t, held) - - // A second holder must not get in while the first has it. - _, err = b.lockZone(t.Context(), "spx3.net") - require.Error(t, err) - assert.ErrorContains(t, err, "contended") - - held.Release(t.Context()) - - // Released, so the peer now succeeds. - second, err := b.lockZone(t.Context(), "spx3.net") - require.NoError(t, err) - require.NotNil(t, second) - second.Release(t.Context()) -} - -func TestZoneLockDistinctZonesDoNotBlock(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, 50*time.Millisecond) - - a := newZoneLocker(nc, "node1") - b := newZoneLocker(nc, "node2") - - first, err := a.lockZone(t.Context(), "spx3.net") - require.NoError(t, err) - defer first.Release(t.Context()) - - // A held zone must never stall a write to an unrelated zone. - other, err := b.lockZone(t.Context(), "compute.internal") - require.NoError(t, err) - require.NotNil(t, other) - other.Release(t.Context()) -} - -func TestZoneLockSerialisesConcurrentHolders(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, 5*time.Second) - - const holders = 8 - var ( - mu sync.Mutex - inside int - maxSeen int - acquired int - ) - - var wg sync.WaitGroup - for i := range holders { - wg.Go(func() { - l := newZoneLocker(nc, "node"+string(rune('a'+i))) - lock, err := l.lockZone(t.Context(), "spx3.net") - if err != nil { - return - } - defer lock.Release(t.Context()) - - mu.Lock() - inside++ - acquired++ - maxSeen = max(maxSeen, inside) - mu.Unlock() - - // Hold long enough that an unserialised peer would overlap here. - time.Sleep(5 * time.Millisecond) - - mu.Lock() - inside-- - mu.Unlock() - }) - } - wg.Wait() - - assert.Equal(t, 1, maxSeen, "at most one holder may be inside the lock at a time") - assert.Equal(t, holders, acquired, "every holder should get the lock within the wait budget") -} - -func TestZoneLockWithoutNATSIsNoop(t *testing.T) { - // A writer with no connection has no queue-group peer, so there is nothing to - // serialise against and locking must not fail the write. - l := newZoneLocker(nil, "node1") - lock, err := l.lockZone(t.Context(), "spx3.net") - require.NoError(t, err) - assert.Nil(t, lock) - lock.Release(t.Context()) // nil receiver must be safe -} - -func TestZoneLockBindConnUpgradesLocker(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - shrinkZoneLockWaits(t, 50*time.Millisecond) - - // Built without a conn (so a no-op), then given one at Subscribe time. - l := newZoneLocker(nil, "node1") - l.bindConn(nc) - - lock, err := l.lockZone(t.Context(), "spx3.net") - require.NoError(t, err) - require.NotNil(t, lock, "a locker that adopted a conn must really lock") - defer lock.Release(t.Context()) - - _, err = newZoneLocker(nc, "node2").lockZone(t.Context(), "spx3.net") - assert.ErrorContains(t, err, "contended") -} - -func TestZoneLockLeaseExpiryIsDetectable(t *testing.T) { - held := &zoneLock{acquired: time.Now().Add(-2 * zoneLockTTL)} - assert.True(t, held.Expired(), "a write must abort rather than land outside its lease") - - fresh := &zoneLock{acquired: time.Now()} - assert.False(t, fresh.Expired()) - - var absent *zoneLock - assert.False(t, absent.Expired(), "a no-op lock never blocks a write") -} - -func TestZoneLockKeyNormalises(t *testing.T) { - // DNS names are already legal KV keys, so this only normalises case and - // guards odd input. - assert.Equal(t, "spx3.net", zoneLockKey("SPX3.Net.")) - assert.Equal(t, "compute.internal", zoneLockKey(" compute.internal ")) - assert.Equal(t, "_apex", zoneLockKey("")) - assert.Equal(t, "_apex", zoneLockKey(".")) - assert.Equal(t, "a_b.net", zoneLockKey("a b.net")) - assert.Equal(t, "zone-1.net", zoneLockKey("zone-1.net")) - - // Distinct zones must never collide onto one key. - assert.NotEqual(t, zoneLockKey("a.spx3.net"), zoneLockKey("b.spx3.net")) -} diff --git a/spinifex/handlers/ec2/instance/dns_hook_test.go b/spinifex/handlers/ec2/instance/dns_hook_test.go deleted file mode 100644 index eb30b9b59..000000000 --- a/spinifex/handlers/ec2/instance/dns_hook_test.go +++ /dev/null @@ -1,98 +0,0 @@ -package handlers_ec2_instance - -import ( - "context" - "testing" - "time" - - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/config" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" - "github.com/mulgadc/spinifex/spinifex/vm" - "github.com/nats-io/nats.go" - "github.com/stretchr/testify/require" -) - -// publishDNS must be a safe no-op when northstar is not configured (no base -// domain) and must tolerate a nil NATS connection while building changes. -func TestPublishDNS(t *testing.T) { - inst := &vm.VM{ - PublicIP: "1.2.3.4", - Instance: &ec2.Instance{PrivateIpAddress: aws.String("172.31.26.216")}, - } - - // Disabled: no base domain → no panic, no publish attempt. - disabled := &InstanceServiceImpl{config: &config.Config{Region: "ap-southeast-2"}} - disabled.publishDNS("123456789012", handlers_dns.ActionUpsert, []*vm.VM{inst}) - - // Enabled base domain but nil NATS conn: builds changes, best-effort publish - // is a no-op for a nil connection (must not panic). - enabled := &InstanceServiceImpl{ - config: &config.Config{Region: "ap-southeast-2"}, - dnsBaseDomain: "spx3.net", - } - enabled.publishDNS("123456789012", handlers_dns.ActionUpsert, []*vm.VM{inst}) - enabled.publishDNS("123456789012", handlers_dns.ActionDelete, []*vm.VM{inst}) - enabled.publishDNS("123456789012", handlers_dns.ActionUpsert, nil) -} - -// The race the check exists for: terminate lands while the deferred launch is -// still publishing the UPSERT, so Status has not caught up and only the -// deletion mark says the record is going away. -func TestTerminateRacedLaunch(t *testing.T) { - now := time.Now().UTC() - tests := []struct { - name string - v *vm.VM - want bool - }{ - {name: "running, not marked", v: &vm.VM{ID: "i-1", Status: vm.StateRunning}, want: false}, - {name: "running, marked for deletion", v: &vm.VM{ID: "i-1", Status: vm.StateRunning, DeletionTimestamp: &now}, want: true}, - {name: "settled terminated", v: &vm.VM{ID: "i-1", Status: vm.StateTerminated}, want: true}, - {name: "operator stopped", v: &vm.VM{ID: "i-1", Status: vm.StateStopped, DesiredState: vm.DesiredStopped}, want: false}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{tt.v.ID: tt.v})} - require.Equal(t, tt.want, svc.terminateRacedLaunch(tt.v)) - }) - } -} - -func TestPrepareRunInstancesDoesNotPublishDNS(t *testing.T) { - eni := &fakeENICreator{ - defaultSubnet: &SubnetInfo{SubnetID: "subnet-default", VpcID: "vpc-1"}, - createOut: &ec2.CreateNetworkInterfaceOutput{NetworkInterface: &ec2.NetworkInterface{ - NetworkInterfaceId: aws.String("eni-1"), - MacAddress: aws.String("aa:bb:cc:dd:ee:ff"), - PrivateIpAddress: aws.String("10.0.0.10"), - VpcId: aws.String("vpc-1"), - }}, - } - svc, _ := prepareSvcWithENI(t, eni, nil) - svc.dnsBaseDomain = "spx3.net" - - published := make(chan struct{}, 1) - sub, err := svc.natsConn.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) { - published <- struct{}{} - _ = msg.Respond([]byte(`{"applied":1}`)) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - require.NoError(t, svc.natsConn.Flush()) - - _, _, _, err = svc.PrepareRunInstances(context.Background(), &ec2.RunInstancesInput{ - InstanceType: aws.String("t3.micro"), - ImageId: aws.String("ami-1"), - MinCount: aws.Int64(1), - MaxCount: aws.Int64(1), - }, "acc", "") - require.NoError(t, err) - - select { - case <-published: - t.Fatal("PrepareRunInstances must not publish DNS before VM launch succeeds") - case <-time.After(100 * time.Millisecond): - } -} diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index ee8e86d91..e7286cb09 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -863,25 +863,6 @@ func (s *InstanceServiceImpl) PrepareRunInstances(ctx context.Context, input *ec return reservation, instances, instanceType, nil } -// publishDNS registers (or withdraws) the public + private A records for a batch -// of instances with the control-plane DNS writer. Best-effort: a failure is -// logged by the writer helper and never blocks the lifecycle operation; the -// reconcile loop repairs any miss. No-op when northstar is not configured. -func (s *InstanceServiceImpl) publishDNS(accountID string, action handlers_dns.Action, instances []*vm.VM) { - if s.dnsBaseDomain == "" || len(instances) == 0 { - return - } - var changes []handlers_dns.Change - for _, instance := range instances { - privateIP := "" - if instance.Instance != nil && instance.Instance.PrivateIpAddress != nil { - privateIP = *instance.Instance.PrivateIpAddress - } - changes = append(changes, handlers_dns.EC2Changes(action, s.config.Region, s.dnsBaseDomain, s.dnsInternalDomain, instance.PublicIP, privateIP)...) - } - handlers_dns.PublishChangesBestEffort(s.natsConn, accountID, changes) -} - // attachPreCreatedENI verifies the ENI is available, attaches it as device-0, // and populates the VM + ec2.Instance. Public-IP auto-assignment is skipped; // the caller manages that out-of-band. @@ -939,7 +920,6 @@ func (s *InstanceServiceImpl) attachPreCreatedENI(ctx context.Context, accountID // into vmMgr: volume prep, GPU claim, vmMgr.Run. Partial failures are tolerated. func (s *InstanceServiceImpl) LaunchRunInstances(ctx context.Context, instances []*vm.VM, input *ec2.RunInstancesInput, instanceType *ec2.InstanceTypeInfo) { var successCount int - launchedByAccount := make(map[string][]*vm.VM) for _, instance := range instances { // Skip if a concurrent terminate raced with prepare. status := s.vmMgr.Status(instance) @@ -1005,36 +985,12 @@ func (s *InstanceServiceImpl) LaunchRunInstances(ctx context.Context, instances s.vmMgr.UpdateGuestDeviceNames(instance) successCount++ - launchedByAccount[instance.AccountID] = append(launchedByAccount[instance.AccountID], instance) slog.InfoContext(ctx, "LaunchRunInstances: launched instance", "instanceId", instance.ID) } - withdrawByAccount := make(map[string][]*vm.VM) - for accountID, launched := range launchedByAccount { - s.publishDNS(accountID, handlers_dns.ActionUpsert, launched) - for _, instance := range launched { - if s.terminateRacedLaunch(instance) { - withdrawByAccount[accountID] = append(withdrawByAccount[accountID], instance) - } - } - } - for accountID, instances := range withdrawByAccount { - s.publishDNS(accountID, handlers_dns.ActionDelete, instances) - } slog.InfoContext(ctx, "LaunchRunInstances: completed", "requested", len(instances), "launched", successCount) } -// terminateRacedLaunch reports whether a terminate raced the deferred launch -// UPSERT: it reads the terminate flag stamped under the lock (plus status) since -// the async state transition may not have landed, which a status-only check misses. -func (s *InstanceServiceImpl) terminateRacedLaunch(instance *vm.VM) bool { - withdraw := false - s.vmMgr.Inspect(instance, func(v *vm.VM) { - withdraw = v.DeletionTimestamp != nil || !handlers_dns.InstanceRetainsRecords(v.Status) - }) - return withdraw -} - // RunInstances is for non-daemon callers (tests). The daemon calls // PrepareRunInstances + LaunchRunInstances directly to respond before launching. func (s *InstanceServiceImpl) RunInstances(ctx context.Context, input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) { @@ -1196,12 +1152,6 @@ func (s *InstanceServiceImpl) StopOrTerminateInstance(ctx context.Context, insta return errors.New(awserrors.ErrorIncorrectInstanceState) } - // Withdraw the instance's DNS records on terminate; stop/start retains - // the IP and the name, so they are a no-op here. - if isTerminate { - s.publishDNS(instance.AccountID, handlers_dns.ActionDelete, []*vm.VM{instance}) - } - go func(id, ownerAccountID string) { //nolint:gosec // detached lifecycle op: stop/terminate continues after the API ack; request ctx would cancel it var err error if isTerminate { diff --git a/spinifex/handlers/eks/dns_reconcile_test.go b/spinifex/handlers/eks/dns_reconcile_test.go index 621b8de8c..c6a631e23 100644 --- a/spinifex/handlers/eks/dns_reconcile_test.go +++ b/spinifex/handlers/eks/dns_reconcile_test.go @@ -4,9 +4,7 @@ import ( "context" "testing" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" "github.com/mulgadc/spinifex/spinifex/testutil" - "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -15,14 +13,7 @@ func TestCreateCluster_EndpointDNSNameUsesOwnerAccount(t *testing.T) { fixture := newEKSServiceFixture(t) fixture.svc.baseDomain = "spx3.net" - sub, err := fixture.svc.deps.NATSConn.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) { - _ = msg.Respond([]byte(`{"applied":1}`)) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - require.NoError(t, fixture.svc.deps.NATSConn.Flush()) - - _, err = fixture.svc.CreateCluster(context.Background(), createInput("alpha"), testAccountID, "") + _, err := fixture.svc.CreateCluster(context.Background(), createInput("alpha"), testAccountID, "") require.NoError(t, err) fixture.svc.WaitLaunches() diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index 7ffa52675..d1796a1c6 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -831,8 +831,6 @@ func (s *EKSServiceImpl) launchClusterInfra(ctx context.Context, lc clusterLaunc s.failClusterLaunch(ctx, acctKV, name, accountID, meta, "persist endpoint metadata", err) return } - // Register the endpoint A record (best-effort; reconcile repairs a miss). - s.publishEKSDNS(accountID, meta, handlers_dns.ActionUpsert) oidcIssuer, err := ClusterOIDCIssuer(s.deps.GatewayBaseURL, region, accountID, name) if err != nil { @@ -1248,9 +1246,6 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name teardownErrs = append(teardownErrs, fmt.Errorf("zeroize OIDC key: %w", err)) } - // Withdraw the endpoint A record alongside the NLB teardown (best-effort). - s.publishEKSDNS(accountID, meta, handlers_dns.ActionDelete) - if meta.NLBArn != "" { // Deregister is best-effort: DeleteClusterNLB tears down the whole NLB // + target group, so a stale target registration cannot leak past it. @@ -2103,18 +2098,6 @@ func clusterJoinEndpoint(meta *ClusterMeta) string { return meta.Endpoint } -// publishEKSDNS registers or withdraws the cluster's account-qualified apiserver -// endpoint A record ({cluster}.{accountID}.{region}.eks.{baseDomain} → EndpointIP) -// with the control-plane DNS writer. Best-effort and a no-op when northstar is not configured; the reconcile -// loop repairs any miss and it never blocks the cluster operation. -func (s *EKSServiceImpl) publishEKSDNS(accountID string, meta *ClusterMeta, action handlers_dns.Action) { - if s.baseDomain == "" || meta == nil || meta.EndpointDNSName == "" { - return - } - changes := handlers_dns.EKSChanges(action, meta.EndpointDNSName, s.baseDomain, meta.EndpointIP) - handlers_dns.PublishChangesBestEffort(s.deps.NATSConn, accountID, changes) -} - // DesiredDNSChanges returns the UPSERT records for every endpoint-ready cluster // across all account buckets, plus whether the enumeration was authoritative. Clusters // live in per-account KV buckets, so a complete cross-tenant view requires diff --git a/spinifex/handlers/elbv2/dns_register_test.go b/spinifex/handlers/elbv2/dns_register_test.go index fb006f91b..1b96ef54a 100644 --- a/spinifex/handlers/elbv2/dns_register_test.go +++ b/spinifex/handlers/elbv2/dns_register_test.go @@ -3,7 +3,6 @@ package handlers_elbv2 import ( "testing" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -35,20 +34,6 @@ func TestLBFrontendIP(t *testing.T) { assert.Equal(t, "10.0.0.9", lbFrontendIP(internal)) } -func TestPublishLBDNS_NoopWhenDisabled(t *testing.T) { - t.Parallel() - rec := &LoadBalancerRecord{ - Scheme: SchemeInternal, - DNSName: "internal-web-abc.ap-southeast-2.elb.spx3.net", - VPCIP: "10.0.0.1", - AccountID: "000000000000", - } - // No base domain → no-op; must not panic on the nil NATS connection. - (&ELBv2ServiceImpl{}).publishLBDNS(rec, handlers_dns.ActionUpsert) - // Base domain set but nil NATS conn → PublishChangesBestEffort tolerates it. - (&ELBv2ServiceImpl{dnsBaseDomain: "spx3.net"}).publishLBDNS(rec, handlers_dns.ActionDelete) -} - func TestDesiredDNSChanges_IncludesEndpointReadyProvisioningLoadBalancers(t *testing.T) { t.Parallel() svc := setupTestService(t) diff --git a/spinifex/handlers/elbv2/service_impl.go b/spinifex/handlers/elbv2/service_impl.go index 06808290e..d4d5b5966 100644 --- a/spinifex/handlers/elbv2/service_impl.go +++ b/spinifex/handlers/elbv2/service_impl.go @@ -1547,8 +1547,6 @@ func (s *ELBv2ServiceImpl) provisionLBDataPlane(ctx context.Context, lc lbLaunch if putErr := s.store.PutLoadBalancer(ctx, record); putErr != nil { return fmt.Errorf("persist launch result for %s: %w", lc.lbArn, putErr) } - // Register the frontend A record now that the serving IP is allocated. - s.publishLBDNS(record, handlers_dns.ActionUpsert) return nil } @@ -1561,18 +1559,6 @@ func lbFrontendIP(r *LoadBalancerRecord) string { return r.VPCIP } -// publishLBDNS registers or withdraws the load balancer's frontend A record with -// the control-plane DNS writer. Best-effort and a no-op when northstar is not -// configured or no frontend IP has been allocated; the reconcile loop repairs -// any miss and never blocks the LB operation. -func (s *ELBv2ServiceImpl) publishLBDNS(record *LoadBalancerRecord, action handlers_dns.Action) { - if s.dnsBaseDomain == "" || record == nil { - return - } - changes := handlers_dns.ELBChanges(action, record.DNSName, s.dnsBaseDomain, lbFrontendIP(record)) - handlers_dns.PublishChangesBestEffort(s.nc, record.AccountID, changes) -} - // DesiredDNSChanges returns the UPSERT records for every endpoint-ready load // balancer across all accounts, plus whether the enumeration was authoritative. The KV // store spans every tenant, so a successful list is a complete cross-account @@ -1803,9 +1789,6 @@ func (s *ELBv2ServiceImpl) DeleteLoadBalancer(ctx context.Context, input *elbv2. // key, so a delete that races the record removal still converges. s.releaseLBNameClaim(ctx, lb.Name, accountID) - // Withdraw the frontend A record (best-effort; reconcile repairs a miss). - s.publishLBDNS(lb, handlers_dns.ActionDelete) - slog.InfoContext(ctx, "DeleteLoadBalancer completed", "lbArn", *input.LoadBalancerArn, "enis", len(lb.ENIs), "accountID", accountID) return &elbv2.DeleteLoadBalancerOutput{}, nil diff --git a/spinifex/handlers/rds/create.go b/spinifex/handlers/rds/create.go index fce939a3a..ee7f0925c 100644 --- a/spinifex/handlers/rds/create.go +++ b/spinifex/handlers/rds/create.go @@ -9,7 +9,6 @@ import ( "github.com/aws/aws-sdk-go/service/rds" "github.com/mulgadc/spinifex/spinifex/awserrors" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go/jetstream" @@ -134,7 +133,6 @@ func (s *Service) CreateDBInstance(ctx context.Context, input *rds.CreateDBInsta // Published as soon as the ENI IP is known rather than on available, so the // name resolves the moment the engine comes up; the security group is what // gates connectivity until then. - s.publishDNS(ctx, accountID, stored, handlers_dns.ActionUpsert) s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, req.Identifier, "DB instance created.", EventCategoryCreation) diff --git a/spinifex/handlers/rds/create_test.go b/spinifex/handlers/rds/create_test.go index a0018a63e..11ba8a1f3 100644 --- a/spinifex/handlers/rds/create_test.go +++ b/spinifex/handlers/rds/create_test.go @@ -2,7 +2,6 @@ package handlers_rds import ( "context" - "encoding/json" "errors" "slices" "strings" @@ -140,9 +139,6 @@ type createHarness struct { network *fakeNetwork iam *iammock.SystemInstanceRoleEnsurer nc *nats.Conn - - // dnsChanges collects what the endpoint publish put on the bus. - dnsChanges chan handlers_dns.ChangeBatch } func newCreateHarness(t *testing.T, baseDomain string) *createHarness { @@ -150,11 +146,10 @@ func newCreateHarness(t *testing.T, baseDomain string) *createHarness { _, nc, _ := testutil.StartTestJetStream(t) h := &createHarness{ - launch: newLaunchHarness(), - network: newFakeNetwork(), - iam: iammock.New(), - nc: nc, - dnsChanges: make(chan handlers_dns.ChangeBatch, 4), + launch: newLaunchHarness(), + network: newFakeNetwork(), + iam: iammock.New(), + nc: nc, } // The storage key is configured on a real cluster, so the volume comes back // encrypted; the unencrypted case is its own test. @@ -162,7 +157,6 @@ func newCreateHarness(t *testing.T, baseDomain string) *createHarness { // The DNS writer is a request-reply consumer, so without a responder the // best-effort publish would sit out its own timeout on every create. - h.stubDNSWriter(t) h.svc = NewService(nc, testRegion).WithDeps(Deps{ LoadCA: newTestCA(t), @@ -176,21 +170,6 @@ func newCreateHarness(t *testing.T, baseDomain string) *createHarness { return h } -func (h *createHarness) stubDNSWriter(t *testing.T) { - t.Helper() - sub, err := h.nc.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) { - var batch handlers_dns.ChangeBatch - if err := json.Unmarshal(msg.Data, &batch); err == nil { - h.dnsChanges <- batch - } - if err := msg.Respond([]byte(`{}`)); err != nil { - t.Logf("respond on %s: %v", handlers_dns.SubjectRecordsetChange, err) - } - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) -} - func (h *createHarness) record(t *testing.T, id string) DBInstanceRecord { t.Helper() kv, err := h.svc.bucket(t.Context(), testAccountID) @@ -322,16 +301,19 @@ func TestCreateDBInstance_IAMFailurePrecedesReservationAndLaunch(t *testing.T) { assert.Empty(t, h.launch.volumes.created) } -func TestCreateDBInstance_PublishesEndpointRecord(t *testing.T) { +// A created instance has to be in the desired set immediately, because the +// desired set is the only thing that puts its endpoint into the zone. +func TestCreateDBInstance_ContributesEndpointRecordToDesiredSet(t *testing.T) { t.Parallel() h := newCreateHarness(t, testBaseDomain) _, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID) require.NoError(t, err) - batch := <-h.dnsChanges - require.Len(t, batch.Changes, 1) - change := batch.Changes[0] + changes, authoritative := h.svc.DesiredDNSChanges() + require.True(t, authoritative) + require.Len(t, changes, 1) + change := changes[0] assert.Equal(t, handlers_dns.ActionUpsert, change.Action) assert.Equal(t, testBaseDomain, change.Zone) assert.Equal(t, "A", change.Type) @@ -358,11 +340,8 @@ func TestCreateDBInstance_EndpointFallsBackToENIIPWithoutBaseDomain(t *testing.T require.NotNil(t, out.DBInstance.Endpoint) assert.Equal(t, rec.ENIPrivateIP, aws.StringValue(out.DBInstance.Endpoint.Address)) - select { - case batch := <-h.dnsChanges: - t.Fatalf("no base domain means no record to publish, got %+v", batch) - default: - } + changes, _ := h.svc.DesiredDNSChanges() + assert.Empty(t, changes, "no base domain means there is no name to put in a zone") } func TestCreateDBInstance_DuplicateIdentifierIsRejected(t *testing.T) { diff --git a/spinifex/handlers/rds/delete.go b/spinifex/handlers/rds/delete.go index 596178069..829495c84 100644 --- a/spinifex/handlers/rds/delete.go +++ b/spinifex/handlers/rds/delete.go @@ -11,7 +11,6 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/rds" "github.com/mulgadc/spinifex/spinifex/awserrors" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" "github.com/mulgadc/spinifex/spinifex/handlers/sysinstance" "github.com/mulgadc/spinifex/spinifex/tags" "github.com/mulgadc/spinifex/spinifex/utils" @@ -202,7 +201,6 @@ func (s *Service) teardownDBInstance(ctx context.Context, kv jetstream.KeyValue, } s.deleteInstanceENIs(ctx, accountID, rec) - s.publishDNS(ctx, accountID, rec, handlers_dns.ActionDelete) if rec.InstanceID != "" { if err := s.DeleteInstanceIndex(ctx, rec.InstanceID); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { diff --git a/spinifex/handlers/rds/dns.go b/spinifex/handlers/rds/dns.go index de5831d8a..e6ddac796 100644 --- a/spinifex/handlers/rds/dns.go +++ b/spinifex/handlers/rds/dns.go @@ -17,18 +17,6 @@ func (s *Service) dnsName(accountID, dbInstanceIdentifier string) string { return handlers_dns.RDSName(dbInstanceIdentifier, accountID, s.region, s.deps.BaseDomain) } -// Best-effort: a miss is repaired by the reconcile backstop within a cycle, and -// a create must not fail because northstar was briefly unreachable. -func (s *Service) publishDNS(ctx context.Context, accountID string, rec *DBInstanceRecord, action handlers_dns.Action) { - if s.deps.BaseDomain == "" || rec == nil || rec.DNSName == "" || rec.ENIPrivateIP == "" { - return - } - changes := handlers_dns.RDSChanges(action, rec.DNSName, s.deps.BaseDomain, rec.ENIPrivateIP) - slog.DebugContext(ctx, "rds: publishing endpoint record", - "dbInstance", rec.DBInstanceIdentifier, "name", rec.DNSName, "action", action) - handlers_dns.PublishChangesBestEffort(s.nc, accountID, changes) -} - // The UPSERTs for every endpoint-ready DB instance across all account buckets, // plus whether the enumeration was authoritative. Instances live in per-account // buckets, so a complete cross-tenant view requires reading every one: any diff --git a/spinifex/handlers/rds/restore.go b/spinifex/handlers/rds/restore.go index f31db5d48..71fbe2853 100644 --- a/spinifex/handlers/rds/restore.go +++ b/spinifex/handlers/rds/restore.go @@ -11,7 +11,6 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/rds" "github.com/mulgadc/spinifex/spinifex/awserrors" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" "github.com/mulgadc/spinifex/spinifex/tags" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go/jetstream" @@ -148,7 +147,6 @@ func (s *Service) RestoreDBInstanceFromDBSnapshot(ctx context.Context, input *rd s.unwindLaunched(ctx, launched) return nil, fmt.Errorf("rds: write instance index for %s: %w", req.Identifier, err) } - s.publishDNS(ctx, accountID, stored, handlers_dns.ActionUpsert) s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, req.Identifier, fmt.Sprintf("Restored from DB snapshot %s.", snapshot.DBSnapshotIdentifier), diff --git a/spinifex/handlers/rds/snapshot_test.go b/spinifex/handlers/rds/snapshot_test.go index 893d4713a..209c59117 100644 --- a/spinifex/handlers/rds/snapshot_test.go +++ b/spinifex/handlers/rds/snapshot_test.go @@ -9,7 +9,6 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/rds" "github.com/mulgadc/spinifex/spinifex/awserrors" - handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" iammock "github.com/mulgadc/spinifex/spinifex/handlers/iam/mock" "github.com/mulgadc/spinifex/spinifex/tags" "github.com/mulgadc/spinifex/spinifex/testutil" @@ -46,7 +45,6 @@ func newSnapshotHarness(t *testing.T, agentFails bool) *snapshotHarness { h.agent = newStubAgent(t, nc, testAccountID, testDBID, agentFails) // Without a responder the best-effort endpoint publish would sit out its own // timeout on every restore. - stubDNSWriter(t, nc) h.svc = NewService(nc, testRegion).WithDeps(Deps{ LoadCA: newTestCA(t), @@ -60,17 +58,6 @@ func newSnapshotHarness(t *testing.T, agentFails bool) *snapshotHarness { return h } -func stubDNSWriter(t *testing.T, nc *nats.Conn) { - t.Helper() - sub, err := nc.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) { - if err := msg.Respond([]byte(`{}`)); err != nil { - t.Logf("respond on %s: %v", handlers_dns.SubjectRecordsetChange, err) - } - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) -} - func (h *snapshotHarness) instance(t *testing.T, id string) DBInstanceRecord { t.Helper() kv, err := h.svc.bucket(t.Context(), testAccountID)