mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
feat: add kvstore watch and the reconciler loop (#900)
Watch is used zero times in spinifex against 33 ticker loops. This adds the primitive on Bucket, which already owns the memoised handle, plus the watch/debounce/resync/re-establish loop, so each converted caller changes only its trigger. Watches are established before the first pass: a change landing between the two would otherwise wait for the next resync, since the pass that would have seen it ran before the watch existed.
This commit is contained in:
Josh Siviour
@@ -6,6 +6,7 @@ package kvstore
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -46,6 +47,10 @@ func NewOpenBucket(kv jetstream.KeyValue, cfg Config) *Bucket {
|
||||
return &Bucket{kv: kv, cfg: cfg}
|
||||
}
|
||||
|
||||
// Name returns the configured bucket name, for callers holding several buckets
|
||||
// that need to tell them apart in a map or a log line.
|
||||
func (b *Bucket) Name() string { return b.cfg.Name }
|
||||
|
||||
// Configured reports whether the bucket has anything to open against, for the
|
||||
// callers whose absent-KV path is a legitimate fallback rather than an error.
|
||||
func (b *Bucket) Configured() bool {
|
||||
@@ -75,6 +80,25 @@ func (b *Bucket) KV(ctx context.Context) (jetstream.KeyValue, error) {
|
||||
return kv, nil
|
||||
}
|
||||
|
||||
// Watch returns a watcher over the keys matching filter, which takes NATS
|
||||
// subject wildcards ("node.*", "lb.*", ">" for everything).
|
||||
//
|
||||
// Updates only: the caller reconciles once at startup, so replaying every
|
||||
// existing key would fire a redundant pass, and replaying them again on every
|
||||
// re-establish would defeat the debounce. The cost is that a disconnect gap is
|
||||
// invisible, which is what the caller's periodic resync exists to cover.
|
||||
func (b *Bucket) Watch(ctx context.Context, filter string) (jetstream.KeyWatcher, error) {
|
||||
kv, err := b.KV(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
w, err := kv.Watch(ctx, filter, jetstream.UpdatesOnly())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("kvstore: watch %s %s: %w", b.cfg.Name, filter, err)
|
||||
}
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// open picks the kvutil helper matching the configured TTL and replica count.
|
||||
func (b *Bucket) open(ctx context.Context) (jetstream.KeyValue, error) {
|
||||
switch {
|
||||
|
||||
@@ -0,0 +1,295 @@
|
||||
// Package reconciler runs a reconcile function on change rather than on a
|
||||
// timer: it watches the KV buckets the function reads, coalesces a burst of
|
||||
// updates into one pass, and falls back to a periodic resync so a gap in the
|
||||
// watch cannot leave the world unconverged.
|
||||
package reconciler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/otelsetup"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// Defaults for Config. The resync is the correctness backstop, so it is slow;
|
||||
// the debounce only has to outlast a multi-key write, so it is short.
|
||||
const (
|
||||
DefaultResync = 5 * time.Minute
|
||||
DefaultDebounce = 250 * time.Millisecond
|
||||
// maxDebounce bounds how long a continuous stream of updates can defer a
|
||||
// pass, so a busy bucket still reconciles rather than starving.
|
||||
maxDebounce = 2 * time.Second
|
||||
// retryWatch is how long to wait before re-opening a watcher that could not
|
||||
// be established, so a bucket that is briefly unreachable is retried
|
||||
// without spinning.
|
||||
retryWatch = 5 * time.Second
|
||||
)
|
||||
|
||||
// Source names the buckets to watch and the keys within them that matter.
|
||||
type Source interface {
|
||||
// Buckets is re-evaluated on every resync, so a bucket that appears after
|
||||
// startup — a new account's, say — is picked up without a restart.
|
||||
Buckets(ctx context.Context) ([]*kvstore.Bucket, error)
|
||||
// Filter is a NATS subject filter over key names, e.g. "node.*".
|
||||
Filter() string
|
||||
}
|
||||
|
||||
// Config describes one reconcile loop.
|
||||
type Config struct {
|
||||
// Name identifies the loop in logs.
|
||||
Name string
|
||||
// Sources are watched for changes. An empty Sources is allowed: the loop
|
||||
// then runs on the resync alone, which is the pre-watch behaviour.
|
||||
Sources []Source
|
||||
// Reconcile performs one pass. It is never called concurrently with itself.
|
||||
Reconcile func(ctx context.Context) error
|
||||
// Resync bounds how stale the world can get when no update arrives, and is
|
||||
// also when Sources are re-enumerated. Zero means DefaultResync.
|
||||
Resync time.Duration
|
||||
// Debounce is how long to wait for a burst to settle. Zero means
|
||||
// DefaultDebounce.
|
||||
Debounce time.Duration
|
||||
}
|
||||
|
||||
// fixDefaults fills in the zero-valued durations.
|
||||
func (c *Config) fixDefaults() {
|
||||
if c.Resync <= 0 {
|
||||
c.Resync = DefaultResync
|
||||
}
|
||||
if c.Debounce <= 0 {
|
||||
c.Debounce = DefaultDebounce
|
||||
}
|
||||
}
|
||||
|
||||
// staticSource is one fixed bucket, for the callers whose bucket set does not
|
||||
// change while the process runs.
|
||||
type staticSource struct {
|
||||
bucket *kvstore.Bucket
|
||||
filter string
|
||||
}
|
||||
|
||||
func (s staticSource) Buckets(context.Context) ([]*kvstore.Bucket, error) {
|
||||
return []*kvstore.Bucket{s.bucket}, nil
|
||||
}
|
||||
|
||||
func (s staticSource) Filter() string { return s.filter }
|
||||
|
||||
// Fixed returns a Source over one bucket whose identity is known at startup.
|
||||
func Fixed(bucket *kvstore.Bucket, filter string) Source {
|
||||
return staticSource{bucket: bucket, filter: filter}
|
||||
}
|
||||
|
||||
var _ Source = staticSource{}
|
||||
|
||||
// Run reconciles once, then on every change and every resync until ctx is done.
|
||||
// It returns only when ctx is cancelled.
|
||||
func Run(ctx context.Context, cfg Config) {
|
||||
cfg.fixDefaults()
|
||||
if cfg.Reconcile == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Buffered by one: a change arriving mid-pass sets the pending flag rather
|
||||
// than blocking the watcher goroutine, and is served by the next pass.
|
||||
changes := make(chan struct{}, 1)
|
||||
w := &watchSet{cfg: cfg, changes: changes}
|
||||
defer w.stop()
|
||||
|
||||
// Watches go up before the first pass, not after: a change landing between
|
||||
// the two would otherwise be invisible until the next resync, because the
|
||||
// pass that would have seen it ran before the watch existed.
|
||||
w.resync(ctx)
|
||||
pass(ctx, cfg, "startup")
|
||||
|
||||
resync := time.NewTicker(cfg.Resync)
|
||||
defer resync.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-resync.C:
|
||||
// Re-enumerating before the pass means a bucket that appeared since
|
||||
// the last resync is watched from here on, and the pass that
|
||||
// follows covers whatever it already held.
|
||||
w.resync(ctx)
|
||||
pass(ctx, cfg, "resync")
|
||||
case <-changes:
|
||||
if settle(ctx, changes, cfg.Debounce) {
|
||||
pass(ctx, cfg, "change")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// settle waits for a burst to stop arriving, so one multi-key write produces
|
||||
// one pass rather than one per key. It gives up waiting after maxDebounce so a
|
||||
// continuous stream of updates cannot defer the pass indefinitely. It reports
|
||||
// false only when ctx ended first.
|
||||
func settle(ctx context.Context, changes <-chan struct{}, debounce time.Duration) bool {
|
||||
quiet := time.NewTimer(debounce)
|
||||
defer quiet.Stop()
|
||||
limit := time.NewTimer(maxDebounce)
|
||||
defer limit.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-limit.C:
|
||||
return true
|
||||
case <-quiet.C:
|
||||
return true
|
||||
case <-changes:
|
||||
if !quiet.Stop() {
|
||||
<-quiet.C
|
||||
}
|
||||
quiet.Reset(debounce)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// pass runs one reconcile, logging rather than returning a failure: the loop
|
||||
// outlives any single pass, and the next resync retries.
|
||||
func pass(ctx context.Context, cfg Config, trigger string) {
|
||||
start := time.Now()
|
||||
if err := cfg.Reconcile(ctx); err != nil {
|
||||
slog.WarnContext(ctx, "reconcile pass failed, retrying next cycle",
|
||||
"loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err)
|
||||
return
|
||||
}
|
||||
slog.DebugContext(ctx, "reconcile pass complete",
|
||||
"loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)))
|
||||
}
|
||||
|
||||
// watchSet holds one watcher per bucket currently being watched, keyed by
|
||||
// bucket name so a resync can tell an already-watched bucket from a new one.
|
||||
type watchSet struct {
|
||||
cfg Config
|
||||
changes chan struct{}
|
||||
active map[string]*watcher
|
||||
}
|
||||
|
||||
// watcher is one bucket's watcher plus the goroutine draining it.
|
||||
type watcher struct {
|
||||
kw jetstream.KeyWatcher
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// resync opens a watcher for every bucket a Source now names and drops those it
|
||||
// no longer does. A bucket that cannot be watched is left out and retried on
|
||||
// the next resync, so one unreachable bucket does not stop the others.
|
||||
func (w *watchSet) resync(ctx context.Context) {
|
||||
if w.active == nil {
|
||||
w.active = map[string]*watcher{}
|
||||
}
|
||||
wanted := map[string]struct{}{}
|
||||
for _, src := range w.cfg.Sources {
|
||||
buckets, err := src.Buckets(ctx)
|
||||
if err != nil {
|
||||
// Leave the existing watchers in place: a failed enumeration is not
|
||||
// evidence that the buckets went away.
|
||||
slog.WarnContext(ctx, "reconcile: enumerate watch buckets failed",
|
||||
"loop", w.cfg.Name, "err", err)
|
||||
for name := range w.active {
|
||||
wanted[name] = struct{}{}
|
||||
}
|
||||
continue
|
||||
}
|
||||
for _, bucket := range buckets {
|
||||
name := bucket.Name()
|
||||
wanted[name] = struct{}{}
|
||||
if _, ok := w.active[name]; ok {
|
||||
continue
|
||||
}
|
||||
if watch := w.open(ctx, bucket, src.Filter()); watch != nil {
|
||||
w.active[name] = watch
|
||||
}
|
||||
}
|
||||
}
|
||||
for name, watch := range w.active {
|
||||
if _, ok := wanted[name]; !ok {
|
||||
watch.stop()
|
||||
delete(w.active, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// open establishes one watcher and starts draining it. A nil return means the
|
||||
// bucket could not be watched this cycle.
|
||||
func (w *watchSet) open(ctx context.Context, bucket *kvstore.Bucket, filter string) *watcher {
|
||||
watchCtx, cancel := context.WithCancel(ctx)
|
||||
kw, err := bucket.Watch(watchCtx, filter)
|
||||
if err != nil {
|
||||
cancel()
|
||||
slog.WarnContext(ctx, "reconcile: watch unavailable, falling back to resync until it recovers",
|
||||
"loop", w.cfg.Name, "bucket", bucket.Name(), "filter", filter, "err", err)
|
||||
return nil
|
||||
}
|
||||
watch := &watcher{kw: kw, cancel: cancel}
|
||||
go w.drain(watchCtx, bucket, filter, watch)
|
||||
return watch
|
||||
}
|
||||
|
||||
// drain forwards updates until the watcher's channel closes, then re-opens it.
|
||||
// A closed channel means the connection dropped, so re-establishment is itself
|
||||
// a change signal: UpdatesOnly hides whatever happened during the gap.
|
||||
func (w *watchSet) drain(ctx context.Context, bucket *kvstore.Bucket, filter string, watch *watcher) {
|
||||
for {
|
||||
for update := range watch.kw.Updates() {
|
||||
// A nil update marks the end of the initial replay, not a change.
|
||||
if update == nil {
|
||||
continue
|
||||
}
|
||||
w.signal()
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
slog.InfoContext(ctx, "reconcile: watch dropped, re-establishing",
|
||||
"loop", w.cfg.Name, "bucket", bucket.Name())
|
||||
kw, err := bucket.Watch(ctx, filter)
|
||||
if err != nil {
|
||||
slog.WarnContext(ctx, "reconcile: re-establish watch failed",
|
||||
"loop", w.cfg.Name, "bucket", bucket.Name(), "err", err)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(retryWatch):
|
||||
}
|
||||
continue
|
||||
}
|
||||
watch.kw = kw
|
||||
w.signal()
|
||||
}
|
||||
}
|
||||
|
||||
// signal reports a change without blocking: the channel's single slot already
|
||||
// means "at least one change is pending", which is all a full-recompute
|
||||
// reconcile needs to know.
|
||||
func (w *watchSet) signal() {
|
||||
select {
|
||||
case w.changes <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// stop tears down every watcher.
|
||||
func (w *watchSet) stop() {
|
||||
for name, watch := range w.active {
|
||||
watch.stop()
|
||||
delete(w.active, name)
|
||||
}
|
||||
}
|
||||
|
||||
// stop cancels the draining goroutine and closes the underlying watcher.
|
||||
// Cancelling already tears the subscription down, so Stop routinely reports an
|
||||
// invalid subscription on the way out; that is the expected shutdown path.
|
||||
func (w *watcher) stop() {
|
||||
w.cancel()
|
||||
if err := w.kw.Stop(); err != nil {
|
||||
slog.Debug("reconcile: stopping watcher", "err", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,362 @@
|
||||
package reconciler_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/reconciler"
|
||||
"github.com/mulgadc/spinifex/spinifex/testutil"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// record is a stand-in for whatever the watched bucket holds; the loop never
|
||||
// decodes it, only notices that it changed.
|
||||
type record struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
// passes counts reconcile calls and lets a test wait for the count to reach a
|
||||
// value, so no test sleeps for a fixed duration hoping a pass happened.
|
||||
type passes struct {
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond
|
||||
count int
|
||||
}
|
||||
|
||||
func newPasses() *passes {
|
||||
p := &passes{}
|
||||
p.cond = sync.NewCond(&p.mu)
|
||||
return p
|
||||
}
|
||||
|
||||
func (p *passes) record() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.count++
|
||||
p.cond.Broadcast()
|
||||
}
|
||||
|
||||
func (p *passes) now() int {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.count
|
||||
}
|
||||
|
||||
// waitFor blocks until at least n passes have run, failing the test rather than
|
||||
// hanging if they do not.
|
||||
func (p *passes) waitFor(t *testing.T, n int, why string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for p.count < n {
|
||||
p.mu.Unlock()
|
||||
if time.Now().After(deadline) {
|
||||
p.mu.Lock()
|
||||
t.Fatalf("timed out waiting for %d passes (%s); saw %d", n, why, p.count)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
p.mu.Lock()
|
||||
}
|
||||
}
|
||||
|
||||
// stillAt asserts no further pass ran during a short settling window, for the
|
||||
// tests whose point is that something did NOT trigger a pass.
|
||||
func (p *passes) stillAt(t *testing.T, n int, why string) {
|
||||
t.Helper()
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
assert.Equal(t, n, p.now(), why)
|
||||
}
|
||||
|
||||
// newBucket starts an embedded JetStream server and returns a store over a
|
||||
// bucket on it, plus the connection so a test can drop it.
|
||||
func newBucket(t *testing.T, name string) (*kvstore.Store[record], *nats.Conn) {
|
||||
t.Helper()
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
return kvstore.New[record](testutil.NewJetStream(t, nc), kvstore.Config{
|
||||
Name: name,
|
||||
History: 1,
|
||||
}), nc
|
||||
}
|
||||
|
||||
// run starts the loop under a context the test cancels on cleanup.
|
||||
func run(t *testing.T, cfg reconciler.Config) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
reconciler.Run(ctx, cfg)
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
<-done
|
||||
})
|
||||
}
|
||||
|
||||
func TestRun_ReconcilesOnceAtStartup(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-startup")
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "startup",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: time.Hour,
|
||||
})
|
||||
|
||||
p.waitFor(t, 1, "the startup pass")
|
||||
p.stillAt(t, 1, "an idle bucket must not produce further passes")
|
||||
}
|
||||
|
||||
func TestRun_AnUpdateTriggersAPass(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-update")
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "update",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: time.Hour,
|
||||
})
|
||||
p.waitFor(t, 1, "the startup pass")
|
||||
|
||||
require.NoError(t, store.Set(t.Context(), "one", &record{Name: "one"}))
|
||||
p.waitFor(t, 2, "the pass triggered by the write")
|
||||
}
|
||||
|
||||
// TestRun_ABurstCollapsesIntoOnePass is the whole point of the debounce: a
|
||||
// multi-key write is one logical change, and a full-recompute reconcile gains
|
||||
// nothing from running once per key.
|
||||
func TestRun_ABurstCollapsesIntoOnePass(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-burst")
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "burst",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: time.Hour,
|
||||
Debounce: 200 * time.Millisecond,
|
||||
})
|
||||
p.waitFor(t, 1, "the startup pass")
|
||||
|
||||
for _, key := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
|
||||
require.NoError(t, store.Set(t.Context(), key, &record{Name: key}))
|
||||
}
|
||||
|
||||
p.waitFor(t, 2, "the pass covering the burst")
|
||||
p.stillAt(t, 2, "eight writes in one burst must produce one pass, not eight")
|
||||
}
|
||||
|
||||
func TestRun_ResyncFiresWithNoWatchTraffic(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-resync")
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "resync",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: 150 * time.Millisecond,
|
||||
})
|
||||
|
||||
p.waitFor(t, 4, "repeated resync passes against a bucket nobody is writing to")
|
||||
}
|
||||
|
||||
// TestRun_NoSourcesStillReconcilesOnTheResync pins the degenerate case: a loop
|
||||
// with nothing watchable is the pre-watch behaviour, not a dead loop.
|
||||
func TestRun_NoSourcesStillReconcilesOnTheResync(t *testing.T) {
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "no-sources",
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: 150 * time.Millisecond,
|
||||
})
|
||||
|
||||
p.waitFor(t, 3, "resync passes with no sources configured")
|
||||
}
|
||||
|
||||
// TestRun_AFailedPassDoesNotStopTheLoop covers the reason Reconcile's error is
|
||||
// logged rather than returned: the loop outlives any single pass.
|
||||
func TestRun_AFailedPassDoesNotStopTheLoop(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-failure")
|
||||
p := newPasses()
|
||||
var calls atomic.Int64
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "failure",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
|
||||
Reconcile: func(context.Context) error {
|
||||
p.record()
|
||||
if calls.Add(1) == 1 {
|
||||
return assert.AnError
|
||||
}
|
||||
return nil
|
||||
},
|
||||
Resync: 150 * time.Millisecond,
|
||||
})
|
||||
|
||||
p.waitFor(t, 3, "passes after the first one failed")
|
||||
}
|
||||
|
||||
// TestRun_AFilterExcludesOtherKeys stops an unrelated write to the same bucket
|
||||
// waking a loop that does not care about it — the ELBv2 bucket holds listeners
|
||||
// and rules alongside the load balancers DNS reconciles on.
|
||||
func TestRun_AFilterExcludesOtherKeys(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-filter")
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "filter",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, "lb.*")},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: time.Hour,
|
||||
})
|
||||
p.waitFor(t, 1, "the startup pass")
|
||||
|
||||
require.NoError(t, store.Set(t.Context(), "rule.one", &record{Name: "rule"}))
|
||||
p.stillAt(t, 1, "a write outside the filter must not trigger a pass")
|
||||
|
||||
require.NoError(t, store.Set(t.Context(), "lb.one", &record{Name: "lb"}))
|
||||
p.waitFor(t, 2, "the pass triggered by a write inside the filter")
|
||||
}
|
||||
|
||||
// dynamicSource is the per-account bucket case: the set of buckets is not known
|
||||
// at startup and is re-read on every resync.
|
||||
type dynamicSource struct {
|
||||
mu sync.Mutex
|
||||
buckets []*kvstore.Bucket
|
||||
}
|
||||
|
||||
func (s *dynamicSource) Buckets(context.Context) ([]*kvstore.Bucket, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return append([]*kvstore.Bucket(nil), s.buckets...), nil
|
||||
}
|
||||
|
||||
func (s *dynamicSource) Filter() string { return ">" }
|
||||
|
||||
func (s *dynamicSource) add(b *kvstore.Bucket) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.buckets = append(s.buckets, b)
|
||||
}
|
||||
|
||||
// TestRun_ABucketAppearingAfterStartupIsPickedUp is the per-account bucket
|
||||
// case: EKS and RDS create a bucket per account, and JetStream has no
|
||||
// bucket-created event, so discovery rides on the resync.
|
||||
func TestRun_ABucketAppearingAfterStartupIsPickedUp(t *testing.T) {
|
||||
_, nc, _ := testutil.StartTestJetStream(t)
|
||||
js := testutil.NewJetStream(t, nc)
|
||||
first := kvstore.New[record](js, kvstore.Config{Name: "reconciler-acct-a", History: 1})
|
||||
second := kvstore.New[record](js, kvstore.Config{Name: "reconciler-acct-b", History: 1})
|
||||
|
||||
src := &dynamicSource{}
|
||||
src.add(first.Bucket)
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "dynamic",
|
||||
Sources: []reconciler.Source{src},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: 200 * time.Millisecond,
|
||||
})
|
||||
p.waitFor(t, 1, "the startup pass")
|
||||
|
||||
// Force the second bucket into existence, then publish it to the source.
|
||||
require.NoError(t, second.Set(t.Context(), "seed", &record{Name: "seed"}))
|
||||
src.add(second.Bucket)
|
||||
|
||||
// One resync to discover it, then a write it must now notice.
|
||||
before := p.now()
|
||||
p.waitFor(t, before+2, "resync passes covering the discovery of the new bucket")
|
||||
|
||||
before = p.now()
|
||||
require.NoError(t, second.Set(t.Context(), "later", &record{Name: "later"}))
|
||||
p.waitFor(t, before+1, "a pass triggered by the newly discovered bucket")
|
||||
}
|
||||
|
||||
// TestRun_AnUnwatchableSourceStillResyncs covers the fallback: a bucket with no
|
||||
// JetStream behind it cannot be watched, and the loop must degrade to the timer
|
||||
// rather than stop.
|
||||
func TestRun_AnUnwatchableSourceStillResyncs(t *testing.T) {
|
||||
unwatchable := kvstore.NewBucket(nil, kvstore.Config{
|
||||
Name: "reconciler-unwatchable",
|
||||
Missing: "no JetStream configured",
|
||||
})
|
||||
p := newPasses()
|
||||
|
||||
run(t, reconciler.Config{
|
||||
Name: "unwatchable",
|
||||
Sources: []reconciler.Source{reconciler.Fixed(unwatchable, ">")},
|
||||
Reconcile: func(context.Context) error { p.record(); return nil },
|
||||
Resync: 150 * time.Millisecond,
|
||||
})
|
||||
|
||||
p.waitFor(t, 3, "resync passes despite the watch never establishing")
|
||||
}
|
||||
|
||||
func TestFixed_ReportsItsBucketAndFilter(t *testing.T) {
|
||||
bucket := kvstore.NewBucket(nil, kvstore.Config{Name: "b"})
|
||||
src := reconciler.Fixed(bucket, "node.*")
|
||||
|
||||
buckets, err := src.Buckets(t.Context())
|
||||
require.NoError(t, err)
|
||||
require.Len(t, buckets, 1)
|
||||
assert.Equal(t, "b", buckets[0].Name())
|
||||
assert.Equal(t, "node.*", src.Filter())
|
||||
}
|
||||
|
||||
func TestBucket_WatchOnANilJetStreamReportsTheConfiguredMessage(t *testing.T) {
|
||||
bucket := kvstore.NewBucket(nil, kvstore.Config{
|
||||
Name: "b",
|
||||
Missing: "watch test: no JetStream client configured",
|
||||
})
|
||||
|
||||
_, err := bucket.Watch(t.Context(), ">")
|
||||
require.EqualError(t, err, "watch test: no JetStream client configured")
|
||||
}
|
||||
|
||||
// TestBucket_WatchDeliversUpdatesOnly pins UpdatesOnly at the kvstore seam: a
|
||||
// watcher established over a bucket that already holds keys must not replay
|
||||
// them, or every re-establish would fire a redundant pass per existing key.
|
||||
func TestBucket_WatchDeliversUpdatesOnly(t *testing.T) {
|
||||
store, _ := newBucket(t, "reconciler-updatesonly")
|
||||
require.NoError(t, store.Set(t.Context(), "existing", &record{Name: "existing"}))
|
||||
|
||||
watcher, err := store.Watch(t.Context(), ">")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = watcher.Stop() })
|
||||
|
||||
require.NoError(t, store.Set(t.Context(), "fresh", &record{Name: "fresh"}))
|
||||
|
||||
key := waitForUpdate(t, watcher)
|
||||
assert.Equal(t, "fresh", key, "a pre-existing key must not be replayed")
|
||||
}
|
||||
|
||||
// waitForUpdate returns the key of the next real update, skipping the nil that
|
||||
// marks the end of the initial replay.
|
||||
func waitForUpdate(t *testing.T, watcher jetstream.KeyWatcher) string {
|
||||
t.Helper()
|
||||
deadline := time.After(10 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case update := <-watcher.Updates():
|
||||
if update == nil {
|
||||
continue
|
||||
}
|
||||
return update.Key()
|
||||
case <-deadline:
|
||||
t.Fatal("timed out waiting for a watch update")
|
||||
return ""
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user