textmachine/platform/internal/metrics/metrics.go

// Package metrics is the platform's telemetry: the numbers an operator needs to see that the
// control plane is doing its job, in the form the ecosystem scrapes.
//
// # Why a Prometheus client and not expvar
//
// stdlib `expvar` is the alternative the zone's own norm points at first ("stdlib before a library",
// ENGINEERING_STANDARDS §1), and it does not carry this job: it has no labels, so "requests by route
// and status" cannot be expressed at all; no histograms, so a latency question is answerable only as
// an average, which is the one statistic that hides the tail; and its JSON is not a format any
// scraper reads without a translator. What it would save is a dependency, and what it would cost is
// writing the missing three by hand — which is the self-written path the same norm rules out.
//
// `prometheus/client_golang` v1.24.1 (released 24.07.2026, pin verified live) is the de-facto
// standard exposition, and the OpenTelemetry alternative is heavier for what this deployment is: a
// collector process to run, an export protocol to configure, and a scrape endpoint at the end of it
// anyway. The baseline this axis is measured against — the external one PD-115 asks for — is
// Prometheus's own naming practices (base units, `_total` on counters, no units in labels) and the
// four golden signals for WHAT to expose.
//
// # Cardinality
//
// Labels carry the route PATTERN, never the path: a raw path carries book and run ids, which is a
// user's library in an operator's index (PD-3) and an unbounded number of time series besides.
package metrics

import (
	"math"
	"net/http"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/collectors"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

// namespace prefixes every series of this service.
const namespace = "tm_platform"

// Metrics is the registry and the instruments of one process.
type Metrics struct {
	registry *prometheus.Registry

	// The runner's state, refreshed by the sweep. Gauges rather than collectors that query on
	// scrape: the sweep already reads the database on a timer, and a scrape that ran its own queries
	// would let anyone with access to the endpoint set the load on the control plane's database.
	queueDepth      prometheus.Gauge
	oldestHold      prometheus.Gauge
	quarantined     prometheus.Gauge
	parked          prometheus.Gauge
	liveRuns        prometheus.Gauge
	tailerLag       prometheus.Gauge
	booksInIntake   *prometheus.GaugeVec
	sweepDuration   *prometheus.HistogramVec
	sweepUnfinished *prometheus.CounterVec
	// stalledRuns and abandonedSurfaces are gauges rather than counters: the question is "how many
	// are stuck RIGHT NOW", and a counter of crossings answers a different one — it never comes back
	// down when the operator has dealt with them.
	stalledRuns       prometheus.Gauge
	abandonedSurfaces prometheus.Gauge
	// backupAge is how long ago the newest restore point was taken. In SECONDS, which is the base
	// unit Prometheus naming asks for, and a GAUGE rather than a counter of successes: a job that
	// has failed every night for a week and one that was never configured both show a flat success
	// counter, while age answers the question an alert is actually for — how much of the paid work
	// and of the credit ledger cannot be recovered right now.
	//
	// ⚠ It is set to +Inf when there is no restore point at all. NaN would be the other candidate
	// and it is the wrong one: `> threshold` is false for NaN, so an instance that never made a
	// backup would sit under every alert ever written for this series.
	backupAge prometheus.Gauge

	// The host's cap on concurrent engine cuts, and how hard it is being hit. Without these a cap is
	// indistinguishable from latency an operator has to guess the cause of — which is what the intake
	// looked like when it had no cap at all.
	cutsInFlight prometheus.Gauge
	cutsWaiting  prometheus.Gauge
	cutSlots     prometheus.Gauge
	cutWaits     prometheus.Counter
	cutGiveUps   prometheus.Counter
	// What the two counters above have already been told. The service counts cumulatively and a
	// Prometheus counter takes increments, so the difference is what is added; both are touched only
	// from the telemetry pass, which is one goroutine (cmd/tmplatformd, observe).
	countedCutWaits, countedCutGiveUps uint64

	requests *prometheus.CounterVec
	latency  *prometheus.HistogramVec
}

// New builds the instruments on a registry of this process's own.
//
// Its own, and not the package-global default: a global registry is shared with every library that
// ever registers into it, and a duplicate registration there is a panic at init in a process whose
// job is to stay up.
func New() *Metrics {
	m := &Metrics{registry: prometheus.NewRegistry()}
	m.queueDepth = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "queue_depth",
		Help: "Jobs in the queue that have not finished (pending, available, running, scheduled or retryable).",
	})
	m.oldestHold = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "oldest_open_hold_seconds",
		Help: "Age of the oldest reservation still open. A hold outlives its run only when a settlement could not be made.",
	})
	m.quarantined = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "quarantined_attempts",
		Help: "Live attempts whose journal is no longer being materialized. Such a run keeps going and keeps spending.",
	})
	m.parked = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "parked_attempts",
		Help: "Live attempts whose journal continued under another stream id, so their projection stopped there. Such a run keeps going and keeps spending; unlike a quarantine, nobody has to lift this.",
	})
	m.liveRuns = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "live_runs",
		Help: "Runs the database believes are still going.",
	})
	m.tailerLag = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "tailer_lag_bytes",
		Help: "Largest number of bytes any live run's journal holds beyond the platform's cursor.",
	})
	m.booksInIntake = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: namespace, Name: "books_in_intake",
		Help: "Books that have not finished intake, by status.",
	}, []string{"status"})
	m.sweepDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Namespace: namespace, Name: "sweep_duration_seconds",
		Help:    "How long one pass of a sweep took.",
		Buckets: sweepBuckets,
	}, []string{"sweep"})
	m.sweepUnfinished = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: namespace, Name: "sweep_unfinished_total",
		Help: "Passes that ran out of their time budget with work left. A rising count is starvation: the tail of the list is never reached (register row PD-169).",
	}, []string{"sweep"})
	m.stalledRuns = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "runs_stalled",
		Help: "Run attempts the reconciler has failed on enough times running to need an operator, whether the run is still live or has ENDED with its money unsettled (tmplatformctl runs).",
	})
	m.abandonedSurfaces = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "reading_surfaces_abandoned",
		Help: "Books whose reading surface could not be materialized and was written off; their text is whatever was materialized before (tmplatformctl books --abandoned).",
	})
	m.backupAge = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "backup_age_seconds",
		Help: "Age of the newest restore point; +Inf when this deployment has none.",
	})
	m.backupAge.Set(math.Inf(1))
	m.requests = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: namespace, Name: "http_requests_total",
		Help: "Requests served, by route pattern, method and status.",
	}, []string{"route", "method", "code"})
	m.latency = prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Namespace: namespace, Name: "http_request_duration_seconds",
		Help:    "Time to serve a request, by route pattern and method.",
		Buckets: latencyBuckets,
	}, []string{"route", "method"})
	m.cutsInFlight = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "cuts_in_flight",
		Help: "Engine cuts running right now, across the upload that cuts its own book, the queue's workers and the backstop sweep.",
	})
	m.cutsWaiting = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "cuts_waiting",
		Help: "Cuts queued for a slot right now. Sustained above zero means the cap, not the engine, is what shapes intake latency.",
	})
	m.cutSlots = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace, Name: "cut_slots",
		Help: "The cap itself (TM_PLATFORM_MAX_CUTS), published so saturation can be read without knowing the deployment's configuration.",
	})
	m.cutWaits = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: namespace, Name: "cut_waits_total",
		Help: "Cuts that found every slot taken and had to wait for one.",
	})
	m.cutGiveUps = prometheus.NewCounter(prometheus.CounterOpts{
		Namespace: namespace, Name: "cut_slot_timeouts_total",
		Help: "Cuts whose caller ran out of budget while waiting for a slot. An upload counted here is accepted `parsing` and finished by the queue, not refused.",
	})
	m.registry.MustRegister(m.queueDepth, m.oldestHold, m.quarantined, m.parked, m.liveRuns, m.tailerLag,
		m.booksInIntake, m.sweepDuration, m.sweepUnfinished, m.stalledRuns, m.abandonedSurfaces,
		m.backupAge, m.requests, m.latency,
		m.cutsInFlight, m.cutsWaiting, m.cutSlots, m.cutWaits, m.cutGiveUps,
		// The runtime and the process itself: memory, goroutines, file descriptors, CPU. They are
		// what answers "is this instance healthy" when none of the numbers above has moved.
		collectors.NewGoCollector(), collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
	return m
}

