mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
refactor: apply the DNS reconcile batch on the elected node instead of publishing it (#965)
The reconcile elected a leader, computed a converging batch, then published it to the queue group, which load-balances rather than serialises. The batch was therefore applied by an arbitrary peer, not by the node that won the election. That split the read that decides which records are stale from the write that removes them. A lifecycle upsert landing between the two was invisible to the second, so a record re-added after the reconcile read it was deleted by a decision that predated it. Applying locally closes the window, and makes the elected node the only writer the reconcile puts on a zone. Deleted with their tests: the batch splitter, the payload limits and the partial-acknowledgement check, none of which have anything left to bound now that no batch crosses a NATS connection. The lifecycle publishers and the zone lock are untouched: they are still a second writer, and retiring them is separate.
This commit is contained in:
Josh Siviour
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user