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) + } +}