diff --git a/changelog/fragments/1785309171-tcp-rejected-rate-metric.yaml b/changelog/fragments/1785309171-tcp-rejected-rate-metric.yaml new file mode 100644 index 0000000000..9003b14eff --- /dev/null +++ b/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 diff --git a/internal/pkg/api/metrics.go b/internal/pkg/api/metrics.go index 72dae34ea3..6e712001c6 100644 --- a/internal/pkg/api/metrics.go +++ b/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. diff --git a/internal/pkg/api/metrics_test.go b/internal/pkg/api/metrics_test.go index 06a6a31853..3686c325d2 100644 --- a/internal/pkg/api/metrics_test.go +++ b/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 diff --git a/internal/pkg/api/router.go b/internal/pkg/api/router.go index af70d0e734..64da8aea73 100644 --- a/internal/pkg/api/router.go +++ b/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. diff --git a/internal/pkg/api/router_test.go b/internal/pkg/api/router_test.go index fa348894f5..d9a23b7e78 100644 --- a/internal/pkg/api/router_test.go +++ b/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()) + }) +} diff --git a/internal/pkg/server/fleet.go b/internal/pkg/server/fleet.go index 055a3dcb68..819ecfd367 100644 --- a/internal/pkg/server/fleet.go +++ b/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)