// sweepBuckets span one tick to well past the sweep's own budget: the question they answer is "is a
// pass still finishing", and the interesting shape is entirely in the tail.
var sweepBuckets = []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300}

// latencyBuckets carry both kinds of request this service serves: a read that must be milliseconds
// and an upload that is legitimately minutes.
var latencyBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300}

// Runner is what the reconciler's sweep observed about the world. Written as one struct because the
// numbers are read together, in one query, and a set of them from different moments would describe a
// state that never existed.
type Runner struct {
	QueueDepth          int64
	OldestHoldSeconds   float64
	QuarantinedAttempts int64
	LiveRuns            int64
	BooksUploading      int64
	BooksParsing        int64
	// StalledRuns and AbandonedSurfaces are the two states an operator has to ACT on, as opposed to
	// wait out. `sweep_unfinished_total` said a pass ran out of time and never said on what, so it
	// could rise for days with nothing to look at (register row PD-169); these name the state, and
	// `tmplatformctl runs` / `books --abandoned` name the rows behind each.
	StalledRuns       int64
	AbandonedSurfaces int64
	// ParkedAttempts counts live attempts whose projection parked. Its own series: a park and a
	// quarantine freeze a run's figures the same way and take opposite actions (PD-438).
	ParkedAttempts int64
}

// ObserveBackupAge publishes how old the newest restore point is. `has` false means there is none,
// and the series then reads +Inf — see the field.
func (m *Metrics) ObserveBackupAge(age time.Duration, has bool) {
	if m == nil {
		return
	}
	if !has {
		m.backupAge.Set(math.Inf(1))
		return
	}
	m.backupAge.Set(age.Seconds())
}

