From cc53c1a3e49cc1f441ef35e15042114b71f8090c Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Fri, 7 Aug 2026 04:24:57 +0000 Subject: [PATCH] =?UTF-8?q?feat(P01):=20metrics=20endpoint=20=E2=80=94=20h?= =?UTF-8?q?and-rolled=20Prometheus=20text=20exposition?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit internal/transport/metrics.go: Metrics struct with counters/gauges, WritePrometheus text exposition; internal/cli/metrics.go: orca metrics HTTP server on :9100 serving /metrics + /healthz. No client_golang dep. ---ci--- project: orca phase: 01 milestone: v0.11 status: execute ---/ci--- --- internal/cli/metrics.go | 133 +++++++++++++++++++++++++++++ internal/cli/metrics_test.go | 116 +++++++++++++++++++++++++ internal/transport/metrics.go | 122 ++++++++++++++++++++++++++ internal/transport/metrics_test.go | 126 +++++++++++++++++++++++++++ 4 files changed, 497 insertions(+) create mode 100644 internal/cli/metrics.go create mode 100644 internal/cli/metrics_test.go create mode 100644 internal/transport/metrics.go create mode 100644 internal/transport/metrics_test.go diff --git a/internal/cli/metrics.go b/internal/cli/metrics.go new file mode 100644 index 0000000..d3b287f --- /dev/null +++ b/internal/cli/metrics.go @@ -0,0 +1,133 @@ +package cli + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/http" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "github.com/spf13/cobra" + + "git.cloudinit.dev/coreci/orca/internal/store" + "git.cloudinit.dev/coreci/orca/internal/transport" +) + +var metricsAddr string + +var metricsCmd = &cobra.Command{ + Use: "metrics", + Short: "Run the orca metrics endpoint (Prometheus text exposition)", + Long: `Start a standalone HTTP server exposing Prometheus text-exposition +metrics at /metrics and a liveness probe at /healthz. Designed to run on the +CLI host or a designated metrics host; polls cluster state periodically and +updates gauges. No orca daemon required (R-001).`, + RunE: func(cmd *cobra.Command, args []string) error { + log := newLogger() + m := transport.NewMetrics() + + db, closer, err := openDB() + if err != nil { + log.Warn("metrics: open db failed, gauges will stay 0", "err", err) + db = nil + closer = func() error { return nil } + } + defer closer() + + var pollWG sync.WaitGroup + pollCtx, pollCancel := context.WithCancel(cmd.Context()) + defer func() { + pollCancel() + pollWG.Wait() + }() + + pollWG.Add(1) + go func() { + defer pollWG.Done() + pollLoop(pollCtx, m, db, log) + }() + + mux := http.NewServeMux() + mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + if err := m.WritePrometheus(w); err != nil { + log.Warn("metrics: write exposition failed", "err", err) + } + }) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok\n")) + }) + + srv := &http.Server{ + Addr: metricsAddr, + Handler: mux, + } + + errCh := make(chan error, 1) + go func() { + err := srv.ListenAndServe() + if err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + fmt.Fprintf(cmd.OutOrStdout(), "✓ orca metrics listening on %s\n", metricsAddr) + fmt.Fprintln(cmd.OutOrStdout(), " /metrics - Prometheus text exposition") + fmt.Fprintln(cmd.OutOrStdout(), " /healthz - liveness probe") + fmt.Fprintln(cmd.OutOrStdout(), " press Ctrl+C to stop") + + ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer stop() + + select { + case <-ctx.Done(): + fmt.Fprintln(cmd.OutOrStdout(), "\nshutting down...") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return srv.Shutdown(shutdownCtx) + case err := <-errCh: + return fmt.Errorf("metrics server: %w", err) + } + }, +} + +func pollLoop(ctx context.Context, m *transport.Metrics, db *sql.DB, log interface{ Warn(string, ...any) }) { + if db == nil { + return + } + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + refresh(ctx, m, db, log) + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + refresh(ctx, m, db, log) + } + } +} + +func refresh(ctx context.Context, m *transport.Metrics, db *sql.DB, log interface{ Warn(string, ...any) }) { + if nodes, err := store.NewNodeRepo(db).List(ctx); err != nil { + log.Warn("metrics: node list failed", "err", err) + } else { + m.SetGauge("nodes_total", float64(len(nodes))) + } + if jobs, err := store.NewJobRepo(db).List(ctx); err != nil { + log.Warn("metrics: job list failed", "err", err) + } else { + m.SetGauge("allocs_total", float64(len(jobs))) + } +} + +func init() { + metricsCmd.Flags().StringVar(&metricsAddr, "addr", ":9100", "listen address for the metrics HTTP server") + rootCmd.AddCommand(metricsCmd) +} diff --git a/internal/cli/metrics_test.go b/internal/cli/metrics_test.go new file mode 100644 index 0000000..566ea89 --- /dev/null +++ b/internal/cli/metrics_test.go @@ -0,0 +1,116 @@ +package cli + +import ( + "bytes" + "context" + "io" + "net" + "net/http" + "strings" + "testing" + "time" +) + +func TestMetricsCmdRegistered(t *testing.T) { + found := false + for _, c := range rootCmd.Commands() { + if c.Name() == "metrics" { + found = true + break + } + } + if !found { + t.Fatal("metricsCmd not registered on root") + } +} + +func TestMetricsAddrFlagDefault(t *testing.T) { + f := metricsCmd.Flags().Lookup("addr") + if f == nil { + t.Fatal("--addr flag not registered on metricsCmd") + } + if f.DefValue != ":9100" { + t.Errorf("--addr default = %q, want %q", f.DefValue, ":9100") + } +} + +func TestMetricsEndpoints(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + resetRootFlags(t) + + // Pick a free port by briefly listening then closing. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen probe: %v", err) + } + addr := ln.Addr().String() + _ = ln.Close() + + metricsAddr = addr + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cmd := metricsCmd + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(&out) + cmd.SetContext(ctx) + + errCh := make(chan error, 1) + go func() { + errCh <- cmd.RunE(cmd, nil) + }() + + deadline := time.Now().Add(5 * time.Second) + var resp *http.Response + for time.Now().Before(deadline) { + resp, err = http.Get("http://" + addr + "/healthz") + if err == nil { + break + } + time.Sleep(20 * time.Millisecond) + } + if err != nil { + t.Fatalf("GET /healthz: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Errorf("/healthz status = %d, want 200", resp.StatusCode) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if !strings.HasPrefix(string(body), "ok") { + t.Errorf("/healthz body = %q, want \"ok\"", string(body)) + } + + resp2, err := http.Get("http://" + addr + "/metrics") + if err != nil { + t.Fatalf("GET /metrics: %v", err) + } + defer resp2.Body.Close() + if resp2.StatusCode != http.StatusOK { + t.Errorf("/metrics status = %d, want 200", resp2.StatusCode) + } + mbody, _ := io.ReadAll(resp2.Body) + ms := string(mbody) + for _, name := range []string{ + "txns_applied_total", + "txns_drifted_total", + "drifts_remediated_total", + "peers_total", + "nodes_total", + "allocs_total", + } { + if !strings.Contains(ms, name) { + t.Errorf("/metrics missing %q\n---\n%s", name, ms) + } + } + + cancel() + select { + case <-errCh: + case <-time.After(3 * time.Second): + t.Fatal("metrics command did not stop after cancel") + } +} diff --git a/internal/transport/metrics.go b/internal/transport/metrics.go new file mode 100644 index 0000000..2049f3d --- /dev/null +++ b/internal/transport/metrics.go @@ -0,0 +1,122 @@ +package transport + +import ( + "fmt" + "io" + "strconv" + "sync" + "sync/atomic" +) + +type metricDef struct { + help string + typ string +} + +var metricOrder = []string{ + "txns_applied_total", + "txns_drifted_total", + "drifts_remediated_total", + "peers_total", + "nodes_total", + "allocs_total", +} + +var metricMeta = map[string]metricDef{ + "txns_applied_total": {"Total transactions applied", "counter"}, + "txns_drifted_total": {"Total transactions drifted", "counter"}, + "drifts_remediated_total": {"Total drifts remediated", "counter"}, + "peers_total": {"Current peer count", "gauge"}, + "nodes_total": {"Current node count", "gauge"}, + "allocs_total": {"Current allocation count", "gauge"}, +} + +type Metrics struct { + cmu sync.RWMutex + counters map[string]*atomic.Int64 + + gmu sync.RWMutex + gauges map[string]float64 +} + +func NewMetrics() *Metrics { + m := &Metrics{ + counters: make(map[string]*atomic.Int64), + gauges: make(map[string]float64), + } + for name, meta := range metricMeta { + if meta.typ == "counter" { + m.counters[name] = new(atomic.Int64) + } else { + m.gauges[name] = 0 + } + } + return m +} + +func (m *Metrics) counter(name string) *atomic.Int64 { + m.cmu.RLock() + c := m.counters[name] + m.cmu.RUnlock() + if c != nil { + return c + } + m.cmu.Lock() + defer m.cmu.Unlock() + if c = m.counters[name]; c != nil { + return c + } + c = new(atomic.Int64) + m.counters[name] = c + return c +} + +func (m *Metrics) IncCounter(name string) { + m.counter(name).Add(1) +} + +func (m *Metrics) AddCounter(name string, delta int64) { + m.counter(name).Add(delta) +} + +func (m *Metrics) SetGauge(name string, value float64) { + m.gmu.Lock() + m.gauges[name] = value + m.gmu.Unlock() +} + +func (m *Metrics) WritePrometheus(w io.Writer) error { + for _, name := range metricOrder { + meta, ok := metricMeta[name] + if !ok { + continue + } + if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, meta.help); err != nil { + return err + } + if _, err := fmt.Fprintf(w, "# TYPE %s %s\n", name, meta.typ); err != nil { + return err + } + switch meta.typ { + case "counter": + m.cmu.RLock() + c := m.counters[name] + m.cmu.RUnlock() + var v int64 + if c != nil { + v = c.Load() + } + if _, err := fmt.Fprintf(w, "%s %d\n", name, v); err != nil { + return err + } + case "gauge": + m.gmu.RLock() + v := m.gauges[name] + m.gmu.RUnlock() + if _, err := fmt.Fprintf(w, "%s %s\n", name, strconv.FormatFloat(v, 'g', -1, 64)); err != nil { + return err + } + } + } + return nil +} diff --git a/internal/transport/metrics_test.go b/internal/transport/metrics_test.go new file mode 100644 index 0000000..f9c9e6d --- /dev/null +++ b/internal/transport/metrics_test.go @@ -0,0 +1,126 @@ +package transport + +import ( + "context" + "io" + "net" + "net/http" + "strings" + "sync" + "testing" + "time" +) + +func TestWritePrometheusFormat(t *testing.T) { + m := NewMetrics() + m.IncCounter("txns_applied_total") + m.AddCounter("txns_applied_total", 41) + m.SetGauge("peers_total", 5) + m.SetGauge("nodes_total", 3.5) + + var sb strings.Builder + if err := m.WritePrometheus(&sb); err != nil { + t.Fatalf("WritePrometheus: %v", err) + } + out := sb.String() + + want := []string{ + "# HELP txns_applied_total Total transactions applied", + "# TYPE txns_applied_total counter", + "txns_applied_total 42", + "# HELP txns_drifted_total Total transactions drifted", + "# TYPE txns_drifted_total counter", + "txns_drifted_total 0", + "# HELP drifts_remediated_total Total drifts remediated", + "# TYPE drifts_remediated_total counter", + "drifts_remediated_total 0", + "# HELP peers_total Current peer count", + "# TYPE peers_total gauge", + "peers_total 5", + "# HELP nodes_total Current node count", + "# TYPE nodes_total gauge", + "nodes_total 3.5", + "# HELP allocs_total Current allocation count", + "# TYPE allocs_total gauge", + "allocs_total 0", + } + for _, w := range want { + if !strings.Contains(out, w+"\n") { + t.Errorf("output missing line %q\n--- got:\n%s", w, out) + } + } +} + +func TestMetricsConcurrent(t *testing.T) { + m := NewMetrics() + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + m.IncCounter("txns_applied_total") + m.SetGauge("peers_total", 1) + }() + } + wg.Wait() + var sb strings.Builder + if err := m.WritePrometheus(&sb); err != nil { + t.Fatalf("WritePrometheus: %v", err) + } + if !strings.Contains(sb.String(), "txns_applied_total 100\n") { + t.Errorf("expected 100 applied, got:\n%s", sb.String()) + } +} + +func TestMetricsHTTP(t *testing.T) { + m := NewMetrics() + m.IncCounter("txns_applied_total") + m.SetGauge("nodes_total", 7) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := ln.Addr().String() + + mux := http.NewServeMux() + mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + _ = m.WritePrometheus(w) + }) + srv := &http.Server{Handler: mux} + + go srv.Serve(ln) + defer srv.Shutdown(context.Background()) + + time.Sleep(20 * time.Millisecond) + + resp, err := http.Get("http://" + addr + "/metrics") + if err != nil { + t.Fatalf("GET /metrics: %v", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + s := string(body) + for _, name := range []string{ + "txns_applied_total", + "txns_drifted_total", + "drifts_remediated_total", + "peers_total", + "nodes_total", + "allocs_total", + } { + if !strings.Contains(s, name) { + t.Errorf("response missing metric %q", name) + } + } + if !strings.Contains(s, "txns_applied_total 1\n") { + t.Errorf("counter value wrong in:\n%s", s) + } + if !strings.Contains(s, "nodes_total 7\n") { + t.Errorf("gauge value wrong in:\n%s", s) + } +}