Files
fancy-dns/internal/app/app.go
T
developer b0e6b3bb92
ci / Console typecheck and build (push) Successful in 14s
ci / Go build, vet and test (push) Failing after 45s
ci / Binary with the console embedded (push) Skipped
ci / Container image (push) Skipped
Add the memory envelope, pprof, and a per-packet fix on the UDP path
A cgroup OOM kill is the worst outcome available to a resolver: an outage
rather than a cache miss, with no chance to serve stale on the way out. So:

- memory.limit / memory.reserve / memory.cache_max_bytes, a new config section.
  limit becomes a GOMEMLIMIT through debug.SetMemoryLimit, reduced by the
  reserve. The reserve exists because the Go heap does not account for
  goroutine stacks, runtime metadata, or the bbolt mmap backing the cache
  snapshot and the query log — handing the runtime the full cgroup limit is how
  you get killed by a mechanism that believed it had headroom.
- config.Size, so a budget reads as "512MiB" rather than a byte count converted
  by hand. Strict about junk: "512MiBx" is an error.
- The cache byte budget is now actually fed to the LRU. It already enforced
  maxBytes per shard on insert, but had no setter and nothing ever called one,
  so the footprint of cache.size was unbounded: one cached ANY or TXT answer
  can be tens of kilobytes, so 500k entries is anywhere from 150MB to several GB
  depending on what the traffic asked for.

- api.pprof mounts net/http/pprof on the admin listener. The pprof handlers sit
  on the app-level mux, not inside the API's own handler, so they would bypass
  the auth middleware entirely; the option is therefore refused unless every
  admin address is loopback. Profiling means port-forwarding, so this is safe
  by construction rather than by policy, and cmdline cannot leak an argument
  that happens to be a key.

- serveUDP was calling net.ResolveUDPAddr per packet. The peer already is a
  *net.UDPAddr; that call builds a string, re-parses the port and allocates on
  the busiest path in the process.

Also fixes a flaky test: TestParallelPolicyAnswersFromTheFastestUpstream
asserted the slow upstreams were hit, but the fan-out cancels on the first
answer, so whether they were dispatched before cancellation is a real race. It
now asserts the answer came from the fast upstream and that nothing waited.
2026-09-27 13:41:48 +00:00

832 lines
23 KiB
Go