// ObserveRunner publishes one reading of the runner's state.
func (m *Metrics) ObserveRunner(r Runner) {
	if m == nil {
		return
	}
	m.queueDepth.Set(float64(r.QueueDepth))
	m.oldestHold.Set(r.OldestHoldSeconds)
	m.quarantined.Set(float64(r.QuarantinedAttempts))
	m.parked.Set(float64(r.ParkedAttempts))
	m.liveRuns.Set(float64(r.LiveRuns))
	m.booksInIntake.WithLabelValues("uploading").Set(float64(r.BooksUploading))
	m.booksInIntake.WithLabelValues("parsing").Set(float64(r.BooksParsing))
	m.stalledRuns.Set(float64(r.StalledRuns))
	m.abandonedSurfaces.Set(float64(r.AbandonedSurfaces))
}

// Cuts is one reading of the host's cap on concurrent engine cuts. Waited and GaveUp are cumulative
// since the process started, which is what lets this publish increments to a counter.
type Cuts struct {
	Limit    int
	InFlight int
	Waiting  int
	Waited   uint64
	GaveUp   uint64
}

// ObserveCuts publishes where the host's cut capacity stands.
func (m *Metrics) ObserveCuts(c Cuts) {
	if m == nil {
		return
	}
	m.cutSlots.Set(float64(c.Limit))
	m.cutsInFlight.Set(float64(c.InFlight))
	m.cutsWaiting.Set(float64(c.Waiting))
	// Only ever forward. A reading below what was already counted would mean the process restarted
	// under the same registry, which cannot happen — but a counter that went backwards would break
	// every rate() over it, so the guard costs one comparison and removes the class.
	if c.Waited > m.countedCutWaits {
		m.cutWaits.Add(float64(c.Waited - m.countedCutWaits))
		m.countedCutWaits = c.Waited
	}
	if c.GaveUp > m.countedCutGiveUps {
		m.cutGiveUps.Add(float64(c.GaveUp - m.countedCutGiveUps))
		m.countedCutGiveUps = c.GaveUp
	}
}

// ObserveTailerLag publishes how far the furthest-behind live run's cursor is from the end of its
// journal, in bytes.
func (m *Metrics) ObserveTailerLag(bytes int64) {
	if m == nil {
		return
	}
	m.tailerLag.Set(float64(bytes))
}

// ObserveSweep records one pass. unfinished means the pass ran out of time with work left, which is
// the starvation the per-run budget bounds and this counter makes visible.
func (m *Metrics) ObserveSweep(name string, d time.Duration, unfinished bool) {
	if m == nil {
		return
	}
	m.sweepDuration.WithLabelValues(name).Observe(d.Seconds())
	if unfinished {
		m.sweepUnfinished.WithLabelValues(name).Inc()
	}
}

// Handler serves the exposition endpoint.
func (m *Metrics) Handler() http.Handler {
	return promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{})
}

// Middleware counts and times every request.
//
// It sits where the access log sits — around the mux, so `Request.Pattern` is filled by the time it
// reads it, and outside the panic recovery, so a request that panicked is counted as the 500 it
// became.
func (m *Metrics) Middleware() func(http.Handler) http.Handler {
	return func(next http.Handler) http.Handler {
		if m == nil {
			return next
		}
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			start := time.Now()
			rec := &recorder{ResponseWriter: w, status: http.StatusOK}
			next.ServeHTTP(rec, r)
			route := r.Pattern
			if route == "" {
				// One series for everything unmatched, never the path: an unrouted request is where
				// an attacker chooses the label, and a label an attacker chooses is unbounded
				// cardinality.
				route = "(unmatched)"
			}
			method := knownMethod(r.Method)
			m.requests.WithLabelValues(route, method, strconv.Itoa(rec.status)).Inc()
			m.latency.WithLabelValues(route, method).Observe(time.Since(start).Seconds())
		})
	}
}

// knownMethod folds anything that is not a method this service serves into one label. A method is a
// token the CALLER chooses, and an unrouted request carries whatever they sent — so without this the
// series count is theirs to grow, exactly like the raw path would be.
func knownMethod(m string) string {
	switch m {
	case http.MethodGet, http.MethodHead, http.MethodPost, http.MethodPut, http.MethodPatch,
		http.MethodDelete, http.MethodOptions:
		return m
	default:
		return "(other)"
	}
}

// recorder captures the status code.
//
// Unwrap is NOT optional: http.ResponseController walks the wrappers through it, and the upload
// route reaches its own read deadline that way. A wrapper without it silently turns that call into
// "not supported" — and the upload then dies at the server's 30-second ReadTimeout instead.
type recorder struct {
	http.ResponseWriter
	status int
}

func (r *recorder) WriteHeader(code int) {
	r.status = code
	r.ResponseWriter.WriteHeader(code)
}

func (r *recorder) Unwrap() http.ResponseWriter { return r.ResponseWriter }