spb/internetpressure

1// Package health keeps the agent counters and serves the local /healthz and /metrics endpoints.2package health34import (5	"context"6	"encoding/json"7	"fmt"8	"net"9	"net/http"10	"runtime"11	"sort"12	"sync"13	"sync/atomic"14	"time"1516	"internetpressure.io/probe-agent/internal/protocol"17)1819// Gauges are read live from the rest of the agent.20type Gauges struct {21	Buffered      func() int22	SpoolBytes    func() int6423	ClockOffsetMs func() int6424	Targets       func() int25	Identity      func() *protocol.Identity26}2728// State is the shared counter set.29type State struct {30	ProbeID      string31	Version      string32	Capabilities []string33	Started      time.Time34	Gauges       Gauges3536	measurementsTotal atomic.Uint6437	errorsTotal       atomic.Uint6438	flushFailures     atomic.Uint6439	panics            atomic.Uint6440	lastFlush         atomic.Int64 // unix nano, 0 = never41	lastConfig        atomic.Int6442	lastConfigVersion atomic.Pointer[string]4344	mu      sync.Mutex45	byKind  map[string]uint6446	byError map[string]uint6447}4849// New creates a State.50func New(probeID, version string, caps []string, g Gauges) *State {51	s := &State{ProbeID: probeID, Version: version, Capabilities: caps, Started: time.Now(), Gauges: g,52		byKind: map[string]uint64{}, byError: map[string]uint64{}}53	empty := ""54	s.lastConfigVersion.Store(&empty)55	return s56}5758// Record counts one measurement.59func (s *State) Record(m protocol.Measurement) {60	s.measurementsTotal.Add(1)61	s.mu.Lock()62	s.byKind[m.Kind]++63	if !m.OK {64		s.errorsTotal.Add(1)65		code := m.Error66		if code == "" {67			code = protocol.ErrOther68		}69		s.byError[code]++70	}71	s.mu.Unlock()72}7374// RecordPanic counts a recovered panic in a check.75func (s *State) RecordPanic() { s.panics.Add(1) }7677// RecordFlushFailure counts a failed batch POST.78func (s *State) RecordFlushFailure() { s.flushFailures.Add(1) }7980// SetLastFlush marks a successful delivery.81func (s *State) SetLastFlush(t time.Time) { s.lastFlush.Store(t.UnixNano()) }8283// SetLastConfig marks a successful config fetch.84func (s *State) SetLastConfig(t time.Time, version string) {85	s.lastConfig.Store(t.UnixNano())86	v := version87	s.lastConfigVersion.Store(&v)88}8990// Uptime in seconds.91func (s *State) Uptime() int64 { return int64(time.Since(s.Started).Seconds()) }9293// Totals returns (measurements, errors).94func (s *State) Totals() (uint64, uint64) { return s.measurementsTotal.Load(), s.errorsTotal.Load() }9596// RSSMb returns the resident-set estimate (heap sys + stacks, in MiB) from runtime.MemStats.97func RSSMb() float64 {98	var ms runtime.MemStats99	runtime.ReadMemStats(&ms)100	return float64(ms.Sys-ms.HeapReleased) / (1024 * 1024)101}102103func fmtTime(unixNano int64) string {104	if unixNano == 0 {105		return ""106	}107	return protocol.FormatTime(time.Unix(0, unixNano))108}109110// Snapshot builds the protocol health block.111func (s *State) Snapshot() *protocol.Health {112	meas, errs := s.Totals()113	h := &protocol.Health{114		TS:                protocol.FormatTime(time.Now()),115		AgentVersion:      s.Version,116		UptimeS:           s.Uptime(),117		MeasurementsTotal: meas,118		ErrorsTotal:       errs,119		RSSMb:             round1(RSSMb()),120		Goroutines:        runtime.NumGoroutine(),121		Capabilities:      s.Capabilities,122		OS:                runtime.GOOS,123		Arch:              runtime.GOARCH,124	}125	if g := s.Gauges; true {126		if g.Buffered != nil {127			h.Buffered = g.Buffered()128		}129		if g.SpoolBytes != nil {130			h.SpoolBytes = g.SpoolBytes()131		}132		if g.ClockOffsetMs != nil {133			h.ClockOffsetMs = g.ClockOffsetMs()134		}135		if g.Identity != nil {136			h.Identity = g.Identity()137		}138	}139	return h140}141142func round1(v float64) float64 { return float64(int64(v*10+0.5)) / 10 }143144// Handler serves /healthz and /metrics.145func (s *State) Handler() http.Handler {146	mux := http.NewServeMux()147	mux.HandleFunc("/healthz", s.healthz)148	mux.HandleFunc("/metrics", s.metrics)149	return mux150}151152func (s *State) healthz(w http.ResponseWriter, _ *http.Request) {153	targets := 0154	if s.Gauges.Targets != nil {155		targets = s.Gauges.Targets()156	}157	var buffered int158	var spool, offset int64159	if s.Gauges.Buffered != nil {160		buffered = s.Gauges.Buffered()161	}162	if s.Gauges.SpoolBytes != nil {163		spool = s.Gauges.SpoolBytes()164	}165	if s.Gauges.ClockOffsetMs != nil {166		offset = s.Gauges.ClockOffsetMs()167	}168	w.Header().Set("Content-Type", "application/json")169	json.NewEncoder(w).Encode(map[string]any{170		"ok":              true,171		"probe_id":        s.ProbeID,172		"version":         s.Version,173		"buffered":        buffered,174		"spool_bytes":     spool,175		"last_flush":      fmtTime(s.lastFlush.Load()),176		"last_config":     fmtTime(s.lastConfig.Load()),177		"config_version":  *s.lastConfigVersion.Load(),178		"targets":         targets,179		"clock_offset_ms": offset,180		"uptime_s":        s.Uptime(),181		"capabilities":    s.Capabilities,182	})183}184185func (s *State) metrics(w http.ResponseWriter, _ *http.Request) {186	w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")187	s.mu.Lock()188	kinds := make([]string, 0, len(s.byKind))189	for k := range s.byKind {190		kinds = append(kinds, k)191	}192	sort.Strings(kinds)193	codes := make([]string, 0, len(s.byError))194	for k := range s.byError {195		codes = append(codes, k)196	}197	sort.Strings(codes)198	byKind := make(map[string]uint64, len(kinds))199	byErr := make(map[string]uint64, len(codes))200	for k, v := range s.byKind {201		byKind[k] = v202	}203	for k, v := range s.byError {204		byErr[k] = v205	}206	s.mu.Unlock()207208	fmt.Fprintln(w, "# TYPE ip_probe_measurements_total counter")209	for _, k := range kinds {210		fmt.Fprintf(w, "ip_probe_measurements_total{kind=%q} %d\n", k, byKind[k])211	}212	fmt.Fprintln(w, "# TYPE ip_probe_errors_total counter")213	for _, k := range codes {214		fmt.Fprintf(w, "ip_probe_errors_total{code=%q} %d\n", k, byErr[k])215	}216	fmt.Fprintln(w, "# TYPE ip_probe_flush_failures_total counter")217	fmt.Fprintf(w, "ip_probe_flush_failures_total %d\n", s.flushFailures.Load())218	fmt.Fprintln(w, "# TYPE ip_probe_check_panics_total counter")219	fmt.Fprintf(w, "ip_probe_check_panics_total %d\n", s.panics.Load())220	fmt.Fprintln(w, "# TYPE ip_probe_buffered gauge")221	if s.Gauges.Buffered != nil {222		fmt.Fprintf(w, "ip_probe_buffered %d\n", s.Gauges.Buffered())223	}224	fmt.Fprintln(w, "# TYPE ip_probe_spool_bytes gauge")225	if s.Gauges.SpoolBytes != nil {226		fmt.Fprintf(w, "ip_probe_spool_bytes %d\n", s.Gauges.SpoolBytes())227	}228	fmt.Fprintln(w, "# TYPE ip_probe_clock_offset_ms gauge")229	if s.Gauges.ClockOffsetMs != nil {230		fmt.Fprintf(w, "ip_probe_clock_offset_ms %d\n", s.Gauges.ClockOffsetMs())231	}232	fmt.Fprintln(w, "# TYPE ip_probe_targets gauge")233	if s.Gauges.Targets != nil {234		fmt.Fprintf(w, "ip_probe_targets %d\n", s.Gauges.Targets())235	}236	fmt.Fprintln(w, "# TYPE ip_probe_uptime_seconds gauge")237	fmt.Fprintf(w, "ip_probe_uptime_seconds %d\n", s.Uptime())238	fmt.Fprintln(w, "# TYPE ip_probe_rss_mb gauge")239	fmt.Fprintf(w, "ip_probe_rss_mb %.1f\n", RSSMb())240	fmt.Fprintln(w, "# TYPE ip_probe_goroutines gauge")241	fmt.Fprintf(w, "ip_probe_goroutines %d\n", runtime.NumGoroutine())242	fmt.Fprintln(w, "# TYPE ip_probe_info gauge")243	fmt.Fprintf(w, "ip_probe_info{probe_id=%q,version=%q,os=%q,arch=%q} 1\n", s.ProbeID, s.Version, runtime.GOOS, runtime.GOARCH)244}245246// Serve runs the local HTTP server on listen until ctx is cancelled.247func (s *State) Serve(ctx context.Context, listen string) error {248	ln, err := net.Listen("tcp", listen)249	if err != nil {250		return err251	}252	srv := &http.Server{Handler: s.Handler(), ReadHeaderTimeout: 5 * time.Second}253	go func() {254		<-ctx.Done()255		shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)256		defer cancel()257		srv.Shutdown(shutdownCtx)258	}()259	if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed {260		return err261	}262	return nil263}264