internal/engine, internal/config and internal/metrics had no test file at all, and writing them turned up five real bugs. internal/engine (23 tests, new harness_test.go): a wired engine over a fake upstream on loopback, a local zone, a blocklist, a cache, a query log and an injectable clock. It pins the pipeline order end to end -- ACL deny before the local zone, local zone before blocklist, blocklist before cache, drop sends nothing, the concurrency and rate-limit refusals, serve-stale served with TTL 1 and refreshed once per key, truncation against a 512-byte client, and exactly one metrics record and one query-log entry per query. Five fixes: - config.ParseSize ignored the unit multiplier, so 512MiB parsed as 512 bytes and the whole memory envelope was inert for anyone who wrote a unit. - finalize (EDNS0 and the UDP size) ran only on the cache-hit path, so a large uncached answer left oversized for a 512-byte client. It now runs on the upstream, local-zone and blocklist paths too. - The upstream UDP client read buffer was hard-coded to 1232 while we advertise dns.max_udp_size, so with a larger buffer configured every answer above 1232 bytes was read truncated and became SERVFAIL. The pool takes the size now. - LRU.prefetchGate was a 64-slot global cap, not the single flight its comment promised: a burst of stale hits launched up to 64 duplicate refreshes of one name. It is a per-key latch now; distinct names still refresh in parallel. - metrics.RecordMiss and RecordStaleHit moved only the Prometheus collector, so the Cache page's misses and stale hits read 0 for the life of the process. AGENTS.md gains the trap that cost the most time: this version of miekg/dns reports a character-string over 255 bytes as zero length, so such a message silently fails to pack, and the write failure looks like a client timeout.
832 lines
23 KiB
Go
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.NewPoolWithUDPSize(cfg.Upstream, cfg.DNS.MaxUDPSize)
|
|
}
|
|
|
|
// 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
|
|
}
|