package app import ( "context" "errors" "fmt" "net/netip" "sort" "strings" "sync" "time" "github.com/miekg/dns" "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/config" "git.k8s-git-k6n.net/mats/fancy-dns/internal/engine" "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/zonefile" ) // hooks builds the api.Hooks that connect the admin API to the live // components. Everything the UI can read or change goes through here. func (a *App) hooks() api.Hooks { return api.Hooks{ Store: a.cfg, Logger: a.log, Version: a.ver, StartedAt: a.startedAt, Dashboard: a.dashboard, Status: a.status, Health: a.health, BlocklistState: a.blocklistState, BlocklistStats: a.blocklistStats, RefreshBlocklists: a.refreshBlocklists, ToggleBlocklist: a.toggleBlocklist, DomainCheck: a.domainCheck, Resolve: a.resolveTest, ListDomains: a.listDomains, AddDomain: a.addDomain, DeleteDomain: a.deleteDomain, ListZones: a.listZones, AddZoneRecord: a.addZoneRecord, PutZoneRecord: a.putZoneRecord, DeleteZoneRecord: a.deleteZoneRecord, DeleteZone: a.deleteZone, AddZoneDelegation: a.addZoneDelegation, DeleteZoneDelegation: a.deleteZoneDelegation, LocalZones: a.listLocalZones, SaveLocalZone: a.saveLocalZone, DeleteLocalZone: a.deleteLocalZone, GetACL: a.getACL, PutACL: a.putACL, GetUpstream: a.getUpstreams, PutUpstream: a.putUpstream, SearchLog: a.searchLog, LogStats: a.logStats, SubscribeLog: a.subscribeLog, ClearLog: a.clearLog, CacheStats: a.cacheStats, CacheTop: a.cacheTop, FlushCache: a.flushCache, Flush: a.flush, } } func (a *App) listLocalZones(ctx context.Context) ([]api.ZoneFile, error) { if a.zones == nil { return nil, fmt.Errorf("local zone editor is available only in Kubernetes") } zones, err := a.zones.List(ctx) if err != nil { return nil, err } out := make([]api.ZoneFile, 0, len(zones)) for _, z := range zones { out = append(out, api.ZoneFile{Name: z.Name, Content: z.Content}) } sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) return out, nil } func (a *App) saveLocalZone(ctx context.Context, name, content string) error { if a.zones == nil { return fmt.Errorf("local zone editor is available only in Kubernetes") } apex := strings.ToLower(dns.Fqdn(strings.TrimSuffix(strings.TrimSpace(name), "."))) _, valid := dns.IsDomainName(apex) if !valid || apex == "." || len(apex) > 243 { return fmt.Errorf("invalid zone name %q", name) } if strings.TrimSpace(content) == "" { return fmt.Errorf("zone file cannot be empty") } parsed, err := zonefile.Parse(strings.NewReader(content), zonefile.Options{Origin: apex, File: name + ".zone"}) if err != nil { return fmt.Errorf("invalid zone file: %w", err) } if len(parsed.Records) == 0 { return fmt.Errorf("zone file must contain at least one record") } for _, record := range parsed.Records { owner := strings.ToLower(dns.Fqdn(record.Name)) if owner != strings.ToLower(apex) && !strings.HasSuffix(owner, "."+strings.ToLower(apex)) { return fmt.Errorf("record %q is outside zone %s", owner, apex) } } if err := a.zones.Save(ctx, strings.TrimSuffix(apex, "."), content); err != nil { return err } if a.cache != nil { a.cache.Flush() } return nil } func (a *App) deleteLocalZone(ctx context.Context, name string) error { if a.zones == nil { return fmt.Errorf("local zone editor is available only in Kubernetes") } name = strings.TrimSuffix(strings.TrimSpace(name), ".") if err := a.zones.Delete(ctx, name); err != nil { return err } if a.cache != nil { a.cache.Flush() } return nil } // health is the unauthenticated probe payload. func (a *App) health() any { status := "ok" problems := a.Degraded() switch { case len(problems) > 0: status = "degraded" } if a.dns == nil || len(a.dns.Listeners()) == 0 { status = "down" } if a.logger != nil && a.logger.ErrorCount() > 0 && status == "ok" { status = "degraded" problems = append(problems, "query log write errors") } return map[string]any{ "status": status, "version": a.ver, "uptime_s": int64(time.Since(a.startedAt).Seconds()), "config_version": a.cfg.Get().Version, "problems": problems, } } // status is the per-component runtime view. func (a *App) status() any { out := map[string]any{ "listeners": a.dns.Listeners(), "transports": a.dns.ByTransport(), "degraded": a.Degraded(), } if a.blocks != nil { st := a.blocks.State() if st != nil { out["blocklist"] = map[string]any{ "rules": st.Set.Len(), "sources": len(st.Sources), "duration": st.Duration, "built_at": st.Set.BuiltAt(), "errors": st.Errors, "digest": st.Digest, "has_wild": st.Set.HasWildcards(), "duplicates": st.Set.DuplicatesN(), } } } if a.local != nil { if store := a.local.Get(); store != nil { out["local_zones"] = map[string]any{ "zones": store.Zones(), "records": store.Count(), "files": store.Files(), } } } if a.cache != nil { out["cache"] = a.cacheStats() } if a.logger != nil { out["qlog"] = a.logger.Stats() } if a.pool != nil { out["upstream"] = a.getUpstreams() } if a.aclH != nil { out["acl"] = a.getACL() } if a.cl != nil { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) out["cluster"] = a.cl.Status(ctx) cancel() } if a.engine != nil { out["engine"] = map[string]any{ "inflight": a.engine.Inflight(), "dropped": a.engine.Dropped(), "group_stats": a.engine.GroupStats(), } } return out } // blocklistState is the full per-source view for the UI. func (a *App) blocklistState() any { if a.blocks == nil { return map[string]any{ "items": []any{}, "stats": map[string]any{"total": 0}, } } st := a.blocks.State() items := make([]blocklist.Source, 0, len(st.Sources)) for _, s := range st.Sources { items = append(items, s) } sort.Slice(items, func(i, j int) bool { return items[i].Name < items[j].Name }) return map[string]any{ "items": items, "stats": blocklistSourceStats(st), "errors": st.Errors, "digest": st.Digest, } } func blocklistSourceStats(st *blocklist.State) map[string]any { if st == nil || st.Set == nil { return map[string]any{"total": 0} } s := st.Set.Stats() return map[string]any{ "total": s.Total, "exact": s.Exact, "subdomain_rules": s.Subdomains, "root_rules": s.RootRules, "allow": s.Allow, "block": s.Block, "passthrough": s.Passthru, "drop": s.Drop, "duplicates": s.Duplicates, "sources": s.Sources, "built_at": s.BuiltAt, "sources_detail": s.PerSource, } } func (a *App) blocklistStats() any { if a.blocks == nil { return map[string]any{"total": 0} } return blocklistSourceStats(a.blocks.State()) } // invalidateBlocklistCache drops every cached answer whose verdict the new // rules might change. Without this, a name that has just become blocked keeps // serving its cached good answer until the TTL expires, which is exactly the // bug an operator hits right after clicking "add rule". func (a *App) invalidateBlocklistCache() { if a.cache == nil { return } a.cache.Flush() } // refreshBlocklists recompiles the list sources. func (a *App) refreshBlocklists(ctx context.Context, names []string) error { if a.blocks == nil { return fmt.Errorf("blocklists are disabled") } a.invalidateBlocklistCache() go func() { if len(names) == 0 { if _, err := a.blocks.RefreshAll(ctx); err != nil { a.log.Warn("blocklist refresh failed", "error", err) } return } // The manager compiles everything in one pass, which is cheaper than // a pass per name and keeps the compiled set internally consistent. if _, err := a.blocks.RefreshAll(ctx); err != nil { a.log.Warn("blocklist refresh failed", "error", err, "names", names) } }() // Tell the cluster leader so every instance ends up on the same content. if a.cl != nil { _ = a.cl.BroadcastRefresh(ctx, names) } return nil } // toggleBlocklist flips one source's enabled flag and persists it. func (a *App) toggleBlocklist(name string, enabled bool) error { if a.blocks == nil { return fmt.Errorf("blocklists are disabled") } err := a.mutateConfig(func(c *config.Config) error { found := false for i := range c.Block.Lists { if c.Block.Lists[i].Name == name { c.Block.Lists[i].Enabled = enabled found = true } } if !found { return fmt.Errorf("no blocklist named %q", name) } return nil }) if err != nil { return err } a.invalidateBlocklistCache() go func() { if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil { a.log.Warn("refresh after toggle failed", "error", rerr) } }() return nil } // mutateConfig clones, mutates, validates and commits the config, returning // the validation error when the change is rejected. func (a *App) mutateConfig(fn func(*config.Config) error) error { cur := a.cfg.Get() next := *cur // A shallow copy is not enough: the slices and maps we touch need their own // backing arrays, or a rejected change would corrupt the live config. next.Block = cur.Block next.Block.Lists = append([]config.ListSource(nil), cur.Block.Lists...) next.Block.LocalZones = append([]string(nil), cur.Block.LocalZones...) next.Block.BlockEntries = append([]string(nil), cur.Block.BlockEntries...) next.Block.AllowEntries = append([]string(nil), cur.Block.AllowEntries...) next.ACL = cur.ACL next.ACL.Groups = append([]config.Group(nil), cur.ACL.Groups...) next.Upstream = cur.Upstream next.Upstream.Servers = append([]config.UpstreamServer(nil), cur.Upstream.Servers...) next.Upstream.Bootstraps = append([]config.UpstreamBootstrap(nil), cur.Upstream.Bootstraps...) if err := fn(&next); err != nil { return err } if err := next.Validate(); err != nil { return err } if a.api == nil { return a.cfg.Swap(&next) } return a.api.Commit(&next) } // domainCheck explains how a name would be treated. func (a *App) domainCheck(name string) (any, error) { entry := (*blocklist.Entry)(nil) if a.blocks != nil { if set := a.blocks.Holder().Get(); set != nil { entry = set.Match(name) } } out := map[string]any{ "name": name, "query": strings.TrimSuffix(name, "."), } if entry == nil { out["matched"] = false out["action"] = "none" out["source"] = "" return out, nil } kind := "exact" pattern := entry.Name if set := a.blocks.Holder().Get(); set != nil { // Recover which of the three maps matched so the UI can say *why*. switch set.MatchPattern(name) { case blocklist.PatternSub: kind = "subdomain" pattern = "*." + entry.Name case blocklist.PatternRoot: kind = "domain" pattern = "**." + entry.Name } } out["matched"] = true out["action"] = entry.Action.String() out["source"] = entry.Source out["categories"] = entry.Categories out["hits"] = entry.Hits() out["rule"] = pattern out["matched_pattern"] = pattern out["match_kind"] = kind return out, nil } // resolveTest runs one query through the live pipeline. It goes through the // engine rather than straight to an upstream: a test that skipped the // pipeline would report a healthy upstream while the blocklist, the ACL or the // local zones were what a client actually hit. func (a *App) resolveTest(ctx context.Context, name, qtype string) (any, error) { if a.engine == nil { return nil, errors.New("the query engine is not running") } qt := dns.TypeA if qtype = strings.TrimSpace(qtype); qtype != "" { var ok bool qt, ok = dns.StringToType[strings.ToUpper(qtype)] if !ok { return nil, fmt.Errorf("unknown record type %q", qtype) } } q := new(dns.Msg) q.SetQuestion(dns.Fqdn(name), qt) q.RecursionDesired = true res := a.engine.Handle(ctx, engine.Request{ Msg: q, Remote: netip.AddrFrom4([4]byte{127, 0, 0, 1}), Transport: "api", Start: time.Now(), }) out := map[string]any{ "name": dns.Fqdn(name), "type": dns.TypeToString[qt], "rcode": "SERVFAIL", "action": string(res.Action), "duration_ms": res.Duration.Milliseconds(), "upstream": res.Upstream, "group": res.Group, "cache": res.CacheKind, "local": res.Local, "truncated": res.Truncated, "blocked": res.BlockSource != "", "block_source": res.BlockSource, "answers": []string{}, } if res.Msg == nil { out["error"] = "no response" return out, nil } out["rcode"] = dns.RcodeToString[res.Msg.Rcode] answers := make([]string, 0, len(res.Msg.Answer)) for _, rr := range res.Msg.Answer { answers = append(answers, strings.TrimSpace(rr.String())) } out["answers"] = answers return out, nil } // listDomains pages the compiled rules. // request would be too slow, so the filter runs over a bounded sample unless // the query is a prefix/exact lookup, which the LRU can answer directly. func (a *App) listDomains(q api.DomainQuery) (any, error) { if a.blocks == nil { return map[string]any{"items": []any{}, "total": 0, "page": 1, "per_page": q.PerPage}, nil } set := a.blocks.Holder().Get() if set == nil { return map[string]any{"items": []any{}, "total": 0, "page": 1, "per_page": q.PerPage}, nil } if q.Text != "" && (q.Action == "" && q.List == "") { if e := set.Lookup(q.Text); e != nil { return map[string]any{ "items": []blocklist.Entry{*e}, "total": 1, "page": 1, "per_page": q.PerPage, }, nil } } type row struct { Name string `json:"name"` Action string `json:"action"` Source string `json:"source"` Categories []string `json:"categories"` Hits uint64 `json:"hits"` Rule string `json:"rule"` } text := strings.ToLower(strings.TrimSpace(q.Text)) action := strings.ToLower(q.Action) // The cap keeps a request bounded on a large set; the UI pages anyway. const scanCap = 400000 matches := make([]row, 0, 256) total := 0 set.Each(func(pattern string, e *blocklist.Entry, kind blocklist.PatternKind) bool { if len(matches) >= scanCap { return false } if action != "" && e.Action.String() != action { return true } if q.List != "" && e.Source != q.List { return true } if text != "" && !strings.Contains(pattern, text) { return true } total++ matches = append(matches, row{ Name: pattern, Action: e.Action.String(), Source: e.Source, Categories: e.Categories, Hits: e.Hits(), // The wire form, so the console can re-add this exact rule // (with its wildcard) instead of degrading it to an exact name. Rule: kind.WireName(pattern), }) return true }) sort.Slice(matches, func(i, j int) bool { if matches[i].Hits != matches[j].Hits { return matches[i].Hits > matches[j].Hits } return matches[i].Name < matches[j].Name }) start := (q.Page - 1) * q.PerPage if start > len(matches) { start = len(matches) } end := start + q.PerPage if end > len(matches) { end = len(matches) } return map[string]any{ "items": matches[start:end], "total": total, "page": q.Page, "per_page": q.PerPage, "truncated": total > len(matches), }, nil } // addDomain appends an inline rule and persists it. func (a *App) addDomain(_ context.Context, in api.DomainInput) (any, error) { name := strings.TrimSpace(in.Name) if name == "" { return nil, fmt.Errorf("name is required") } act, ok := blocklist.ParseAction(in.Action) if !ok { return nil, fmt.Errorf("unknown action %q", in.Action) } // A bulk paste goes in as several entries; a single name goes in as one. entries := []string{} if in.Content != "" { entries = strings.Split(in.Content, "\n") } else { entries = []string{name} } err := a.mutateConfig(func(c *config.Config) error { for _, e := range entries { e = strings.TrimSpace(e) if e == "" || strings.HasPrefix(e, "#") { continue } entry := e if act == blocklist.ActionBlock { c.Block.BlockEntries = append(c.Block.BlockEntries, entry) } else { c.Block.AllowEntries = append(c.Block.AllowEntries, entry) } } return nil }) if err != nil { return nil, err } if a.blocks != nil { if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil { a.log.Warn("refresh after add failed", "error", rerr) } } // A new rule changes the verdict for the whole subtree it covers, and an // allow rule can rescue a name many other lists blocked, so the honest // thing is to drop the affected answers rather than guess. a.invalidateBlocklistCache() return map[string]any{"ok": true, "name": name, "count": len(entries)}, nil } // deleteDomain removes a rule and persists the change. func (a *App) deleteDomain(_ context.Context, name, action, list string) error { target := strings.ToLower(strings.TrimSpace(name)) removed := 0 err := a.mutateConfig(func(c *config.Config) error { strip := func(entries []string) []string { out := entries[:0] for _, e := range entries { normal := strings.ToLower(strings.TrimSpace(e)) normal = strings.TrimPrefix(strings.TrimPrefix(normal, "*."), "**.") normal = strings.TrimSuffix(normal, ".") if normal == target { removed++ continue } out = append(out, e) } return out } c.Block.BlockEntries = strip(c.Block.BlockEntries) c.Block.AllowEntries = strip(c.Block.AllowEntries) // A file-backed list is edited by editing the file, so the API only // reports that rather than pretending it removed something. if list != "" && removed == 0 { return fmt.Errorf("%q comes from list %q; edit that list instead", name, list) } if removed == 0 { return fmt.Errorf("no inline rule matches %q", name) } return nil }) if err != nil { return err } if a.blocks != nil { if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil { a.log.Warn("refresh after delete failed", "error", rerr) } } a.invalidateBlocklistCache() return nil } // getACL returns the ACL groups. func (a *App) getACL() any { groups := a.engine.GroupStats() return map[string]any{ "groups": groups, "default_policy": a.cfg.Get().ACL.DefaultPolicy, "default_rate_limit": a.cfg.Get().ACL.DefaultRateLimit, } } // putACL validates and persists new ACL groups. func (a *App) putACL(_ context.Context, groups []config.Group, def string) (int64, error) { err := a.mutateConfig(func(c *config.Config) error { c.ACL.Groups = groups if def != "" { c.ACL.DefaultPolicy = def } return nil }) if err != nil { return 0, err } // An allow_domains or deny_domains change alters the verdict for names // already in the cache, so the cached answers are no longer trustworthy. if a.cache != nil { a.cache.Flush() } return a.cfg.Get().Version, nil } // getUpstreams returns the pool state. func (a *App) getUpstreams() any { cfg := a.cfg.Get() servers := make([]resolver.Stats, 0, len(a.pool.All())) for _, s := range a.pool.Stats() { servers = append(servers, s) } return map[string]any{ "policy": cfg.Upstream.Policy, "servers": servers, // Durations on the wire are duration strings, everywhere: this is the // same spelling config.yaml uses and the same one Duration.MarshalJSON // writes for bootstraps[].refresh. It used to be a float of seconds // while the console declared the field a string. "bootstraps": cfg.Upstream.Bootstraps, "healthcheck_interval": cfg.Upstream.HealthCheck, "max_attempts": cfg.Upstream.MaxAttempts, "breaker_threshold": cfg.Upstream.BreakerThreshold, "breaker_cooldown": cfg.Upstream.BreakerCooldown, } } // putUpstream validates and persists a new upstream config. func (a *App) putUpstream(_ context.Context, up config.Upstream) (int64, error) { err := a.mutateConfig(func(c *config.Config) error { c.Upstream = up return nil }) if err != nil { return 0, err } // The pool rebinds itself on the config swap; do it now so the UI sees // the new servers immediately rather than after the next reload. a.pool.Rebind(up) return a.cfg.Get().Version, nil } // searchLog pages the query log. func (a *App) searchLog(_ context.Context, q api.LogQuery) (any, error) { if a.logger == nil { return map[string]any{"items": []any{}, "total": 0, "offset": 0}, nil } page, err := a.logger.Search(qlog.Query{ Limit: q.Limit, Offset: q.Offset, Text: q.Text, Action: q.Action, Client: q.Client, RCode: q.RCode, BlockedOnly: q.BlockedOnly, Since: q.Since, }) if err != nil { return nil, err } out := make([]api.LogEntry, 0, len(page.Items)) for _, e := range page.Items { out = append(out, toLogEntry(e)) } return map[string]any{ "items": out, "total": page.Total, "offset": page.Offset, "limit": page.Limit, "has_more": len(out) == page.Limit, }, nil } func toLogEntry(e qlog.Entry) api.LogEntry { return api.LogEntry{ ID: e.ID, TS: e.TS, Client: e.Client, Group: e.Group, Name: e.Name, QType: e.QType, QTypeStr: e.QTypeStr, RCode: e.RCode, RCodeStr: e.RCodeName(), Action: string(e.Action), Response: e.Response, Duration: e.Duration, Upstream: e.Upstream, List: e.List, Protocol: e.Protocol, Blocked: e.Blocked, } } // logStats is the live stats frame pushed over SSE. func (a *App) logStats() any { t := metrics.TotalsNow() var hitRate float64 if t.Hits+t.Misses > 0 { hitRate = float64(t.Hits) / float64(t.Hits+t.Misses) } var blockedPct float64 if t.Queries > 0 { blockedPct = float64(t.Blocked) / float64(t.Queries) } return map[string]any{ "qps": a.series.qps(), "blocked_pct": blockedPct, "cache_hit_rate": hitRate, "queries_total": t.Queries, "blocked_total": t.Blocked, "errors_total": t.Errors, "inflight": t.Concurrent, "uptime_s": int64(time.Since(a.startedAt).Seconds()), } } // subscribeLog streams new entries to one SSE client. func (a *App) subscribeLog() (<-chan api.LogEntry, func()) { if a.logger == nil { ch := make(chan api.LogEntry) close(ch) return ch, func() {} } src, cancel := a.logger.Subscribe() out := make(chan api.LogEntry, 256) done := make(chan struct{}) go func() { defer close(out) for { select { case <-done: return case e, ok := <-src: if !ok { return } select { case out <- toLogEntry(e): case <-done: return } } } }() return out, func() { close(done) cancel() } } // clearLog empties the query log. func (a *App) clearLog() error { if a.logger == nil { return nil } return a.logger.Clear() } // cacheStats returns the cache snapshot. func (a *App) cacheStats() any { if a.cache == nil { return map[string]any{"enabled": false} } st := a.cache.Stats() return map[string]any{ "enabled": st.Enabled, "size": st.Size, "entries": st.Entries, "shards": st.Shards, "min_ttl_ms": st.MinTTLMS, "max_ttl_ms": st.MaxTTLMS, "serve_stale_ms": st.ServeStaleM, "negative_ttl_ms": st.NegativeMS, "hit_rate": st.HitRate, "hits": st.Hits, "misses": st.Misses, "stale_hits": st.StaleHits, "evictions": st.Evictions, "bytes": st.Bytes, "bytes_human": st.MemoryHuman, "persisted": st.Persist, } } // cacheTop returns the hottest cached names. func (a *App) cacheTop(limit int) any { if a.cache == nil { return []any{} } entries := a.cache.TopEntries(limit) out := make([]map[string]any, 0, len(entries)) for _, e := range entries { out = append(out, map[string]any{ "name": e.Name, "qtype": e.Type, "qtype_name": dns.Type(e.Type).String(), "hits": e.Hits, "bytes": e.Bytes, }) } return out } // flushCache empties the cache. func (a *App) flushCache() int64 { if a.cache == nil { return 0 } st := a.cache.Stats() a.cache.Flush() return int64(st.Entries) } // flush persists buffered state. func (a *App) flush() { if a.logger != nil { a.logger.Flush() } a.snapshotCache() } // seriesStore keeps the rolling counters the dashboard charts read. type seriesStore struct { mu sync.Mutex queries *metrics.Ring blocked *metrics.Ring errors *metrics.Ring cacheHit *metrics.Ring // The previous totals, used to derive per-bucket deltas. lastQueries uint64 lastBlocked uint64 lastErrors uint64 lastHits uint64 lastAt time.Time qpsValue float64 } func newSeriesStore() *seriesStore { return &seriesStore{ queries: metrics.NewRing(300, 5*time.Second), blocked: metrics.NewRing(300, 5*time.Second), errors: metrics.NewRing(300, 5*time.Second), cacheHit: metrics.NewRing(300, 5*time.Second), lastAt: time.Now(), } } // sample folds the delta since the last sample into the rings. Sampling on a // timer rather than per query keeps the query path free of any extra work and // is accurate enough for a chart. func (s *seriesStore) sample() { t := metrics.TotalsNow() s.mu.Lock() now := time.Now() dt := now.Sub(s.lastAt).Seconds() if dt > 0 { s.qpsValue = float64(t.Queries-s.lastQueries) / dt } // Ring buckets are 5s wide; scale the interval delta accordingly. scale := dt / 5 if scale < 0 { scale = 0 } dq := int64(float64(t.Queries-s.lastQueries) * scale) db := int64(float64(t.Blocked-s.lastBlocked) * scale) de := int64(float64(t.Errors-s.lastErrors) * scale) dc := int64(float64(t.Hits-s.lastHits) * scale) if dq > 0 { s.queries.Add(dq) } if db > 0 { s.blocked.Add(db) } if de > 0 { s.errors.Add(de) } if dc > 0 { s.cacheHit.Add(dc) } s.lastQueries, s.lastBlocked, s.lastErrors, s.lastHits = t.Queries, t.Blocked, t.Errors, t.Hits s.lastAt = now s.mu.Unlock() } func (s *seriesStore) qps() float64 { s.mu.Lock() defer s.mu.Unlock() return s.qpsValue } // dashboard builds the aggregated payload. func (a *App) dashboard(_ context.Context, rangeKey string) (any, error) { spec, ok := ranges[rangeKey] if !ok { spec = ranges["1h"] } t := metrics.TotalsNow() hitRate := 0.0 if t.Hits+t.Misses > 0 { hitRate = float64(t.Hits) / float64(t.Hits+t.Misses) } blockedPct := 0.0 if t.Queries > 0 { blockedPct = float64(t.Blocked) / float64(t.Queries) * 100 } // Build the series by walking the rings oldest-first and dropping anything // older than the requested range. var series []map[string]any now := time.Now() qts, qv := a.series.queries.Snapshot() _, bv := a.series.blocked.Snapshot() _, ev := a.series.errors.Snapshot() _, cv := a.series.cacheHit.Snapshot() series = make([]map[string]any, 0, len(qts)) for i := range qts { age := now.Sub(time.Unix(qts[i], 0)) if age > spec.duration { break } series = append(series, map[string]any{ "t": qts[i], "queries": qv[i], "blocked": bv[i], "errors": ev[i], "cache_hits": cv[i], }) } topBlocked := []any{} topQueried := []any{} if a.logger != nil { for _, d := range a.logger.TopBlocked(15) { topBlocked = append(topBlocked, map[string]any{ "domain": d.Name, "count": d.Count, "list": "", "qtype": d.QType, }) } for _, d := range a.logger.TopQueried(15) { topQueried = append(topQueried, map[string]any{ "domain": d.Name, "count": d.Count, "qtype": d.QType, "blocked": d.Blocked > 0, }) } } problems := a.Degraded() status := "ok" if len(problems) > 0 { status = "degraded" } out := map[string]any{ "health": status, "degraded_reason": strings.Join(problems, "; "), "version": a.ver, "node_id": a.clusterNodeID(), "uptime_seconds": int64(time.Since(a.startedAt).Seconds()), "queries_total": t.Queries, "queries_per_sec": a.series.qps(), "cache_hit_rate": hitRate, "blocked_total": t.Blocked, "blocked_pct": blockedPct, "errors_total": t.Errors, "inflight": t.Concurrent, "series": series, "top_blocked": topBlocked, "top_queried": topQueried, "upstreams": a.pool.Stats(), "listeners": a.dns.Listeners(), "transports": a.dns.ByTransport(), "blocklist_rules": a.blocklistRuleCount(), "cache_entries": a.cacheEntries(), "config_version": a.cfg.Get().Version, "generation": a.cfg.Generation(), } if a.logger != nil { recent := a.logger.Recent(20) items := make([]api.LogEntry, 0, len(recent)) for _, e := range recent { items = append(items, toLogEntry(e)) } out["recent"] = items } return out, nil } func (a *App) clusterNodeID() string { if a.cl != nil { return a.cl.NodeID() } return "" } func (a *App) blocklistRuleCount() int { if a.blocks == nil { return 0 } if st := a.blocks.State(); st != nil && st.Set != nil { return st.Set.Len() } return 0 } func (a *App) cacheEntries() int { if a.cache == nil { return 0 } return a.cache.Stats().Entries } // rangeSpec is one dashboard time range. type rangeSpec struct { duration time.Duration buckets int } var ranges = map[string]rangeSpec{ "15m": {duration: 15 * time.Minute, buckets: 45}, "1h": {duration: time.Hour, buckets: 60}, "6h": {duration: 6 * time.Hour, buckets: 72}, "24h": {duration: 24 * time.Hour, buckets: 96}, }