From 20c48783cc6608e610849a8db8259335d31b8c42 Mon Sep 17 00:00:00 2001 From: jedarden Date: Thu, 26 Mar 2026 00:50:10 -0400 Subject: [PATCH] Add Prometheus metrics endpoint to match worker Adds a metrics HTTP server to acb-worker exposing Prometheus text format at /metrics, plus /health and /ready K8s probe endpoints. Tracks counters (matches, errors, jobs, replays, polls, heartbeats) and histograms (match duration, replay upload duration, replay size). Instruments the full worker execution flow. Fixes .gitignore binary patterns to use root-anchored paths so cmd/ subdirectories aren't incorrectly excluded. Co-Authored-By: Claude Opus 4.6 (1M context) --- .gitignore | 8 +- PROGRESS.md | 33 +++-- cmd/acb-worker/main.go | 38 +++++ cmd/acb-worker/metrics.go | 255 +++++++++++++++++++++++++++++++++ cmd/acb-worker/metrics_test.go | 239 ++++++++++++++++++++++++++++++ 5 files changed, 558 insertions(+), 15 deletions(-) create mode 100644 cmd/acb-worker/metrics.go create mode 100644 cmd/acb-worker/metrics_test.go diff --git a/.gitignore b/.gitignore index c990ec2..db1c7ca 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,7 @@ -# Binaries -acb-local -acb-mapgen -acb-worker +# Binaries (root-level only) +/acb-local +/acb-mapgen +/acb-worker # Node modules node_modules/ diff --git a/PROGRESS.md b/PROGRESS.md index b6ec516..5afb835 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -4,20 +4,25 @@ **Status: 🔄 In Progress** -**Last Updated: 2026-03-24** +**Last Updated: 2026-03-26** -### Recent Changes (2026-03-24) +### Recent Changes (2026-03-26) +- Added Prometheus-compatible metrics endpoint to match worker (`cmd/acb-worker/metrics.go`) + - Counters: matches_total, match_errors_total, jobs_claimed/failed, replays_uploaded, poll_cycles, heartbeats + - Histograms: match_duration_seconds, replay_upload_duration_seconds, replay_size_bytes + - Worker info gauge with worker_id label + - `/health` and `/ready` endpoints on metrics HTTP server (default :9090) + - Configurable via `ACB_METRICS_ADDR` environment variable +- Instrumented worker execution flow with metrics recording +- Added comprehensive tests (`cmd/acb-worker/metrics_test.go`) + - Health/ready endpoint tests, counter accuracy, histogram bucket correctness + - Concurrency safety test (10 goroutines x 100 operations) +- All tests pass (engine + worker) + +### Previous Changes (2026-03-24) - Added GitHub Actions CI workflow (`.github/workflows/ci.yml`) - - Runs Go engine tests with race detector - - Runs TypeScript tests for worker-api and indexer - - Builds web app and uploads artifacts - - Builds all Go CLI tools - Added `README.md` with project overview and quick start guide -- Added `.gitignore` for proper repository hygiene -- Added `package-lock.json` files for reproducible builds -- Verified all tests pass (glicko2.test.ts, generator.test.ts) -- Verified web build succeeds -- Verified compiled binaries work (acb-local, acb-mapgen, acb-worker) +- Added `.gitignore` and `package-lock.json` files ### Phase 6 Progress @@ -50,6 +55,12 @@ - `/health` - Liveness probe (always returns 200) - `/ready` - Readiness probe (checks database connectivity, returns 503 if unavailable) - Documented in DEPLOYMENT.md +- [x] Prometheus metrics endpoint (`cmd/acb-worker/metrics.go`) + - Counters: matches, errors, jobs, replays, polls, heartbeats + - Histograms: match duration, replay upload duration, replay size + - Worker info gauge with labels + - Separate HTTP server on configurable port (default :9090) + - Integrated into worker execution flow with full instrumentation - [x] GitHub Actions CI workflow - `.github/workflows/ci.yml` for automated testing - Go tests with race detector diff --git a/cmd/acb-worker/main.go b/cmd/acb-worker/main.go index bfb8e69..3983477 100644 --- a/cmd/acb-worker/main.go +++ b/cmd/acb-worker/main.go @@ -12,6 +12,7 @@ import ( "fmt" "log" "math/rand" + "net/http" "os" "os/signal" "syscall" @@ -81,16 +82,33 @@ func main() { r2Client = NewR2Client(cfg) } + // Create metrics + metrics := NewMetrics(cfg.WorkerID) + // Create worker worker := &Worker{ cfg: cfg, api: apiClient, r2: r2Client, + metrics: metrics, logger: log.New(os.Stdout, fmt.Sprintf("[worker-%s] ", cfg.WorkerID), log.LstdFlags), rng: rand.New(rand.NewSource(time.Now().UnixNano())), heartbeat: *heartbeat, } + // Start metrics HTTP server + metricsAddr := getEnv("ACB_METRICS_ADDR", ":9090") + metricsServer := &http.Server{ + Addr: metricsAddr, + Handler: metrics.Handler(), + } + go func() { + worker.logger.Printf("Metrics server listening on %s", metricsAddr) + if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + worker.logger.Printf("Metrics server error: %v", err) + } + }() + // Set up signal handling ctx, cancel := context.WithCancel(context.Background()) sigChan := make(chan os.Signal, 1) @@ -104,6 +122,11 @@ func main() { // Run worker loop worker.Run(ctx) + + // Shut down metrics server gracefully + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutdownCancel() + metricsServer.Shutdown(shutdownCtx) } // getEnv gets an environment variable with a default value. @@ -124,6 +147,7 @@ type Worker struct { cfg *Config api *APIClient r2 *R2Client + metrics *Metrics logger *log.Logger rng *rand.Rand heartbeat time.Duration @@ -151,6 +175,8 @@ func (w *Worker) Run(ctx context.Context) { // pollAndExecute polls for a job and executes it if available. func (w *Worker) pollAndExecute(ctx context.Context) error { + w.metrics.RecordPollCycle() + // Get next pending job job, err := w.api.GetNextJob(ctx) if err != nil { @@ -172,27 +198,36 @@ func (w *Worker) pollAndExecute(ctx context.Context) error { return fmt.Errorf("failed to claim job %s: %w", job.ID, err) } + w.metrics.RecordJobClaimed() w.logger.Printf("Claimed job %s, executing match...", job.ID) // Execute the match + matchStart := time.Now() result, replay, err := w.executeMatch(ctx, claimResp) if err != nil { + w.metrics.RecordMatchError() w.logger.Printf("Match execution failed: %v", err) // Mark job as failed if failErr := w.api.FailJob(ctx, job.ID, w.cfg.WorkerID, err.Error()); failErr != nil { + w.metrics.RecordJobFailed() w.logger.Printf("Failed to mark job as failed: %v", failErr) } return err } + w.metrics.RecordMatch(time.Since(matchStart)) // Upload replay to R2 replayURL := "" if w.r2 != nil { + uploadStart := time.Now() + replayData, _ := json.Marshal(replay) replayURL, err = w.uploadReplay(ctx, claimResp.Match.ID, replay) if err != nil { + w.metrics.RecordReplayUploadError() w.logger.Printf("Failed to upload replay: %v", err) // Continue without replay URL - match result is more important } else { + w.metrics.RecordReplayUpload(time.Since(uploadStart), len(replayData)) w.logger.Printf("Uploaded replay to %s", replayURL) } } @@ -316,7 +351,10 @@ func (w *Worker) sendHeartbeats(ctx context.Context, jobID string) { return case <-ticker.C: if err := w.api.Heartbeat(ctx, jobID, w.cfg.WorkerID); err != nil { + w.metrics.RecordHeartbeatError() w.logger.Printf("Heartbeat failed: %v", err) + } else { + w.metrics.RecordHeartbeat() } } } diff --git a/cmd/acb-worker/metrics.go b/cmd/acb-worker/metrics.go new file mode 100644 index 0000000..69ed8d8 --- /dev/null +++ b/cmd/acb-worker/metrics.go @@ -0,0 +1,255 @@ +// Metrics collection and HTTP server for monitoring. +// +// Exposes Prometheus text format metrics at /metrics, plus +// /health and /ready endpoints for K8s probes. +package main + +import ( + "fmt" + "net/http" + "sort" + "strings" + "sync" + "sync/atomic" + "time" +) + +// Metrics collects operational metrics for the match worker. +type Metrics struct { + // Counters + matchesTotal atomic.Int64 + matchErrorsTotal atomic.Int64 + jobsClaimedTotal atomic.Int64 + jobsFailedTotal atomic.Int64 + replaysUploaded atomic.Int64 + replayUploadErrs atomic.Int64 + pollCycles atomic.Int64 + heartbeatsSent atomic.Int64 + heartbeatErrors atomic.Int64 + + // Histograms (stored as individual observations) + mu sync.Mutex + matchDurations []float64 // seconds + replayUploadDurations []float64 // seconds + replaySizes []float64 // bytes + + // State + startTime time.Time + workerID string + ready atomic.Bool +} + +// NewMetrics creates a new Metrics instance. +func NewMetrics(workerID string) *Metrics { + m := &Metrics{ + startTime: time.Now(), + workerID: workerID, + matchDurations: make([]float64, 0, 1024), + replayUploadDurations: make([]float64, 0, 1024), + replaySizes: make([]float64, 0, 1024), + } + m.ready.Store(true) + return m +} + +// RecordMatch records a completed match. +func (m *Metrics) RecordMatch(duration time.Duration) { + m.matchesTotal.Add(1) + m.mu.Lock() + m.matchDurations = append(m.matchDurations, duration.Seconds()) + m.mu.Unlock() +} + +// RecordMatchError records a match execution error. +func (m *Metrics) RecordMatchError() { + m.matchErrorsTotal.Add(1) +} + +// RecordJobClaimed records a job claim. +func (m *Metrics) RecordJobClaimed() { + m.jobsClaimedTotal.Add(1) +} + +// RecordJobFailed records a job failure report. +func (m *Metrics) RecordJobFailed() { + m.jobsFailedTotal.Add(1) +} + +// RecordReplayUpload records a successful replay upload. +func (m *Metrics) RecordReplayUpload(duration time.Duration, sizeBytes int) { + m.replaysUploaded.Add(1) + m.mu.Lock() + m.replayUploadDurations = append(m.replayUploadDurations, duration.Seconds()) + m.replaySizes = append(m.replaySizes, float64(sizeBytes)) + m.mu.Unlock() +} + +// RecordReplayUploadError records a replay upload error. +func (m *Metrics) RecordReplayUploadError() { + m.replayUploadErrs.Add(1) +} + +// RecordPollCycle records a poll cycle. +func (m *Metrics) RecordPollCycle() { + m.pollCycles.Add(1) +} + +// RecordHeartbeat records a heartbeat sent. +func (m *Metrics) RecordHeartbeat() { + m.heartbeatsSent.Add(1) +} + +// RecordHeartbeatError records a heartbeat error. +func (m *Metrics) RecordHeartbeatError() { + m.heartbeatErrors.Add(1) +} + +// SetReady sets the worker readiness state. +func (m *Metrics) SetReady(ready bool) { + m.ready.Store(ready) +} + +// Handler returns an http.Handler serving metrics and health endpoints. +func (m *Metrics) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/metrics", m.handleMetrics) + mux.HandleFunc("/health", m.handleHealth) + mux.HandleFunc("/ready", m.handleReady) + return mux +} + +func (m *Metrics) handleHealth(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + fmt.Fprintf(w, `{"status":"ok","worker_id":%q,"uptime_seconds":%.0f}`, + m.workerID, time.Since(m.startTime).Seconds()) +} + +func (m *Metrics) handleReady(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if m.ready.Load() { + w.WriteHeader(http.StatusOK) + fmt.Fprintf(w, `{"status":"ready","worker_id":%q}`, m.workerID) + } else { + w.WriteHeader(http.StatusServiceUnavailable) + fmt.Fprintf(w, `{"status":"not_ready","worker_id":%q}`, m.workerID) + } +} + +func (m *Metrics) handleMetrics(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + + var b strings.Builder + + // Worker info + writeGauge(&b, "acb_worker_info", "Worker metadata", 1, + "worker_id", m.workerID) + writeGauge(&b, "acb_worker_uptime_seconds", "Time since worker started", + time.Since(m.startTime).Seconds()) + + // Counters + writeCounter(&b, "acb_matches_total", "Total matches executed", + float64(m.matchesTotal.Load())) + writeCounter(&b, "acb_match_errors_total", "Total match execution errors", + float64(m.matchErrorsTotal.Load())) + writeCounter(&b, "acb_jobs_claimed_total", "Total jobs claimed", + float64(m.jobsClaimedTotal.Load())) + writeCounter(&b, "acb_jobs_failed_total", "Total jobs reported as failed", + float64(m.jobsFailedTotal.Load())) + writeCounter(&b, "acb_replays_uploaded_total", "Total replays uploaded to R2", + float64(m.replaysUploaded.Load())) + writeCounter(&b, "acb_replay_upload_errors_total", "Total replay upload errors", + float64(m.replayUploadErrs.Load())) + writeCounter(&b, "acb_poll_cycles_total", "Total job poll cycles", + float64(m.pollCycles.Load())) + writeCounter(&b, "acb_heartbeats_sent_total", "Total heartbeats sent", + float64(m.heartbeatsSent.Load())) + writeCounter(&b, "acb_heartbeat_errors_total", "Total heartbeat errors", + float64(m.heartbeatErrors.Load())) + + // Histograms (snapshot under lock) + m.mu.Lock() + matchDurs := make([]float64, len(m.matchDurations)) + copy(matchDurs, m.matchDurations) + uploadDurs := make([]float64, len(m.replayUploadDurations)) + copy(uploadDurs, m.replayUploadDurations) + replaySizes := make([]float64, len(m.replaySizes)) + copy(replaySizes, m.replaySizes) + m.mu.Unlock() + + matchBuckets := []float64{1, 5, 10, 30, 60, 120, 300, 600} + writeHistogram(&b, "acb_match_duration_seconds", + "Match execution duration in seconds", matchDurs, matchBuckets) + + uploadBuckets := []float64{0.1, 0.5, 1, 2, 5, 10, 30} + writeHistogram(&b, "acb_replay_upload_duration_seconds", + "Replay upload duration in seconds", uploadDurs, uploadBuckets) + + sizeBuckets := []float64{1024, 10240, 102400, 1048576, 10485760} + writeHistogram(&b, "acb_replay_size_bytes", + "Replay file size in bytes", replaySizes, sizeBuckets) + + w.Write([]byte(b.String())) +} + +// writeCounter writes a counter metric in Prometheus text format. +func writeCounter(b *strings.Builder, name, help string, value float64) { + fmt.Fprintf(b, "# HELP %s %s\n", name, help) + fmt.Fprintf(b, "# TYPE %s counter\n", name) + fmt.Fprintf(b, "%s %g\n\n", name, value) +} + +// writeGauge writes a gauge metric in Prometheus text format. +func writeGauge(b *strings.Builder, name, help string, value float64, labels ...string) { + fmt.Fprintf(b, "# HELP %s %s\n", name, help) + fmt.Fprintf(b, "# TYPE %s gauge\n", name) + if len(labels) > 0 { + labelStr := formatLabels(labels) + fmt.Fprintf(b, "%s{%s} %g\n\n", name, labelStr, value) + } else { + fmt.Fprintf(b, "%s %g\n\n", name, value) + } +} + +// writeHistogram writes a histogram metric in Prometheus text format. +func writeHistogram(b *strings.Builder, name, help string, observations []float64, buckets []float64) { + fmt.Fprintf(b, "# HELP %s %s\n", name, help) + fmt.Fprintf(b, "# TYPE %s histogram\n", name) + + sort.Float64s(buckets) + + var sum float64 + for _, v := range observations { + sum += v + } + + sorted := make([]float64, len(observations)) + copy(sorted, observations) + sort.Float64s(sorted) + + count := len(sorted) + for _, boundary := range buckets { + n := countLE(sorted, boundary) + fmt.Fprintf(b, "%s_bucket{le=\"%g\"} %d\n", name, boundary, n) + } + fmt.Fprintf(b, "%s_bucket{le=\"+Inf\"} %d\n", name, count) + fmt.Fprintf(b, "%s_sum %g\n", name, sum) + fmt.Fprintf(b, "%s_count %d\n\n", name, count) +} + +// countLE counts values <= boundary in a sorted slice. +func countLE(sorted []float64, boundary float64) int { + i := sort.Search(len(sorted), func(i int) bool { + return sorted[i] > boundary + }) + return i +} + +// formatLabels formats label key-value pairs for Prometheus output. +func formatLabels(labels []string) string { + var parts []string + for i := 0; i+1 < len(labels); i += 2 { + parts = append(parts, fmt.Sprintf(`%s=%q`, labels[i], labels[i+1])) + } + return strings.Join(parts, ",") +} diff --git a/cmd/acb-worker/metrics_test.go b/cmd/acb-worker/metrics_test.go new file mode 100644 index 0000000..21753a8 --- /dev/null +++ b/cmd/acb-worker/metrics_test.go @@ -0,0 +1,239 @@ +package main + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestMetricsHealth(t *testing.T) { + m := NewMetrics("test-worker") + handler := m.Handler() + + req := httptest.NewRequest("GET", "/health", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + resp := w.Result() + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", resp.StatusCode) + } + + body, _ := io.ReadAll(resp.Body) + if !strings.Contains(string(body), `"status":"ok"`) { + t.Fatalf("expected ok status, got: %s", body) + } + if !strings.Contains(string(body), `"worker_id":"test-worker"`) { + t.Fatalf("expected worker_id, got: %s", body) + } +} + +func TestMetricsReady(t *testing.T) { + m := NewMetrics("test-worker") + handler := m.Handler() + + // Ready by default + req := httptest.NewRequest("GET", "/ready", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Result().StatusCode != http.StatusOK { + t.Fatalf("expected 200 when ready, got %d", w.Result().StatusCode) + } + + // Set not ready + m.SetReady(false) + w = httptest.NewRecorder() + handler.ServeHTTP(w, httptest.NewRequest("GET", "/ready", nil)) + + if w.Result().StatusCode != http.StatusServiceUnavailable { + t.Fatalf("expected 503 when not ready, got %d", w.Result().StatusCode) + } +} + +func TestMetricsCounters(t *testing.T) { + m := NewMetrics("test-worker") + + m.RecordMatch(5 * time.Second) + m.RecordMatch(10 * time.Second) + m.RecordMatchError() + m.RecordJobClaimed() + m.RecordJobClaimed() + m.RecordJobClaimed() + m.RecordJobFailed() + m.RecordPollCycle() + m.RecordPollCycle() + m.RecordHeartbeat() + m.RecordHeartbeatError() + m.RecordReplayUpload(500*time.Millisecond, 50000) + m.RecordReplayUploadError() + + handler := m.Handler() + req := httptest.NewRequest("GET", "/metrics", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + body := w.Body.String() + + assertMetric(t, body, "acb_matches_total", "2") + assertMetric(t, body, "acb_match_errors_total", "1") + assertMetric(t, body, "acb_jobs_claimed_total", "3") + assertMetric(t, body, "acb_jobs_failed_total", "1") + assertMetric(t, body, "acb_replays_uploaded_total", "1") + assertMetric(t, body, "acb_replay_upload_errors_total", "1") + assertMetric(t, body, "acb_poll_cycles_total", "2") + assertMetric(t, body, "acb_heartbeats_sent_total", "1") + assertMetric(t, body, "acb_heartbeat_errors_total", "1") +} + +func TestMetricsHistogram(t *testing.T) { + m := NewMetrics("test-worker") + + // Record match durations: 2s, 8s, 15s + m.RecordMatch(2 * time.Second) + m.RecordMatch(8 * time.Second) + m.RecordMatch(15 * time.Second) + + handler := m.Handler() + req := httptest.NewRequest("GET", "/metrics", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + body := w.Body.String() + + // Check histogram buckets: 2 <= 5, 8 <= 10, 15 <= 30 + assertContains(t, body, `acb_match_duration_seconds_bucket{le="5"} 1`) + assertContains(t, body, `acb_match_duration_seconds_bucket{le="10"} 2`) + assertContains(t, body, `acb_match_duration_seconds_bucket{le="30"} 3`) + assertContains(t, body, `acb_match_duration_seconds_bucket{le="+Inf"} 3`) + assertContains(t, body, `acb_match_duration_seconds_sum 25`) + assertContains(t, body, `acb_match_duration_seconds_count 3`) +} + +func TestMetricsReplayHistogram(t *testing.T) { + m := NewMetrics("test-worker") + + m.RecordReplayUpload(100*time.Millisecond, 5000) + m.RecordReplayUpload(2*time.Second, 200000) + + handler := m.Handler() + req := httptest.NewRequest("GET", "/metrics", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + body := w.Body.String() + + // Upload durations: 0.1s <= 0.1, 2s <= 2 + assertContains(t, body, `acb_replay_upload_duration_seconds_bucket{le="0.1"} 1`) + assertContains(t, body, `acb_replay_upload_duration_seconds_bucket{le="2"} 2`) + assertContains(t, body, `acb_replay_upload_duration_seconds_count 2`) + + // Replay sizes: 5000 <= 10240, 200000 <= 1.0486e+06 + assertContains(t, body, `acb_replay_size_bytes_bucket{le="10240"} 1`) + assertContains(t, body, `acb_replay_size_bytes_count 2`) +} + +func TestMetricsContentType(t *testing.T) { + m := NewMetrics("test-worker") + handler := m.Handler() + + req := httptest.NewRequest("GET", "/metrics", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + ct := w.Result().Header.Get("Content-Type") + if !strings.HasPrefix(ct, "text/plain") { + t.Fatalf("expected text/plain content type, got: %s", ct) + } +} + +func TestMetricsWorkerInfo(t *testing.T) { + m := NewMetrics("my-worker-42") + handler := m.Handler() + + req := httptest.NewRequest("GET", "/metrics", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + body := w.Body.String() + assertContains(t, body, `acb_worker_info{worker_id="my-worker-42"} 1`) +} + +func TestCountLE(t *testing.T) { + sorted := []float64{1, 2, 3, 5, 10, 20} + + tests := []struct { + boundary float64 + want int + }{ + {0.5, 0}, + {1, 1}, + {3, 3}, + {4, 3}, + {10, 5}, + {100, 6}, + } + + for _, tt := range tests { + got := countLE(sorted, tt.boundary) + if got != tt.want { + t.Errorf("countLE(%v, %g) = %d, want %d", sorted, tt.boundary, got, tt.want) + } + } +} + +func TestFormatLabels(t *testing.T) { + got := formatLabels([]string{"a", "1", "b", "2"}) + want := `a="1",b="2"` + if got != want { + t.Errorf("formatLabels = %q, want %q", got, want) + } +} + +func TestMetricsConcurrency(t *testing.T) { + m := NewMetrics("test-worker") + + done := make(chan struct{}) + for i := 0; i < 10; i++ { + go func() { + for j := 0; j < 100; j++ { + m.RecordMatch(time.Duration(j) * time.Millisecond) + m.RecordPollCycle() + m.RecordHeartbeat() + m.RecordReplayUpload(time.Millisecond, 1000) + } + done <- struct{}{} + }() + } + + for i := 0; i < 10; i++ { + <-done + } + + if m.matchesTotal.Load() != 1000 { + t.Fatalf("expected 1000 matches, got %d", m.matchesTotal.Load()) + } + if m.pollCycles.Load() != 1000 { + t.Fatalf("expected 1000 poll cycles, got %d", m.pollCycles.Load()) + } +} + +// assertMetric checks a simple counter line like "metric_name 42" +func assertMetric(t *testing.T, body, metric, value string) { + t.Helper() + expected := metric + " " + value + if !strings.Contains(body, expected) { + t.Errorf("expected %q in metrics output, got:\n%s", expected, body) + } +} + +// assertContains checks that body contains substr. +func assertContains(t *testing.T, body, substr string) { + t.Helper() + if !strings.Contains(body, substr) { + t.Errorf("expected %q in output, got:\n%s", substr, body) + } +}