mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat: gateway-side instance record cache (Phase 2) (#992)
* feat: add read-only instance record cache (Phase 2) Adds spinifex/instancecache: a revision-fenced snapshot+watch informer over the instance record KV space, with a per-account index and OTel metrics. Wired up and running in awsgw at startup, but not yet consulted by any read path. * fix: cover the fence with real syncs, bound decode-failure cardinality, expose degraded Adds end-to-end fencing tests that drive freshSync/periodicResync against the embedded NATS server via a test-only postSnapshotHook seam, since drainBuffered unit tests alone don't exercise the buffer-before-snapshot ordering. Drops the per-key attribute from the decode_failures counter to keep it O(1) time series; the key is still logged where the decode happens. Adds a Degraded accessor set on sync failure and cleared on the next success, per D7, without changing what List or Get answer. * refactor: replace instancecache's buffering fence with ResumeFromRevision The watch/snapshot fence no longer needs a side buffer, watcher modes, or an overflow failure mode. Every sync now snapshots first, then opens a watcher with kvstore.Bucket.WatchFrom (ResumeFromRevision), and replays onto a private candidate map until the end-of-replay marker arrives. Only a fully caught-up candidate is swapped in as the live map, together with its watcher becoming active; the previous watcher is stopped and joined only after, so two watchers never apply to the same map. kvstore.Bucket gains WatchFrom as a sibling to Watch, opened through the same withKV wrapper, for a caller that needs to replay from a snapshot's high-water mark rather than an UpdatesOnly live feed. * fix: cover the fence with production entry points, not a reimplementation The four fence tests recomputed hw+1 and called kvstore.WatchFrom/replay themselves instead of going through syncOnce, so they verified nats.go's resume behavior (already covered by TestResumeFromRevision_...) rather than the cache's own revision handling. A mutation inside syncOnce (hw+1 -> hw+1_000_000) passed all four. Reinstate a postSnapshotHook seam (nil in production) so the initial-sync pair can call syncOnce/installSync directly and the periodic-resync pair can call periodicResync directly, landing the write inside syncOnce's internal snapshot-to-watch window without duplicating it. The same mutation now fails all four; reverting passes all four. * feat: count the records a resumed watch replays, so the fence is observable live
This commit is contained in:
Josh Siviour
@@ -0,0 +1,546 @@
|
||||
// Package instancecache is a read-only informer over the live instance record
|
||||
// space: a cache of *vm.VM projections kept current by a JetStream KV watch
|
||||
// and backstopped by a periodic whole-set resync. It answers List and Get
|
||||
// from memory; nothing here fans out over NATS.
|
||||
//
|
||||
// Entries are immutable. An update replaces a map entry's pointer; nothing
|
||||
// mutates a *vm.VM in place, and a caller must not either.
|
||||
//
|
||||
// The watch and a snapshot are fenced against each other by sequencing, not
|
||||
// buffering: every sync — initial, periodic, or a dead watcher's replacement
|
||||
// — takes a Snapshot first, then opens a new watcher that resumes from just
|
||||
// past the snapshot's high-water mark and replays onto a private candidate
|
||||
// map until it catches up. Only once that candidate is complete does it
|
||||
// become the live map, at the same moment the new watcher becomes the active
|
||||
// one; the old watcher, if there was one, is stopped and joined only after.
|
||||
// Because the candidate is a map nobody else can see until that swap, a
|
||||
// failed sync costs nothing: the live map and the previously active watcher
|
||||
// were never touched.
|
||||
//
|
||||
// A record that will not decode never removes an entry: the cache only ever
|
||||
// removes a key because the record space said it is gone, never because one
|
||||
// read of it failed.
|
||||
package instancecache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/mulgadc/spinifex/spinifex/vm"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// Defaults for Config.
|
||||
const (
|
||||
// DefaultResyncInterval is the periodic resync cadence: the completeness
|
||||
// backstop for a delete or TTL expiry the watch missed.
|
||||
DefaultResyncInterval = 30 * time.Second
|
||||
|
||||
// DefaultRetryInterval is how long a failed sync waits before trying
|
||||
// again, so an unreachable bucket is retried without spinning.
|
||||
DefaultRetryInterval = 5 * time.Second
|
||||
)
|
||||
|
||||
// Config configures the KV bucket a Cache watches and its clocks. Bucket and
|
||||
// Prefix are supplied by the caller so this package never has to know the
|
||||
// daemon's bucket name or key-space constants.
|
||||
type Config struct {
|
||||
// Bucket names and sizes the KV bucket to open. The caller resolves the
|
||||
// name and replica count, matching how every other reconcile loop in
|
||||
// awsgw opens the same bucket.
|
||||
Bucket kvstore.Config
|
||||
|
||||
// Prefix is the key prefix identifying an instance record, e.g. "i.".
|
||||
// The watch and snapshot filter is Prefix+"*", and an entry's map key is
|
||||
// its record key with Prefix trimmed.
|
||||
Prefix string
|
||||
|
||||
ResyncInterval time.Duration
|
||||
RetryInterval time.Duration
|
||||
}
|
||||
|
||||
// liveWatcher is one KeyWatcher plus the context that owns it and the
|
||||
// goroutine draining it once it is live.
|
||||
type liveWatcher struct {
|
||||
kw jetstream.KeyWatcher
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// Cache is the informer: a live map of instance ID to *vm.VM, an account
|
||||
// index over it, and the goroutine that keeps both current. Construct with
|
||||
// New and start with Run.
|
||||
type Cache struct {
|
||||
store *kvstore.Store[vm.InstanceRecord]
|
||||
prefix string
|
||||
filter string
|
||||
|
||||
resyncInterval time.Duration
|
||||
retryInterval time.Duration
|
||||
|
||||
mu sync.RWMutex
|
||||
entries map[string]*vm.VM
|
||||
index map[string]map[string]struct{}
|
||||
ready bool
|
||||
|
||||
watcherMu sync.Mutex
|
||||
active *liveWatcher
|
||||
|
||||
watcherLost chan struct{}
|
||||
lastResync atomic.Int64 // unix nanos of the last successful sync, 0 = never
|
||||
degraded atomic.Bool // true from a failed sync until the next one succeeds
|
||||
|
||||
metrics *cacheMetrics
|
||||
|
||||
// postSnapshotHook, when set, runs inside syncOnce after snapshotCandidate
|
||||
// returns and before the resuming watcher is opened. Nil in production; a
|
||||
// test seam for landing a write deterministically inside the window a
|
||||
// resuming watcher has to close, which nothing else can reach because it
|
||||
// is internal to syncOnce.
|
||||
postSnapshotHook func()
|
||||
}
|
||||
|
||||
// New returns a Cache over the bucket cfg describes. Call Run to start it;
|
||||
// until the first sync completes, List reports the cache not ready.
|
||||
func New(js jetstream.JetStream, cfg Config) *Cache {
|
||||
resync := cfg.ResyncInterval
|
||||
if resync <= 0 {
|
||||
resync = DefaultResyncInterval
|
||||
}
|
||||
retry := cfg.RetryInterval
|
||||
if retry <= 0 {
|
||||
retry = DefaultRetryInterval
|
||||
}
|
||||
c := &Cache{
|
||||
store: kvstore.New[vm.InstanceRecord](js, cfg.Bucket),
|
||||
prefix: cfg.Prefix,
|
||||
filter: cfg.Prefix + "*",
|
||||
resyncInterval: resync,
|
||||
retryInterval: retry,
|
||||
entries: map[string]*vm.VM{},
|
||||
index: map[string]map[string]struct{}{},
|
||||
watcherLost: make(chan struct{}, 1),
|
||||
}
|
||||
c.metrics = newCacheMetrics(c)
|
||||
return c
|
||||
}
|
||||
|
||||
// Run drives the informer until ctx is cancelled: the initial sync, then the
|
||||
// periodic resync and watcher-replacement loop. Intended to run in its own
|
||||
// goroutine, matching the other reconcile loops awsgw starts.
|
||||
func (c *Cache) Run(ctx context.Context) {
|
||||
lw := c.initialSync(ctx)
|
||||
if lw == nil {
|
||||
return
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(c.resyncInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
active := c.getActive()
|
||||
if active != nil {
|
||||
active.cancel()
|
||||
_ = active.kw.Stop()
|
||||
}
|
||||
return
|
||||
case <-c.watcherLost:
|
||||
active := c.getActive()
|
||||
if active != nil {
|
||||
c.replaceWatcher(ctx, active)
|
||||
}
|
||||
case <-ticker.C:
|
||||
c.periodicResync(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ready reports whether the initial whole-set sync has completed. Before it
|
||||
// has, List and Get answer from an empty or partial view and must never be
|
||||
// read as proof an instance does not exist.
|
||||
func (c *Cache) Ready() bool {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return c.ready
|
||||
}
|
||||
|
||||
// Degraded reports whether the most recent sync attempt (initial, periodic,
|
||||
// or watcher-replacement) failed. The cache keeps serving its last-known
|
||||
// contents regardless, but a cache that has been degraded for a while cannot
|
||||
// be trusted about absence. Not consulted by List or Get in this phase.
|
||||
func (c *Cache) Degraded() bool {
|
||||
return c.degraded.Load()
|
||||
}
|
||||
|
||||
// List returns the cached instances visible to accountID and whether the
|
||||
// cache is ready to be believed about absence. IsInstanceVisibleToCaller is
|
||||
// applied after the index lookup as defence in depth: the index is a
|
||||
// performance structure, not the authorisation boundary.
|
||||
func (c *Cache) List(_ context.Context, accountID string) ([]*vm.VM, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
ids := c.index[indexKeyFor(accountID)]
|
||||
out := make([]*vm.VM, 0, len(ids))
|
||||
for id := range ids {
|
||||
v, ok := c.entries[id]
|
||||
if !ok || !instance.IsInstanceVisibleToCaller(accountID, v) {
|
||||
continue
|
||||
}
|
||||
out = append(out, v)
|
||||
}
|
||||
return out, c.ready
|
||||
}
|
||||
|
||||
// Get returns one instance by ID, consulting KV directly on a cache miss. A
|
||||
// nil instance with a nil error means the record space has no record of it.
|
||||
func (c *Cache) Get(ctx context.Context, instanceID string) (*vm.VM, error) {
|
||||
c.mu.RLock()
|
||||
v, ok := c.entries[instanceID]
|
||||
c.mu.RUnlock()
|
||||
if ok {
|
||||
return v, nil
|
||||
}
|
||||
|
||||
rec, _, err := c.store.Get(ctx, c.prefix+instanceID)
|
||||
if err != nil {
|
||||
if errors.Is(err, kvstore.ErrNotFound) {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return vm.VMFromRecord(rec), nil
|
||||
}
|
||||
|
||||
// getActive and setActive guard the watcher pointer separately from the map,
|
||||
// so a List or Get never blocks behind a watcher transition.
|
||||
func (c *Cache) getActive() *liveWatcher {
|
||||
c.watcherMu.Lock()
|
||||
defer c.watcherMu.Unlock()
|
||||
return c.active
|
||||
}
|
||||
|
||||
func (c *Cache) setActive(lw *liveWatcher) {
|
||||
c.watcherMu.Lock()
|
||||
c.active = lw
|
||||
c.watcherMu.Unlock()
|
||||
}
|
||||
|
||||
func (c *Cache) markResynced() {
|
||||
c.lastResync.Store(time.Now().UnixNano())
|
||||
c.degraded.Store(false)
|
||||
}
|
||||
|
||||
// resyncAge reports how long ago the last successful sync completed. ok is
|
||||
// false when no sync has ever succeeded, which the resync-age metric skips
|
||||
// rather than reporting a misleading zero.
|
||||
func (c *Cache) resyncAge() (time.Duration, bool) {
|
||||
n := c.lastResync.Load()
|
||||
if n == 0 {
|
||||
return 0, false
|
||||
}
|
||||
return time.Since(time.Unix(0, n)), true
|
||||
}
|
||||
|
||||
// entriesByState groups the live map by raw instance state, for the
|
||||
// entries-by-state gauge.
|
||||
func (c *Cache) entriesByState() map[string]int64 {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
out := make(map[string]int64, len(c.entries))
|
||||
for _, v := range c.entries {
|
||||
out[string(v.Status)]++
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// sleep waits out the retry interval, reporting false if ctx ended first.
|
||||
func (c *Cache) sleep(ctx context.Context) bool {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-time.After(c.retryInterval):
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// syncFailed records a failed sync attempt: the metric, the degraded flag,
|
||||
// and a warning log, shared by every sync path so a failure looks the same
|
||||
// regardless of which one hit it.
|
||||
func (c *Cache) syncFailed(ctx context.Context, err error) {
|
||||
c.metrics.resyncFailed()
|
||||
c.degraded.Store(true)
|
||||
slog.WarnContext(ctx, "instancecache: sync failed", "err", err)
|
||||
}
|
||||
|
||||
// drain applies every event lw's watcher delivers, live, once the sync that
|
||||
// installed lw has already caught it up: a channel that closes while lw's
|
||||
// context is still live means the connection dropped out from under it,
|
||||
// which is the signal Run replaces the watcher on; a channel that closes
|
||||
// because the context was cancelled is an intentional teardown and reports
|
||||
// nothing. An event delivered after lw has been superseded as the active
|
||||
// watcher is dropped rather than applied: the watcher that replaced it is
|
||||
// independently live-subscribed and already covers it, so nothing is lost,
|
||||
// and this keeps it true that only the active watcher ever writes to the
|
||||
// live map.
|
||||
func (c *Cache) drain(lw *liveWatcher) {
|
||||
defer close(lw.done)
|
||||
for {
|
||||
select {
|
||||
case <-lw.ctx.Done():
|
||||
return
|
||||
case entry, ok := <-lw.kw.Updates():
|
||||
if !ok {
|
||||
if lw.ctx.Err() == nil && c.getActive() == lw {
|
||||
select {
|
||||
case c.watcherLost <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
if entry == nil {
|
||||
continue
|
||||
}
|
||||
if c.getActive() != lw {
|
||||
continue
|
||||
}
|
||||
c.applyLive(entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) applyLive(entry jetstream.KeyValueEntry) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.apply(c.entries, c.index, entry)
|
||||
}
|
||||
|
||||
// apply decodes one watch entry into entries/index. A put that will not
|
||||
// decode leaves whatever is already at that key untouched and counts a
|
||||
// decode failure; a delete or purge always removes the key. Absence is
|
||||
// established only by the record space saying so, never by a decode failure.
|
||||
func (c *Cache) apply(entries map[string]*vm.VM, index map[string]map[string]struct{}, entry jetstream.KeyValueEntry) {
|
||||
id := strings.TrimPrefix(entry.Key(), c.prefix)
|
||||
switch entry.Operation() {
|
||||
case jetstream.KeyValueDelete, jetstream.KeyValuePurge:
|
||||
removeFrom(entries, index, id)
|
||||
default:
|
||||
var rec vm.InstanceRecord
|
||||
if err := json.Unmarshal(entry.Value(), &rec); err != nil {
|
||||
c.metrics.decodeFailure()
|
||||
slog.Warn("instancecache: undecodable record, keeping previous value",
|
||||
"key", entry.Key(), "err", err)
|
||||
return
|
||||
}
|
||||
putInto(entries, index, id, vm.VMFromRecord(&rec))
|
||||
}
|
||||
}
|
||||
|
||||
// snapshotCandidate reads the whole record space through one consumer,
|
||||
// mutually consistent per kvstore.Store.Snapshot, and builds a fresh
|
||||
// entries/index pair from it.
|
||||
func (c *Cache) snapshotCandidate(ctx context.Context) (map[string]*vm.VM, map[string]map[string]struct{}, uint64, error) {
|
||||
items, highWater, err := c.store.Snapshot(ctx, c.filter)
|
||||
if err != nil {
|
||||
return nil, nil, 0, err
|
||||
}
|
||||
entries := make(map[string]*vm.VM, len(items))
|
||||
index := make(map[string]map[string]struct{})
|
||||
for i := range items {
|
||||
id := strings.TrimPrefix(items[i].Key, c.prefix)
|
||||
putInto(entries, index, id, vm.VMFromRecord(&items[i].Value))
|
||||
}
|
||||
return entries, index, highWater, nil
|
||||
}
|
||||
|
||||
// replay applies every event kw delivers onto entries/index until the
|
||||
// end-of-replay marker (nil) arrives: the candidate then reflects everything
|
||||
// between the snapshot's high-water mark and the moment it caught up, which
|
||||
// is the whole fence — no buffer, no mode, just sequencing.
|
||||
func (c *Cache) replay(ctx context.Context, kw jetstream.KeyWatcher, entries map[string]*vm.VM, index map[string]map[string]struct{}) (int, error) {
|
||||
applied := 0
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return applied, ctx.Err()
|
||||
case entry, ok := <-kw.Updates():
|
||||
if !ok {
|
||||
return applied, errors.New("instancecache: watcher closed before catching up")
|
||||
}
|
||||
if entry == nil {
|
||||
return applied, nil
|
||||
}
|
||||
c.apply(entries, index, entry)
|
||||
applied++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// syncOnce runs the whole build algorithm once: snapshot the record space,
|
||||
// open a watcher that resumes from just past the snapshot's high-water mark,
|
||||
// and replay onto the candidate until it catches up. It touches neither the
|
||||
// live map nor the active watcher — installSync does that, and only once an
|
||||
// attempt fully succeeds, so a failed attempt has cost nothing.
|
||||
func (c *Cache) syncOnce(ctx context.Context) (*liveWatcher, map[string]*vm.VM, map[string]map[string]struct{}, error) {
|
||||
entries, index, hw, err := c.snapshotCandidate(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
if c.postSnapshotHook != nil {
|
||||
c.postSnapshotHook()
|
||||
}
|
||||
|
||||
watchCtx, cancel := context.WithCancel(ctx)
|
||||
kw, err := c.store.WatchFrom(watchCtx, c.filter, hw+1)
|
||||
if err != nil {
|
||||
cancel()
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
|
||||
replayed, err := c.replay(watchCtx, kw, entries, index)
|
||||
if err != nil {
|
||||
cancel()
|
||||
_ = kw.Stop()
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
c.metrics.replayApplied(replayed)
|
||||
slog.Debug("instancecache: sync caught up", "snapshot_entries", len(entries),
|
||||
"high_water", hw, "replayed", replayed)
|
||||
|
||||
return &liveWatcher{kw: kw, ctx: watchCtx, cancel: cancel, done: make(chan struct{})}, entries, index, nil
|
||||
}
|
||||
|
||||
// installSync swaps entries/index in as the live map and lw in as the active
|
||||
// watcher together, starts lw's live-apply goroutine, then stops and joins
|
||||
// oldLw. In that order, so there is never a moment where two watchers both
|
||||
// apply to the live map: oldLw, while it was still active, only ever applied
|
||||
// to the map installSync is now replacing, and it stops being active in the
|
||||
// same step that replacement happens.
|
||||
func (c *Cache) installSync(lw *liveWatcher, entries map[string]*vm.VM, index map[string]map[string]struct{}, oldLw *liveWatcher) {
|
||||
c.mu.Lock()
|
||||
c.entries, c.index = entries, index
|
||||
c.ready = true
|
||||
c.mu.Unlock()
|
||||
c.setActive(lw)
|
||||
go c.drain(lw)
|
||||
|
||||
if oldLw != nil {
|
||||
oldLw.cancel()
|
||||
_ = oldLw.kw.Stop()
|
||||
<-oldLw.done
|
||||
}
|
||||
}
|
||||
|
||||
// resyncUntilSuccess retries syncOnce against the retry interval until it
|
||||
// succeeds or ctx ends, installing the result over oldLw. Used by the
|
||||
// initial sync and watcher replacement, which both must block until a live
|
||||
// watcher exists again; a periodic resync tries only once per tick instead.
|
||||
func (c *Cache) resyncUntilSuccess(ctx context.Context, oldLw *liveWatcher) *liveWatcher {
|
||||
for {
|
||||
lw, entries, index, err := c.syncOnce(ctx)
|
||||
if err != nil {
|
||||
c.syncFailed(ctx, err)
|
||||
if !c.sleep(ctx) {
|
||||
return nil
|
||||
}
|
||||
continue
|
||||
}
|
||||
c.installSync(lw, entries, index, oldLw)
|
||||
c.markResynced()
|
||||
return lw
|
||||
}
|
||||
}
|
||||
|
||||
// initialSync is resyncUntilSuccess with no old watcher to fence against,
|
||||
// plus making the result the active watcher and recording the first
|
||||
// successful resync.
|
||||
func (c *Cache) initialSync(ctx context.Context) *liveWatcher {
|
||||
return c.resyncUntilSuccess(ctx, nil)
|
||||
}
|
||||
|
||||
// periodicResync tries the build algorithm once per tick. The active watcher
|
||||
// keeps applying to the live map for the whole attempt, since installSync is
|
||||
// the only thing that ever changes what "the live map" is, so a failure
|
||||
// costs nothing but a discarded candidate; the next tick tries again rather
|
||||
// than retrying in a loop here.
|
||||
func (c *Cache) periodicResync(ctx context.Context) {
|
||||
oldLw := c.getActive()
|
||||
if oldLw == nil {
|
||||
return
|
||||
}
|
||||
lw, entries, index, err := c.syncOnce(ctx)
|
||||
if err != nil {
|
||||
c.syncFailed(ctx, err)
|
||||
return
|
||||
}
|
||||
c.installSync(lw, entries, index, oldLw)
|
||||
c.markResynced()
|
||||
}
|
||||
|
||||
// replaceWatcher retires a dead watcher and installs a fresh one under the
|
||||
// same build algorithm, retrying against the retry interval until it
|
||||
// succeeds or ctx ends.
|
||||
func (c *Cache) replaceWatcher(ctx context.Context, oldLw *liveWatcher) {
|
||||
lw := c.resyncUntilSuccess(ctx, oldLw)
|
||||
if lw != nil {
|
||||
c.metrics.watchReconnected()
|
||||
}
|
||||
}
|
||||
|
||||
// indexKeyFor is the account index's bucket for v's owner: Global for an
|
||||
// empty owner, so a legacy pre-account record files under the same bucket a
|
||||
// platform-managed record does, and never under every account at once.
|
||||
func indexKeyFor(accountID string) string {
|
||||
if accountID == "" {
|
||||
return utils.GlobalAccountID
|
||||
}
|
||||
return accountID
|
||||
}
|
||||
|
||||
func putInto(entries map[string]*vm.VM, index map[string]map[string]struct{}, id string, v *vm.VM) {
|
||||
if old, ok := entries[id]; ok {
|
||||
removeFromIndex(index, indexKeyFor(old.AccountID), id)
|
||||
}
|
||||
entries[id] = v
|
||||
addToIndex(index, indexKeyFor(v.AccountID), id)
|
||||
}
|
||||
|
||||
func removeFrom(entries map[string]*vm.VM, index map[string]map[string]struct{}, id string) {
|
||||
old, ok := entries[id]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(entries, id)
|
||||
removeFromIndex(index, indexKeyFor(old.AccountID), id)
|
||||
}
|
||||
|
||||
func addToIndex(index map[string]map[string]struct{}, key, id string) {
|
||||
set, ok := index[key]
|
||||
if !ok {
|
||||
set = make(map[string]struct{})
|
||||
index[key] = set
|
||||
}
|
||||
set[id] = struct{}{}
|
||||
}
|
||||
|
||||
func removeFromIndex(index map[string]map[string]struct{}, key, id string) {
|
||||
set, ok := index[key]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(set, id)
|
||||
if len(set) == 0 {
|
||||
delete(index, key)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,637 @@
|
||||
// internal watcher and lifecycle state directly (there is no exported
|
||||
// surface for "one live watcher" or "the candidate before install"), and the
|
||||
// account-index defence-in-depth test deliberately mis-files a record by
|
||||
// writing straight into the unexported maps.
|
||||
//
|
||||
//test:in-package — the fencing and lifecycle tests assert on the watcher's
|
||||
package instancecache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/resource"
|
||||
"github.com/mulgadc/spinifex/spinifex/testutil"
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/mulgadc/spinifex/spinifex/vm"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.opentelemetry.io/otel"
|
||||
sdkmetric "go.opentelemetry.io/otel/sdk/metric"
|
||||
"go.opentelemetry.io/otel/sdk/metric/metricdata"
|
||||
)
|
||||
|
||||
const (
|
||||
testPrefix = "i."
|
||||
acctA = "111122223333"
|
||||
acctB = "444455556666"
|
||||
)
|
||||
|
||||
// newTestCache starts an embedded JetStream server, opens a KV bucket for it
|
||||
// and returns a Cache over that bucket plus the raw handles a test needs to
|
||||
// drive the underlying key space directly.
|
||||
func newTestCache(t *testing.T, mutate ...func(*Config)) (*Cache, jetstream.KeyValue, *cacheConn) {
|
||||
t.Helper()
|
||||
_, nc, js := testutil.StartTestJetStream(t)
|
||||
|
||||
bucket := fmt.Sprintf("instances-%d", time.Now().UnixNano())
|
||||
kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: bucket, History: 1})
|
||||
require.NoError(t, err)
|
||||
|
||||
cfg := Config{
|
||||
Bucket: kvstore.Config{Name: bucket, History: 1, Replicas: 1},
|
||||
Prefix: testPrefix,
|
||||
RetryInterval: 20 * time.Millisecond,
|
||||
}
|
||||
for _, m := range mutate {
|
||||
m(&cfg)
|
||||
}
|
||||
return New(js, cfg), kv, &cacheConn{nc: nc, js: js}
|
||||
}
|
||||
|
||||
// cacheConn bundles the connection handles a test occasionally needs beyond
|
||||
// the Cache and its bucket, named to avoid importing nats.go just for a type.
|
||||
type cacheConn struct {
|
||||
nc interface{ NumSubscriptions() int }
|
||||
js jetstream.JetStream
|
||||
}
|
||||
|
||||
func testRecord(id, accountID string, state vm.InstanceState) *vm.InstanceRecord {
|
||||
return &vm.InstanceRecord{
|
||||
Metadata: resource.Metadata{Name: id, AccountID: accountID},
|
||||
Spec: vm.InstanceSpec{InstanceType: "m7i.large"},
|
||||
Status: vm.InstanceStatus{Status: state, LastNode: "node-1"},
|
||||
}
|
||||
}
|
||||
|
||||
func putRecord(t *testing.T, kv jetstream.KeyValue, id string, rec *vm.InstanceRecord) uint64 {
|
||||
t.Helper()
|
||||
data, err := json.Marshal(rec)
|
||||
require.NoError(t, err)
|
||||
rev, err := kv.Put(context.Background(), testPrefix+id, data)
|
||||
require.NoError(t, err)
|
||||
return rev
|
||||
}
|
||||
|
||||
func putGarbage(t *testing.T, kv jetstream.KeyValue, id string) uint64 {
|
||||
t.Helper()
|
||||
rev, err := kv.Put(context.Background(), testPrefix+id, []byte("not-json"))
|
||||
require.NoError(t, err)
|
||||
return rev
|
||||
}
|
||||
|
||||
func idsOf(list []*vm.VM) []string {
|
||||
out := make([]string, len(list))
|
||||
for i, v := range list {
|
||||
out[i] = v.ID
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func waitFor(t *testing.T, timeout time.Duration, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("condition not met within %s", timeout)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func waitForReady(t *testing.T, c *Cache) {
|
||||
t.Helper()
|
||||
waitFor(t, 2*time.Second, c.Ready)
|
||||
}
|
||||
|
||||
func waitForListLen(t *testing.T, c *Cache, accountID string, n int) []*vm.VM {
|
||||
t.Helper()
|
||||
var list []*vm.VM
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
list, _ = c.List(context.Background(), accountID)
|
||||
return len(list) == n
|
||||
})
|
||||
return list
|
||||
}
|
||||
|
||||
// Process-wide manual metric reader, installed once per test binary so the
|
||||
// resync-failure and decode-failure counters can be observed. Cumulative for
|
||||
// the life of the binary, so tests read a before/after delta.
|
||||
var (
|
||||
metricsReaderOnce sync.Once
|
||||
metricsReader *sdkmetric.ManualReader
|
||||
)
|
||||
|
||||
func installMetricsReader(t *testing.T) {
|
||||
t.Helper()
|
||||
metricsReaderOnce.Do(func() {
|
||||
metricsReader = sdkmetric.NewManualReader()
|
||||
otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(metricsReader)))
|
||||
})
|
||||
}
|
||||
|
||||
func counterSum(t *testing.T, name string) int64 {
|
||||
t.Helper()
|
||||
var rm metricdata.ResourceMetrics
|
||||
require.NoError(t, metricsReader.Collect(context.Background(), &rm))
|
||||
var total int64
|
||||
for _, sm := range rm.ScopeMetrics {
|
||||
for _, m := range sm.Metrics {
|
||||
if m.Name != name {
|
||||
continue
|
||||
}
|
||||
sum, ok := m.Data.(metricdata.Sum[int64])
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
for _, dp := range sum.DataPoints {
|
||||
total += dp.Value
|
||||
}
|
||||
}
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
// --- Unit, in instancecache ------------------------------------------------
|
||||
|
||||
func TestList_PutAppearsAfterWatchDelivers(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
|
||||
list := waitForListLen(t, c, acctA, 1)
|
||||
require.Equal(t, "i-1", list[0].ID)
|
||||
}
|
||||
|
||||
func TestList_DeleteAndPurgeRemove(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
putRecord(t, kv, "i-2", testRecord("i-2", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 2)
|
||||
|
||||
require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-1"))
|
||||
list := waitForListLen(t, c, acctA, 1)
|
||||
require.Equal(t, "i-2", list[0].ID)
|
||||
|
||||
require.NoError(t, kv.Purge(context.Background(), testPrefix+"i-2"))
|
||||
waitForListLen(t, c, acctA, 0)
|
||||
}
|
||||
|
||||
func TestPeriodicResync_RemovesEntryTheWatchNeverToldItAbout(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-real", testRecord("i-real", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
// Simulate exactly what a per-key queue cannot cover: a key is gone from
|
||||
// the record space, but nothing was watching when it went, so the live
|
||||
// map still shows it.
|
||||
c.mu.Lock()
|
||||
putInto(c.entries, c.index, "i-ghost", vm.VMFromRecord(testRecord("i-ghost", acctA, vm.StateRunning)))
|
||||
c.mu.Unlock()
|
||||
waitForListLen(t, c, acctA, 2)
|
||||
|
||||
c.periodicResync(ctx)
|
||||
|
||||
list := waitForListLen(t, c, acctA, 1)
|
||||
require.Equal(t, "i-real", list[0].ID)
|
||||
}
|
||||
|
||||
func TestSync_LeaksNothingOnRepeatedSnapshotFailure(t *testing.T) {
|
||||
c, kv, conn := newTestCache(t)
|
||||
putGarbage(t, kv, "i-bad")
|
||||
|
||||
ctx := t.Context()
|
||||
baseline := conn.nc.NumSubscriptions()
|
||||
go c.Run(ctx)
|
||||
|
||||
// The initial sync's snapshot fails on the garbage record every retry, so
|
||||
// the replay watcher is never opened: nothing persists to leak. Snapshot
|
||||
// itself briefly opens and closes its own probe watcher on every attempt,
|
||||
// so the assertion is that the count always settles back to baseline,
|
||||
// not that it never blips.
|
||||
for range 10 {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
waitFor(t, time.Second, func() bool { return conn.nc.NumSubscriptions() == baseline })
|
||||
}
|
||||
require.False(t, c.Ready())
|
||||
}
|
||||
|
||||
func TestPeriodicResync_TTLExpiredRecordRemovedWithNoWatchEvent(t *testing.T) {
|
||||
_, _, js := testutil.StartTestJetStream(t)
|
||||
bucket := fmt.Sprintf("instances-ttl-%d", time.Now().UnixNano())
|
||||
kv, err := js.CreateKeyValue(context.Background(),
|
||||
jetstream.KeyValueConfig{Bucket: bucket, History: 1, TTL: 200 * time.Millisecond})
|
||||
require.NoError(t, err)
|
||||
|
||||
c := New(js, Config{
|
||||
Bucket: kvstore.Config{Name: bucket, History: 1, Replicas: 1},
|
||||
Prefix: testPrefix,
|
||||
RetryInterval: 20 * time.Millisecond,
|
||||
})
|
||||
putRecord(t, kv, "i-ttl", testRecord("i-ttl", acctA, vm.StateTerminated))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
time.Sleep(500 * time.Millisecond) // past the bucket TTL; no delete event is ever delivered
|
||||
|
||||
c.mu.RLock()
|
||||
_, stillThere := c.entries["i-ttl"]
|
||||
c.mu.RUnlock()
|
||||
require.True(t, stillThere, "the watch alone must not have removed a TTL-expired key")
|
||||
|
||||
c.periodicResync(ctx)
|
||||
waitForListLen(t, c, acctA, 0)
|
||||
}
|
||||
|
||||
func TestPeriodicResync_UndecodableRecordMarksDegraded(t *testing.T) {
|
||||
installMetricsReader(t)
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
before := counterSum(t, "mulga.instancecache.resync_failures")
|
||||
beforeAge, hadAge := c.resyncAge()
|
||||
require.True(t, hadAge)
|
||||
|
||||
putGarbage(t, kv, "i-bad")
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
c.periodicResync(ctx)
|
||||
|
||||
require.Equal(t, before+1, counterSum(t, "mulga.instancecache.resync_failures"))
|
||||
require.True(t, c.Degraded(), "a failed resync must mark the cache degraded")
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Len(t, list, 1)
|
||||
require.Equal(t, "i-1", list[0].ID)
|
||||
|
||||
afterAge, hadAge2 := c.resyncAge()
|
||||
require.True(t, hadAge2)
|
||||
require.GreaterOrEqual(t, afterAge, beforeAge, "a failed resync must not refresh the last-successful-resync mark")
|
||||
}
|
||||
|
||||
func TestDegraded_ClearedByNextSuccessfulResync(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
require.False(t, c.Degraded(), "a cache that has never failed a sync must not report degraded")
|
||||
|
||||
putGarbage(t, kv, "i-bad")
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
c.periodicResync(ctx)
|
||||
require.True(t, c.Degraded())
|
||||
|
||||
require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-bad"))
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
c.periodicResync(ctx)
|
||||
require.False(t, c.Degraded(), "the next successful resync must clear degraded")
|
||||
}
|
||||
|
||||
func TestWatch_UndecodableUpdateLeavesExistingEntry(t *testing.T) {
|
||||
installMetricsReader(t)
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
before := counterSum(t, "mulga.instancecache.decode_failures")
|
||||
putGarbage(t, kv, "i-1")
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
return counterSum(t, "mulga.instancecache.decode_failures") > before
|
||||
})
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Len(t, list, 1)
|
||||
require.Equal(t, vm.StateRunning, list[0].Status,
|
||||
"an undecodable watch update must not remove or corrupt the existing entry")
|
||||
}
|
||||
|
||||
func TestList_NotReadyBeforeFirstSync(t *testing.T) {
|
||||
c, _, _ := newTestCache(t)
|
||||
list, ready := c.List(context.Background(), acctA)
|
||||
require.False(t, ready)
|
||||
require.Empty(t, list)
|
||||
}
|
||||
|
||||
// --- Fencing: replay closes the snapshot-to-watch window --------------------
|
||||
//
|
||||
// The window the fence has to cover — after Snapshot has already read the
|
||||
// record space, before the resuming watcher opens — is internal to syncOnce,
|
||||
// so postSnapshotHook is the seam that lands a write inside it. Each test
|
||||
// drives the real production path (syncOnce/installSync or periodicResync)
|
||||
// rather than reimplementing the window's steps itself, so a wrong revision
|
||||
// or a dropped WatchFrom call in production is what these catch, not
|
||||
// whatever a test independently recomputes.
|
||||
|
||||
func TestFence_PutDuringInitialSnapshot_SurvivesInstall(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning))
|
||||
|
||||
c.postSnapshotHook = func() {
|
||||
putRecord(t, kv, "i-new", testRecord("i-new", acctA, vm.StateRunning))
|
||||
}
|
||||
|
||||
lw, entries, index, err := c.syncOnce(context.Background())
|
||||
require.NoError(t, err)
|
||||
c.installSync(lw, entries, index, nil)
|
||||
|
||||
list, ready := c.List(context.Background(), acctA)
|
||||
require.True(t, ready)
|
||||
require.ElementsMatch(t, []string{"i-seed", "i-new"}, idsOf(list),
|
||||
"a put landing after the snapshot but before the watcher opened must be replayed")
|
||||
}
|
||||
|
||||
func TestFence_DeleteDuringInitialSnapshot_RemovedFromInstall(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning))
|
||||
putRecord(t, kv, "i-doomed", testRecord("i-doomed", acctA, vm.StateRunning))
|
||||
|
||||
c.postSnapshotHook = func() {
|
||||
require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-doomed"))
|
||||
}
|
||||
|
||||
lw, entries, index, err := c.syncOnce(context.Background())
|
||||
require.NoError(t, err)
|
||||
c.installSync(lw, entries, index, nil)
|
||||
|
||||
list, ready := c.List(context.Background(), acctA)
|
||||
require.True(t, ready)
|
||||
require.Equal(t, []string{"i-seed"}, idsOf(list),
|
||||
"a delete landing after the snapshot but before the watcher opened must not survive as the snapshot's stale copy")
|
||||
}
|
||||
|
||||
func TestFence_PutDuringPeriodicResync_SurvivesInstall(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
c.postSnapshotHook = func() {
|
||||
putRecord(t, kv, "i-new", testRecord("i-new", acctA, vm.StateRunning))
|
||||
}
|
||||
c.periodicResync(ctx)
|
||||
c.postSnapshotHook = nil
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.ElementsMatch(t, []string{"i-seed", "i-new"}, idsOf(list),
|
||||
"a put landing after a periodic resync's snapshot but before the new watcher opened must be replayed")
|
||||
}
|
||||
|
||||
func TestFence_DeleteDuringPeriodicResync_RemovedFromInstall(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning))
|
||||
putRecord(t, kv, "i-doomed", testRecord("i-doomed", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 2)
|
||||
|
||||
c.postSnapshotHook = func() {
|
||||
require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-doomed"))
|
||||
}
|
||||
c.periodicResync(ctx)
|
||||
c.postSnapshotHook = nil
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Equal(t, []string{"i-seed"}, idsOf(list),
|
||||
"a delete landing after a periodic resync's snapshot but before the new watcher opened must not be overwritten by the snapshot's stale copy")
|
||||
}
|
||||
|
||||
// --- Lifecycle ---------------------------------------------------------------
|
||||
|
||||
func TestPeriodicResync_FailureLeavesLiveWatcherAndMapServing(t *testing.T) {
|
||||
c, kv, conn := newTestCache(t)
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
baseline := conn.nc.NumSubscriptions()
|
||||
putGarbage(t, kv, "i-bad")
|
||||
|
||||
for range 10 {
|
||||
c.periodicResync(ctx)
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Len(t, list, 1)
|
||||
require.Equal(t, "i-1", list[0].ID)
|
||||
require.Equal(t, baseline, conn.nc.NumSubscriptions(),
|
||||
"a failed periodic resync must not leave a stray subscription behind")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPeriodicResync_FailureLosesNothing(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
putGarbage(t, kv, "i-bad") // every future resync now fails against the same bad key
|
||||
|
||||
for i := range 10 {
|
||||
id := fmt.Sprintf("i-live-%d", i)
|
||||
putRecord(t, kv, id, testRecord(id, acctA, vm.StateRunning))
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
list, _ := c.List(ctx, acctA)
|
||||
for _, v := range list {
|
||||
if v.ID == id {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
|
||||
c.periodicResync(ctx)
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Contains(t, idsOf(list), id)
|
||||
require.Contains(t, idsOf(list), "i-seed")
|
||||
}
|
||||
|
||||
require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-seed"))
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
list, _ := c.List(ctx, acctA)
|
||||
for _, v := range list {
|
||||
if v.ID == "i-seed" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
func TestPeriodicResync_TwentyConsecutiveSuccessesOneWatcherEach(t *testing.T) {
|
||||
c, kv, conn := newTestCache(t)
|
||||
putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
baseline := conn.nc.NumSubscriptions()
|
||||
|
||||
// Each periodic resync opens a fresh watcher and retires the last one, so
|
||||
// the subscription count stays flat even though the watcher identity
|
||||
// changes every time.
|
||||
prev := c.getActive()
|
||||
for range 20 {
|
||||
c.periodicResync(ctx)
|
||||
require.Equal(t, baseline, conn.nc.NumSubscriptions())
|
||||
cur := c.getActive()
|
||||
require.NotSame(t, prev, cur, "each successful periodic resync installs a new watcher")
|
||||
prev = cur
|
||||
}
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Equal(t, []string{"i-seed"}, idsOf(list))
|
||||
}
|
||||
|
||||
func TestReplaceWatcher_OldStoppedAndJoinedBeforeNewGoesLive(t *testing.T) {
|
||||
c, kv, conn := newTestCache(t)
|
||||
putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
|
||||
oldLw := c.getActive()
|
||||
baseline := conn.nc.NumSubscriptions()
|
||||
|
||||
require.NoError(t, oldLw.kw.Stop()) // simulate the connection dropping this watcher
|
||||
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
select {
|
||||
case <-oldLw.done:
|
||||
default:
|
||||
return false
|
||||
}
|
||||
next := c.getActive()
|
||||
return next != nil && next != oldLw
|
||||
})
|
||||
|
||||
select {
|
||||
case <-oldLw.done:
|
||||
default:
|
||||
t.Fatal("old watcher must be joined once replacement completes")
|
||||
}
|
||||
|
||||
waitFor(t, 2*time.Second, func() bool { return conn.nc.NumSubscriptions() == baseline })
|
||||
|
||||
list, ready := c.List(ctx, acctA)
|
||||
require.True(t, ready)
|
||||
require.Len(t, list, 1)
|
||||
require.Equal(t, "i-1", list[0].ID)
|
||||
|
||||
putRecord(t, kv, "i-2", testRecord("i-2", acctA, vm.StateRunning))
|
||||
waitForListLen(t, c, acctA, 2)
|
||||
}
|
||||
|
||||
// --- The account index ------------------------------------------------------
|
||||
|
||||
func TestList_ReturnsOnlyThatAccountsInstances(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-a", testRecord("i-a", acctA, vm.StateRunning))
|
||||
putRecord(t, kv, "i-b", testRecord("i-b", acctB, vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitForListLen(t, c, acctA, 1)
|
||||
waitForListLen(t, c, acctB, 1)
|
||||
|
||||
listA, _ := c.List(ctx, acctA)
|
||||
require.Equal(t, []string{"i-a"}, idsOf(listA))
|
||||
|
||||
listB, _ := c.List(ctx, acctB)
|
||||
require.Equal(t, []string{"i-b"}, idsOf(listB))
|
||||
}
|
||||
|
||||
func TestList_GlobalDoesNotSeeOtherAccountsInstances(t *testing.T) {
|
||||
c, kv, _ := newTestCache(t)
|
||||
putRecord(t, kv, "i-cust", testRecord("i-cust", acctA, vm.StateRunning))
|
||||
putRecord(t, kv, "i-global", testRecord("i-global", utils.GlobalAccountID, vm.StateRunning))
|
||||
putRecord(t, kv, "i-legacy", testRecord("i-legacy", "", vm.StateRunning))
|
||||
|
||||
ctx := t.Context()
|
||||
go c.Run(ctx)
|
||||
waitForReady(t, c)
|
||||
waitFor(t, 2*time.Second, func() bool {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return len(c.entries) == 3
|
||||
})
|
||||
|
||||
listGlobal, ready := c.List(ctx, utils.GlobalAccountID)
|
||||
require.True(t, ready)
|
||||
require.ElementsMatch(t, []string{"i-global", "i-legacy"}, idsOf(listGlobal))
|
||||
|
||||
listCust, _ := c.List(ctx, acctA)
|
||||
require.Equal(t, []string{"i-cust"}, idsOf(listCust))
|
||||
}
|
||||
|
||||
func TestList_VisibilityAppliedAfterIndexLookup(t *testing.T) {
|
||||
c, _, _ := newTestCache(t)
|
||||
|
||||
// Deliberately mis-file: the index says i-misfiled belongs to acctA, but
|
||||
// the record itself is owned by acctB. IsInstanceVisibleToCaller must
|
||||
// still reject it at the List seam.
|
||||
c.mu.Lock()
|
||||
c.entries["i-misfiled"] = vm.VMFromRecord(testRecord("i-misfiled", acctB, vm.StateRunning))
|
||||
addToIndex(c.index, acctA, "i-misfiled")
|
||||
c.ready = true
|
||||
c.mu.Unlock()
|
||||
|
||||
list, ready := c.List(context.Background(), acctA)
|
||||
require.True(t, ready)
|
||||
require.Empty(t, list)
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
package instancecache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/metric"
|
||||
)
|
||||
|
||||
// meterName is the instrumentation scope for every instancecache metric.
|
||||
const meterName = "github.com/mulgadc/spinifex/spinifex/instancecache"
|
||||
|
||||
// cacheMetrics holds one Cache's instruments. Registered per instance, not as
|
||||
// a package-level singleton, because the gauges read that instance's own
|
||||
// state through a callback closure.
|
||||
type cacheMetrics struct {
|
||||
resyncFailures metric.Int64Counter
|
||||
watchReconnects metric.Int64Counter
|
||||
decodeFailures metric.Int64Counter
|
||||
replayEvents metric.Int64Counter
|
||||
}
|
||||
|
||||
// newCacheMetrics registers c's instruments against the global meter
|
||||
// provider. A registration failure is handled by otel.Handle and leaves the
|
||||
// affected instrument nil; every recorder below tolerates a nil instrument.
|
||||
func newCacheMetrics(c *Cache) *cacheMetrics {
|
||||
meter := otel.Meter(meterName)
|
||||
m := &cacheMetrics{}
|
||||
|
||||
var err error
|
||||
m.resyncFailures, err = meter.Int64Counter("mulga.instancecache.resync_failures",
|
||||
metric.WithDescription("Count of failed initial, periodic or watcher-replacement syncs."),
|
||||
metric.WithUnit("{failure}"))
|
||||
if err != nil {
|
||||
otel.Handle(err)
|
||||
}
|
||||
m.watchReconnects, err = meter.Int64Counter("mulga.instancecache.watch_reconnects",
|
||||
metric.WithDescription("Count of times the KV watch was replaced after dying."),
|
||||
metric.WithUnit("{reconnect}"))
|
||||
if err != nil {
|
||||
otel.Handle(err)
|
||||
}
|
||||
m.replayEvents, err = meter.Int64Counter("mulga.instancecache.replay_events",
|
||||
metric.WithDescription("Count of records the snapshot missed and the resumed watch replayed before the cache went live."),
|
||||
metric.WithUnit("{record}"))
|
||||
if err != nil {
|
||||
otel.Handle(err)
|
||||
}
|
||||
m.decodeFailures, err = meter.Int64Counter("mulga.instancecache.decode_failures",
|
||||
metric.WithDescription("Count of instance records that failed to decode. The offending key is logged, not attributed, to bound cardinality."),
|
||||
metric.WithUnit("{failure}"))
|
||||
if err != nil {
|
||||
otel.Handle(err)
|
||||
}
|
||||
|
||||
if _, err := meter.Int64ObservableGauge("mulga.instancecache.entries",
|
||||
metric.WithDescription("Cached instance count by raw instance state."),
|
||||
metric.WithUnit("{instance}"),
|
||||
metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error {
|
||||
for state, n := range c.entriesByState() {
|
||||
o.Observe(n, metric.WithAttributeSet(attribute.NewSet(attribute.String("state", state))))
|
||||
}
|
||||
return nil
|
||||
})); err != nil {
|
||||
otel.Handle(err)
|
||||
}
|
||||
|
||||
if _, err := meter.Float64ObservableGauge("mulga.instancecache.resync_age_seconds",
|
||||
metric.WithDescription("Age of the last successful resync. Absent until the first one completes."),
|
||||
metric.WithUnit("s"),
|
||||
metric.WithFloat64Callback(func(_ context.Context, o metric.Float64Observer) error {
|
||||
if age, ok := c.resyncAge(); ok {
|
||||
o.Observe(age.Seconds())
|
||||
}
|
||||
return nil
|
||||
})); err != nil {
|
||||
otel.Handle(err)
|
||||
}
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) resyncFailed() {
|
||||
if m.resyncFailures == nil {
|
||||
return
|
||||
}
|
||||
m.resyncFailures.Add(context.Background(), 1)
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) watchReconnected() {
|
||||
if m.watchReconnects == nil {
|
||||
return
|
||||
}
|
||||
m.watchReconnects.Add(context.Background(), 1)
|
||||
}
|
||||
|
||||
// replayApplied records how many records the resumed watch had to replay.
|
||||
// A zero count is recorded too: it is the difference between a quiet cluster
|
||||
// and an instrument that never fired.
|
||||
func (m *cacheMetrics) replayApplied(n int) {
|
||||
if m.replayEvents == nil {
|
||||
return
|
||||
}
|
||||
m.replayEvents.Add(context.Background(), int64(n))
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) decodeFailure() {
|
||||
if m.decodeFailures == nil {
|
||||
slog.Warn("instancecache: decode failure counter unavailable")
|
||||
return
|
||||
}
|
||||
m.decodeFailures.Add(context.Background(), 1)
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package instancecache_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/testutil"
|
||||
"github.com/mulgadc/spinifex/spinifex/vm"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestResumeFromRevision_ReplaysWhatTheSnapshotMissed documents the server
|
||||
// behaviour the whole Cache design rests on: a watcher opened AFTER a
|
||||
// snapshot, resuming from just past the snapshot's high-water mark, replays
|
||||
// exactly the events the snapshot missed — even though nothing was watching
|
||||
// while they happened — and nothing the snapshot already covered. If a
|
||||
// nats.go upgrade ever changes this, this test fails loudly and points
|
||||
// straight at the cause, rather than the cache's own tests failing in a way
|
||||
// that looks like a cache bug.
|
||||
func TestResumeFromRevision_ReplaysWhatTheSnapshotMissed(t *testing.T) {
|
||||
_, _, js := testutil.StartTestJetStream(t)
|
||||
ctx := context.Background()
|
||||
|
||||
store := kvstore.New[vm.InstanceRecord](js, kvstore.Config{Name: "resume-bucket", History: 1, Replicas: 1})
|
||||
|
||||
rec := func(id string) *vm.InstanceRecord {
|
||||
r := &vm.InstanceRecord{}
|
||||
r.Metadata.Name = id
|
||||
return r
|
||||
}
|
||||
|
||||
// Three keys exist before the snapshot.
|
||||
for i := range 3 {
|
||||
require.NoError(t, store.Set(ctx, fmt.Sprintf("i.pre-%d", i), rec(fmt.Sprintf("pre-%d", i))))
|
||||
}
|
||||
|
||||
items, highWater, err := store.Snapshot(ctx, "i.*")
|
||||
require.NoError(t, err)
|
||||
require.Len(t, items, 3, "snapshot should see the three pre-existing keys")
|
||||
|
||||
// These land AFTER the snapshot and BEFORE any watcher exists. This is
|
||||
// exactly the window a buffering fence would otherwise have to cover.
|
||||
require.NoError(t, store.Set(ctx, "i.post-1", rec("post-1")))
|
||||
require.NoError(t, store.Delete(ctx, "i.pre-0"))
|
||||
|
||||
// Open the watcher only now, resuming from just past the snapshot.
|
||||
kw, err := js.KeyValue(ctx, "resume-bucket")
|
||||
require.NoError(t, err)
|
||||
w, err := kw.Watch(ctx, "i.*", jetstream.ResumeFromRevision(highWater+1))
|
||||
require.NoError(t, err)
|
||||
defer func() { _ = w.Stop() }()
|
||||
|
||||
seen := map[string]jetstream.KeyValueOp{}
|
||||
timeout := time.After(10 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case e := <-w.Updates():
|
||||
if e == nil {
|
||||
// End-of-replay marker: everything the resume owed us has arrived.
|
||||
require.Equal(t, jetstream.KeyValuePut, seen["i.post-1"],
|
||||
"a put made after the snapshot, before the watcher existed, must be replayed")
|
||||
require.Equal(t, jetstream.KeyValueDelete, seen["i.pre-0"],
|
||||
"a delete made after the snapshot, before the watcher existed, must be replayed")
|
||||
require.NotContains(t, seen, "i.pre-1", "keys already in the snapshot must not be replayed")
|
||||
return
|
||||
}
|
||||
seen[e.Key()] = e.Operation()
|
||||
case <-timeout:
|
||||
t.Fatalf("no end-of-replay marker within 10s; saw %v", seen)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -271,6 +271,30 @@ func (b *Bucket) Watch(ctx context.Context, filter string) (jetstream.KeyWatcher
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// WatchFrom returns a watcher over the keys matching filter that resumes
|
||||
// from revision: it replays every value at or after revision, followed by
|
||||
// the end-of-replay marker, before turning into an ordinary live feed. It
|
||||
// takes no UpdatesOnly, unlike Watch, because the whole point is to replay
|
||||
// what a Snapshot's high-water mark missed — a caller rebuilding a cache
|
||||
// from a snapshot uses this to close that window without a buffering fence.
|
||||
func (b *Bucket) WatchFrom(ctx context.Context, filter string, revision uint64) (jetstream.KeyWatcher, error) {
|
||||
var w jetstream.KeyWatcher
|
||||
// Wrapped inside the closure, so a bucket that could not be opened at all
|
||||
// reports the configured reason rather than a watch failure on top of it.
|
||||
err := b.withKV(ctx, func(kv jetstream.KeyValue) error {
|
||||
var err error
|
||||
w, err = kv.Watch(ctx, filter, jetstream.ResumeFromRevision(revision))
|
||||
if err != nil {
|
||||
return fmt.Errorf("kvstore: watch %s %s: %w", b.cfg.Name, filter, err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, 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 {
|
||||
|
||||
@@ -30,6 +30,7 @@ import (
|
||||
handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector"
|
||||
handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota"
|
||||
handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts"
|
||||
"github.com/mulgadc/spinifex/spinifex/instancecache"
|
||||
"github.com/mulgadc/spinifex/spinifex/kvstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/network/reconcile"
|
||||
"github.com/mulgadc/spinifex/spinifex/objectstore"
|
||||
@@ -570,6 +571,15 @@ func launchService(config *config.ClusterConfig) error {
|
||||
// disabled.
|
||||
go gw.Quota.RunBedrockRPMSync(janitorCtx, js, len(config.Nodes))
|
||||
|
||||
// Instance cache: a read-only informer over the live instance record space,
|
||||
// started now so it is warm well before anything reads it. Nothing consults
|
||||
// it yet; the describe path still serves from the fan-out and KV.
|
||||
instanceCache := instancecache.New(js, instancecache.Config{
|
||||
Bucket: kvstore.Config{Name: daemon.InstanceStateBucket, History: 1, Replicas: len(config.Nodes)},
|
||||
Prefix: daemon.InstanceRecordPrefix,
|
||||
})
|
||||
go instanceCache.Run(janitorCtx)
|
||||
|
||||
handler := gw.SetupRoutes()
|
||||
|
||||
// Load TLS certificate
|
||||
|
||||
Reference in New Issue
Block a user