feat(multiplayer): expose allocator quota metrics

server/allocator/metrics.go

@@ -0,0 +1,99 @@
+package allocator
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+)
+
+// Metrics is a bounded allocator-role collector. Region is the only label so
+// a bad request cannot create unbounded Prometheus cardinality.
+type Metrics struct {
+	mu      sync.Mutex
+	regions map[string]*allocationMetric
+}
+
+type allocationMetric struct {
+	attempts uint64
+	success  uint64
+	failure  uint64
+	denied   uint64
+}
+
+func NewMetrics() *Metrics {
+	return &Metrics{regions: map[string]*allocationMetric{"EU": {}, "NA": {}}}
+}
+
+func (m *Metrics) ObserveAttempt(region string) {
+	if metric := m.metric(region); metric != nil {
+		m.mu.Lock()
+		metric.attempts++
+		m.mu.Unlock()
+	}
+}
+
+func (m *Metrics) ObserveSuccess(region string) {
+	if metric := m.metric(region); metric != nil {
+		m.mu.Lock()
+		metric.success++
+		m.mu.Unlock()
+	}
+}
+
+func (m *Metrics) ObserveFailure(region string) {
+	if metric := m.metric(region); metric != nil {
+		m.mu.Lock()
+		metric.failure++
+		m.mu.Unlock()
+	}
+}
+
+func (m *Metrics) ObserveDenied(region string) {
+	if metric := m.metric(region); metric != nil {
+		m.mu.Lock()
+		metric.denied++
+		m.mu.Unlock()
+	}
+}
+
+func (m *Metrics) metric(region string) *allocationMetric {
+	if m == nil || (region != "EU" && region != "NA") {
+		return nil
+	}
+	return m.regions[region]
+}
+
+func (m *Metrics) WritePrometheus(w io.Writer) error {
+	if m == nil {
+		return nil
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if _, err := io.WriteString(w, "# TYPE cosmic_clash_allocator_allocation_attempts_total counter\n# TYPE cosmic_clash_allocator_allocations_total counter\n# TYPE cosmic_clash_allocator_allocation_failures_total counter\n# TYPE cosmic_clash_allocator_quota_denials_total counter\n"); err != nil {
+		return err
+	}
+	for _, region := range []string{"EU", "NA"} {
+		metric := m.regions[region]
+		labels := fmt.Sprintf(`region="%s"`, region)
+		if _, err := fmt.Fprintf(w, "cosmic_clash_allocator_allocation_attempts_total{%s} %d\ncosmic_clash_allocator_allocations_total{%s} %d\ncosmic_clash_allocator_allocation_failures_total{%s} %d\ncosmic_clash_allocator_quota_denials_total{%s} %d\n", labels, metric.attempts, labels, metric.success, labels, metric.failure, labels, metric.denied); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// MetricsHandler exposes only the read-only Prometheus endpoint. The caller
+// owns the listener and can bind it to a private metrics network.
+func MetricsHandler(metrics *Metrics) http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			w.WriteHeader(http.StatusMethodNotAllowed)
+			return
+		}
+		w.Header().Set("Content-Type", "text/plain; version=0.0.4")
+		_ = metrics.WritePrometheus(w)
+	})
+	return mux
+}

server/allocator/metrics_test.go

@@ -0,0 +1,60 @@
+package allocator
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestMetricsExportsFixedRegionalCounters(t *testing.T) {
+	metrics := NewMetrics()
+	metrics.ObserveAttempt("EU")
+	metrics.ObserveSuccess("EU")
+	metrics.ObserveFailure("EU")
+	metrics.ObserveDenied("EU")
+	metrics.ObserveAttempt("APAC")
+	var output strings.Builder
+	if err := metrics.WritePrometheus(&output); err != nil {
+		t.Fatal(err)
+	}
+	text := output.String()
+	for _, fragment := range []string{
+		`cosmic_clash_allocator_allocation_attempts_total{region="EU"} 1`,
+		`cosmic_clash_allocator_allocations_total{region="EU"} 1`,
+		`cosmic_clash_allocator_allocation_failures_total{region="EU"} 1`,
+		`cosmic_clash_allocator_quota_denials_total{region="EU"} 1`,
+		`region="NA"`,
+	} {
+		if !strings.Contains(text, fragment) {
+			t.Fatalf("metrics missing %q: %s", fragment, text)
+		}
+	}
+	if strings.Contains(text, "APAC") {
+		t.Fatal("unbounded region label escaped into metrics")
+	}
+}
+
+func TestNilMetricsAreSafe(t *testing.T) {
+	var metrics *Metrics
+	metrics.ObserveAttempt("EU")
+	if err := metrics.WritePrometheus(&strings.Builder{}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestMetricsHandlerIsReadOnlyAndScoped(t *testing.T) {
+	metrics := NewMetrics()
+	metrics.ObserveAttempt("NA")
+	handler := MetricsHandler(metrics)
+	get := httptest.NewRecorder()
+	handler.ServeHTTP(get, httptest.NewRequest(http.MethodGet, "/metrics", nil))
+	if get.Code != http.StatusOK || !strings.Contains(get.Body.String(), `region="NA"`) {
+		t.Fatalf("GET /metrics status=%d body=%s", get.Code, get.Body.String())
+	}
+	post := httptest.NewRecorder()
+	handler.ServeHTTP(post, httptest.NewRequest(http.MethodPost, "/metrics", nil))
+	if post.Code != http.StatusMethodNotAllowed {
+		t.Fatalf("POST /metrics status=%d, want 405", post.Code)
+	}
+}

server/allocator/service.go

@@ -34,12 +34,20 @@ type SharedAllocationQuota interface {
 	Consume(context.Context, string, time.Time) error
 }
 
+type AllocationMetrics interface {
+	ObserveAttempt(string)
+	ObserveSuccess(string)
+	ObserveFailure(string)
+	ObserveDenied(string)
+}
+
 type Service struct {
 	Provider Provider
 	Durable  Durable
 	Roster   RosterPublisher
 	Budget   AllocationBudget
 	Quota    SharedAllocationQuota
+	Metrics  AllocationMetrics
 	Now      func() time.Time
 }
 
@@ -88,23 +96,41 @@ func (s Service) Allocate(ctx context.Context, request domain.AllocationRequest,
 		return agones.AllocatedServer{}, errNotConfigured
 	}
 	now := s.Now()
+	if s.Metrics != nil {
+		s.Metrics.ObserveAttempt(request.Region)
+	}
 	if s.Budget != nil {
 		if err := s.Budget.Allow(request.Region, now); err != nil {
+			if s.Metrics != nil {
+				s.Metrics.ObserveDenied(request.Region)
+			}
 			return agones.AllocatedServer{}, err
 		}
 	}
 	if s.Quota != nil {
 		if err := s.Quota.Consume(ctx, request.Region, now); err != nil {
+			if s.Metrics != nil {
+				s.Metrics.ObserveDenied(request.Region)
+			}
 			return agones.AllocatedServer{}, err
 		}
 	}
 	result, err := s.Provider.Allocate(ctx, request, labels, now)
 	if err != nil {
+		if s.Metrics != nil {
+			s.Metrics.ObserveFailure(request.Region)
+		}
 		return agones.AllocatedServer{}, err
 	}
 	if _, err := s.Durable.RecordProviderAllocation(ctx, result.Allocation, now); err != nil {
+		if s.Metrics != nil {
+			s.Metrics.ObserveFailure(request.Region)
+		}
 		return agones.AllocatedServer{}, err
 	}
+	if s.Metrics != nil {
+		s.Metrics.ObserveSuccess(request.Region)
+	}
 	return result, nil
 }
 
@@ -114,10 +140,21 @@ func (s Service) RecordProviderAllocation(ctx context.Context, result agones.All
 	}
 	if s.Quota != nil {
 		if err := s.Quota.Consume(ctx, result.Allocation.Region, now); err != nil {
+			if s.Metrics != nil {
+				s.Metrics.ObserveDenied(result.Allocation.Region)
+			}
 			return domain.Allocation{}, err
 		}
 	}
-	return s.Durable.RecordProviderAllocation(ctx, result.Allocation, now)
+	allocation, err := s.Durable.RecordProviderAllocation(ctx, result.Allocation, now)
+	if s.Metrics != nil {
+		if err != nil {
+			s.Metrics.ObserveFailure(result.Allocation.Region)
+		} else {
+			s.Metrics.ObserveSuccess(result.Allocation.Region)
+		}
+	}
+	return allocation, err
 }
 
 var errNotConfigured = &configurationError{}

server/allocator/service_test.go

@@ -3,6 +3,7 @@ package allocator
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 	"time"
 
