feat(P01): metrics endpoint — hand-rolled Prometheus text exposition
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---
This commit is contained in:
Jon Chery
@@ -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)
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user