jarl/internal/stats/poll.go

package stats

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"
)

// Snapshot is named's cumulative counters at one moment, flattened to the
// keys Jarl stores (see Counters in docs/stats.md).
type Snapshot struct {
	BootTime string
	Values   map[string]int64
	// Gauges are named's current values (memory, cache size, sockets),
	// not counters.
	Gauges     map[string]int64
	Version    string
	ConfigTime string
}

// serverJSON is the part of /json/v1 Jarl uses.
type serverJSON struct {
	BootTime   string           `json:"boot-time"`
	ConfigTime string           `json:"config-time"`
	Version    string           `json:"version"`
	SockStats  map[string]int64 `json:"sockstats"`
	Memory     struct {
		InUse int64 `json:"InUse"`
	} `json:"memory"`
	Opcodes map[string]int64 `json:"opcodes"`
	RCodes  map[string]int64 `json:"rcodes"`
	QTypes  map[string]int64 `json:"qtypes"`
	NSStats map[string]int64 `json:"nsstats"`
	Views   map[string]struct {
		Resolver struct {
			Stats      map[string]int64 `json:"stats"`
			CacheStats map[string]int64 `json:"cachestats"`
			Cache      map[string]int64 `json:"cache"` // RRsets in the cache by type
		} `json:"resolver"`
	} `json:"views"`
}

// nsstatsKeys maps named's server counters to Jarl's counter names.
var nsstatsKeys = map[string]string{
	"QryAuthAns":   "answer:authoritative",
	"QryNoauthAns": "answer:recursive",
	"QryRecursion": "recursion",
	"QryDropped":   "dropped",
	"RateDropped":  "ratelimit:dropped",
	"RateSlipped":  "ratelimit:slipped",
	"QryTCP":       "transport:tcp",
	"QryUDP":       "transport:udp",
	"QryDuplicate": "duplicate",
}

// rttKeys maps named's upstream round-trip buckets to readable names.
var rttKeys = map[string]string{
	"QryRTT10":    "rtt:<10ms",
	"QryRTT100":   "rtt:10-100ms",
	"QryRTT500":   "rtt:100-500ms",
	"QryRTT800":   "rtt:500-800ms",
	"QryRTT1600":  "rtt:800-1600ms",
	"QryRTT1600+": "rtt:>1600ms",
}

var dnssecKeys = map[string]string{"ValOk": "dnssec:valid", "ValNegOk": "dnssec:valid-negative", "ValFail": "dnssec:failed"}

// Flatten turns the statistics channel's JSON into a snapshot.
func Flatten(body []byte) (*Snapshot, error) {
	var j serverJSON
	if err := json.Unmarshal(body, &j); err != nil {
		return nil, fmt.Errorf("unexpected statistics data: %w", err)
	}
	v := map[string]int64{"queries": j.Opcodes["QUERY"]}
	for k, n := range j.RCodes {
		if n != 0 && !strings.HasPrefix(k, "RESERVED") && (k[0] < '0' || k[0] > '9') {
			v["rcode:"+k] = n
		}
	}
	for k, n := range j.QTypes {
		v["qtype:"+k] = n
	}
	for k, name := range nsstatsKeys {
		if n := j.NSStats[k]; n != 0 {
			v[name] = n
		}
	}
	g := map[string]int64{"named:mem": j.Memory.InUse,
		"named:recursion-highwater": j.NSStats["RecursHighwater"], "named:tcp-highwater": j.NSStats["TCPConnHighWater"]}
	for k, n := range j.SockStats {
		switch {
		case strings.HasPrefix(k, "UDP") && strings.HasSuffix(k, "Active"):
			g["sockets:udp"] += n
		case strings.HasPrefix(k, "TCP") && strings.HasSuffix(k, "Active"):
			g["sockets:tcp"] += n
		}
	}
	for _, view := range j.Views {
		g["cache:mem"] += view.Resolver.CacheStats["TreeMemInUse"] + view.Resolver.CacheStats["HeapMemInUse"]
		for _, n := range view.Resolver.Cache {
			g["cache:rrsets"] += n
		}
		for k, name := range rttKeys {
			v[name] += view.Resolver.Stats[k]
		}
		for k, name := range dnssecKeys {
			v[name] += view.Resolver.Stats[k]
		}
		v["cache:hits"] += view.Resolver.CacheStats["QueryHits"]
		v["cache:misses"] += view.Resolver.CacheStats["QueryMisses"]
	}
	for k, n := range v {
		if n == 0 && k != "queries" {
			delete(v, k)
		}
	}
	return &Snapshot{BootTime: j.BootTime, Values: v, Gauges: g, Version: j.Version, ConfigTime: j.ConfigTime}, nil
}

// Delta returns the change from prev to cur. When named restarted (its
// boot time changed), the counters started again from zero, so the
// current values are the change. Without a restart, a counter can still
// go down: the per-view ones (cache, upstream times, validation) are
// summed over named's views, and a reconfig that adds, removes or
// recreates views changes that sum. Such a counter counts 0 for the
// minute; it must never make every counter count from zero, which would
// add everything since named started (millions of "queries" in a minute).
func Delta(prev, cur *Snapshot) map[string]int64 {
	out := map[string]int64{}
	restarted := prev.BootTime != cur.BootTime
	for k, n := range cur.Values {
		d := n
		if !restarted {
			d = max(n-prev.Values[k], 0)
		}
		if d != 0 || k == "queries" {
			out[k] = d
		}
	}
	return out
}

// Poller reads named's statistics channel.
type Poller struct {
	URL    string // e.g. http://127.0.0.1:8058/json/v1 (server, memory and sockets)
	Client *http.Client
}

// NewPoller returns a poller for a statistics channel address.
func NewPoller(addr string) *Poller {
	return &Poller{URL: "http://" + addr + "/json/v1", Client: &http.Client{Timeout: 10 * time.Second}}
}

// Poll fetches a snapshot.
func (p *Poller) Poll(ctx context.Context) (*Snapshot, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.URL, nil)
	if err != nil {
		return nil, err
	}
	resp, err := p.Client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("named's statistics channel is not reachable (%v); is the configuration with statistics applied and named running?", err)
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(io.LimitReader(resp.Body, 64<<20))
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("named's statistics channel answered %s", resp.Status)
	}
	return Flatten(body)
}