libnovel/backend/internal/runner/metrics.go

package runner

// metrics.go — Prometheus metrics HTTP endpoint for the runner.
//
// GET /metrics returns a Prometheus text/plain scrape response.
// Exposes:
//   - Standard Go runtime metrics (via promhttp)
//   - Runner task counters (tasks_running, tasks_completed, tasks_failed)
//   - Asynq queue metrics (registered in asynq_runner.go when Redis is enabled)
//
// GET /health — simple liveness probe.

import (
	"context"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"time"

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

// metricsServer serves GET /metrics and GET /health for the runner process.
type metricsServer struct {
	addr string
	r    *Runner
	log  *slog.Logger
}

func newMetricsServer(addr string, r *Runner, log *slog.Logger) *metricsServer {
	ms := &metricsServer{addr: addr, r: r, log: log}
	ms.registerCollectors()
	return ms
}

// registerCollectors registers runner-specific Prometheus collectors.
// Called once at construction; Asynq queue collector is registered separately
// in asynq_runner.go after the Redis connection is established.
func (ms *metricsServer) registerCollectors() {
	// Runner task gauges / counters backed by the atomic fields on Runner.
	ms.r.metricsRegistry.MustRegister(prometheus.NewGaugeFunc(
		prometheus.GaugeOpts{
			Namespace: "runner",
			Name:      "tasks_running",
			Help:      "Number of tasks currently being processed.",
		},
		func() float64 { return float64(ms.r.tasksRunning.Load()) },
	))
	ms.r.metricsRegistry.MustRegister(prometheus.NewCounterFunc(
		prometheus.CounterOpts{
			Namespace: "runner",
			Name:      "tasks_completed_total",
			Help:      "Total number of tasks completed successfully since startup.",
		},
		func() float64 { return float64(ms.r.tasksCompleted.Load()) },
	))
	ms.r.metricsRegistry.MustRegister(prometheus.NewCounterFunc(
		prometheus.CounterOpts{
			Namespace: "runner",
			Name:      "tasks_failed_total",
			Help:      "Total number of tasks that ended in failure since startup.",
		},
		func() float64 { return float64(ms.r.tasksFailed.Load()) },
	))
	ms.r.metricsRegistry.MustRegister(prometheus.NewGaugeFunc(
		prometheus.GaugeOpts{
			Namespace: "runner",
			Name:      "uptime_seconds",
			Help:      "Seconds since the runner process started.",
		},
		func() float64 { return time.Since(ms.r.startedAt).Seconds() },
	))
}

// ListenAndServe starts the HTTP server and blocks until ctx is cancelled or
// a fatal listen error occurs.
func (ms *metricsServer) ListenAndServe(ctx context.Context) error {
	mux := http.NewServeMux()
	mux.Handle("GET /metrics", promhttp.HandlerFor(ms.r.metricsRegistry, promhttp.HandlerOpts{}))
	mux.HandleFunc("GET /health", ms.handleHealth)

	srv := &http.Server{
		Addr:         ms.addr,
		Handler:      mux,
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 10 * time.Second,
		BaseContext:  func(_ net.Listener) context.Context { return ctx },
	}

	errCh := make(chan error, 1)
	go func() {
		ms.log.Info("runner: metrics server listening", "addr", ms.addr)
		errCh <- srv.ListenAndServe()
	}()

	select {
	case <-ctx.Done():
		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutCtx)
		return nil
	case err := <-errCh:
		return fmt.Errorf("runner: metrics server: %w", err)
	}
}

// handleHealth handles GET /health — simple liveness probe.
func (ms *metricsServer) handleHealth(w http.ResponseWriter, _ *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	_, _ = w.Write([]byte(`{"status":"ok"}`))
}