From 0a70332adf04477e0fb6c52ff623248c2504fb8c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 16 Jun 2026 16:29:30 -0700 Subject: [PATCH] s3api: add optional request interceptor to circuit breaker (#9994) * s3api: add optional request interceptor to circuit breaker Add an optional Interceptor func(next http.HandlerFunc, action string) http.HandlerFunc field on CircuitBreaker, applied at the very top of Limit() -- before upload concurrency limiting and before the 'if !cb.Enabled' early return -- so it runs for every route regardless of breaker state. It is nil by default (no behavior change). An interceptor may reject a request (write its own response and skip next) or wrap next to observe/shape it. This provides a single per-request extension point at the existing per-route chokepoint (cb.Limit already wraps nearly every S3 route), useful for tracing, auditing, or request rate limiting without touching the route table. * s3api: evaluate circuit breaker interceptor per request Move the Interceptor nil-check into the returned handler so it is consulted per request instead of captured at registration time. This: - lets the interceptor be installed after registerRouter runs (handlers are built during construction, before request-time dependencies exist), and - avoids dereferencing a nil CircuitBreaker at registration time, which panicked tests that register routes on a server with a nil cb (e.g. TestRouting_STSWithQueryParams). Adds a regression test for installing the interceptor after Limit() returns. --- weed/s3api/s3api_circuit_breaker.go | 24 +++++- weed/s3api/s3api_circuit_breaker_test.go | 97 ++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 1 deletion(-) diff --git a/weed/s3api/s3api_circuit_breaker.go b/weed/s3api/s3api_circuit_breaker.go index 9296036af..786542d75 100644 --- a/weed/s3api/s3api_circuit_breaker.go +++ b/weed/s3api/s3api_circuit_breaker.go @@ -25,6 +25,14 @@ type CircuitBreaker struct { counters map[string]*int64 limitations map[string]int64 s3a *S3ApiServer + + // Interceptor, if set, wraps the per-route handler ahead of ALL + // circuit-breaker logic (upload limiting and the breaker checks) and runs + // even when the breaker is disabled. It is a general per-request + // interceptor seam: an implementation may reject the request (write its own + // response and not call next) or wrap next to observe/shape it. Nil by + // default, so it is a no-op unless explicitly set. + Interceptor func(next http.HandlerFunc, action string) http.HandlerFunc } func NewCircuitBreaker(option *S3ApiServerOption) *CircuitBreaker { @@ -92,7 +100,7 @@ func (cb *CircuitBreaker) loadCircuitBreakerConfig(cfg *s3_pb.S3CircuitBreakerCo } func (cb *CircuitBreaker) Limit(f func(w http.ResponseWriter, r *http.Request), action string) (http.HandlerFunc, Action) { - return func(w http.ResponseWriter, r *http.Request) { + inner := func(w http.ResponseWriter, r *http.Request) { // Apply upload limiting for write actions if configured if cb.s3a != nil && (action == s3_constants.ACTION_WRITE) && (cb.s3a.option.ConcurrentUploadLimit != 0 || cb.s3a.option.ConcurrentFileUploadLimit != 0) { @@ -161,6 +169,20 @@ func (cb *CircuitBreaker) Limit(f func(w http.ResponseWriter, r *http.Request), return } s3err.WriteErrorResponse(w, r, errCode) + } + + // The interceptor is consulted per request rather than captured here, so it + // can be installed after the routes are registered (e.g. once the server and + // its dependencies are constructed) and so a nil CircuitBreaker is never + // dereferenced at registration time. When unset this is just a nil check. + // It runs outermost: before upload limiting and the breaker checks, and + // regardless of cb.Enabled. + return func(w http.ResponseWriter, r *http.Request) { + if cb.Interceptor != nil { + cb.Interceptor(inner, action)(w, r) + return + } + inner(w, r) }, Action(action) } diff --git a/weed/s3api/s3api_circuit_breaker_test.go b/weed/s3api/s3api_circuit_breaker_test.go index 72e166558..cf5e7817d 100644 --- a/weed/s3api/s3api_circuit_breaker_test.go +++ b/weed/s3api/s3api_circuit_breaker_test.go @@ -2,6 +2,7 @@ package s3api import ( "net/http" + "net/http/httptest" "sync" "sync/atomic" "testing" @@ -106,3 +107,99 @@ func doLimit(circuitBreaker *CircuitBreaker, routineCount int, r *http.Request, } return successCounter } + +// TestLimitInterceptor verifies the optional request interceptor: it is a no-op +// when nil, runs ahead of the (disabled) breaker logic, and can either reject a +// request or pass it through to the wrapped handler. +func TestLimitInterceptor(t *testing.T) { + readAction := s3_constants.ACTION_READ + newCB := func() *CircuitBreaker { + // Enabled defaults to false and s3a is nil, so without an interceptor + // Limit's handler falls straight through to the wrapped handler. + return &CircuitBreaker{counters: make(map[string]*int64), limitations: make(map[string]int64)} + } + + // 1. nil interceptor must not change behavior: the handler still runs. + t.Run("nil interceptor is a no-op", func(t *testing.T) { + cb := newCB() + called := false + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { called = true }, readAction) + h(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/bucket/object", nil)) + if !called { + t.Fatal("handler should run when no interceptor is set") + } + }) + + // 2. a rejecting interceptor runs first and prevents the handler, even + // though the breaker itself is disabled. + t.Run("interceptor can reject", func(t *testing.T) { + cb := newCB() + var order []string + cb.Interceptor = func(next http.HandlerFunc, action string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + order = append(order, "interceptor") + s3err.WriteErrorResponse(w, r, s3err.ErrTooManyRequest) // reject; do not call next + } + } + handlerRan := false + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { + handlerRan = true + order = append(order, "handler") + }, readAction) + rec := httptest.NewRecorder() + h(rec, httptest.NewRequest(http.MethodGet, "/bucket/object", nil)) + if handlerRan { + t.Fatal("a rejecting interceptor must prevent the handler from running") + } + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503 from ErrTooManyRequest, got %d", rec.Code) + } + if len(order) != 1 || order[0] != "interceptor" { + t.Fatalf("interceptor should run alone, got %v", order) + } + }) + + // 3. a pass-through interceptor runs before the handler and sees the action. + t.Run("interceptor can pass through", func(t *testing.T) { + cb := newCB() + var order []string + var seenAction string + cb.Interceptor = func(next http.HandlerFunc, action string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + seenAction = action + order = append(order, "interceptor") + next(w, r) + } + } + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) { + order = append(order, "handler") + }, readAction) + h(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/bucket/object", nil)) + if seenAction != readAction { + t.Fatalf("interceptor should receive the route action, got %q", seenAction) + } + if len(order) != 2 || order[0] != "interceptor" || order[1] != "handler" { + t.Fatalf("interceptor must run before the handler, got %v", order) + } + }) + + // 4. installed AFTER Limit() returns: still takes effect, because the + // interceptor is consulted per request rather than captured at + // registration time (the handlers are built during router registration, + // before dependencies that need the running server exist). + t.Run("interceptor installed after registration takes effect", func(t *testing.T) { + cb := newCB() + h, _ := cb.Limit(func(w http.ResponseWriter, r *http.Request) {}, readAction) // built while nil + ran := false + cb.Interceptor = func(next http.HandlerFunc, action string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + ran = true + next(w, r) + } + } + h(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/bucket/object", nil)) + if !ran { + t.Fatal("interceptor set after Limit() must still run (request-time evaluation)") + } + }) +}