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") +}