Make the resolver pure DNS: JSON access log on stdout instead of a log store
The resolver kept its own query-log store (a bbolt ring, a live ring with SSE fanout, rolling per-domain counters and dashboard time-series) for the console. In a multi-node deployment each pod held a different slice of that, so every view disagreed with itself, and the cost sat on the query path. Charts belong in Grafana and log search in the log collector, so the store is gone. internal/accesslog writes one JSON line per notable query to stdout. The default is what was blocked and what failed (blocked, redirected, SERVFAIL, dropped, refused); log_cached and sample_rate add ordinary answers for a debugging session. Lines go through a bounded buffer to a single writer goroutine and are dropped, and counted in fancydns_access_log_dropped_total, when the stream cannot keep up, so a stalled log pipe can never stall a query. The config key is access_log (the old qlog block is ignored, like any unknown key). The metrics gain what the dashboard used to compute: fancydns_group_queries_total by ACL group and outcome and fancydns_responses_total by RCODE. Removed with the store: /api/dashboard, /api/logs, /api/logs/stream, /api/logs/clear, /api/cache/top, the series rings, and the cache's per-name hit table, which took a mutex and built a string on every cache hit. Tests that read the old log now read the access-log lines; the contract test drops the removed endpoints. The web console still calls them and is changed in the next commit. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
matsandClaude Sonnet 5.5 committed
1 parent
dd1117fd91
commit
c6c5ca4ff7
31 files changed
+818
-3389
No files matched your search
@@ -0,0 +1,67 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/miekg/dns"
|
||||
)
|
||||
|
||||
type lockedBuf struct {
|
||||
mu sync.Mutex
|
||||
b bytes.Buffer
|
||||
}
|
||||
|
||||
func (l *lockedBuf) Write(p []byte) (int, error) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return l.b.Write(p)
|
||||
}
|
||||
func (l *lockedBuf) lines() []map[string]any {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
var out []map[string]any
|
||||
for _, ln := range strings.Split(strings.TrimSpace(l.b.String()), "\n") {
|
||||
if ln == "" {
|
||||
continue
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal([]byte(ln), &m); err != nil {
|
||||
panic("access log line is not JSON: " + ln)
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// TestAccessLogDefaultsToBlockedAndFailed runs the whole resolver with the
|
||||
// shipped defaults: a blocked query leaves a line, an ordinary answer and a
|
||||
// cache hit leave none, and the line carries what a log search needs.
|
||||
func TestAccessLogDefaultsToBlockedAndFailed(t *testing.T) {
|
||||
out := &lockedBuf{}
|
||||
h := newHarness(t, nil, func(o *Options) { o.AccessLogOut = out })
|
||||
|
||||
h.query("www.example.test", dns.TypeA) // allowed
|
||||
h.query("www.example.test", dns.TypeA) // cached
|
||||
if m := h.query("ads.example.com", dns.TypeA); m.Rcode != dns.RcodeNameError {
|
||||
t.Fatalf("setup: ads.example.com rcode %d, want a block", m.Rcode)
|
||||
}
|
||||
h.app.logger.Flush()
|
||||
|
||||
lines := out.lines()
|
||||
if len(lines) != 1 {
|
||||
t.Fatalf("%d access-log lines, want exactly the blocked query: %v", len(lines), lines)
|
||||
}
|
||||
l := lines[0]
|
||||
for k, want := range map[string]any{
|
||||
"log": "access", "msg": "query", "name": "ads.example.com.", "action": "block",
|
||||
"list": "test-hosts", "qtype": "A", "client": "127.0.0.1", "protocol": "udp",
|
||||
} {
|
||||
if l[k] != want {
|
||||
t.Errorf("%s = %v, want %v (line %v)", k, l[k], want, l)
|
||||
}
|
||||
}
|
||||
}
|
||||
+19
-36
@@ -7,6 +7,7 @@ import (
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
@@ -19,6 +20,7 @@ import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.k8s-git-k6n.net/mats/fancy-dns/internal/accesslog"
|
||||
"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"
|
||||
@@ -30,7 +32,6 @@ import (
|
||||
"git.k8s-git-k6n.net/mats/fancy-dns/internal/kubezones"
|
||||
"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"
|
||||
@@ -51,6 +52,8 @@ type Options struct {
|
||||
LogFormat string
|
||||
// Kube turns on the Kubernetes source for groups, lists and zones.
|
||||
Kube KubeOptions
|
||||
// AccessLogOut receives the access log; nil means os.Stdout.
|
||||
AccessLogOut io.Writer
|
||||
}
|
||||
|
||||
// App is the running process.
|
||||
@@ -64,7 +67,7 @@ type App struct {
|
||||
overrideMu sync.Mutex
|
||||
cache *cache.Manager
|
||||
cstore *cache.Store
|
||||
logger *qlog.Logger
|
||||
logger *accesslog.Logger
|
||||
aclH *acl.Holder
|
||||
local *localzone.Holder
|
||||
// zones is the Kubernetes-sourced zone client.
|
||||
@@ -83,7 +86,6 @@ type App struct {
|
||||
dns *server.Server
|
||||
api *api.Server
|
||||
cl *cluster.Cluster
|
||||
series *seriesStore
|
||||
|
||||
apiServers []*http.Server
|
||||
apiLn []net.Listener
|
||||
@@ -292,29 +294,20 @@ func apiTLS(cfg *config.Config) (*tls.Config, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
// startLogger opens the query log.
|
||||
// startLogger starts the access log: JSON lines on stdout for blocked and
|
||||
// failed queries (and more, if configured). It stores nothing.
|
||||
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)
|
||||
if !cfg.AccessLog.Enabled {
|
||||
return nil
|
||||
}
|
||||
a.logger = l
|
||||
a.logger = accesslog.New(accesslog.Options{
|
||||
Writer: a.opt.AccessLogOut,
|
||||
Buffer: cfg.AccessLog.Buffer,
|
||||
LogBlocked: cfg.AccessLog.LogBlocked,
|
||||
LogErrors: cfg.AccessLog.LogErrors,
|
||||
LogCached: cfg.AccessLog.LogCached,
|
||||
SampleRate: cfg.AccessLog.SampleRate,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -441,7 +434,6 @@ func (a *App) startDNS(ctx context.Context) {
|
||||
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 {
|
||||
@@ -632,22 +624,16 @@ func (a *App) startBackground(ctx context.Context, cfg *config.Config) {
|
||||
}
|
||||
// Upstream health probes.
|
||||
go a.healthLoop(ctx)
|
||||
// Live gauges and the dashboard series sampler.
|
||||
// The in-flight gauge.
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
@@ -838,11 +824,8 @@ func (a *App) Stop(ctx context.Context) error {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
if a.logger != nil {
|
||||
a.logger.Flush()
|
||||
if err := a.logger.Close(); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
if err := a.logger.Close(); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
if a.cstore != nil {
|
||||
// Snapshot, close, then clear the pointer: the background snapshot
|
||||
|
||||
@@ -122,25 +122,6 @@ func TestAdminEndpointKeySets(t *testing.T) {
|
||||
wantKeys("components", "config", "started", "uptime_s", "version"))
|
||||
})
|
||||
|
||||
t.Run("dashboard", func(t *testing.T) {
|
||||
code, body := h.get("/api/dashboard?range=1h")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("GET /api/dashboard = %d: %s", code, body)
|
||||
}
|
||||
requireKeys(t, "GET /api/dashboard", keysOf(t, body), wantKeys(
|
||||
"blocked_pct", "blocked_total", "blocklist_rules", "cache_entries",
|
||||
"cache_hit_rate", "config_version", "degraded_reason", "errors_total",
|
||||
"generation", "health", "inflight", "listeners", "node_id",
|
||||
"queries_per_sec", "queries_total", "recent", "series", "top_blocked",
|
||||
"top_queried", "transports", "upstreams", "uptime_seconds", "version",
|
||||
))
|
||||
requireKeys(t, "GET /api/dashboard recent[0]", keysOfField(t, body, "recent", "0"), wantKeys(
|
||||
"action", "blocked", "client", "client_group", "duration_ms", "id",
|
||||
"list", "name", "protocol", "qtype", "qtype_name", "rcode", "rcode_name",
|
||||
"response", "ts", "upstream",
|
||||
))
|
||||
})
|
||||
|
||||
t.Run("blocklists", func(t *testing.T) {
|
||||
code, body := h.get("/api/blocklists")
|
||||
if code != http.StatusOK {
|
||||
@@ -210,23 +191,6 @@ func TestAdminEndpointKeySets(t *testing.T) {
|
||||
"weight"))
|
||||
})
|
||||
|
||||
t.Run("logs", func(t *testing.T) {
|
||||
code, body := h.get("/api/logs?limit=10")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("GET /api/logs = %d: %s", code, body)
|
||||
}
|
||||
requireKeys(t, "GET /api/logs", keysOf(t, body),
|
||||
wantKeys("has_more", "items", "limit", "offset", "total"))
|
||||
// Every key on every row. These were omitempty, so a row without an
|
||||
// ACL group or a blocklist arrived without the key and the console
|
||||
// rendered "undefined" instead of an empty cell.
|
||||
requireKeys(t, "GET /api/logs items[0]", keysOfField(t, body, "items", "0"), wantKeys(
|
||||
"action", "blocked", "client", "client_group", "duration_ms", "id",
|
||||
"list", "name", "protocol", "qtype", "qtype_name", "rcode", "rcode_name",
|
||||
"response", "ts", "upstream",
|
||||
))
|
||||
})
|
||||
|
||||
t.Run("cache", func(t *testing.T) {
|
||||
code, body := h.get("/api/cache")
|
||||
if code != http.StatusOK {
|
||||
@@ -243,19 +207,6 @@ func TestAdminEndpointKeySets(t *testing.T) {
|
||||
))
|
||||
})
|
||||
|
||||
t.Run("cache top", func(t *testing.T) {
|
||||
code, body := h.get("/api/cache/top?limit=5")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("GET /api/cache/top = %d: %s", code, body)
|
||||
}
|
||||
requireKeys(t, "GET /api/cache/top", keysOf(t, body), wantKeys("items"))
|
||||
// There is no `count` and no `last_seen`: the cache tracks a hit
|
||||
// counter and a byte count per name, and the console's "Lookups" and
|
||||
// "Last seen" columns were reading fields that never existed.
|
||||
requireKeys(t, "GET /api/cache/top items[0]", keysOfField(t, body, "items", "0"),
|
||||
wantKeys("bytes", "hits", "name", "qtype", "qtype_name"))
|
||||
})
|
||||
|
||||
t.Run("settings", func(t *testing.T) {
|
||||
code, body := h.get("/api/settings")
|
||||
if code != http.StatusOK {
|
||||
@@ -466,14 +417,6 @@ func TestRemainingEndpointKeySets(t *testing.T) {
|
||||
))
|
||||
})
|
||||
|
||||
t.Run("logs clear", func(t *testing.T) {
|
||||
code, body := h.do(http.MethodPost, "/api/logs/clear", "")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("POST /api/logs/clear = %d: %s", code, body)
|
||||
}
|
||||
requireKeys(t, "POST /api/logs/clear", keysOf(t, body), wantKeys("ok"))
|
||||
})
|
||||
|
||||
t.Run("settings validate", func(t *testing.T) {
|
||||
// A document that is valid: the issue lists are present and empty.
|
||||
code, body := h.do(http.MethodPost, "/api/settings/validate",
|
||||
@@ -516,43 +459,6 @@ func TestRemainingEndpointKeySets(t *testing.T) {
|
||||
requireKeys(t, "POST /api/settings/reload", keysOf(t, body), wantKeys("ok", "version"))
|
||||
})
|
||||
|
||||
t.Run("log stream ready frame", func(t *testing.T) {
|
||||
// The stream never ends, so read the first frame and hang up: the
|
||||
// console only needs to know the stream opened and what it carries.
|
||||
req, _ := http.NewRequest("GET", "http://"+h.api+"/api/logs/stream", nil)
|
||||
req.Header.Set("X-API-Key", h.key)
|
||||
res, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("GET /api/logs/stream: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode != http.StatusOK {
|
||||
t.Fatalf("GET /api/logs/stream = %d", res.StatusCode)
|
||||
}
|
||||
br := newLineReader(res.Body)
|
||||
var event, data string
|
||||
for range 2 {
|
||||
line, err := br.readLine()
|
||||
if err != nil {
|
||||
t.Fatalf("read stream: %v", err)
|
||||
}
|
||||
switch {
|
||||
case line == "":
|
||||
case len(line) > 7 && line[:7] == "event: ":
|
||||
event = line[7:]
|
||||
case len(line) > 6 && line[:6] == "data: ":
|
||||
data = line[6:]
|
||||
}
|
||||
if event != "" && data != "" {
|
||||
break
|
||||
}
|
||||
}
|
||||
if event != "ready" {
|
||||
t.Fatalf("first event = %q, want ready", event)
|
||||
}
|
||||
requireKeys(t, "GET /api/logs/stream ready frame", keysOf(t, []byte(data)),
|
||||
wantKeys("at", "version"))
|
||||
})
|
||||
}
|
||||
|
||||
// quote renders a Go string as a JSON string literal.
|
||||
|
||||
@@ -22,9 +22,9 @@ func TestDumpShapes(t *testing.T) {
|
||||
h := newHarness(t, nil)
|
||||
h.query("hot.test.", 1)
|
||||
for _, p := range []string{
|
||||
"/api/status", "/api/dashboard?range=1h", "/api/blocklists",
|
||||
"/api/status", "/api/blocklists",
|
||||
"/api/domains?per_page=3", "/api/clients", "/api/upstreams",
|
||||
"/api/logs?limit=3", "/api/cache", "/api/cache/top?limit=3",
|
||||
"/api/cache",
|
||||
"/api/settings", "/api/whoami",
|
||||
} {
|
||||
code, body := h.get(p)
|
||||
|
||||
@@ -140,7 +140,6 @@ func newHarness(t *testing.T, mutate func(*config.Config), opts ...func(*Options
|
||||
cfg.Upstream.HealthCheck = 0
|
||||
cfg.Upstream.Servers = []config.UpstreamServer{{Addr: up.addr, Proto: "udp", Weight: 1, HealthName: "test"}}
|
||||
cfg.Cache.Persist.Path = filepath.Join(dir, "cache.db")
|
||||
cfg.QueryLog.Path = filepath.Join(dir, "queries.db")
|
||||
cfg.Block.Lists = []config.ListSource{{
|
||||
Name: "test-hosts", Format: "hosts", Path: hosts, Enabled: true, Refresh: 0,
|
||||
}}
|
||||
@@ -159,7 +158,9 @@ func newHarness(t *testing.T, mutate func(*config.Config), opts ...func(*Options
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
opt := Options{ConfigPath: path, Version: "test"}
|
||||
// The access log goes to stdout by design; keep it out of the test output
|
||||
// unless a test captures it.
|
||||
opt := Options{ConfigPath: path, Version: "test", AccessLogOut: io.Discard}
|
||||
for _, o := range opts {
|
||||
o(&opt)
|
||||
}
|
||||
|
||||
+8
-335
@@ -8,7 +8,6 @@ import (
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/miekg/dns"
|
||||
@@ -17,8 +16,6 @@ import (
|
||||
"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"
|
||||
)
|
||||
@@ -31,7 +28,6 @@ func (a *App) hooks() api.Hooks {
|
||||
Logger: a.log,
|
||||
Version: a.ver,
|
||||
StartedAt: a.startedAt,
|
||||
Dashboard: a.dashboard,
|
||||
Status: a.status,
|
||||
Health: a.health,
|
||||
BlocklistState: a.blocklistState,
|
||||
@@ -61,12 +57,7 @@ func (a *App) hooks() api.Hooks {
|
||||
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,
|
||||
}
|
||||
@@ -269,9 +260,11 @@ func (a *App) health() any {
|
||||
if a.dns == nil || len(a.dns.Listeners()) == 0 {
|
||||
status = "down"
|
||||
}
|
||||
if a.logger != nil && a.logger.ErrorCount() > 0 && status == "ok" {
|
||||
if a.logger.Dropped() > 0 && status == "ok" {
|
||||
// Lines are dropped when the log stream cannot keep up. DNS is fine, but
|
||||
// whoever reads the log is missing events, which is worth a mention.
|
||||
status = "degraded"
|
||||
problems = append(problems, "query log write errors")
|
||||
problems = append(problems, "access log is dropping lines")
|
||||
}
|
||||
return map[string]any{
|
||||
"status": status,
|
||||
@@ -316,8 +309,9 @@ func (a *App) status() any {
|
||||
if a.cache != nil {
|
||||
out["cache"] = a.cacheStats()
|
||||
}
|
||||
if a.logger != nil {
|
||||
out["qlog"] = a.logger.Stats()
|
||||
out["access_log"] = map[string]any{
|
||||
"enabled": a.logger != nil,
|
||||
"dropped": a.logger.Dropped(),
|
||||
}
|
||||
if a.pool != nil {
|
||||
out["upstream"] = a.getUpstreams()
|
||||
@@ -866,106 +860,6 @@ func (a *App) putUpstream(_ context.Context, up config.Upstream) (int64, error)
|
||||
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 {
|
||||
@@ -992,25 +886,6 @@ func (a *App) cacheStats() any {
|
||||
}
|
||||
}
|
||||
|
||||
// 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 {
|
||||
@@ -1023,208 +898,6 @@ func (a *App) flushCache() int64 {
|
||||
|
||||
// flush persists buffered state.
|
||||
func (a *App) flush() {
|
||||
if a.logger != nil {
|
||||
a.logger.Flush()
|
||||
}
|
||||
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},
|
||||
}
|
||||
@@ -801,87 +801,34 @@ func TestBlocklistToggle(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
// TestDashboardAndStatus checks the aggregate endpoints return the fields the
|
||||
// UI reads, so a frontend change cannot silently drift from the API.
|
||||
func TestDashboardAndStatus(t *testing.T) {
|
||||
// TestStatus checks the status and health endpoints carry the sections the
|
||||
// console's status page reads.
|
||||
func TestStatus(t *testing.T) {
|
||||
h := newHarness(t, nil)
|
||||
h.query("www.example.test", dns.TypeA)
|
||||
h.query("ads.example.com", dns.TypeA)
|
||||
|
||||
code, out := h.get("/api/dashboard?range=1h")
|
||||
if code != 200 {
|
||||
t.Fatalf("dashboard = %d", code)
|
||||
code, out := h.get("/api/healthz")
|
||||
if code != 200 || !strings.Contains(string(out), `"status":"ok"`) {
|
||||
t.Fatalf("healthz = %d %s", code, out)
|
||||
}
|
||||
var d struct {
|
||||
Health string `json:"health"`
|
||||
Queries uint64 `json:"queries_total"`
|
||||
Blocked uint64 `json:"blocked_total"`
|
||||
HitRate any `json:"cache_hit_rate"`
|
||||
Series []any `json:"series"`
|
||||
TopBlocked []struct {
|
||||
Domain string `json:"domain"`
|
||||
Count uint64 `json:"count"`
|
||||
} `json:"top_blocked"`
|
||||
Upstreams []struct {
|
||||
Addr string `json:"addr"`
|
||||
Health string `json:"health"`
|
||||
} `json:"upstreams"`
|
||||
}
|
||||
if err := json.Unmarshal(out, &d); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if d.Health != "ok" {
|
||||
t.Errorf("health = %q, want ok", d.Health)
|
||||
}
|
||||
if d.Queries == 0 {
|
||||
t.Error("queries_total is zero after two queries")
|
||||
}
|
||||
if d.Blocked == 0 {
|
||||
t.Error("blocked_total is zero after a blocked query")
|
||||
}
|
||||
if len(d.Upstreams) == 0 {
|
||||
t.Error("upstreams empty in dashboard")
|
||||
}
|
||||
if len(d.Series) == 0 {
|
||||
t.Error("series empty in dashboard")
|
||||
}
|
||||
_ = d.HitRate
|
||||
_ = d.TopBlocked
|
||||
|
||||
code, out = h.get("/api/status")
|
||||
if code != 200 {
|
||||
t.Fatalf("status = %d", code)
|
||||
}
|
||||
if !strings.Contains(string(out), "listeners") {
|
||||
t.Error("status has no listeners section")
|
||||
var st struct {
|
||||
Components map[string]json.RawMessage `json:"components"`
|
||||
}
|
||||
}
|
||||
|
||||
// TestLogsAndClear covers the query log endpoint and the clear action.
|
||||
func TestLogsAndClear(t *testing.T) {
|
||||
h := newHarness(t, nil)
|
||||
h.query("www.example.test", dns.TypeA)
|
||||
h.query("ads.example.com", dns.TypeA)
|
||||
waitFor(t, 3*time.Second, func() bool {
|
||||
code, out := h.get("/api/logs?limit=50")
|
||||
if code != 200 {
|
||||
return false
|
||||
if err := json.Unmarshal(out, &st); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, want := range []string{"listeners", "blocklist", "cache", "upstream", "access_log"} {
|
||||
if _, ok := st.Components[want]; !ok {
|
||||
t.Errorf("status has no %q section", want)
|
||||
}
|
||||
return strings.Contains(string(out), "example.test")
|
||||
})
|
||||
code, out := h.get("/api/logs?q=ads.example.com&blocked=1")
|
||||
if code != 200 {
|
||||
t.Fatalf("logs = %d", code)
|
||||
}
|
||||
if !strings.Contains(string(out), "ads.example.com") {
|
||||
t.Error("filtered log did not include the blocked query")
|
||||
}
|
||||
if code, _ := h.do("POST", "/api/logs/clear", ""); code != 200 {
|
||||
t.Errorf("clear logs = %d, want 200", code)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCacheEndpoints covers the cache stats/top/flush trio.
|
||||
// TestCacheEndpoints covers the cache stats and flush.
|
||||
func TestCacheEndpoints(t *testing.T) {
|
||||
h := newHarness(t, nil)
|
||||
h.query("www.example.test", dns.TypeA)
|
||||
@@ -900,13 +847,6 @@ func TestCacheEndpoints(t *testing.T) {
|
||||
if c.Shards <= 0 {
|
||||
t.Error("shards is zero")
|
||||
}
|
||||
code, out = h.get("/api/cache/top?limit=10")
|
||||
if code != 200 {
|
||||
t.Fatalf("cache top = %d", code)
|
||||
}
|
||||
if !strings.Contains(string(out), "example.test") {
|
||||
t.Error("cache top does not list the queried name")
|
||||
}
|
||||
if code, _ := h.do("POST", "/api/cache/flush", ""); code != 200 {
|
||||
t.Errorf("flush = %d, want 200", code)
|
||||
}
|
||||
@@ -949,62 +889,6 @@ func TestReloadFromDisk(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestLogStream checks the SSE endpoint emits a ready frame and at least one
|
||||
// entry frame, and that it terminates when the client disconnects.
|
||||
func TestLogStream(t *testing.T) {
|
||||
h := newHarness(t, nil)
|
||||
req, err := httpNewRequest("GET", "http://"+h.api+"/api/logs/stream", h.key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// A streaming client must not time out while idle, so give it a generous
|
||||
// deadline; the test closes the body when it has seen what it needs.
|
||||
streamClient := &http.Client{Timeout: 20 * time.Second}
|
||||
resp, err := streamClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != 200 {
|
||||
t.Fatalf("stream = %d", resp.StatusCode)
|
||||
}
|
||||
if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
|
||||
t.Errorf("content-type = %q, want text/event-stream", ct)
|
||||
}
|
||||
if resp.Header.Get("X-Accel-Buffering") != "no" {
|
||||
t.Error("missing X-Accel-Buffering: no, a proxy would buffer the stream")
|
||||
}
|
||||
// Generate a query and read frames until we see an entry.
|
||||
go func() {
|
||||
time.Sleep(150 * time.Millisecond)
|
||||
h.query("www.example.test", dns.TypeA)
|
||||
}()
|
||||
buf := make([]byte, 4096)
|
||||
seen := map[string]bool{}
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
n, err := resp.Body.Read(buf)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
chunk := string(buf[:n])
|
||||
for _, ev := range []string{"event: ready", "event: entry", "event: stats"} {
|
||||
if strings.Contains(chunk, ev) {
|
||||
seen[ev] = true
|
||||
}
|
||||
}
|
||||
if seen["event: ready"] && seen["event: entry"] {
|
||||
break
|
||||
}
|
||||
}
|
||||
if !seen["event: ready"] {
|
||||
t.Error("no ready event")
|
||||
}
|
||||
if !seen["event: entry"] {
|
||||
t.Error("no entry event for a query issued while streaming")
|
||||
}
|
||||
}
|
||||
|
||||
// TestMetricsEndpoint checks Prometheus output is served and protected.
|
||||
func TestMetricsEndpoint(t *testing.T) {
|
||||
h := newHarness(t, nil)
|
||||
@@ -1066,13 +950,13 @@ func TestConcurrentQueries(t *testing.T) {
|
||||
apiDone := make(chan error, workers)
|
||||
for w := 0; w < workers; w++ {
|
||||
go func() {
|
||||
res, err := httpGet("http://"+h.api+"/api/dashboard", h.key)
|
||||
res, err := httpGet("http://"+h.api+"/api/status", h.key)
|
||||
if err != nil {
|
||||
apiDone <- err
|
||||
return
|
||||
}
|
||||
if res.status != 200 {
|
||||
apiDone <- fmt.Errorf("dashboard = %d", res.status)
|
||||
apiDone <- fmt.Errorf("status = %d", res.status)
|
||||
return
|
||||
}
|
||||
apiDone <- nil
|
||||
|
||||
Reference in new issue
Block a user