// Package app wires the components together and owns the process lifecycle:
// configuration, listeners, background loops, the admin API and shutdown.
package app
import (
"context"
"crypto/tls"
"errors"
"fmt"
"log/slog"
"net"
"net/http"
"net/http/pprof"
"net/netip"
"os"
"path/filepath"
"strings"
"sync"
"time"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/acl"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/api"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/blocklist"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/cache"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/cluster"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/engine"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/localzone"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/metrics"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/qlog"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/resolver"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/server"
"git.k8s-git-k6n.net/mats/fancy-dns/internal/web"
"github.com/miekg/dns"
)
// Version is the build version, overridden at link time with -ldflags.
var Version = "dev"
// Options configures an App.
type Options struct {
// ConfigPath is the configuration file; "" runs on defaults.
ConfigPath string
// Version is the reported build version.
Version string
// LogLevel and LogFormat override the config's logging settings.
LogLevel string
LogFormat string
}
// App is the running process.
type App struct {
opt Options
log *slog.Logger
cfg *config.Store
ver string
blocks *blocklist.Manager
cache *cache.Manager
cstore *cache.Store
logger *qlog.Logger
aclH *acl.Holder
local *localzone.Holder
pool *resolver.Pool
reverse *engine.ReverseCache
engine *engine.Engine
dns *server.Server
api *api.Server
cl *cluster.Cluster
series *seriesStore
apiServers []*http.Server
apiLn []net.Listener
// reloadMu serialises reload work, since SIGHUP, the API and the cluster
// can all trigger one at the same time.
reloadMu sync.Mutex
startedAt time.Time
// degraded is set when a background component failed; surfaced in health.
degradedMu sync.RWMutex
degraded []string
}
// New builds an App from options. It does not bind anything yet; call Start.
func New(opt Options) (*App, error) {
if opt.Version == "" {
opt.Version = Version
}
cfg, err := config.Load(opt.ConfigPath)
if err != nil {
return nil, err
}
log, err := newLogger(cfg, opt)
if err != nil {
return nil, err
}
store := config.NewStore(cfg, opt.ConfigPath)
return &App{opt: opt, log: log, cfg: store, ver: opt.Version}, nil
}
// newLogger builds the structured logger from config plus CLI overrides.
func newLogger(cfg *config.Config, opt Options) (*slog.Logger, error) {
level := cfg.Log.Level
if opt.LogLevel != "" {
level = opt.LogLevel
}
var lvl slog.Level
switch level {
case "debug":
lvl = slog.LevelDebug
case "warn":
lvl = slog.LevelWarn
case "error":
lvl = slog.LevelError
default:
lvl = slog.LevelInfo
}
format := cfg.Log.Format
if opt.LogFormat != "" {
format = opt.LogFormat
}
handlerOpts := &slog.HandlerOptions{Level: lvl}
var w *os.File = os.Stderr
if cfg.Log.Addr != "" {
if strings.HasPrefix(cfg.Log.Addr, "/") {
f, err := os.OpenFile(cfg.Log.Addr, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return nil, fmt.Errorf("open log file: %w", err)
}
w = f
}
}
var h slog.Handler
if format == "json" {
h = slog.NewJSONHandler(w, handlerOpts)
} else {
h = slog.NewTextHandler(w, handlerOpts)
}
return slog.New(h).With("service", "fancy-dns", "version", opt.Version), nil
}
// Start brings up every component in dependency order.
func (a *App) Start(ctx context.Context) error {
a.startedAt = time.Now()
cfg := a.cfg.Get()
metrics.SetUp(true)
if err := a.startCache(cfg); err != nil {
return err
}
if err := a.startLogger(cfg); err != nil {
return err
}
a.startBlocklist()
a.startLocalZones(cfg)
a.startUpstream(cfg)
a.startDNS(ctx)
if err := a.startAPI(ctx); err != nil {
return err
}
a.startCluster(ctx, cfg)
a.startBackground(ctx, cfg)
a.cfg.OnSwap(a.onConfigSwap)
// The real bound admin address, not the configured one: with :0 the config
// is useless for telling the operator where to point a browser.
a.log.Info("fancy-dns started",
"config", a.cfg.Path(),
"dns", a.dns.Listeners(),
"api", a.Addr())
return nil
}
// startCache opens the bbolt snapshot and restores a warm start.
func (a *App) startCache(cfg *config.Config) error {
// The memory envelope is a process-wide setting, not a component one, so
// it is applied on every (re)start and on every SIGHUP reload. Doing it
// here means a changed limit takes effect with the config that set it.
applyMemoryLimit(cfg.Memory, a.log)
opts := cache.Options{
Size: cfg.Cache.Size,
Shards: cfg.Cache.Shards,
MaxBytes: cfg.Memory.EffectiveCacheBytes(),
MinTTL: cfg.Cache.MinTTL.D(),
MaxTTL: cfg.Cache.MaxTTL.D(),
NegativeTTL: cfg.Cache.NegativeTTL.D(),
ServeStale: cfg.Cache.ServeStale.D(),
Prefetch: cfg.Cache.Prefetch,
}
if cfg.Cache.Enabled && cfg.Cache.Persist.Enabled {
st, err := cache.OpenStore(cache.StoreOptions{
Path: cfg.CachePath(),
MaxEntries: cfg.Cache.Persist.MaxEntries,
})
if err != nil {
a.markDegraded("cache persistence: " + err.Error())
a.log.Warn("cache persistence unavailable, continuing without it", "error", err)
} else {
a.cstore = st
if cfg.Cache.Persist.LoadOnStart {
lru := cache.NewLRU(cfg.Cache.Size, cfg.Cache.Shards)
loaded, skipped, lerr := st.Load(lru, time.Now())
if lerr != nil {
a.log.Warn("cache snapshot load failed", "error", lerr)
} else if loaded > 0 {
a.log.Info("cache snapshot restored", "entries", loaded, "skipped", skipped)
// Re-seed the real LRU from the temporary one.
a.cache = cache.NewManager(opts, st)
lru.Walk(func(k cache.Key, e *cache.Entry) bool {
a.cache.PutRaw(k, e.Wire, e.TTL, e.Origin, e.Upstream, e.Neg)
return true
})
}
}
}
}
if a.cache == nil {
a.cache = cache.NewManager(opts, a.cstore)
}
return nil
}
// prewarm resolves a few names at boot in the background, so a restart does
// not look like an outage for the names that matter most.
func (a *App) prewarm(c context.Context, specs []string) {
go func() {
for _, spec := range specs {
name, qtypeStr, ok := strings.Cut(spec, "/")
if !ok {
name, qtypeStr = spec, "A"
}
qtype, ok := dns.StringToType[strings.ToUpper(qtypeStr)]
if !ok {
a.log.Warn("prewarm: unknown record type", "spec", spec)
continue
}
m := new(dns.Msg)
m.SetQuestion(dns.Fqdn(name), qtype)
m.RecursionDesired = true
res := a.engine.Handle(c, engine.Request{Msg: m, Transport: "internal", Start: time.Now()})
if res.Msg != nil && res.Msg.Rcode == dns.RcodeSuccess {
a.log.Debug("prewarmed", "name", name, "type", qtypeStr)
}
}
}()
}
// tlsCfgType aliases the TLS config so the listener helpers read cleanly.
type tlsCfgType = tls.Config
// apiTLS builds the admin listener's TLS config when one is configured.
func apiTLS(cfg *config.Config) (*tls.Config, error) {
if cfg.API.CertFile == "" || cfg.API.KeyFile == "" {
return nil, nil
}
cert, err := tls.LoadX509KeyPair(cfg.API.CertFile, cfg.API.KeyFile)
if err != nil {
return nil, fmt.Errorf("load admin certificate: %w", err)
}
return &tls.Config{
Certificates: []tls.Certificate{cert},
MinVersion: tls.VersionTLS12,
NextProtos: []string{"h2", "http/1.1"},
}, nil
}
// startLogger opens the query log.
func (a *App) startLogger(cfg *config.Config) error {
opts := qlog.DefaultOptions()
opts.Path = cfg.QueryLogPath()
opts.MaxEntries = cfg.QueryLog.MaxEntries
opts.LiveBuffer = cfg.QueryLog.LiveBuffer
opts.FlushInterval = cfg.QueryLog.FlushInterval.D()
opts.SampleRate = cfg.QueryLog.SampleRate
opts.LogBlocked = cfg.QueryLog.LogBlocked
opts.LogCached = cfg.QueryLog.LogCached
opts.LogErrors = cfg.QueryLog.LogErrors
opts.TopDomains = cfg.QueryLog.TopDomains
opts.TopTTL = cfg.QueryLog.TopTTL.D()
if !cfg.QueryLog.Enabled {
opts.Path = ""
}
l, err := qlog.New(opts)
if err != nil {
a.markDegraded("query log: " + err.Error())
a.log.Warn("query log unavailable, continuing without persistence", "error", err)
return nil
}
a.logger = l
return nil
}
// startBlocklist starts the refresh loop and does the first compile.
func (a *App) startBlocklist() {
cfg := a.cfg.Get()
if !cfg.Block.Enabled {
a.log.Info("blocklists disabled")
return
}
a.blocks = blocklist.NewManager(blocklist.ManagerOptions{
Config: a.cfg,
Logger: a.log,
})
// Compile once before serving so the first queries are not unprotected.
if _, err := a.blocks.RefreshAll(context.Background()); err != nil {
a.markDegraded("blocklist: " + err.Error())
a.log.Warn("initial blocklist compile failed", "error", err)
}
}
// startLocalZones compiles the local master files.
func (a *App) startLocalZones(cfg *config.Config) {
a.local = localzone.NewHolder(nil)
if len(cfg.Block.LocalZones) == 0 {
return
}
a.reloadLocalZones()
}
// reloadLocalZones recompiles the local zone store. A failure keeps the
// previous store: serving stale local data beats serving nothing.
func (a *App) reloadLocalZones() {
cfg := a.cfg.Get()
if len(cfg.Block.LocalZones) == 0 {
a.local.Set(nil)
return
}
paths := make([]string, 0, len(cfg.Block.LocalZones))
for _, p := range cfg.Block.LocalZones {
paths = append(paths, cfg.AbsPath(p))
}
store, err := localzone.Load(paths, localzone.Options{})
if err != nil {
a.markDegraded("local zones: " + err.Error())
a.log.Error("local zone load failed, keeping previous data", "error", err)
return
}
a.local.Set(store)
metrics.SetLocalRecords(store.Count())
a.log.Info("local zones loaded", "zones", store.Zones(), "records", store.Count())
}
// startUpstream builds the resolver pool.
func (a *App) startUpstream(cfg *config.Config) {
a.pool = resolver.NewPool(cfg.Upstream)
}
// startDNS builds the engine and the listeners.
func (a *App) startDNS(ctx context.Context) {
a.aclH = acl.NewHolder(acl.Compile(cfgACL(a.cfg.Get())))
a.reverse = engine.NewReverseCache(engine.ReverseOptions{Max: 8192})
a.engine = engine.New(engine.Options{
Config: a.cfg,
Logger: a.logger,
Blocklists: a.blocksHolder(),
Cache: a.cache,
ACL: a.aclH,
Local: a.local,
Pool: a.pool,
Reverse: a.reverse,
})
a.engine.SetFlusher(a.flush)
a.dns = server.New(server.Options{
Engine: a.engine,
Config: a.cfg,
Logger: a.log,
Version: a.ver,
})
a.dns.SetIdentity(
func(ip string) []string { names, _ := a.reverse.Names(ip); return names },
func(ip string) []string {
for _, n := range mustNames(a.reverse, ip) {
if tags := acl.TagsFromName(n); len(tags) > 0 {
return tags
}
}
return nil
},
)
if err := a.dns.Start(ctx); err != nil {
a.log.Error("dns listeners failed to start", "error", err)
a.markDegraded("dns listeners: " + err.Error())
}
a.series = newSeriesStore()
}
func mustNames(rc *engine.ReverseCache, ip string) []string {
names, _ := rc.Names(ip)
return names
}
func (a *App) blocksHolder() *blocklist.Holder {
if a.blocks == nil {
return blocklist.NewHolder(nil)
}
return a.blocks.Holder()
}
func cfgACL(cfg *config.Config) config.ACL { return cfg.ACL }
// startAPI binds the admin listener and, when configured, the embedded UI.
func (a *App) startAPI(ctx context.Context) error {
cfg := a.cfg.Get()
if !cfg.API.Enabled {
return nil
}
api.SetMetricsHandler(metrics.Handler())
hooks := a.hooks()
srv, err := api.New(hooks)
if err != nil {
return err
}
a.api = srv
// One mux: the API claims its base path, the SPA claims the rest. The API
// is registered first so /api/* never falls through to the SPA's
// index.html fallback (which would return HTML to a JSON client).
mux := http.NewServeMux()
// The api package already registers its routes with the full base path, so
// no StripPrefix here: a more specific pattern than the SPA's "/" wins, so
// /api/* never falls through to index.html.
mux.Handle(cfg.API.BasePath+"/", srv.Handler())
mux.Handle(cfg.API.BasePath, srv.Handler())
if cfg.API.UI {
ui, err := web.Handler("")
if err != nil {
a.log.Warn("web console unavailable", "error", err)
} else {
mux.Handle("/", ui)
if web.Available() {
a.log.Info("web console enabled")
} else {
a.log.Warn("web console assets missing from this binary; run 'make web' and rebuild")
}
}
}
// pprof on the top-level mux would sit OUTSIDE srv.Handler(), so the API's
// auth would never see it — /debug/pprof/profile starts a CPU profile for
// anyone who can reach the port. Gate it on loopback instead: profiling
// means ssh port-forwarding, and that keeps the unauthenticated surface
// off a routable interface by construction rather than by policy.
if cfg.API.Pprof {
if !onlyLoopback(cfg.API.Listen) || !onlyLoopback(cfg.API.TLSListen) {
a.log.Warn("api.pprof is set but not every admin address is loopback; refusing to mount pprof")
} else {
mux.HandleFunc("/debug/pprof/", pprof.Index)
mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
a.log.Info("pprof mounted on the admin listener", "path", "/debug/pprof/")
}
}
tlsCfg, tlsErr := apiTLS(cfg)
var errs []error
for _, addr := range cfg.API.Listen {
ln, err := net.Listen("tcp", addr)
if err != nil {
errs = append(errs, fmt.Errorf("bind api %s: %w", addr, err))
continue
}
a.startHTTPServer(mux, ln, nil, "api")
}
if len(cfg.API.TLSListen) > 0 {
if tlsErr != nil {
errs = append(errs, tlsErr)
} else {
for _, addr := range cfg.API.TLSListen {
ln, err := net.Listen("tcp", addr)
if err != nil {
errs = append(errs, fmt.Errorf("bind api tls %s: %w", addr, err))
continue
}
a.startHTTPServer(mux, ln, tlsCfg, "api-tls")
}
}
}
if err := errors.Join(errs...); err != nil {
a.markDegraded("admin api: " + err.Error())
a.log.Error("admin api failed to bind", "error", err)
}
_ = ctx
return nil
}
// onlyLoopback reports whether every address is a loopback one. An empty list
// counts as loopback: nothing is bound, so nothing is exposed.
func onlyLoopback(addrs []string) bool {
for _, a := range addrs {
host, _, err := net.SplitHostPort(a)
if err != nil {
host = a
}
ip, err := netip.ParseAddr(host)
if err != nil {
// A hostname, or a bare "*": not provably loopback.
return false
}
if !ip.IsLoopback() && !ip.IsUnspecified() {
return false
}
}
return true
}
func (a *App) startHTTPServer(h http.Handler, ln net.Listener, tlsCfg *tlsCfgType, name string) {
srv := &http.Server{
Handler: h,
ReadHeaderTimeout: 15 * time.Second,
IdleTimeout: 120 * time.Second,
}
if tlsCfg != nil {
srv.TLSConfig = tlsCfg
ln = tls.NewListener(ln, tlsCfg)
}
a.apiServers = append(a.apiServers, srv)
a.apiLn = append(a.apiLn, ln)
a.log.Info("admin api listening", "addr", ln.Addr().String(), "tls", tlsCfg != nil, "name", name)
go func() {
if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
a.log.Error("admin api stopped", "error", err, "name", name)
}
}()
}
// startCluster connects to Valkey and wires cluster events into a reload.
func (a *App) startCluster(ctx context.Context, cfg *config.Config) {
if !cfg.Cluster.Enabled {
a.cl = cluster.New(cluster.Options{Config: a.cfg, Logger: a.log, Version: a.ver})
return
}
a.cl = cluster.New(cluster.Options{
Config: a.cfg,
Logger: a.log,
Version: a.ver,
StartedAt: a.startedAt,
Addr: cfg.Cluster.Advertise,
StateFunc: a.nodeState,
Digests: a.nodeDigests,
Handler: a.onClusterEvent,
})
a.cl.Start(ctx)
}
// startBackground runs the periodic jobs: cache snapshots, upstream health
// probes and the metrics gauge.
func (a *App) startBackground(ctx context.Context, cfg *config.Config) {
// Cache snapshots.
if a.cstore != nil {
interval := cfg.Cache.Persist.Interval.D()
if interval <= 0 {
interval = 30 * time.Second
}
go func() {
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
a.snapshotCache()
return
case <-t.C:
a.snapshotCache()
}
}
}()
}
// Upstream health probes.
go a.healthLoop(ctx)
// Live gauges and the dashboard series sampler.
go func() {
t := time.NewTicker(2 * time.Second)
defer t.Stop()
sample := time.NewTicker(5 * time.Second)
defer sample.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
metrics.SetConcurrent(int(a.engine.Inflight()))
case <-sample.C:
if a.series != nil {
a.series.sample()
}
}
}
}()
}
func (a *App) snapshotCache() {
if a.cstore == nil || a.cache == nil {
return
}
n, err := a.cstore.Snapshot(a.cache.LRU(), time.Now())
if err != nil {
a.log.Warn("cache snapshot failed", "error", err)
return
}
a.log.Debug("cache snapshot written", "entries", n)
}
func (a *App) healthLoop(ctx context.Context) {
for {
interval := a.cfg.Get().Upstream.HealthCheck.D()
if interval <= 0 {
interval = 30 * time.Second
}
select {
case <-ctx.Done():
return
case <-time.After(interval):
}
pool := a.pool
if pool == nil {
continue
}
hctx, cancel := context.WithTimeout(ctx, 5*time.Second)
ok := pool.HealthCheck(hctx, 3*time.Second)
cancel()
total := len(pool.All())
for _, u := range pool.All() {
healthy := u.Health() == resolver.HealthUp
metrics.SetUpstreamHealth(u.Addr(), u.Config().Proto, healthy)
}
if total > 0 && ok == 0 {
a.markDegraded("no upstream is answering health probes")
} else {
a.clearDegraded("no upstream is answering health probes")
}
}
}
// markDegraded records a component problem for the health endpoint.
func (a *App) markDegraded(reason string) {
a.degradedMu.Lock()
defer a.degradedMu.Unlock()
for _, r := range a.degraded {
if r == reason {
return
}
}
a.degraded = append(a.degraded, reason)
}
func (a *App) clearDegraded(reason string) {
a.degradedMu.Lock()
defer a.degradedMu.Unlock()
out := a.degraded[:0]
for _, r := range a.degraded {
if r != reason {
out = append(out, r)
}
}
a.degraded = out
}
// Degraded returns the current problem list.
func (a *App) Degraded() []string {
a.degradedMu.RLock()
defer a.degradedMu.RUnlock()
return append([]string(nil), a.degraded...)
}
// onConfigSwap reacts to a new configuration.
func (a *App) onConfigSwap(cfg *config.Config) {
a.log.Info("configuration applied", "version", cfg.Version)
if a.pool != nil {
a.pool.Rebind(cfg.Upstream)
}
a.reloadLocalZones()
if a.blocks != nil && !cfg.Block.Enabled {
a.log.Info("blocklists disabled by configuration change")
}
}
// onClusterEvent reacts to a peer message.
func (a *App) onClusterEvent(ctx context.Context, ev cluster.Event) {
switch ev.Kind {
case cluster.EventReload:
a.log.Info("peer requested a config reload", "from", ev.NodeID)
if err := a.Reload(); err != nil {
a.log.Warn("cluster-triggered reload failed", "error", err)
}
case cluster.EventBlocklists:
// Only the lease holder fetches, so N instances do not hammer a mirror
// N times; the others keep serving their compiled copy.
if a.cl != nil && a.cl.IsLeader() {
if a.blocks != nil {
a.blocks.Trigger(firstOr(ev.Sources, ""))
}
}
case cluster.EventConfig:
// A peer pushed a full document. We do not adopt it silently: a
// follower adopting a document it cannot see on disk would drift. The
// operator re-runs a deploy or hits "reload" instead.
a.log.Info("peer published a new config version", "from", ev.NodeID, "version", ev.Version)
}
}
func firstOr(s []string, def string) string {
if len(s) > 0 && s[0] != "" {
return s[0]
}
return def
}
// nodeState supplies the live state published to peers.
func (a *App) nodeState() *cluster.NodeState {
addrs := make([]string, 0, 4)
for _, l := range a.dns.Listeners() {
addrs = append(addrs, l.Transport+":"+l.Addr)
}
return &cluster.NodeState{
Addr: a.opt.ConfigPath,
Version: a.ver,
StartedAt: a.startedAt,
Listeners: addrs,
Generation: a.cfg.Generation(),
}
}
// nodeDigests supplies the digests peers compare.
func (a *App) nodeDigests() (string, int, time.Time, string) {
var (
bl string
rules int
built time.Time
)
if a.blocks != nil {
st := a.blocks.State()
if st != nil {
bl = st.Digest
if st.Set != nil {
rules = st.Set.Len()
built = st.Set.BuiltAt()
}
}
}
return bl, rules, built, cluster.Digest(a.cfg.Get())
}
// Reload re-reads the config file and applies it.
func (a *App) Reload() error {
a.reloadMu.Lock()
defer a.reloadMu.Unlock()
return a.cfg.Reload()
}
// Stop shuts everything down in reverse order, with a deadline.
func (a *App) Stop(ctx context.Context) error {
a.log.Info("shutting down")
metrics.SetUp(false)
var errs []error
if a.cl != nil {
a.cl.Stop()
}
if a.blocks != nil {
a.blocks.Stop()
}
if a.dns != nil {
if err := a.dns.Stop(ctx); err != nil {
errs = append(errs, err)
}
}
for _, srv := range a.apiServers {
shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
err := srv.Shutdown(shutCtx)
cancel()
if err != nil {
errs = append(errs, err)
}
}
if a.logger != nil {
a.logger.Flush()
if err := a.logger.Close(); err != nil {
errs = append(errs, err)
}
}
if a.cstore != nil {
// Snapshot, close, then clear the pointer: the background snapshot
// goroutine also fires on ctx.Done, and a nil check is how it learns
// to stop rather than writing to a closed file.
a.snapshotCache()
if err := a.cstore.Close(); err != nil {
errs = append(errs, err)
}
a.cstore = nil
}
if err := errors.Join(errs...); err != nil {
a.log.Warn("shutdown completed with errors", "error", err)
} else {
a.log.Info("shutdown complete")
}
return errors.Join(errs...)
}
// Addr returns the admin API address, which a preview needs.
func (a *App) Addr() string {
for _, ln := range a.apiLn {
return ln.Addr().String()
}
return ""
}
// DataDir is where the bbolt files live.
func (a *App) DataDir() string {
return filepath.Dir(a.cfg.Get().CachePath())
}
// Log exposes the logger so the command can report signals.
func (a *App) Log() *slog.Logger { return a.log }
// SetAdminAddr overrides the admin API listen address without touching the
// config file, which is what a container platform's -admin-addr flag wants.
func (a *App) SetAdminAddr(addr string) error {
next := a.cfg.Get()
cp := *next
cp.API = next.API
cp.API.Listen = []string{addr}
if err := a.cfg.Swap(&cp); err != nil {
return err
}
return nil
}