monitoring/internal/fleet/fleet.go

// Package fleet turns what the luxd validators and the Kubernetes API say into
// a Prometheus exposition.
//
// It is separate from the process that serves it for one reason: a metric whose
// meaning you cannot test without a cluster is a metric nobody trusts. Poll
// targets go in, exposition text comes out, and the differential tests drive the
// whole thing against fakes.
package fleet

import (
	"context"
	"fmt"
	"net/http"
	"net/url"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/luxfi/monitoring/internal/kube"
	"github.com/luxfi/monitoring/internal/luxd"
)

// Brand labels every series for white-labelled routing downstream.
const Brand = "lux"

// Networks maps a network to its namespace and luxd HTTP port, per the deployed
// StatefulSets. The ports genuinely differ per net; that is not a mistake to be
// normalised away here.
var Networks = []struct {
	Name string
	NS   string
	Port int
}{
	{"mainnet", "lux-mainnet", 9630},
	{"testnet", "lux-testnet", 9640},
	{"devnet", "lux-devnet", 9650},
}

// logPatterns map a log line to a counter label. Validated against live luxd
// logs: on healthy nodes all three match zero lines; on a crashing node
// "fatal EQUIVOCATION: conflicting finality cert ..." matches all three.
// certRejected REQUIRES the literal cert token next to a bad qualifier so it
// never matches routine "build block failure" noise.
var logPatterns = []struct {
	Name string
	RE   *regexp.Regexp
}{
	{"cert_rejected", regexp.MustCompile(`(?i)(certificate|\bcert\b).{0,60}(reject|invalid|conflict|expired|untrusted|unverified)` +
		`|(reject|invalid|conflict|expired|untrusted).{0,60}(certificate|\bcert\b)`)},
	{"equivocation", regexp.MustCompile(`(?i)equivocat`)},
	{"fatal", regexp.MustCompile(`(?i)\bfatal\b|\bpanic\b`)},
}

// Config is what the deployment sets.
type Config struct {
	Replicas     int
	PollInterval time.Duration
	RPCTimeout   time.Duration
	Headless     string

	// KubeBase and NodeBase override in-cluster addressing. Empty means the
	// real cluster; the differential tests set them to a fake.
	KubeBase string
	NodeBase func(net, ns, pod string, port int) string
}

// sample is one validator's answers for one poll. Every pointer field is
// "did not say", which is a different fact from zero and is rendered as an
// absent series rather than a fabricated number.
type sample struct {
	Net, Idx int
	Up       bool
	Healthy  bool
	Boot     bool
	Failing  []string
	Peers    *int
	AccH     *uint64
	AccHash  string
	AccTS    *uint64
	FinH     *uint64
	FinHash  string
	Restarts *int
	Ready    *bool
	Logs     map[string]int
}

// Exporter owns the poll loop and the current exposition.
type Exporter struct {
	cfg Config

	mu   sync.RWMutex
	text string

	// Cumulative log counters and the last log timestamp seen, per target.
	// Cumulative is the point: a counter that resets on a scrape gap hides
	// exactly the burst it exists to catch.
	logMu    sync.Mutex
	logCount map[string]map[string]int
	logSince map[string]string
}

func NewExporter(cfg Config) *Exporter {
	if cfg.Replicas <= 0 {
		cfg.Replicas = 5
	}
	if cfg.PollInterval <= 0 {
		cfg.PollInterval = 15 * time.Second
	}
	if cfg.RPCTimeout <= 0 {
		cfg.RPCTimeout = 3 * time.Second
	}
	if cfg.Headless == "" {
		cfg.Headless = "luxd-headless"
	}
	return &Exporter{
		cfg:      cfg,
		text:     "# exporter starting\n",
		logCount: map[string]map[string]int{},
		logSince: map[string]string{},
	}
}

// Exposition is the current scrape body.
func (e *Exporter) Exposition() string {
	e.mu.RLock()
	defer e.mu.RUnlock()
	return e.text
}

// Run polls until ctx is done.
func (e *Exporter) Run(ctx context.Context) {
	for {
		text := e.Once(ctx)
		e.mu.Lock()
		e.text = text
		e.mu.Unlock()
		select {
		case <-ctx.Done():
			return
		case <-time.After(e.cfg.PollInterval):
		}
	}
}

