The runtime can now take ClientGroup, BlockList and DNSZone objects (fancydns.k6n.net/v1alpha1) from a cluster with -kube, from a pod's service account or a kubeconfig alike. internal/kubesource watches them with dynamic informers, converts and validates each object on its own, and publishes an immutable snapshot. A bad object is skipped and keeps its last good version, so one broken edit cannot remove a running policy; an invalid merged config is dropped whole. The last snapshot is saved so a restart without apiserver access still serves. config.Store gains an overlay: Base() is what the YAML says, Get() is base plus cluster objects. Everything that writes the config back (mutateConfig, SetAdminAddr, the settings editor) now starts from Base(), otherwise cluster objects would be copied into the file and shadow their own objects. YAML wins every name collision. putACL and toggleBlocklist refuse to edit cluster-owned objects. BlockList has no path field and refuses file:// URLs so a namespaced object cannot pick files the runtime reads or writes. DNSZone is rendered through the zone writer's own prove-by-reparse step, with the SOA serial taken from metadata.generation so every replica renders identical bytes. qlog: Flush reset the pending slice to [:0], so entries logged while it was still storing the batch were appended over the same backing array. The race detector found it through the new Kubernetes integration test. pending is now set to nil, with a regression test that fails without the change. CRD manifests, RBAC and examples are in deploy/k8s/crds; tests pin the schemas to the Go spec structs so the two cannot drift. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
178 lines
4.8 KiB
Go
178 lines
4.8 KiB
Go
package config
|
|
|
|
import (
|
|
"sync"
|
|
"sync/atomic"
|
|
)
|
|
|
|
// Store is the process-wide, hot-reloadable configuration holder. Readers take
|
|
// a snapshot pointer (cheap, no lock) and writers swap a fully-built *Config
|
|
// in atomically, so a request never observes a half-applied config.
|
|
type Store struct {
|
|
cur atomic.Pointer[Config]
|
|
// base is the configuration as the file (or an API write) defined it. cur
|
|
// is base with the overlay applied; without an overlay the two are the
|
|
// same pointer. Writers that persist back to the file must start from
|
|
// base, or they would write overlay content into the file.
|
|
base atomic.Pointer[Config]
|
|
overlay atomic.Pointer[func(*Config) *Config]
|
|
mu sync.Mutex // serialises writers so LastError/Version stay coherent
|
|
// path is the file the config was loaded from, used by the API to
|
|
// persist mutations.
|
|
path string
|
|
// lastError records why the most recent reload attempt failed, so the
|
|
// health endpoint can report a stale-but-running state.
|
|
lastErr atomic.Pointer[string]
|
|
// generation increments on every successful swap.
|
|
generation atomic.Uint64
|
|
// hooks run after a successful swap (server re-binds listeners etc).
|
|
hooksMu sync.RWMutex
|
|
hooks []func(*Config)
|
|
}
|
|
|
|
// NewStore wraps cfg.
|
|
func NewStore(cfg *Config, path string) *Store {
|
|
s := &Store{path: path}
|
|
s.cur.Store(cfg)
|
|
s.base.Store(cfg)
|
|
return s
|
|
}
|
|
|
|
// Base returns the configuration without the overlay: what the file says.
|
|
func (s *Store) Base() *Config {
|
|
if b := s.base.Load(); b != nil {
|
|
return b
|
|
}
|
|
return s.cur.Load()
|
|
}
|
|
|
|
// SetOverlay installs fn, which derives the effective configuration from the
|
|
// base (a cluster source adding groups and lists, say). fn must not modify its
|
|
// argument. Call Reapply afterwards, and again whenever fn's inputs change.
|
|
func (s *Store) SetOverlay(fn func(*Config) *Config) {
|
|
if fn == nil {
|
|
s.overlay.Store(nil)
|
|
return
|
|
}
|
|
s.overlay.Store(&fn)
|
|
}
|
|
|
|
func (s *Store) effective(base *Config) *Config {
|
|
if fn := s.overlay.Load(); fn != nil {
|
|
if out := (*fn)(base); out != nil {
|
|
return out
|
|
}
|
|
}
|
|
return base
|
|
}
|
|
|
|
// Reapply recomputes the effective configuration from the base and swaps it
|
|
// in, running the hooks as for any other swap.
|
|
func (s *Store) Reapply() {
|
|
s.mu.Lock()
|
|
next := s.effective(s.Base())
|
|
s.cur.Store(next)
|
|
s.generation.Add(1)
|
|
s.lastErr.Store(nil)
|
|
s.mu.Unlock()
|
|
s.runHooks(next)
|
|
}
|
|
|
|
// Get returns the current configuration snapshot.
|
|
func (s *Store) Get() *Config { return s.cur.Load() }
|
|
|
|
// Path is the file this config was loaded from ("" when running without one).
|
|
func (s *Store) Path() string { return s.path }
|
|
|
|
// Generation counts successful swaps; the cluster uses it to detect
|
|
// divergence between instances.
|
|
func (s *Store) Generation() uint64 { return s.generation.Load() }
|
|
|
|
// LastError returns the error from the most recent failed reload, or nil.
|
|
func (s *Store) LastError() string {
|
|
if p := s.lastErr.Load(); p != nil {
|
|
return *p
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// Swap installs cfg as the current configuration. Callers are responsible for
|
|
// validating first; Swap itself only guards structural invariants that would
|
|
// otherwise panic subsystems.
|
|
func (s *Store) Swap(cfg *Config) error {
|
|
if cfg == nil {
|
|
return errNilConfig
|
|
}
|
|
if cfg.Cache.Shards <= 0 {
|
|
cfg.Cache.Shards = 32
|
|
}
|
|
s.mu.Lock()
|
|
s.base.Store(cfg)
|
|
eff := s.effective(cfg)
|
|
s.cur.Store(eff)
|
|
s.generation.Add(1)
|
|
s.lastErr.Store(nil)
|
|
s.mu.Unlock()
|
|
s.runHooks(eff)
|
|
return nil
|
|
}
|
|
|
|
// Reload re-reads the backing file and swaps it in on success. On failure the
|
|
// previous configuration stays active and the error is recorded.
|
|
func (s *Store) Reload() error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.path == "" {
|
|
return errNoPath
|
|
}
|
|
cfg, err := Load(s.path)
|
|
if err != nil {
|
|
msg := err.Error()
|
|
s.lastErr.Store(&msg)
|
|
return err
|
|
}
|
|
// Preserve the working dir of the running instance so relative paths keep
|
|
// resolving the same way even if the file moved.
|
|
cfg.Server.WorkingDir = s.Base().Server.WorkingDir
|
|
s.base.Store(cfg)
|
|
eff := s.effective(cfg)
|
|
s.cur.Store(eff)
|
|
s.generation.Add(1)
|
|
s.lastErr.Store(nil)
|
|
s.runHooks(eff)
|
|
return nil
|
|
}
|
|
|
|
// OnSwap registers a callback invoked with the new config after each swap.
|
|
// Hooks run outside the writer lock; a panicking hook is contained so one bad
|
|
// subsystem cannot take down a reload.
|
|
func (s *Store) OnSwap(fn func(*Config)) {
|
|
if fn == nil {
|
|
return
|
|
}
|
|
s.hooksMu.Lock()
|
|
s.hooks = append(s.hooks, fn)
|
|
s.hooksMu.Unlock()
|
|
}
|
|
|
|
func (s *Store) runHooks(cfg *Config) {
|
|
s.hooksMu.RLock()
|
|
hooks := s.hooks
|
|
s.hooksMu.RUnlock()
|
|
for _, h := range hooks {
|
|
func() {
|
|
defer func() { _ = recover() }()
|
|
h(cfg)
|
|
}()
|
|
}
|
|
}
|
|
|
|
type storeError string
|
|
|
|
func (e storeError) Error() string { return string(e) }
|
|
|
|
const (
|
|
errNilConfig = storeError("config: refusing to swap a nil configuration")
|
|
errNoPath = storeError("config: no config file path known, cannot reload")
|
|
)
|