diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 6af6c05cf..ccc47b264 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1559,7 +1559,7 @@ func (d *Daemon) startCluster() error { store := objectstore.NewS3ObjectStoreFromConfig(admin.DialTarget(d.config.Predastore.Host), d.config.Predastore.Region, d.config.Predastore.AccessKey, d.config.Predastore.SecretKey) d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, store, d.vmMgr, d.resourceMgr, d.jsManager) d.dnsWriter = handlers_dns.NewWriter(d.config, d.clusterConfig, d.natsConn) - d.dnsReconciler = handlers_dns.NewReconciler(d.config, d.natsConn, d.dnsDesiredSet, d.dnsWatchSources()...) + d.dnsReconciler = handlers_dns.NewReconciler(d.config, d.natsConn, d.dnsWriter, d.dnsDesiredSet, d.dnsWatchSources()...) d.dnsBaseDomain = handlers_dns.ResolveBaseDomain(d.config) d.dnsInternalDomain = handlers_dns.ResolveInternalDomain(d.config) d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) diff --git a/spinifex/handlers/dns/reconcile.go b/spinifex/handlers/dns/reconcile.go index ec320f766..38acc4a9c 100644 --- a/spinifex/handlers/dns/reconcile.go +++ b/spinifex/handlers/dns/reconcile.go @@ -2,18 +2,15 @@ package dns import ( "context" - "encoding/json" "fmt" "log/slog" "strings" "time" - "unicode/utf8" nsconfig "github.com/mulgadc/northstar/pkg/config" "github.com/mulgadc/spinifex/spinifex/config" reconcilelock "github.com/mulgadc/spinifex/spinifex/network/reconcile" "github.com/mulgadc/spinifex/spinifex/reconciler" - "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -24,15 +21,6 @@ import ( // path by which a normal lifecycle event reaches DNS. const DefaultReconcileInterval = reconciler.DefaultResync -const ( - // maxReconcileNATSPayloadBytes stays below nats.conf's 1 MB max_payload. - maxReconcileNATSPayloadBytes = 900 * 1024 - // reconcileNATSHeaderHeadroom reserves space for account and trace headers - // when a deployment negotiates a lower server payload limit. - reconcileNATSHeaderHeadroom = 4 * 1024 - changeBatchJSONOverhead = len(`{"changes":[]}`) -) - // DesiredFunc returns the full desired managed record set built from the live // resource inventory across all tenants. The daemon supplies it by enumerating // instances, load balancers, and EKS clusters. @@ -70,10 +58,10 @@ type PruneScope struct { // are DELETEd. A pass runs on a change to any watched resource store, and on // the interval as the backstop for drift those stores cannot report. // -// Each pass is gated on a CAS-elected leader so one node publishes the -// converging batch. Without it every node published its own batch, and since -// the queue group load-balances rather than serialises, an idle cluster raced -// its own zone object once per cycle. +// Each pass is gated on a CAS-elected leader, which then writes the zone itself +// rather than publishing the batch to the queue group. Handing its own work to +// a peer put a second writer on the object and left the read that computed the +// deletes in a different process from the write that applied them. // // Only cluster-wide-enumerable records are pruned, and only for the classes a // cycle enumerated in full. Every class the desired set carries is readable @@ -88,31 +76,34 @@ type Reconciler struct { // leaves every stale private record behind. internalDomain string nc *nats.Conn - desired DesiredFunc - interval time.Duration - accountID string - holder string - sources []reconciler.Source + // writer applies the converged batch. The reconcile owns the zone object on + // the node that won the election, so the batch never leaves this process. + writer *Writer + desired DesiredFunc + interval time.Duration + holder string + sources []reconciler.Source } // NewReconciler builds the drift backstop. It is disabled (a no-op) when -// northstar S3 is not configured or no desired-set provider is supplied. -// sources name the buckets whose changes should wake the loop. They are -// optional: with none supplied the loop falls back to the interval alone, -// which is the behaviour that predates the watch. -func NewReconciler(cfg *config.Config, nc *nats.Conn, desired DesiredFunc, sources ...reconciler.Source) *Reconciler { +// northstar S3 is not configured, no desired-set provider is supplied, or the +// writer that applies its batches is unavailable. sources name the buckets whose +// changes should wake the loop. They are optional: with none supplied the loop +// falls back to the interval alone, which is the behaviour that predates the +// watch. +func NewReconciler(cfg *config.Config, nc *nats.Conn, writer *Writer, desired DesiredFunc, sources ...reconciler.Source) *Reconciler { r := &Reconciler{ - nc: nc, - desired: desired, - interval: DefaultReconcileInterval, - accountID: utils.GlobalAccountID, - sources: sources, + nc: nc, + writer: writer, + desired: desired, + interval: DefaultReconcileInterval, + sources: sources, } if cfg != nil { r.holder = cfg.Node } zoneCfg, ok := zoneS3Config(cfg) - if !ok || desired == nil { + if !ok || desired == nil || !writer.Enabled() { return r } r.enabled = true @@ -148,8 +139,10 @@ func (r *Reconciler) Run(ctx context.Context) { }) } -// reconcileOnce computes the converging batch and publishes it best-effort, on -// whichever node wins this cycle's election. +// reconcileOnce computes the converging batch and applies it, on whichever node +// wins this cycle's election. The read that finds the stale records and the +// write that removes them now happen in one process, so a record re-added +// between them is no longer deleted by a batch that predates it. func (r *Reconciler) reconcileOnce(ctx context.Context) { if !r.enabled { return @@ -172,121 +165,12 @@ func (r *Reconciler) reconcileOnce(ctx context.Context) { return } slog.Debug("dns reconcile: converging", "changes", len(batch)) - payloadLimit := reconcilePayloadLimit(r.nc) - if err := publishReconcileBatches(batch, payloadLimit, func(changes []Change) error { - return publishReconcileBatch(r.nc, r.accountID, changes) - }); err != nil { - slog.Warn("dns reconcile: batching or publish failed, retrying next cycle", "changes", len(batch), "error", err) + res, err := r.writer.ApplyBatch(&ChangeBatch{Changes: batch}) + if err != nil { + slog.Warn("dns reconcile: apply failed, retrying next cycle", "changes", len(batch), "error", err) return } - slog.Debug("dns reconcile: converged", "changes", len(batch)) -} - -// publishReconcileBatch requires the writer to acknowledge every submitted -// change, so a missing transport or partial success cannot report convergence. -func publishReconcileBatch(nc *nats.Conn, accountID string, changes []Change) error { - res, err := PublishChanges(nc, accountID, changes) - if err != nil { - return err - } - applied := 0 - if res != nil { - applied = res.Applied - } - if applied != len(changes) { - return fmt.Errorf("writer acknowledged %d of %d changes", applied, len(changes)) - } - return nil -} - -// publishReconcileBatches sends bounded batches sequentially so each is -// acknowledged before the next begins. On failure it stops: the next cycle -// safely rebuilds and retries the complete idempotent desired state. -func publishReconcileBatches(changes []Change, payloadLimit int, publish func([]Change) error) error { - batches, err := splitReconcileChanges(changes, payloadLimit) - if err != nil { - return err - } - for i, batch := range batches { - if err := publish(batch); err != nil { - return fmt.Errorf("publish batch %d of %d: %w", i+1, len(batches), err) - } - } - return nil -} - -// reconcilePayloadLimit respects a lower limit advertised by the connected -// server while retaining the repository policy ceiling. -func reconcilePayloadLimit(nc *nats.Conn) int { - limit := maxReconcileNATSPayloadBytes - if nc == nil || nc.MaxPayload() <= 0 { - return limit - } - serverLimit := max(0, int(nc.MaxPayload())-reconcileNATSHeaderHeadroom) - return min(limit, serverLimit) -} - -// splitReconcileChanges preserves change order while enforcing the Route 53 -// per-zone record/value request limits and the effective NATS payload ceiling. -// UPSERT costs count twice, matching Route 53 semantics. -func splitReconcileChanges(changes []Change, payloadLimit int) ([][]Change, error) { - var batches [][]Change - start := 0 - zoneRecords := map[string]int{} - zoneValueChars := map[string]int{} - currentPayloadBytes := changeBatchJSONOverhead - - flush := func(end int) { - if start == end { - return - } - batches = append(batches, changes[start:end]) - start = end - clear(zoneRecords) - clear(zoneValueChars) - currentPayloadBytes = changeBatchJSONOverhead - } - - for i, change := range changes { - encoded, err := json.Marshal(change) - if err != nil { - return nil, fmt.Errorf("marshal change %d for %s: %w", i+1, change.Name, err) - } - multiplier := 1 - if change.Action == ActionUpsert { - multiplier = 2 - } - records := multiplier - valueChars := multiplier * utf8.RuneCountInString(change.Value) - singlePayloadBytes := changeBatchJSONOverhead + len(encoded) - - if records > MaxRecordsPerChangeRequest { - return nil, fmt.Errorf("change %d for %s has %d record elements; maximum is %d", i+1, change.Name, records, MaxRecordsPerChangeRequest) - } - if valueChars > MaxValueCharsPerChangeRequest { - return nil, fmt.Errorf("change %d for %s has %d value characters; maximum is %d", i+1, change.Name, valueChars, MaxValueCharsPerChangeRequest) - } - if singlePayloadBytes > payloadLimit { - return nil, fmt.Errorf("change %d for %s serializes to %d bytes; payload maximum is %d", i+1, change.Name, singlePayloadBytes, payloadLimit) - } - - payloadBytes := len(encoded) - if i > start { - payloadBytes++ // JSON comma between adjacent changes. - } - if i > start && (zoneRecords[change.Zone]+records > MaxRecordsPerChangeRequest || - zoneValueChars[change.Zone]+valueChars > MaxValueCharsPerChangeRequest || - currentPayloadBytes+payloadBytes > payloadLimit) { - flush(i) - payloadBytes = len(encoded) - } - - zoneRecords[change.Zone] += records - zoneValueChars[change.Zone] += valueChars - currentPayloadBytes += payloadBytes - } - flush(len(changes)) - return batches, nil + slog.Debug("dns reconcile: converged", "changes", res.Applied, "zones", res.Zones) } // computeBatch reads every zone holding prunable records and converges the diff --git a/spinifex/handlers/dns/reconcile_apply_test.go b/spinifex/handlers/dns/reconcile_apply_test.go new file mode 100644 index 000000000..238b05f48 --- /dev/null +++ b/spinifex/handlers/dns/reconcile_apply_test.go @@ -0,0 +1,142 @@ +//test:in-package — these drive reconcileOnce against a real Writer over the +//fake S3, which needs the unexported Reconciler fields and the writer's own +//resolved S3 config. + +package dns + +import ( + "testing" + + nsconfig "github.com/mulgadc/northstar/pkg/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// reconcilerFor pairs a Reconciler with the writer that applies its batches, +// sharing one bucket. nc stays nil so a pass skips the election and runs. +func reconcilerFor(w *Writer, desired func() DesiredSet) *Reconciler { + return &Reconciler{ + enabled: true, + baseDomain: w.baseDomain, + internalDomain: "compute.internal", + s3cfg: w.s3cfg, + writer: w, + desired: desired, + } +} + +// labelsOfType lists the zone-relative labels carried by records of one type, +// read back through the same parser northstar uses. +func labelsOfType(t *testing.T, s3cfg *nsconfig.S3Config, zone string, rtype uint16) []string { + t.Helper() + cfg, exists, err := nsconfig.ReadZoneRaw(s3cfg, zone) + require.NoError(t, err) + require.True(t, exists, "zone %s should exist", zone) + var out []string + for _, rec := range cfg.Records { + if rec.Type == rtype { + out = append(out, rec.Domain) + } + } + return out +} + +// The pass applies its own batch. There is no NATS connection here, so nothing +// could carry a ChangeBatch to a peer: if the record reaches the zone object, +// the leader wrote it. +func TestReconcileOnce_WritesTheZoneWithoutPublishing(t *testing.T) { + w, objects := newTestWriter(t) + r := reconcilerFor(w, 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"`, + "a pass with no transport still has to reach the zone object, or it published instead of writing") +} + +// The desired set holds no apex, SOA, NS or glue record — the writer +// materialises those from cluster config. A pass that pruned everything absent +// from the desired set would take the NS set with it, which is an outage for the +// zone rather than a missing address. +func TestReconcileOnce_StructuralRecordsSurviveAPrune(t *testing.T) { + w, objects := newTestWriter(t) + objects[testBase+".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." +[[records]] +domain = "ns1." +type = 1 +address = "10.0.0.1" +[[records]] +domain = "live.elb." +type = 1 +address = "10.200.1.9" +[[records]] +domain = "stale.elb." +type = 1 +address = "10.200.1.99" +` + + r := reconcilerFor(w, func() DesiredSet { + return DesiredSet{ + Changes: []Change{upsert("live.elb."+testBase, "10.200.1.9")}, + Prunable: PruneScope{ELB: true, EKS: true, RDS: true, EC2: true}, + } + }) + + r.reconcileOnce(t.Context()) + + body := objects[testBase+".toml"] + assert.NotContains(t, body, `address = "10.200.1.99"`, "the stale ELB record is the one thing a pass may delete") + assert.Contains(t, body, `address = "10.200.1.9"`, "the live ELB record must survive") + assert.Contains(t, body, `soa = "ns1.spx3.net."`, "the SOA is not in the desired set and must survive") + + assert.Equal(t, []string{""}, labelsOfType(t, w.s3cfg, testBase, nsconfig.TypeNS), + "the apex NS set is not in the desired set and must survive a pass") + assert.Contains(t, labelsOfType(t, w.s3cfg, testBase, nsconfig.TypeA), "ns1.", + "nameserver glue is not in the desired set and must survive a pass") +} + +// A zone a pass creates has to arrive complete. The private zone does not exist +// until an instance lands in it, so the first pass to write one materialises it, +// and a zone materialised without its apex and NS set does not resolve at all. +func TestReconcileOnce_MaterialisedZoneCarriesApexAndNameservers(t *testing.T) { + w, objects := newMultiNodeTestWriter(t) + r := reconcilerFor(w, func() DesiredSet { + return DesiredSet{Changes: []Change{{ + Action: ActionUpsert, Zone: "compute.internal", + Name: "ip-10-20-0-5.ap-southeast-2.compute.internal", Type: "A", Value: "10.20.0.5", + }}} + }) + + r.reconcileOnce(t.Context()) + + body := objects["compute.internal.toml"] + require.NotEmpty(t, body, "the pass must materialise the zone it writes into") + + // Both cluster nodes, at the apex: a materialised zone with no NS set is a + // zone that does not resolve. + assert.Equal(t, []string{"", ""}, labelsOfType(t, w.s3cfg, "compute.internal", nsconfig.TypeNS), + "every NS record belongs at the apex, one per cluster nameserver") + assert.Contains(t, body, `address = "ns1.compute.internal."`) + assert.Contains(t, body, `address = "ns2.compute.internal."`) + + cfg, _, err := nsconfig.ReadZoneRaw(w.s3cfg, "compute.internal") + require.NoError(t, err) + assert.Equal(t, "compute.internal", cfg.Domain.Domain, "the apex names the zone") + assert.Equal(t, "ns1.compute.internal.", cfg.Domain.SOA, "a zone with no SOA is not answerable") + assert.Subset(t, labelsOfType(t, w.s3cfg, "compute.internal", nsconfig.TypeA), []string{"ns1.", "ns2."}, + "the NS set needs its glue or the delegation is unresolvable") +} diff --git a/spinifex/handlers/dns/reconcile_test.go b/spinifex/handlers/dns/reconcile_test.go index 32b39fa05..c1def0a9c 100644 --- a/spinifex/handlers/dns/reconcile_test.go +++ b/spinifex/handlers/dns/reconcile_test.go @@ -2,10 +2,6 @@ package dns import ( "context" - "encoding/json" - "errors" - "fmt" - "strings" "sync/atomic" "testing" "time" @@ -144,235 +140,6 @@ func TestComputeConvergeRejectsUnsupportedRecordType(t *testing.T) { assert.Contains(t, err.Error(), "unsupported DNS record type") } -func TestSplitReconcileChanges(t *testing.T) { - tests := []struct { - name string - changes func() []Change - }{ - { - name: "record element limit", - changes: func() []Change { - changes := make([]Change, 501) - for i := range changes { - changes[i] = upsert(fmt.Sprintf("record-%03d.spx3.net", len(changes)-i), "192.0.2.1") - } - return changes - }, - }, - { - name: "value character limit", - changes: func() []Change { - return []Change{ - upsert("first.spx3.net", strings.Repeat("a", 9_000)), - upsert("second.spx3.net", strings.Repeat("b", 9_000)), - } - }, - }, - { - name: "serialized payload limit", - changes: func() []Change { - changes := make([]Change, 100) - for i := range changes { - changes[i] = upsert(fmt.Sprintf("%s-%03d.spx3.net", strings.Repeat("n", 10_000), len(changes)-i), "192.0.2.1") - } - return changes - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - changes := tt.changes() - batches, err := splitReconcileChanges(changes, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - require.Greater(t, len(batches), 1) - - var flattened []Change - for _, batch := range batches { - flattened = append(flattened, batch...) - payload, err := json.Marshal(ChangeBatch{Changes: batch}) - require.NoError(t, err) - assert.LessOrEqual(t, len(payload), maxReconcileNATSPayloadBytes) - - records, valueChars := 0, 0 - for _, change := range batch { - multiplier := 1 - if change.Action == ActionUpsert { - multiplier = 2 - } - records += multiplier - valueChars += multiplier * len([]rune(change.Value)) - } - assert.LessOrEqual(t, records, MaxRecordsPerChangeRequest) - assert.LessOrEqual(t, valueChars, MaxValueCharsPerChangeRequest) - } - assert.Equal(t, changes, flattened, "batching must preserve change order") - }) - } -} - -func TestSplitReconcileChangesExactBoundaries(t *testing.T) { - t.Run("record elements", func(t *testing.T) { - changes := make([]Change, MaxRecordsPerChangeRequest/2) - for i := range changes { - changes[i] = upsert(fmt.Sprintf("record-%d.spx3.net", i), "192.0.2.1") - } - batches, err := splitReconcileChanges(changes, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - require.Len(t, batches, 1) - - overLimit := make([]Change, len(changes)+1) - copy(overLimit, changes) - overLimit[len(changes)] = upsert("one-over.spx3.net", "192.0.2.2") - batches, err = splitReconcileChanges(overLimit, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - assert.Len(t, batches, 2) - }) - - t.Run("value characters", func(t *testing.T) { - change := upsert("exact.spx3.net", strings.Repeat("界", MaxValueCharsPerChangeRequest/2)) - batches, err := splitReconcileChanges([]Change{change}, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - require.Len(t, batches, 1) - assert.Equal(t, change, batches[0][0]) - }) - - t.Run("serialized payload", func(t *testing.T) { - change := upsert("", "192.0.2.1") - encoded, err := json.Marshal(change) - require.NoError(t, err) - nameBytes := maxReconcileNATSPayloadBytes - changeBatchJSONOverhead - len(encoded) - require.Positive(t, nameBytes) - change.Name = strings.Repeat("n", nameBytes) - - payload, err := json.Marshal(ChangeBatch{Changes: []Change{change}}) - require.NoError(t, err) - require.Len(t, payload, maxReconcileNATSPayloadBytes) - batches, err := splitReconcileChanges([]Change{change}, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - require.Len(t, batches, 1) - - change.Name += "n" - _, err = splitReconcileChanges([]Change{change}, maxReconcileNATSPayloadBytes) - require.Error(t, err) - assert.Contains(t, err.Error(), "payload maximum") - }) -} - -func TestSplitReconcileChangesTracksRequestLimitsPerZone(t *testing.T) { - changes := make([]Change, MaxRecordsPerChangeRequest) - for i := range changes { - zone := "spx3.net" - if i%2 == 1 { - zone = "compute.internal" - } - changes[i] = Change{Action: ActionUpsert, Zone: zone, Name: fmt.Sprintf("record-%d.%s", i, zone), Type: "A", Value: "192.0.2.1"} - } - - batches, err := splitReconcileChanges(changes, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - require.Len(t, batches, 1, "each zone independently stays at 1,000 record elements") -} - -func TestSplitAndPublishReconcileChangesHandlesEmptyAndSingleton(t *testing.T) { - batches, err := splitReconcileChanges(nil, maxReconcileNATSPayloadBytes) - require.NoError(t, err) - assert.Empty(t, batches) - - calls := 0 - err = publishReconcileBatches(nil, maxReconcileNATSPayloadBytes, func([]Change) error { - calls++ - return nil - }) - require.NoError(t, err) - assert.Zero(t, calls) - - change := upsert("one.spx3.net", "192.0.2.1") - err = publishReconcileBatches([]Change{change}, maxReconcileNATSPayloadBytes, func(batch []Change) error { - calls++ - assert.Equal(t, []Change{change}, batch) - return nil - }) - require.NoError(t, err) - assert.Equal(t, 1, calls) -} - -func TestPublishReconcileBatchRejectsMissingAcknowledgement(t *testing.T) { - err := publishReconcileBatch(nil, "000000000000", []Change{upsert("one.spx3.net", "192.0.2.1")}) - require.Error(t, err) - assert.ErrorContains(t, err, "writer acknowledged 0 of 1 changes") -} - -func TestSplitReconcileChangesRejectsOversizedChange(t *testing.T) { - changes := []Change{upsert("oversized.spx3.net", strings.Repeat("x", MaxValueCharsPerChangeRequest/2+1))} - - _, err := splitReconcileChanges(changes, maxReconcileNATSPayloadBytes) - require.Error(t, err) - assert.Contains(t, err.Error(), "value characters") -} - -func TestPublishReconcileBatchesWaitsForEachAcknowledgement(t *testing.T) { - changes := make([]Change, 501) - for i := range changes { - changes[i] = upsert(fmt.Sprintf("record-%d.spx3.net", i), "192.0.2.1") - } - firstStarted := make(chan struct{}) - secondStarted := make(chan struct{}) - releaseFirst := make(chan struct{}) - done := make(chan error, 1) - calls := 0 - - go func() { - done <- publishReconcileBatches(changes, maxReconcileNATSPayloadBytes, func([]Change) error { - calls++ - switch calls { - case 1: - close(firstStarted) - <-releaseFirst - case 2: - close(secondStarted) - } - return nil - }) - }() - - <-firstStarted - select { - case <-secondStarted: - t.Fatal("second batch started before the first was acknowledged") - case <-time.After(25 * time.Millisecond): - } - close(releaseFirst) - require.NoError(t, <-done) - assert.Equal(t, 2, calls) - select { - case <-secondStarted: - default: - t.Fatal("second batch was not published after the first acknowledgement") - } -} - -func TestPublishReconcileBatchesStopsAfterFailedAcknowledgement(t *testing.T) { - changes := make([]Change, 1_001) - for i := range changes { - changes[i] = upsert("record.spx3.net", "192.0.2.1") - } - publishErr := errors.New("writer unavailable") - calls := 0 - - err := publishReconcileBatches(changes, maxReconcileNATSPayloadBytes, func([]Change) error { - calls++ - if calls == 2 { - return publishErr - } - return nil - }) - - require.ErrorIs(t, err, publishErr) - assert.ErrorContains(t, err, "publish batch 2 of 3") - assert.Equal(t, 2, calls, "later batches must wait for and stop after a failed acknowledgement") -} - func TestReconcilerDisabledIsNoop(t *testing.T) { r := &Reconciler{} // enabled=false assert.False(t, r.Enabled()) @@ -420,7 +187,7 @@ func TestReconcilerCorruptZoneRebuildsWithoutPruning(t *testing.T) { batch, err := r.computeBatch() require.NoError(t, err) assert.Equal(t, []Change{upsert("lb-1.elb."+testBase, "10.200.1.9")}, batch, - "the full desired set must still be published so the writer rebuilds the zone") + "the full desired set must still be applied so the zone is rebuilt") assert.Empty(t, deletesOf(batch), "a corrupt zone must never produce deletes") } @@ -436,7 +203,7 @@ func TestReconcilerBackendErrorStillAborts(t *testing.T) { // able to start the loop unconditionally. Without it the disabled reconciler // would enter the watch loop and block until shutdown instead of returning. func TestReconciler_RunReturnsWhenDisabled(t *testing.T) { - r := NewReconciler(nil, nil, nil) + r := NewReconciler(nil, nil, nil, nil) require.False(t, r.Enabled()) done := make(chan struct{}) @@ -499,8 +266,8 @@ func TestReconciler_RunPerformsStartupPassThenStopsOnCancel(t *testing.T) { func TestNewReconciler_RetainsItsWatchSources(t *testing.T) { bucket := kvstore.NewBucket(nil, kvstore.Config{Name: "b"}) - assert.Empty(t, NewReconciler(nil, nil, nil).sources) - assert.Len(t, NewReconciler(nil, nil, nil, + assert.Empty(t, NewReconciler(nil, nil, nil, nil).sources) + assert.Len(t, NewReconciler(nil, nil, nil, nil, reconciler.Fixed(bucket, "node.*"), reconciler.Fixed(bucket, "lb.*"), ).sources, 2) diff --git a/spinifex/handlers/dns/writer.go b/spinifex/handlers/dns/writer.go index 3150b09dc..f73cc97c2 100644 --- a/spinifex/handlers/dns/writer.go +++ b/spinifex/handlers/dns/writer.go @@ -60,8 +60,9 @@ func NewWriter(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn) return w } -// Enabled reports whether the writer will process changes. -func (w *Writer) Enabled() bool { return w.enabled } +// Enabled reports whether the writer will process changes. Nil-safe: the +// 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