feat: add http_server.tcp_rejected_rate metric for connection-cap saturation

changelog/fragments/1785309171-tcp-rejected-rate-metric.yaml

@@ -0,0 +1,5 @@
+kind: enhancement
+
+summary: Add http_server.tcp_rejected_rate metric for connection-cap saturation signalling
+
+component: fleet-server

internal/pkg/api/metrics.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -33,9 +34,13 @@ import (
 var (
 	registry *metricsRegistry
 
-	cntHTTPNew    *statsCounter
-	cntHTTPClose  *statsCounter
-	cntHTTPActive *statsGauge
+	cntHTTPNew          *statsCounter
+	cntHTTPClose        *statsCounter
+	cntHTTPActive       *statsGauge
+	cntHTTPRejectedRate *statsGauge
+	// cntHTTPRejected is an internal-only accumulator; only the derived rate is
+	// exposed in /stats. Not registered with the metrics registry.
+	cntHTTPRejected atomic.Uint64
 
 	cntCheckin       routeStats
 	cntEnroll        routeStats
@@ -61,6 +66,7 @@ func init() {
 	cntHTTPNew = newCounter(registry, "tcp_open")
 	cntHTTPClose = newCounter(registry, "tcp_close")
 	cntHTTPActive = newGauge(registry, "tcp_active")
+	cntHTTPRejectedRate = newGauge(registry, "tcp_rejected_rate")
 
 	routesRegistry := registry.newRegistry("routes")
 
@@ -326,9 +332,9 @@ func attachPrometheusEndpoint(router metricsRouter, reg *prometheus.Registry, bi
 	router.AddRoute("/metrics", promhttp.InstrumentMetricHandler(reg, h).ServeHTTP)
 }
 
-// CheckinRateSampleInterval is how often RunCheckinRejectionRateSampler recomputes
-// the checkin capacity-rejection rate.
-const CheckinRateSampleInterval = 10 * time.Second
+// RejectionRateSampleInterval is how often the rejection-rate samplers recompute
+// the checkin and connection-cap rejection rates.
+const RejectionRateSampleInterval = 10 * time.Second
 
 // computeRate returns the count of rejections (cur-prev) divided by an elapsed
 // time (dt), rounded to the nearest integer. It returns 0 if dt is non-positive
@@ -352,6 +358,45 @@ func computeRate(prev, cur uint64, dt time.Duration) uint64 {
 	return uint64(float64(cur-prev)/dt.Seconds() + 0.5)
 }
 
+// RunConnRejectionRateSampler periodically samples the connection-cap rejection
+// counter (http_server.tcp_rejected) and publishes the resulting rate as the
+// tcp_rejected_rate gauge.
+//
+// tcp_rejected_rate is a saturation signal that complements tcp_active:
+//   - tcp_active is a utilization signal — EPA uses it to maintain headroom by
+//     scaling before pods fill up. It works well at steady state but understates
+//     demand when connections are short-lived (e.g. an enrollment burst), because
+//     the metric lags behind a burst that arrives and clears between sample points.
+//   - tcp_rejected_rate fires only when the pod is already at its connection
+//     ceiling and actively turning clients away (HTTP 429 from the chi Throttle
+//     middleware). It catches bursts that tcp_active misses, triggering an
+//     immediate scale-up rather than waiting for the next utilization sample.
+//
+// Together, EPA can scale both proactively (tcp_active: maintain headroom during
+// steady-state load) and reactively (tcp_rejected_rate: act the moment a burst
+// overwhelms a pod). EPA takes the maximum desired replica count across all
+// configured metrics, so tcp_rejected_rate can only make scale-up more aggressive,
+// never suppress it.
+//
+// Run blocks until ctx is canceled, following the same lifecycle contract as the
+// other subsystems started in Fleet.runSubsystems.
+func RunConnRejectionRateSampler(ctx context.Context, interval time.Duration) error {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	prev, prevT := cntHTTPRejected.Load(), time.Now()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case now := <-ticker.C:
+			cur := cntHTTPRejected.Load()
+			cntHTTPRejectedRate.Set(computeRate(prev, cur, now.Sub(prevT)))
+			prev, prevT = cur, now
+		}
+	}
+}
+
 // RunCheckinRejectionRateSampler periodically samples the checkin route's
 // limit_max counter (checkins rejected because Fleet Server was at capacity) and
 // publishes the resulting rate as the limit_max_rate gauge.

internal/pkg/api/metrics_test.go

@@ -77,6 +77,58 @@ func TestComputeRate(t *testing.T) {
 	}
 }
 
