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) <noreply@anthropic.com>
This commit is contained in:
jedarden
parent
930f0263c2
commit
20c48783cc
5 changed files with 558 additions and 15 deletions
8
.gitignore
vendored
8
.gitignore
vendored
|
|
@ -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/
|
||||
|
|
|
|||
33
PROGRESS.md
33
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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
255
cmd/acb-worker/metrics.go
Normal file
255
cmd/acb-worker/metrics.go
Normal file
|
|
@ -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, ",")
|
||||
}
|
||||
239
cmd/acb-worker/metrics_test.go
Normal file
239
cmd/acb-worker/metrics_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue