feat: add Prometheus /metrics endpoint for concurrency and queue gauges

Expose sandbox_concurrency_active, sandbox_queue_length,
sandbox_concurrency_max, and sandbox_queue_max as Prometheus
text exposition format gauges for autoscaling custom metrics.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: koki-develop

cmd/serve.go

@@ -95,6 +95,8 @@ func runServe(_ *cobra.Command, _ []string) error {
 
 	h := &handler.Handler{Runner: sandbox.NewRunner(cfg), MaxFiles: flagMaxFiles, MaxFileSize: flagMaxFileSize}
 
+	metrics := &intmw.ConcurrencyMetrics{}
+
 	e := echo.New()
 	e.HTTPErrorHandler = handler.NewHTTPErrorHandler()
 	e.Use(middleware.Recover())
@@ -103,10 +105,12 @@ func runServe(_ *cobra.Command, _ []string) error {
 	e.GET("/healthz", func(c *echo.Context) error {
 		return c.JSON(http.StatusOK, map[string]string{"status": "ok"})
 	})
+	e.GET("/metrics", intmw.MetricsHandler(metrics, flagMaxConcurrency, flagMaxQueueSize))
 	e.POST("/v1/run", h.RunHandler, intmw.ConcurrencyLimiter(intmw.ConcurrencyConfig{
 		MaxConcurrency: flagMaxConcurrency,
 		MaxQueueSize:   flagMaxQueueSize,
 		QueueTimeout:   time.Duration(flagQueueTimeout) * time.Second,
+		Metrics:        metrics,
 	}))
 
 	sc := echo.StartConfig{

internal/middleware/concurrency.go

@@ -9,11 +9,19 @@ import (
 	"github.com/labstack/echo/v5"
 )
 
+// ConcurrencyMetrics exposes live concurrency and queue counters for external
+// consumption (e.g. Prometheus /metrics endpoint).
+type ConcurrencyMetrics struct {
+	Active atomic.Int64
+	Queued atomic.Int64
+}
+
 // ConcurrencyConfig holds parameters for the concurrency limiter middleware.
 type ConcurrencyConfig struct {
 	MaxConcurrency int
 	MaxQueueSize   int
 	QueueTimeout   time.Duration
+	Metrics        *ConcurrencyMetrics
 }
 
 // ConcurrencyLimiter returns an Echo middleware that limits concurrent handler
@@ -22,21 +30,24 @@ type ConcurrencyConfig struct {
 // than QueueTimeout receive 503 (SERVER_BUSY).
 func ConcurrencyLimiter(cfg ConcurrencyConfig) echo.MiddlewareFunc {
 	sem := make(chan struct{}, cfg.MaxConcurrency)
-	var queued atomic.Int64
 
 	return func(next echo.HandlerFunc) echo.HandlerFunc {
 		return func(c *echo.Context) error {
 			// Fast path: try to acquire a semaphore slot without blocking.
 			select {
 			case sem <- struct{}{}:
-				defer func() { <-sem }()
+				cfg.Metrics.Active.Add(1)
+				defer func() {
+					<-sem
+					cfg.Metrics.Active.Add(-1)
+				}()
 				return next(c)
 			default:
 			}
 
 			// Slow path: semaphore is full — enter the queue.
-			q := queued.Add(1)
-			defer queued.Add(-1)
+			q := cfg.Metrics.Queued.Add(1)
+			defer cfg.Metrics.Queued.Add(-1)
 
 			if q > int64(cfg.MaxQueueSize) {
 				return c.JSON(http.StatusServiceUnavailable, handler.ErrorResponse{
@@ -50,7 +61,11 @@ func ConcurrencyLimiter(cfg ConcurrencyConfig) echo.MiddlewareFunc {
 
 			select {
 			case sem <- struct{}{}:
-				defer func() { <-sem }()
+				cfg.Metrics.Active.Add(1)
+				defer func() {
+					<-sem
+					cfg.Metrics.Active.Add(-1)
+				}()
 				return next(c)
 			case <-timer.C:
 				return c.JSON(http.StatusServiceUnavailable, handler.ErrorResponse{

internal/middleware/concurrency_test.go

@@ -13,7 +13,10 @@ import (
 	"github.com/labstack/echo/v5"
 )
 
-func setupEcho(cfg ConcurrencyConfig, gate chan struct{}) (*echo.Echo, echo.MiddlewareFunc) {
+func setupEcho(cfg ConcurrencyConfig, gate chan struct{}) (*echo.Echo, *ConcurrencyMetrics) {
+	if cfg.Metrics == nil {
+		cfg.Metrics = &ConcurrencyMetrics{}
+	}
 	e := echo.New()
 	mw := ConcurrencyLimiter(cfg)
 
@@ -25,7 +28,7 @@ func setupEcho(cfg ConcurrencyConfig, gate chan struct{}) (*echo.Echo, echo.Midd
 	}
 
 	e.POST("/v1/run", h, mw)
-	return e, mw
+	return e, cfg.Metrics
 }
 
 func doRequest(e *echo.Echo) *httptest.ResponseRecorder {
@@ -42,8 +45,18 @@ func doRequestWithContext(e *echo.Echo, ctx context.Context) *httptest.ResponseR
 	return rec
 }
 
+func assertMetricsZero(t *testing.T, metrics *ConcurrencyMetrics) {
+	t.Helper()
+	if v := metrics.Active.Load(); v != 0 {
+		t.Errorf("expected Active=0 after completion, got %d", v)
+	}
+	if v := metrics.Queued.Load(); v != 0 {
+		t.Errorf("expected Queued=0 after completion, got %d", v)
+	}
+}
+
 func TestConcurrencyLimiter_UnderLimit(t *testing.T) {
-	e, _ := setupEcho(ConcurrencyConfig{
+	e, metrics := setupEcho(ConcurrencyConfig{
 		MaxConcurrency: 2,
 		MaxQueueSize:   5,
 		QueueTimeout:   5 * time.Second,
@@ -53,11 +66,12 @@ func TestConcurrencyLimiter_UnderLimit(t *testing.T) {
 	if rec.Code != http.StatusOK {
 		t.Fatalf("expected 200, got %d", rec.Code)
 	}
+	assertMetricsZero(t, metrics)
 }
 
 func TestConcurrencyLimiter_QueueAndSucceed(t *testing.T) {
 	gate := make(chan struct{})
-	e, _ := setupEcho(ConcurrencyConfig{
+	e, metrics := setupEcho(ConcurrencyConfig{
 		MaxConcurrency: 1,
 		MaxQueueSize:   5,
 		QueueTimeout:   5 * time.Second,
@@ -98,11 +112,12 @@ func TestConcurrencyLimiter_QueueAndSucceed(t *testing.T) {
 	if queuedRec.Code != http.StatusOK {
 		t.Fatalf("queued request: expected 200, got %d", queuedRec.Code)
 	}
+	assertMetricsZero(t, metrics)
 }
 
 func TestConcurrencyLimiter_QueueFull(t *testing.T) {
 	gate := make(chan struct{})
-	e, _ := setupEcho(ConcurrencyConfig{
+	e, metrics := setupEcho(ConcurrencyConfig{
 		MaxConcurrency: 1,
 		MaxQueueSize:   1,
 		QueueTimeout:   5 * time.Second,
@@ -142,11 +157,12 @@ func TestConcurrencyLimiter_QueueFull(t *testing.T) {
 	// Cleanup: release gate and wait.
 	close(gate)
 	wg.Wait()
+	assertMetricsZero(t, metrics)
 }
 
 func TestConcurrencyLimiter_QueueTimeout(t *testing.T) {
 	gate := make(chan struct{})
-	e, _ := setupEcho(ConcurrencyConfig{
+	e, metrics := setupEcho(ConcurrencyConfig{
 		MaxConcurrency: 1,
 		MaxQueueSize:   5,
 		QueueTimeout:   100 * time.Millisecond,
@@ -178,11 +194,12 @@ func TestConcurrencyLimiter_QueueTimeout(t *testing.T) {
 	// Cleanup.
 	close(gate)
 	wg.Wait()
+	assertMetricsZero(t, metrics)
 }
 
 func TestConcurrencyLimiter_ClientCancel(t *testing.T) {
 	gate := make(chan struct{})
-	e, _ := setupEcho(ConcurrencyConfig{
+	e, metrics := setupEcho(ConcurrencyConfig{
 		MaxConcurrency: 1,
 		MaxQueueSize:   5,
 		QueueTimeout:   10 * time.Second,
@@ -227,4 +244,5 @@ func TestConcurrencyLimiter_ClientCancel(t *testing.T) {
 	// Cleanup.
 	close(gate)
 	wg.Wait()
+	assertMetricsZero(t, metrics)
 }

internal/middleware/metrics.go

@@ -0,0 +1,30 @@
+package middleware
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/labstack/echo/v5"
+)
+
+// MetricsHandler returns an Echo handler that renders Prometheus text exposition
+// format for the sandbox concurrency and queue gauges.
+func MetricsHandler(metrics *ConcurrencyMetrics, maxConcurrency, maxQueueSize int) echo.HandlerFunc {
+	return func(c *echo.Context) error {
+		body := fmt.Sprintf("# HELP sandbox_concurrency_active Number of requests currently executing.\n"+
+			"# TYPE sandbox_concurrency_active gauge\n"+
+			"sandbox_concurrency_active %d\n"+
+			"# HELP sandbox_queue_length Number of requests waiting in queue.\n"+
+			"# TYPE sandbox_queue_length gauge\n"+
+			"sandbox_queue_length %d\n"+
+			"# HELP sandbox_concurrency_max Configured maximum concurrent executions.\n"+
+			"# TYPE sandbox_concurrency_max gauge\n"+
+			"sandbox_concurrency_max %d\n"+
+			"# HELP sandbox_queue_max Configured maximum queue size.\n"+
+			"# TYPE sandbox_queue_max gauge\n"+
+			"sandbox_queue_max %d\n",
+			metrics.Active.Load(), metrics.Queued.Load(), maxConcurrency, maxQueueSize)
+		c.Response().Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+		return c.String(http.StatusOK, body)
+	}
+}

internal/middleware/metrics_test.go

@@ -0,0 +1,33 @@
+package middleware
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/labstack/echo/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetricsHandler(t *testing.T) {
+	metrics := &ConcurrencyMetrics{}
+	metrics.Active.Store(3)
+	metrics.Queued.Store(7)
+
+	e := echo.New()
+	e.GET("/metrics", MetricsHandler(metrics, 10, 50))
+
+	req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+	rec := httptest.NewRecorder()
+	e.ServeHTTP(rec, req)
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	assert.Contains(t, rec.Header().Get("Content-Type"), "text/plain")
+
+	body := rec.Body.String()
+	assert.Contains(t, body, "sandbox_concurrency_active 3")
+	assert.Contains(t, body, "sandbox_queue_length 7")
+	assert.Contains(t, body, "sandbox_concurrency_max 10")
+	assert.Contains(t, body, "sandbox_queue_max 50")
+}