+func TestRunConnRejectionRateSampler(t *testing.T) {
+	// Swap in a fresh gauge registered under a private namespace to avoid
+	// duplicate-name panics in the Prometheus registry across repeated runs.
+	origRate := cntHTTPRejectedRate
+	t.Cleanup(func() { cntHTTPRejectedRate = origRate })
+	cntHTTPRejectedRate = newGauge(
+		registry.newRegistry(fmt.Sprintf("test_conn_rejection_rate_sampler_%d", testRegistrySeq.Add(1))),
+		"tcp_rejected_rate",
+	)
+
+	ctx, cancel := context.WithCancel(t.Context())
+
+	const interval = 10 * time.Millisecond
+	done := make(chan error, 1)
+	go func() {
+		done <- RunConnRejectionRateSampler(ctx, interval)
+	}()
+
+	injectorCtx, stopInjector := context.WithCancel(ctx)
+	injectorDone := make(chan struct{})
+	go func() {
+		defer close(injectorDone)
+		ticker := time.NewTicker(interval / 2)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-injectorCtx.Done():
+				return
+			case <-ticker.C:
+				cntHTTPRejected.Add(1)
+			}
+		}
+	}()
+
+	t.Cleanup(func() {
+		cancel()
+		stopInjector()
+		require.NoError(t, <-done)
+		<-injectorDone
+	})
+
+	require.Eventually(t, func() bool {
+		return cntHTTPRejectedRate.metric.Get() > 0
+	}, 20*interval, interval/4, "expected the connection rejection rate gauge to become positive while rejections were ongoing")
+
+	stopInjector()
+	<-injectorDone
+	require.Eventually(t, func() bool {
+		return cntHTTPRejectedRate.metric.Get() == 0
+	}, 20*interval, interval/4, "expected the connection rejection rate gauge to return to zero after rejections stopped")
+}
+
 func TestRunCheckinRejectionRateSampler(t *testing.T) {
 	// cntCheckin is a package-level global shared with the checkin route handler
 	// and other tests in this package. Swap in a throwaway routeStats registered

internal/pkg/api/router.go

@@ -28,7 +28,7 @@ func newRouter(cfg *config.ServerLimits, si ServerInterface, tracer *apm.Tracer)
 	r.Use(logger.Middleware) // Attach middlewares to router directly so the occur before any request parsing/validation
 	r.Use(middleware.Recoverer)
 	if cfg.MaxConnections > 0 {
-		r.Use(middleware.Throttle(cfg.MaxConnections))
+		r.Use(throttleWithCount(cfg.MaxConnections))
 	}
 	r.Use(Limiter(cfg).middleware)
 	return HandlerWithOptions(si, ChiServerOptions{
@@ -74,6 +74,47 @@ func Limiter(cfg *config.ServerLimits) *limiter {
 	}
 }
 
+// throttleWithCount wraps chi's Throttle middleware and increments
+// cntHTTPRejected whenever a request is rejected with HTTP 429 due to the
+// per-pod connection cap. The count is rate-sampled by RunConnRejectionRateSampler
+// and exposed as http_server.tcp_rejected_rate for use as an EPA scaling trigger.
+func throttleWithCount(limit int) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		throttled := middleware.Throttle(limit)(next)
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			rw := &statusRecorder{ResponseWriter: w}
+			throttled.ServeHTTP(rw, r)
+			if rw.status == http.StatusTooManyRequests {
+				cntHTTPRejected.Add(1)
+			}
+		})
+	}
+}
+
+// statusRecorder captures the HTTP status code written by a downstream handler.
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+}
+
+func (s *statusRecorder) Write(buf []byte) (int, error) {
+	if s.status == 0 {
+		s.WriteHeader(http.StatusOK)
+	}
+	return s.ResponseWriter.Write(buf)
+}
+
+func (s *statusRecorder) WriteHeader(code int) {
+	if s.status == 0 {
+		s.status = code
+	}
+	s.ResponseWriter.WriteHeader(code)
+}
+
+func (s *statusRecorder) Unwrap() http.ResponseWriter {
+	return s.ResponseWriter
+}
+
 var pgpReg = regexp.MustCompile(`\/api\/agents\/upgrades\/[0-9]+\.[0-9]+\.[0-9]+\/pgp-public-key`)
 
 // pathToOperation determines the endpoint passed on the request path.

internal/pkg/api/router_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/elastic/fleet-server/v7/internal/pkg/config"
 	"github.com/go-chi/chi/v5"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestPathToOperation(t *testing.T) {
@@ -96,7 +97,7 @@ func TestLimiter(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			h := testStatusServer(t, tt.cfg)
 			w := httptest.NewRecorder()
-			r := httptest.NewRequest("GET", "/api/status", nil)
+			r := httptest.NewRequestWithContext(t.Context(), "GET", "/api/status", nil)
 
 			h.ServeHTTP(w, r)
 			resp := w.Result()
@@ -105,3 +106,63 @@ func TestLimiter(t *testing.T) {
 		})
 	}
 }
+
+func TestStatusRecorder(t *testing.T) {
+	t.Run("captures explicit status", func(t *testing.T) {
+		rw := httptest.NewRecorder()
+		rec := &statusRecorder{ResponseWriter: rw}
+		rec.WriteHeader(http.StatusTeapot)
+		require.Equal(t, http.StatusTeapot, rec.status)
+		require.Equal(t, http.StatusTeapot, rw.Code)
+	})
+
+	t.Run("captures implicit success status", func(t *testing.T) {
+		rw := httptest.NewRecorder()
+		rec := &statusRecorder{ResponseWriter: rw}
+		_, err := rec.Write([]byte("ok"))
+		require.NoError(t, err)
+		require.Equal(t, http.StatusOK, rec.status)
+		require.Equal(t, http.StatusOK, rw.Code)
+		require.Equal(t, "ok", rw.Body.String())
+	})
+}
+
+func TestThrottleWithCount(t *testing.T) {
+	t.Run("does not increment counter on success", func(t *testing.T) {
+		before := cntHTTPRejected.Load()
+		h := throttleWithCount(5)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		}))
+		w := httptest.NewRecorder()
+		h.ServeHTTP(w, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
+		require.Equal(t, http.StatusOK, w.Code)
+		require.Equal(t, before, cntHTTPRejected.Load())
+	})
+
+	t.Run("increments counter when connection cap is hit", func(t *testing.T) {
+		before := cntHTTPRejected.Load()
+
+		started := make(chan struct{})
+		release := make(chan struct{})
+		requestDone := make(chan struct{})
+		h := throttleWithCount(1)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			close(started)
+			<-release
+		}))
+
+		go func() {
+			defer close(requestDone)
+			h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
+		}()
+		t.Cleanup(func() {
+			close(release)
+			<-requestDone
+		})
+		<-started // first request has acquired the connection slot
+
+		w := httptest.NewRecorder()
+		h.ServeHTTP(w, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
+		require.Equal(t, http.StatusTooManyRequests, w.Code)
+		require.Equal(t, before+1, cntHTTPRejected.Load())
+	})
+}

internal/pkg/server/fleet.go

@@ -527,7 +527,12 @@ func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgro
 	// Samples the checkin capacity-rejection counter into a rate gauge, exposed via
 	// /stats for use as an autoscaling signal. See api.RunCheckinRejectionRateSampler.
 	g.Go(loggedRunFunc(ctx, "Checkin rejection rate sampler", func(ctx context.Context) error {
-		return api.RunCheckinRejectionRateSampler(ctx, api.CheckinRateSampleInterval)
+		return api.RunCheckinRejectionRateSampler(ctx, api.RejectionRateSampleInterval)
+	}))
+	// Samples the connection-cap rejection counter into a rate gauge, exposed via
+	// /stats for use as an autoscaling signal. See api.RunConnRejectionRateSampler.
+	g.Go(loggedRunFunc(ctx, "Connection rejection rate sampler", func(ctx context.Context) error {
+		return api.RunConnRejectionRateSampler(ctx, api.RejectionRateSampleInterval)
 	}))
 
 	ct, err := api.NewCheckinT(f.verCon, &cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, bulker)