// Once performs one full poll and returns the exposition it produced. Exported
// so a test can take exactly one sample and assert on it.
func (e *Exporter) Once(ctx context.Context) string {
	start := time.Now()

	k := kube.New(5 * time.Second)
	if e.cfg.KubeBase != "" {
		k = kube.NewAt(e.cfg.KubeBase, 5*time.Second)
	}
	hc := &http.Client{Timeout: e.cfg.RPCTimeout}

	var (
		wg  sync.WaitGroup
		mu  sync.Mutex
		out []sample
	)
	for ni := range Networks {
		for idx := 0; idx < e.cfg.Replicas; idx++ {
			wg.Add(1)
			go func(ni, idx int) {
				defer wg.Done()
				s := e.poll(ctx, k, hc, ni, idx)
				mu.Lock()
				out = append(out, s)
				mu.Unlock()
			}(ni, idx)
		}
	}
	wg.Wait()

	return render(out, time.Since(start), e.cfg.Replicas)
}

func (e *Exporter) nodeBase(ni, idx int) string {
	n := Networks[ni]
	pod := fmt.Sprintf("luxd-%d", idx)
	if e.cfg.NodeBase != nil {
		return e.cfg.NodeBase(n.Name, n.NS, pod, n.Port)
	}
	return fmt.Sprintf("http://%s.%s.%s.svc:%d", pod, e.cfg.Headless, n.NS, n.Port)
}

// poll gathers one validator. Pod state and logs come from the apiserver and are
// independent of whether luxd itself answers — that independence is what makes
// "Ready while the RPC is dead" observable at all.
func (e *Exporter) poll(ctx context.Context, k *kube.Client, hc *http.Client, ni, idx int) sample {
	n := Networks[ni]
	pod := fmt.Sprintf("luxd-%d", idx)
	s := sample{Net: ni, Idx: idx}

	if p, err := k.GetPod(ctx, n.NS, pod); err == nil {
		r, ready := p.Restarts(), p.Ready()
		s.Restarts, s.Ready = &r, &ready
	}
	s.Logs = e.scanLogs(ctx, k, n.NS, pod, n.Name)

	c := &luxd.Client{Base: e.nodeBase(ni, idx), HTTP: hc}
	latest, err := c.CBlock(ctx, "latest")
	if err != nil || latest == nil {
		return s
	}
	s.Up = true
	h, ts := latest.Number, latest.Timestamp
	s.AccH, s.AccHash, s.AccTS = &h, latest.Hash, &ts

	if fin, err := c.CBlock(ctx, "finalized"); err == nil && fin != nil {
		fh := fin.Number
		s.FinH, s.FinHash = &fh, fin.Hash
	}
	if p, err := c.Peers(ctx); err == nil {
		s.Peers = &p
	}
	if b, err := c.Bootstrapped(ctx, "C"); err == nil {
		s.Boot = b
	}
	if hh, err := c.Health(ctx); err == nil {
		s.Healthy = hh.Healthy
		s.Failing = hh.FailedChecks()
		sort.Strings(s.Failing)
	}
	return s
}

// scanLogs advances the per-pod log cursor and increments the cumulative
// counters. The first scan primes the last 5 minutes, matching the alert window.
func (e *Exporter) scanLogs(ctx context.Context, k *kube.Client, ns, pod, net string) map[string]int {
	key := net + "/" + pod

	e.logMu.Lock()
	counts := e.logCount[key]
	if counts == nil {
		counts = map[string]int{}
		for _, p := range logPatterns {
			counts[p.Name] = 0
		}
		e.logCount[key] = counts
	}
	since := e.logSince[key]
	e.logMu.Unlock()

	q := "container=luxd&timestamps=true&tailLines=3000"
	if since != "" {
		q += "&sinceTime=" + url.QueryEscape(since)
	} else {
		q += "&sinceSeconds=300"
	}
	text, err := k.Logs(ctx, ns, pod, q)
	if err != nil {
		return copyCounts(counts)
	}

	newest := since
	for _, line := range strings.Split(text, "\n") {
		sp := strings.Index(line, " ")
		if sp <= 0 {
			continue
		}
		ts, msg := line[:sp], line[sp+1:]
		if since != "" && ts <= since {
			continue
		}
		if ts > newest {
			newest = ts
		}
		for _, p := range logPatterns {
			if p.RE.MatchString(msg) {
				e.logMu.Lock()
				counts[p.Name]++
				e.logMu.Unlock()
			}
		}
	}
	if newest == "" {
		newest = time.Now().UTC().Format("2006-01-02T15:04:05Z")
	}
	e.logMu.Lock()
	e.logSince[key] = newest
	out := copyCounts(counts)
	e.logMu.Unlock()
	return out
}

func copyCounts(in map[string]int) map[string]int {
	out := make(map[string]int, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}