@@ -56,11 +57,16 @@ func (d *durableSpy) RecordProviderAllocation(_ context.Context, allocation doma
 func TestServiceDurablyRecordsProviderAllocationBeforeReturning(t *testing.T) {
 	provider := &providerSpy{result: agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: "a", MatchID: "m", ServerID: "gs", State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"}}
 	durable := &durableSpy{}
-	service := Service{Provider: provider, Durable: durable, Now: func() time.Time { return time.Unix(1000, 0) }}
+	metrics := NewMetrics()
+	service := Service{Provider: provider, Durable: durable, Metrics: metrics, Now: func() time.Time { return time.Unix(1000, 0) }}
 	result, err := service.Allocate(context.Background(), domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, map[string]string{"region": "EU"})
 	if err != nil || result.Endpoint == "" || durable.calls != 1 || durable.allocation.ServerID != "gs" {
 		t.Fatalf("result=%+v err=%v durable=%+v", result, err, durable)
 	}
+	var output strings.Builder
+	if err := metrics.WritePrometheus(&output); err != nil || !strings.Contains(output.String(), `allocations_total{region="EU"} 1`) {
+		t.Fatalf("success metric err=%v output=%s", err, output.String())
+	}
 }
 
 func TestServiceDoesNotReturnProviderResultAfterDurableFailure(t *testing.T) {
@@ -76,13 +82,19 @@ func TestServiceDoesNotReturnProviderResultAfterDurableFailure(t *testing.T) {
 func TestServiceConsumesSharedQuotaBeforeFreshProviderCall(t *testing.T) {
 	provider := &providerSpy{result: agones.AllocatedServer{Allocation: domain.Allocation{AllocationID: "a", State: domain.ServerAllocated}, Endpoint: "127.0.0.1:7777"}}
 	quota := &quotaSpy{err: errors.New("quota exhausted")}
-	service := Service{Provider: provider, Durable: &durableSpy{}, Quota: quota, Now: func() time.Time { return time.Unix(1000, 0) }}
+	metrics := NewMetrics()
+	service := Service{Provider: provider, Durable: &durableSpy{}, Quota: quota, Metrics: metrics, Now: func() time.Time { return time.Unix(1000, 0) }}
 	if _, err := service.Allocate(context.Background(), domain.AllocationRequest{AllocationID: "a", MatchID: "m", Region: "EU", Build: "b", Protocol: 1, Transport: "enet"}, nil); err == nil {
 		t.Fatal("quota rejection was ignored")
 	}
 	if quota.calls != 1 || provider.calls != 0 {
 		t.Fatalf("quota/provider calls = %d/%d, want 1/0", quota.calls, provider.calls)
 	}
+	var output strings.Builder
+	_ = metrics.WritePrometheus(&output)
+	if !strings.Contains(output.String(), `quota_denials_total{region="EU"} 1`) {
+		t.Fatalf("quota denial metric missing: %s", output.String())
+	}
 }
 
 func TestServiceConsumesSharedQuotaOnceWhenReconcilingProviderResult(t *testing.T) {

server/cmd/allocator/main.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"flag"
 	"log"
+	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
@@ -27,6 +28,7 @@ func main() {
 	workloadSecret := flag.String("workload-secret", os.Getenv("COSMIC_CLASH_WORKLOAD_SECRET"), "HMAC secret for control-plane-issued workload tokens (see workload/signed_token.go); must match cmd/control-plane's own --workload-secret. Unset skips minting a cosmic-clash.io/workload-token annotation entirely")
 	allocationQuota := flag.Int("allocation-quota", 0, "optional per-replica allocation attempts per region per quota window; zero disables this local guard")
 	allocationQuotaWindow := flag.Duration("allocation-quota-window", time.Minute, "window for --allocation-quota")
+	metricsAddr := flag.String("metrics-addr", envOrDefault("COSMIC_CLASH_ALLOCATOR_METRICS_ADDR", ":9091"), "allocator Prometheus metrics address; empty disables metrics")
 	flag.Parse()
 	if *dsn == "" || *agonesURL == "" {
 		fatalf("--dsn/COSMIC_CLASH_POSTGRES_DSN and --agones-url/COSMIC_CLASH_AGONES_URL are required")
@@ -62,6 +64,7 @@ func main() {
 		}
 		log.Printf("allocator: enabled per-replica regional allocation quota=%d window=%s", *allocationQuota, *allocationQuotaWindow)
 	}
+	metrics := allocator.NewMetrics()
 	client := agones.Client{BaseURL: *agonesURL, Namespace: *namespace, WorkloadSecret: []byte(*workloadSecret)}
 	worker := allocator.Worker{
 		Claims: store.AllocatingMatchClaims{DB: db, Transport: *transport},
@@ -70,12 +73,30 @@ func main() {
 			Durable:  store.AllocationRegistry{DB: db},
 			Quota:    store.AllocationQuota{DB: db},
 			Budget:   budget,
+			Metrics:  metrics,
 			Now:      now,
 		},
 		Now: now,
 	}
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
+	var metricsServer *http.Server
+	if *metricsAddr != "" {
+		metricsServer = &http.Server{Addr: *metricsAddr, Handler: allocator.MetricsHandler(metrics)}
+		go func() {
+			if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+				log.Printf("allocator: metrics server: %v", err)
+			}
+		}()
+		log.Printf("allocator: metrics listening on %s", *metricsAddr)
+	}
+	defer func() {
+		if metricsServer != nil {
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			_ = metricsServer.Shutdown(shutdownCtx)
+		}
+	}()
 	ticker := time.NewTicker(*interval)
 	defer ticker.Stop()
 	for {