diff --git a/internal/observabilityproxy/server.go b/internal/observabilityproxy/server.go index 2e391442c..0f9e533b7 100644 --- a/internal/observabilityproxy/server.go +++ b/internal/observabilityproxy/server.go @@ -266,8 +266,12 @@ 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 +// readForm reads the whole body under the request timeout, so a client +// that sends it slowly holds its slot only that long. ParseForm leaves the +// body of a GET or of a POST that is not a form unread, and net/http +// drains a leftover body with no deadline once the headers go out, so the +// rest is read here as well; draining rather than refusing a declared +// body keeps any client that sends one working. 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. @@ -278,7 +282,13 @@ func (s *Server) readForm(w http.ResponseWriter, r *http.Request) *HTTPError { _ = rc.SetReadDeadline(time.Now().Add(s.cfg.RequestTimeout)) } err := r.ParseForm() + if err == nil { + _, err = io.Copy(io.Discard, r.Body) + } + var tooLarge *http.MaxBytesError switch { + case errors.As(err, &tooLarge): + return errorf(http.StatusRequestEntityTooLarge, "request body larger than %d bytes", tooLarge.Limit) case err == nil: _ = rc.SetReadDeadline(time.Time{}) return nil diff --git a/internal/observabilityproxy/server_deadline_test.go b/internal/observabilityproxy/server_deadline_test.go index afd69ed30..ac62dd782 100644 --- a/internal/observabilityproxy/server_deadline_test.go +++ b/internal/observabilityproxy/server_deadline_test.go @@ -21,6 +21,7 @@ import ( "io" "net" "net/http" + "strings" "sync" "testing" "time" @@ -244,3 +245,55 @@ func TestServerAbortsTruncatedResponse(t *testing.T) { t.Fatal("a truncated store answer reached the client as a complete response") } } + +// TestServerUndeliveredBodyReleasesSlot declares a body that never comes +// on requests whose form parsing does not read it: a GET, a POST that is +// not a form, and a GET tail. net/http would drain that body with no +// deadline once the headers go out, so the slot must still come back. +func TestServerUndeliveredBodyReleasesSlot(t *testing.T) { + for name, raw := range map[string]string{ + "get query": "GET " + logsRoot + "query?query=* HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\nContent-Length: 10\r\n\r\n", + "post text query": "POST " + logsRoot + "query?query=* HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\nContent-Type: text/plain\r\nContent-Length: 10\r\n\r\n", + "get tail": "GET " + logsRoot + "tail?query=* HTTP/1.1\r\nHost: proxy\r\nAuthorization: Bearer good\r\nContent-Length: 10\r\n\r\n", + } { + t.Run(name, func(t *testing.T) { + lines := make(chan string, 1) + done := make(chan struct{}) + h := newHarness(t, Config{ + RequestTimeout: 200 * time.Millisecond, + TailIdleTimeout: 200 * time.Millisecond, + Limiter: LimiterConfig{MaxConcurrentPerUser: 1, MaxTails: 1}, + }, tailUpstream(lines, done)) + t.Cleanup(func() { close(done) }) + pipe := servePipe(t, h.server) + conn := pipe.send(t, raw) + held := func() bool { + h.server.limits.mu.Lock() + defer h.server.limits.mu.Unlock() + return h.server.limits.active["alice"]+h.server.limits.total > 0 + } + eventually(t, time.Second, held, "the request never took a slot") + if resp := readResponse(t, conn); resp.StatusCode != http.StatusRequestTimeout { + t.Fatalf("undelivered body answered %d, want 408", resp.StatusCode) + } + eventually(t, time.Second, func() bool { return !held() }, "an undelivered body keeps its slot") + }) + } +} + +func TestServerOversizedBody(t *testing.T) { + h := newHarness(t, Config{}, nil) + for _, ct := range []string{"application/x-www-form-urlencoded", "text/plain"} { + req, _ := http.NewRequest(http.MethodPost, h.proxy.URL+logsRoot+"query?query=*", strings.NewReader("query="+strings.Repeat("x", maxFormBytes))) + req.Header.Set("Authorization", "Bearer good") + req.Header.Set("Content-Type", ct) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusRequestEntityTooLarge { + t.Errorf("%s body over the cap answered %d, want 413", ct, resp.StatusCode) + } + } +}