923 lines
25 KiB
Go
923 lines
25 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"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/metrics"
|
|
"git.k8s-git-k6n.net/mats/fancy-dns/internal/qlog"
|
|
"git.k8s-git-k6n.net/mats/fancy-dns/internal/resolver"
|
|
)
|
|
|
|
// 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,
|
|
ListDomains: a.listDomains,
|
|
AddDomain: a.addDomain,
|
|
DeleteDomain: a.deleteDomain,
|
|
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,
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// listDomains pages the compiled rules. Walking a million-entry map on every
|
|
// 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(),
|
|
Rule: kind.Pattern(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,
|
|
"bootstraps": cfg.Upstream.Bootstraps,
|
|
"healthcheck_interval": cfg.Upstream.HealthCheck.D().Seconds(),
|
|
"max_attempts": cfg.Upstream.MaxAttempts,
|
|
"breaker_threshold": cfg.Upstream.BreakerThreshold,
|
|
"breaker_cooldown": cfg.Upstream.BreakerCooldown.D().Seconds(),
|
|
}
|
|
}
|
|
|
|
// 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},
|
|
}
|