From f81aaf4d256c3d8ef2554b8a2a029dc36ea500de Mon Sep 17 00:00:00 2001 From: ZverGuy Date: Thu, 5 Nov 2026 00:18:23 +0700 Subject: [PATCH] fix(observability-proxy): admit a request before reading its body The form was parsed before the concurrency slot was taken, and the listener sets no body-read deadline, so a client that never finished a POST body held a handler outside every limit for as long as it liked. Tenant workloads reach the Service directly, so nginx buffering does not cover this path. The slot is now taken first and the body is read under the request timeout; a body that does not arrive in time answers 408 and frees the slot. The deadline is lifted once the body is read, because the server's background read would take a later expiry for a client that went away and cancel a live tail. Assisted-by: LLM Signed-off-by: ZverGuy --- internal/observabilityproxy/server.go | 40 +++-- .../server_deadline_test.go | 137 ++++++++++++++++++ 2 files changed, 168 insertions(+), 9 deletions(-) create mode 100644 internal/observabilityproxy/server_deadline_test.go diff --git a/internal/observabilityproxy/server.go b/internal/observabilityproxy/server.go index 043fd93d5..72d380e21 100644 --- a/internal/observabilityproxy/server.go +++ b/internal/observabilityproxy/server.go @@ -25,6 +25,7 @@ import ( "log/slog" "math" "net/http" + "os" "strconv" "strings" "time" @@ -159,15 +160,6 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTT if !ok { return errorf(http.StatusNotFound, "not found") } - r.Body = http.MaxBytesReader(w, r.Body, maxFormBytes) - if err := r.ParseForm(); err != nil { - return errorf(http.StatusBadRequest, "invalid form: %v", err) - } - params, herr := BuildLogsParams(endpoint, r.Form, route.Namespace, s.now(), s.cfg.Logs) - if herr != nil { - return herr - } - tail := endpoint == "tail" acquire := s.limits.acquire if tail { @@ -178,6 +170,13 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTT return errorf(http.StatusTooManyRequests, "too many concurrent requests") } defer release() + if herr := s.readForm(w, r); herr != nil { + return herr + } + params, herr := BuildLogsParams(endpoint, r.Form, route.Namespace, s.now(), s.cfg.Logs) + if herr != nil { + return herr + } store, err := s.stores.Resolve(route.Namespace) if err != nil { @@ -233,6 +232,29 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTT return nil } +// readForm reads the body under the request timeout, so a client that +// sends it slowly holds its slot only that long. Once the body is read the +// deadline is lifted: the server's background read would take a later +// expiry for a client that went away and cancel a tail. After a failed +// read it stays, so the server's drain of the unread body is bounded too. +func (s *Server) readForm(w http.ResponseWriter, r *http.Request) *HTTPError { + r.Body = http.MaxBytesReader(w, r.Body, maxFormBytes) + rc := http.NewResponseController(w) + if s.cfg.RequestTimeout > 0 { + _ = rc.SetReadDeadline(time.Now().Add(s.cfg.RequestTimeout)) + } + err := r.ParseForm() + switch { + case err == nil: + _ = rc.SetReadDeadline(time.Time{}) + return nil + case errors.Is(err, os.ErrDeadlineExceeded): + return errorf(http.StatusRequestTimeout, "request body not received within %s", s.cfg.RequestTimeout) + default: + return errorf(http.StatusBadRequest, "invalid form: %v", err) + } +} + func (s *Server) authorize(r *http.Request, route Route) (string, *HTTPError) { token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") if !ok { diff --git a/internal/observabilityproxy/server_deadline_test.go b/internal/observabilityproxy/server_deadline_test.go new file mode 100644 index 000000000..6931d8c01 --- /dev/null +++ b/internal/observabilityproxy/server_deadline_test.go @@ -0,0 +1,137 @@ +/* +Copyright 2026 The Cozystack Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package observabilityproxy + +import ( + "bufio" + "net" + "net/http" + "sync" + "testing" + "time" +) + +// pipeListener hands the server one end of a net.Pipe per dial. A pipe has +// no buffer, so a client that stops reading blocks the server's next write +// at once, as a full TCP window eventually does. +type pipeListener struct { + conns chan net.Conn + closed chan struct{} + once sync.Once +} + +type pipeAddr struct{} + +func (pipeAddr) Network() string { return "pipe" } +func (pipeAddr) String() string { return "pipe" } + +func servePipe(t *testing.T, handler http.Handler) *pipeListener { + t.Helper() + l := &pipeListener{conns: make(chan net.Conn), closed: make(chan struct{})} + srv := &http.Server{Handler: handler, ReadHeaderTimeout: 10 * time.Second} + go func() { _ = srv.Serve(l) }() + t.Cleanup(func() { _ = srv.Close() }) + return l +} + +func (l *pipeListener) Accept() (net.Conn, error) { + select { + case c := <-l.conns: + return c, nil + case <-l.closed: + return nil, net.ErrClosed + } +} + +func (l *pipeListener) Close() error { + l.once.Do(func() { close(l.closed) }) + return nil +} + +func (l *pipeListener) Addr() net.Addr { return pipeAddr{} } + +// send opens a connection and writes raw to it without waiting for the +// server to read all of it. +func (l *pipeListener) send(t *testing.T, raw string) net.Conn { + t.Helper() + client, server := net.Pipe() + t.Cleanup(func() { _ = client.Close() }) + select { + case l.conns <- server: + case <-time.After(2 * time.Second): + t.Fatal("pipe server is not accepting") + } + go func() { _, _ = client.Write([]byte(raw)) }() + return client +} + +func eventually(t *testing.T, within time.Duration, cond func() bool, msg string) { + t.Helper() + deadline := time.Now().Add(within) + for !cond() { + if time.Now().After(deadline) { + t.Fatal(msg) + } + time.Sleep(10 * time.Millisecond) + } +} + +func readResponse(t *testing.T, conn net.Conn) *http.Response { + t.Helper() + got := make(chan *http.Response, 1) + go func() { + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + close(got) + return + } + got <- resp + }() + select { + case resp, ok := <-got: + if !ok { + t.Fatal("reading the response failed") + } + return resp + case <-time.After(3 * time.Second): + t.Fatal("no response") + } + return nil +} + +// TestServerSlowBodyHoldsSlotUntilDeadline sends a POST whose body never +// completes: it must count against the user's concurrency, and the read +// must give up at the request timeout instead of holding the handler. +func TestServerSlowBodyHoldsSlotUntilDeadline(t *testing.T) { + h := newHarness(t, Config{ + RequestTimeout: 200 * time.Millisecond, + Limiter: LimiterConfig{MaxConcurrentPerUser: 1}, + }, nil) + pipe := servePipe(t, h.server) + conn := pipe.send(t, "POST "+logsRoot+"query HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\n"+ + "Content-Type: application/x-www-form-urlencoded\r\nContent-Length: 100\r\n\r\nquery=") + + eventually(t, time.Second, func() bool { + return h.status("GET", logsRoot+"query", "good") == http.StatusTooManyRequests + }, "a request with an unfinished body does not hold a concurrency slot") + if resp := readResponse(t, conn); resp.StatusCode != http.StatusRequestTimeout { + t.Fatalf("unfinished body answered %d, want 408", resp.StatusCode) + } + eventually(t, 2*time.Second, func() bool { + return h.status("GET", logsRoot+"query", "good") == http.StatusOK + }, "the slot of an unfinished body was not released") +}