diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0edffe94..2a091e90 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -445,6 +445,45 @@ jobs: - name: middleware/otel run: go test -race -count=1 -timeout=120s ./... working-directory: middleware/otel + # celeris#720, #732: Adapt must hand net/http the request headers, and + # Adapt, adapters.WrapMiddleware and the metrics, otel and logger + # middleware must hand net/http, Prometheus, the OTel SDK and slog + # copies of the request strings those keep, not views of the + # connection's receive buffer. The steps above run these packages, but + # without -v, and each of these tests drops its io_uring arms without a + # word when the probe finds no ring, which happens while another test + # binary's rings are still being freed. So they run here by name, one + # package at a time, with CELERIS_REQUIRE_IOURING_WORKERS=1 turning a + # missing io_uring arm into a failure, and an exact tally: the seven + # top-level tests and all 35 arms (std, and epoll and io_uring with sync + # and async handlers; 14 of them io_uring) must PASS, with no SKIP line, + # and every go test must exit 0. A failing test does not stop the step + # early, and the step runs even when a step above failed. + - name: celeris#720, #732 request strings handed to net/http, Prometheus, OTel, slog (io_uring arms required) + if: ${{ !cancelled() }} + shell: bash + env: + CELERIS_REQUIRE_IOURING_WORKERS: "1" + run: | + set -o pipefail + echo "memlock (KiB): $(ulimit -l)" + : > /tmp/c732.log + failed=0 + go test -race -count=1 -timeout=300s -v -run '^(TestAdaptRequestCarriesHeaders|TestAdaptKeptRequestStringsSurviveNextRequest)$' . 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1)) + go test -race -count=1 -timeout=300s -v -run '^TestWrapMiddlewareKeptStringsSurviveNextRequest$' ./middleware/adapters/ 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1)) + go test -race -count=1 -timeout=300s -v -run '^TestKeptRecordSurvivesNextRequest$' ./middleware/logger/ 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1)) + (cd middleware/metrics && go test -race -count=1 -timeout=300s -v -run '^(TestLabelValuesSurviveNextRequest|TestLabelValuesSurviveConnectionReuse)$' .) 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1)) + (cd middleware/otel && go test -race -count=1 -timeout=300s -v -run '^TestAttributesSurviveNextRequest$' .) 2>&1 | tee -a /tmp/c732.log || failed=$((failed + 1)) + tests='TestAdaptRequestCarriesHeaders|TestAdaptKeptRequestStringsSurviveNextRequest|TestWrapMiddlewareKeptStringsSurviveNextRequest|TestKeptRecordSurvivesNextRequest|TestLabelValuesSurviveNextRequest|TestLabelValuesSurviveConnectionReuse|TestAttributesSurviveNextRequest' + top=$(grep -cE "^--- PASS: (${tests}) \(" /tmp/c732.log || true) + arms=$(grep -cE "^ --- PASS: (${tests})/" /tmp/c732.log || true) + uring=$(grep -cE "^ --- PASS: (${tests})/io_uring" /tmp/c732.log || true) + skipped=$(grep -cE '^[[:space:]]*--- SKIP' /tmp/c732.log || true) + echo "celeris#720/#732 tests: top-level PASS $top (want 7), arm PASS $arms (want 35), io_uring arm PASS $uring (want 14), SKIP lines $skipped, go test runs failed $failed (want 0)" + if [ "$top" -ne 7 ] || [ "$arms" -ne 35 ] || [ "$uring" -ne 14 ] || [ "$skipped" -ne 0 ] || [ "$failed" -ne 0 ]; then + echo "expected all seven tests and all 35 arms (14 of them io_uring) to PASS with no SKIP line" + exit 1 + fi - name: middleware/protobuf run: go test -race -count=1 -timeout=120s ./... working-directory: middleware/protobuf diff --git a/adapt_request_linux_test.go b/adapt_request_linux_test.go new file mode 100644 index 00000000..83d05477 --- /dev/null +++ b/adapt_request_linux_test.go @@ -0,0 +1,299 @@ +//go:build linux + +package celeris_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "sort" + "strconv" + "strings" + "testing" + "time" + + "github.com/goceleris/celeris" + celerisengine "github.com/goceleris/celeris/engine" + "github.com/goceleris/celeris/probe" +) + +// TestAdaptRequestCarriesHeaders pins celeris#720. +// +// On epoll and io_uring the H1 parser builds the header slice lazily: it is +// filled the first time something reads a header. Adapt copied the headers +// from the slice without filling it, so on a route with no header read +// before it, the *http.Request had no headers at all. Nothing runs before +// the adapted handler here. +func TestAdaptRequestCarriesHeaders(t *testing.T) { + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + got := make(chan http.Header, 1) + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async}) + srv.POST("/h", celeris.Adapt(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got <- r.Header.Clone() + _, _ = w.Write([]byte("ok")) + }))) + return srv + }) + defer stop() + + conn, br := keptDial(t, addr) + defer func() { _ = conn.Close() }() + keptRoundTrip(t, conn, br, "POST /h HTTP/1.1\r\nHost: h.example\r\nX-Id: id-1\r\nAuthorization: Bearer t0k3n\r\nCookie: sid=s1\r\nContent-Type: application/json\r\nContent-Length: 2\r\n\r\n{}") + h := <-got + want := map[string]string{ + "X-Id": "id-1", + "Authorization": "Bearer t0k3n", + "Cookie": "sid=s1", + "Content-Type": "application/json", + } + var wrong []string + for k, v := range want { + if h.Get(k) != v { + wrong = append(wrong, fmt.Sprintf("%s=%q (want %q)", k, h.Get(k), v)) + } + } + sort.Strings(wrong) + t.Logf("KEPT720 arm=%s header keys=%d wrong=%d", a.name, len(h), len(wrong)) + if len(wrong) > 0 { + t.Errorf("the adapted handler's request is missing headers: %s", strings.Join(wrong, ", ")) + } + }) + } +} + +// TestAdaptKeptRequestStringsSurviveNextRequest pins the Adapt site of +// celeris#732. +// +// net/http lets a handler keep the request's strings after ServeHTTP +// returns. On epoll and io_uring the method, path, query, Host and header +// strings Adapt handed over were views of the connection's receive buffer, +// which the engine reuses for the connection's next request. Three requests +// with the same layout and different values go over one keep-alive +// connection; the handler keeps its request's strings, and after the third +// request each kept set must still read its own request. The methods are +// ones the H1 parser does not intern, the header name "111-111" is one +// net/http's canonicalization returns unchanged (it has no letters), and no +// request has a query (with one, the URL was a new string and the path a +// copy by accident), so none of them is copied on the way in. +func TestAdaptKeptRequestStringsSurviveNextRequest(t *testing.T) { + methods := []string{"TRACE", "PURGE", "MKCOL"} + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + kept := make(chan map[string]string, len(methods)) + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async}) + h := celeris.Adapt(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Legal in net/http: keep the request's strings. + got := map[string]string{ + "method": r.Method, + "path": r.URL.Path, + "query": r.URL.RawQuery, + "host": r.Host, + "x-id": r.Header.Get("X-Id"), + } + for k := range r.Header { + if strings.HasSuffix(k, "-111") || strings.HasSuffix(k, "-222") || strings.HasSuffix(k, "-333") { + got["digit-key"] = k + } + } + kept <- got + _, _ = w.Write([]byte("ok")) + })) + for _, m := range methods { + srv.Handle(m, "/a/:id", h) + } + return srv + }) + defer stop() + + conn, br := keptDial(t, addr) + defer func() { _ = conn.Close() }() + var want []map[string]string + for i, m := range methods { + v := strings.Repeat(string(rune('a'+i)), 4) + d := strings.Repeat(strconv.Itoa(i+1), 3) + keptRoundTrip(t, conn, br, m+" /a/"+v+" HTTP/1.1\r\nHost: h-"+v+".example\r\nX-Id: id-"+v+"\r\n"+d+"-"+d+": k\r\n\r\n") + want = append(want, map[string]string{ + "method": m, "path": "/a/" + v, "query": "", "host": "h-" + v + ".example", + "x-id": "id-" + v, "digit-key": d + "-" + d, + }) + } + keptCompare(t, a.name, "KEPT732ADAPT", want, kept) + }) + } +} + +// keptCompare reads one kept map per request from kept and reports every +// field that no longer reads its own request. +func keptCompare(t *testing.T, arm, tag string, want []map[string]string, kept <-chan map[string]string) { + t.Helper() + var wrong []string + for i, w := range want { + var got map[string]string + select { + case got = <-kept: + case <-time.After(10 * time.Second): + t.Fatalf("request %d: the handler did not report", i+1) + } + keys := make([]string, 0, len(w)) + for k := range w { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + if got[k] != w[k] { + wrong = append(wrong, fmt.Sprintf("request %d %s: %q (want %q)", i+1, k, got[k], w[k])) + } + } + } + t.Logf("%s arm=%s requests=%d wrong=%d", tag, arm, len(want), len(wrong)) + if len(wrong) > 0 { + t.Errorf("strings kept from a request read other bytes after the connection's later requests:\n %s", strings.Join(wrong, "\n ")) + } +} + +type keptArm struct { + name string + engine celeris.EngineType + async bool +} + +// keptArms returns std, and epoll and io_uring with sync and async +// handlers. With CELERIS_REQUIRE_IOURING_WORKERS=1 a kernel with no usable +// io_uring fails the test instead of dropping the io_uring arms. +func keptArms(t *testing.T) []keptArm { + t.Helper() + arms := []keptArm{ + {"std", celeris.Std, false}, + {"epoll", celeris.Epoll, false}, + {"epoll-async", celeris.Epoll, true}, + } + if ok, p := keptProbeIOUring(); ok { + arms = append(arms, keptArm{"io_uring", celeris.IOUring, false}, keptArm{"io_uring-async", celeris.IOUring, true}) + } else if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") == "1" { + t.Fatalf("io_uring tier=%s kernel=%s, and CELERIS_REQUIRE_IOURING_WORKERS=1 forbids dropping the io_uring arms", p.IOUringTier, p.KernelVersion) + } else { + t.Logf("io_uring tier=%s kernel=%s: io_uring arms not run", p.IOUringTier, p.KernelVersion) + } + return arms +} + +// keptProbeIOUring probes the kernel's io_uring support. With +// CELERIS_REQUIRE_IOURING_WORKERS=1 a probe that finds no usable ring is +// retried for up to 10 s: the probe's ring can fail with ENOMEM against +// RLIMIT_MEMLOCK while the rings of engines stopped moments ago, or of +// another test binary of the same user, are still charged +// (engine/iouring/ring_budget_linux_test.go). +func keptProbeIOUring() (usable bool, p celerisengine.CapabilityProfile) { + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") != "1" { + return usable, p + } + for deadline := time.Now().Add(10 * time.Second); !usable && time.Now().Before(deadline); { + time.Sleep(10 * time.Millisecond) + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + } + return usable, p +} + +// startKeptServer starts the server mk builds on a fresh loopback listener +// and returns its address and a shutdown closure. A start that fails only +// with ENOMEM (io_uring ring memory still charged to RLIMIT_MEMLOCK, see +// keptProbeIOUring) is retried with a new server for up to 10 s. +func startKeptServer(t *testing.T, mk func() *celeris.Server) (string, func()) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + s := mk() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.StartWithListenerAndContext(ctx, ln) }() + addr, err := keptWaitReady(s, done) + if err == nil { + return addr, func() { cancel(); <-done } + } + cancel() + _ = ln.Close() + if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + continue + } + t.Fatalf("server did not start: %v", err) + } +} + +func keptWaitReady(s *celeris.Server, done <-chan error) (string, error) { + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + if err == nil { + err = errors.New("start returned before the server was ready") + } + return "", err + default: + } + if a := s.Addr(); a != nil { + if c, err := net.DialTimeout("tcp", a.String(), 100*time.Millisecond); err == nil { + _ = c.Close() + return a.String(), nil + } + } + time.Sleep(10 * time.Millisecond) + } + return "", errors.New("server not ready within 30s") +} + +func keptDial(t *testing.T, addr string) (net.Conn, *bufio.Reader) { + t.Helper() + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(20 * time.Second)) + return conn, bufio.NewReader(conn) +} + +// keptRoundTrip writes one request and reads its response; it fails the +// test unless the status is 2xx. +func keptRoundTrip(t *testing.T, conn net.Conn, br *bufio.Reader, req string) { + t.Helper() + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatal(err) + } + status, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read status: %v", err) + } + n := 0 + for { + line, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read header: %v", err) + } + if line == "\r\n" { + break + } + if k, v, ok := strings.Cut(line, ":"); ok && strings.EqualFold(k, "content-length") { + n, _ = strconv.Atoi(strings.TrimSpace(v)) + } + } + if _, err := br.Discard(n); err != nil { + t.Fatalf("read body: %v", err) + } + if f := strings.Fields(status); len(f) < 2 || f[1][0] != '2' { + t.Fatalf("status %q for %q", strings.TrimSpace(status), strings.SplitN(req, "\r\n", 2)[0]) + } +} diff --git a/bridge.go b/bridge.go index 910cc36e..a053c6e9 100644 --- a/bridge.go +++ b/bridge.go @@ -52,10 +52,62 @@ func AdaptFunc(h http.HandlerFunc) HandlerFunc { return Adapt(h) } +// buildHTTPRequest builds the *http.Request an adapted handler receives. +// +// Every string it hands to net/http is a copy. On epoll and io_uring the +// method, path, query and header strings are views of the connection's +// receive buffer, which the engine reuses for the connection's next request +// and, once the connection closes, for another connection. net/http lets a +// handler keep the request's strings after ServeHTTP returns (a log line +// queued for later, a map key, a value handed to a goroutine), so a kept view +// would read other request bytes (celeris#732). The copies share one +// allocation. The body is not copied: net/http forbids reading it after +// ServeHTTP returns. func buildHTTPRequest(c *Context) (*http.Request, error) { - url := c.path + // The H1 parser of the native engines defers the header slice until + // something reads a header. Without this, a route with no header read + // before Adapt handed net/http no request headers at all (celeris#720). + c.stream.MaterializeHeaders() + hdrs := c.stream.Headers + + urlLen := len(c.path) + if c.rawQuery != "" { + urlLen += 1 + len(c.rawQuery) + } + authority := "" + n := len(c.method) + urlLen + for _, h := range hdrs { + if strings.HasPrefix(h[0], ":") { + if h[0] == ":authority" && authority == "" { + authority = h[1] + } + continue + } + n += len(h[0]) + len(h[1]) + } + n += len(authority) + + var sb strings.Builder + sb.Grow(n) + sb.WriteString(c.method) + sb.WriteString(c.path) if c.rawQuery != "" { - url += "?" + c.rawQuery + sb.WriteByte('?') + sb.WriteString(c.rawQuery) + } + for _, h := range hdrs { + if !strings.HasPrefix(h[0], ":") { + sb.WriteString(h[0]) + sb.WriteString(h[1]) + } + } + sb.WriteString(authority) + // Cut the copies back out, in the order they were written. + rest := sb.String() + next := func(l int) string { + s := rest[:l] + rest = rest[l:] + return s } var body io.Reader @@ -64,19 +116,21 @@ func buildHTTPRequest(c *Context) (*http.Request, error) { body = bytes.NewReader(data) } - req, err := http.NewRequestWithContext(c.Context(), c.method, url, body) + method := next(len(c.method)) + req, err := http.NewRequestWithContext(c.Context(), method, next(urlLen), body) if err != nil { return nil, err } - for _, h := range c.stream.Headers { + for _, h := range hdrs { if strings.HasPrefix(h[0], ":") { continue } - req.Header.Add(h[0], h[1]) + key := next(len(h[0])) + req.Header.Add(key, next(len(h[1]))) } - if host := c.Header(":authority"); host != "" { + if host := next(len(authority)); host != "" { req.Host = host } diff --git a/middleware/adapters/adapters.go b/middleware/adapters/adapters.go index 62268574..fddf5e6e 100644 --- a/middleware/adapters/adapters.go +++ b/middleware/adapters/adapters.go @@ -103,10 +103,49 @@ func WrapMiddleware(mw func(http.Handler) http.Handler) celeris.HandlerFunc { // buildRequest reconstructs an *http.Request from a celeris Context for use // with stdlib middleware/handlers. +// +// Every string it hands to net/http is a copy, and the copies share one +// allocation. On epoll and io_uring the method, path, query, Host and header +// strings are views of the connection's receive buffer, which the engine +// reuses for the connection's next request and, once the connection closes, +// for another connection; net/http lets a middleware keep the request's +// strings after it returns (celeris#732). The body is not copied: net/http +// forbids reading it after ServeHTTP returns. func buildRequest(c *celeris.Context) *http.Request { - path := c.Path() - if q := c.RawQuery(); q != "" { - path += "?" + q + method, path, query, host := c.Method(), c.Path(), c.RawQuery(), c.Host() + hdrs := c.RequestHeaders() + + urlLen := len(path) + if query != "" { + urlLen += 1 + len(query) + } + n := len(method) + urlLen + len(host) + for _, h := range hdrs { + if !strings.HasPrefix(h[0], ":") { + n += len(h[0]) + len(h[1]) + } + } + var sb strings.Builder + sb.Grow(n) + sb.WriteString(method) + sb.WriteString(path) + if query != "" { + sb.WriteByte('?') + sb.WriteString(query) + } + sb.WriteString(host) + for _, h := range hdrs { + if !strings.HasPrefix(h[0], ":") { + sb.WriteString(h[0]) + sb.WriteString(h[1]) + } + } + // Cut the copies back out, in the order they were written. + rest := sb.String() + next := func(l int) string { + s := rest[:l] + rest = rest[l:] + return s } var body io.Reader @@ -115,16 +154,20 @@ func buildRequest(c *celeris.Context) *http.Request { body = bytes.NewReader(data) } - req, _ := http.NewRequestWithContext(c.Context(), c.Method(), path, body) + method = next(len(method)) + target := next(urlLen) + host = next(len(host)) + req, _ := http.NewRequestWithContext(c.Context(), method, target, body) - for _, h := range c.RequestHeaders() { + for _, h := range hdrs { if strings.HasPrefix(h[0], ":") { continue } - req.Header.Add(h[0], h[1]) + key := next(len(h[0])) + req.Header.Add(key, next(len(h[1]))) } - if host := c.Host(); host != "" { + if host != "" { req.Host = host } diff --git a/middleware/adapters/retained_strings_linux_test.go b/middleware/adapters/retained_strings_linux_test.go new file mode 100644 index 00000000..466f6735 --- /dev/null +++ b/middleware/adapters/retained_strings_linux_test.go @@ -0,0 +1,257 @@ +//go:build linux + +package adapters_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "sort" + "strconv" + "strings" + "testing" + "time" + + "github.com/goceleris/celeris" + celerisengine "github.com/goceleris/celeris/engine" + "github.com/goceleris/celeris/middleware/adapters" + "github.com/goceleris/celeris/probe" +) + +// TestWrapMiddlewareKeptStringsSurviveNextRequest pins the +// middleware/adapters site of celeris#732. +// +// net/http lets a middleware keep the request's strings after it returns (a +// rate limiter's map key, a log line queued for later). On epoll and +// io_uring the method, path, query, Host and header strings buildRequest +// handed over were views of the connection's receive buffer, which the +// engine reuses for the connection's next request. Three requests with the +// same layout and different values go over one keep-alive connection; the +// wrapped middleware keeps its request's strings, and after the third +// request each kept set must still read its own request. The methods are +// ones the H1 parser does not intern, the header name "111-111" is one +// net/http's canonicalization returns unchanged (it has no letters), and no +// request has a query (with one, the URL was a new string and the path a +// copy by accident). +func TestWrapMiddlewareKeptStringsSurviveNextRequest(t *testing.T) { + methods := []string{"TRACE", "PURGE", "MKCOL"} + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + kept := make(chan map[string]string, len(methods)) + mw := func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got := map[string]string{ + "method": r.Method, + "path": r.URL.Path, + "query": r.URL.RawQuery, + "host": r.Host, + "x-id": r.Header.Get("X-Id"), + } + for k := range r.Header { + if strings.HasSuffix(k, "-111") || strings.HasSuffix(k, "-222") || strings.HasSuffix(k, "-333") { + got["digit-key"] = k + } + } + kept <- got + next.ServeHTTP(w, r) + }) + } + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async}) + srv.Use(adapters.WrapMiddleware(mw)) + for _, m := range methods { + srv.Handle(m, "/w/:id", func(c *celeris.Context) error { return c.String(200, "ok") }) + } + return srv + }) + defer stop() + + conn, br := keptDial(t, addr) + defer func() { _ = conn.Close() }() + var want []map[string]string + for i, m := range methods { + v := strings.Repeat(string(rune('a'+i)), 4) + d := strings.Repeat(strconv.Itoa(i+1), 3) + keptRoundTrip(t, conn, br, m+" /w/"+v+" HTTP/1.1\r\nHost: h-"+v+".example\r\nX-Id: id-"+v+"\r\n"+d+"-"+d+": k\r\n\r\n") + want = append(want, map[string]string{ + "method": m, "path": "/w/" + v, "query": "", "host": "h-" + v + ".example", + "x-id": "id-" + v, "digit-key": d + "-" + d, + }) + } + keptCompare(t, a.name, "KEPT732WRAP", want, kept) + }) + } +} + +// keptCompare reads one kept map per request from kept and reports every +// field that no longer reads its own request. +func keptCompare(t *testing.T, arm, tag string, want []map[string]string, kept <-chan map[string]string) { + t.Helper() + var wrong []string + for i, w := range want { + var got map[string]string + select { + case got = <-kept: + case <-time.After(10 * time.Second): + t.Fatalf("request %d: the handler did not report", i+1) + } + keys := make([]string, 0, len(w)) + for k := range w { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + if got[k] != w[k] { + wrong = append(wrong, fmt.Sprintf("request %d %s: %q (want %q)", i+1, k, got[k], w[k])) + } + } + } + t.Logf("%s arm=%s requests=%d wrong=%d", tag, arm, len(want), len(wrong)) + if len(wrong) > 0 { + t.Errorf("strings kept from a request read other bytes after the connection's later requests:\n %s", strings.Join(wrong, "\n ")) + } +} + +type keptArm struct { + name string + engine celeris.EngineType + async bool +} + +// keptArms returns std, and epoll and io_uring with sync and async +// handlers. With CELERIS_REQUIRE_IOURING_WORKERS=1 a kernel with no usable +// io_uring fails the test instead of dropping the io_uring arms. +func keptArms(t *testing.T) []keptArm { + t.Helper() + arms := []keptArm{ + {"std", celeris.Std, false}, + {"epoll", celeris.Epoll, false}, + {"epoll-async", celeris.Epoll, true}, + } + if ok, p := keptProbeIOUring(); ok { + arms = append(arms, keptArm{"io_uring", celeris.IOUring, false}, keptArm{"io_uring-async", celeris.IOUring, true}) + } else if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") == "1" { + t.Fatalf("io_uring tier=%s kernel=%s, and CELERIS_REQUIRE_IOURING_WORKERS=1 forbids dropping the io_uring arms", p.IOUringTier, p.KernelVersion) + } else { + t.Logf("io_uring tier=%s kernel=%s: io_uring arms not run", p.IOUringTier, p.KernelVersion) + } + return arms +} + +// keptProbeIOUring probes the kernel's io_uring support. With +// CELERIS_REQUIRE_IOURING_WORKERS=1 a probe that finds no usable ring is +// retried for up to 10 s: the probe's ring can fail with ENOMEM against +// RLIMIT_MEMLOCK while the rings of engines stopped moments ago, or of +// another test binary of the same user, are still charged +// (engine/iouring/ring_budget_linux_test.go). +func keptProbeIOUring() (usable bool, p celerisengine.CapabilityProfile) { + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") != "1" { + return usable, p + } + for deadline := time.Now().Add(10 * time.Second); !usable && time.Now().Before(deadline); { + time.Sleep(10 * time.Millisecond) + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + } + return usable, p +} + +// startKeptServer starts the server mk builds on a fresh loopback listener +// and returns its address and a shutdown closure. A start that fails only +// with ENOMEM (io_uring ring memory still charged to RLIMIT_MEMLOCK, see +// keptProbeIOUring) is retried with a new server for up to 10 s. +func startKeptServer(t *testing.T, mk func() *celeris.Server) (string, func()) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + s := mk() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.StartWithListenerAndContext(ctx, ln) }() + addr, err := keptWaitReady(s, done) + if err == nil { + return addr, func() { cancel(); <-done } + } + cancel() + _ = ln.Close() + if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + continue + } + t.Fatalf("server did not start: %v", err) + } +} + +func keptWaitReady(s *celeris.Server, done <-chan error) (string, error) { + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + if err == nil { + err = errors.New("start returned before the server was ready") + } + return "", err + default: + } + if a := s.Addr(); a != nil { + if c, err := net.DialTimeout("tcp", a.String(), 100*time.Millisecond); err == nil { + _ = c.Close() + return a.String(), nil + } + } + time.Sleep(10 * time.Millisecond) + } + return "", errors.New("server not ready within 30s") +} + +func keptDial(t *testing.T, addr string) (net.Conn, *bufio.Reader) { + t.Helper() + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(20 * time.Second)) + return conn, bufio.NewReader(conn) +} + +// keptRoundTrip writes one request and reads its response; it fails the +// test unless the status is 2xx. +func keptRoundTrip(t *testing.T, conn net.Conn, br *bufio.Reader, req string) { + t.Helper() + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatal(err) + } + status, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read status: %v", err) + } + n := 0 + for { + line, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read header: %v", err) + } + if line == "\r\n" { + break + } + if k, v, ok := strings.Cut(line, ":"); ok && strings.EqualFold(k, "content-length") { + n, _ = strconv.Atoi(strings.TrimSpace(v)) + } + } + if _, err := br.Discard(n); err != nil { + t.Fatalf("read body: %v", err) + } + if f := strings.Fields(status); len(f) < 2 || f[1][0] != '2' { + t.Fatalf("status %q for %q", strings.TrimSpace(status), strings.SplitN(req, "\r\n", 2)[0]) + } +} diff --git a/middleware/logger/config.go b/middleware/logger/config.go index 0607f2a6..621edbba 100644 --- a/middleware/logger/config.go +++ b/middleware/logger/config.go @@ -67,6 +67,11 @@ type Config struct { // Output is the slog.Logger used to emit log records. // When nil, slog.Default() is used. + // + // A handler other than [FastHandler] receives copies of the string + // values, because slog allows it to keep a record after Handle returns + // and the request strings are only valid during the request on epoll + // and io_uring. The copies share one allocation per logged request. Output *slog.Logger // Level maps an HTTP response status code to a slog.Level, controlling diff --git a/middleware/logger/logger.go b/middleware/logger/logger.go index 9e62d0b2..9b97e009 100644 --- a/middleware/logger/logger.go +++ b/middleware/logger/logger.go @@ -5,6 +5,7 @@ import ( "log/slog" "net/url" "os" + "slices" "strings" "sync" "time" @@ -248,6 +249,14 @@ func New(config ...Config) celeris.HandlerFunc { if fh, ok := handler.(*FastHandler); ok { fh.HandleDirect(ts, level, "request", attrs) } else { + // Any other handler may keep the record: slog lets a Handler + // keep a Record after Handle returns by calling Record.Clone, + // which shares the strings, and asynchronous and batching + // handlers do. The package's own handlers format the record + // before they return, so only the others get copies. + if _, ok := handler.(*groupHandler); !ok { + ownStringValues(attrs) + } r := slog.NewRecord(ts, level, "request", 0) r.AddAttrs(attrs...) _ = handler.Handle(ctx, r) @@ -272,6 +281,48 @@ func New(config ...Config) celeris.HandlerFunc { } } +// ownStringValues replaces every string value in attrs, in place, with a +// copy, descending into groups. The copies of the top-level values share one +// allocation. Keys are left as they are: the middleware's own are constants +// or built fresh, and Fields keys are expected to be constants. Values of +// other kinds (slog.AnyValue, a LogValuer) are left as they are. +// +// On epoll and io_uring the request strings the middleware logs (method, +// path, Host, User-Agent, Referer, query, client IP, request ID) are views +// of the connection's receive buffer, which the engine reuses for the +// connection's next request and, once the connection closes, for another +// connection. A handler that keeps the record would later format other +// request bytes, including another client's headers (celeris#732). +func ownStringValues(attrs []slog.Attr) { + n := 0 + for _, a := range attrs { + if a.Value.Kind() == slog.KindString { + n += len(a.Value.String()) + } + } + var b strings.Builder + b.Grow(n) + for _, a := range attrs { + if a.Value.Kind() == slog.KindString { + b.WriteString(a.Value.String()) + } + } + rest := b.String() + for i, a := range attrs { + switch a.Value.Kind() { + case slog.KindString: + l := len(a.Value.String()) + attrs[i].Value = slog.StringValue(rest[:l]) + rest = rest[l:] + case slog.KindGroup: + // A new slice: a Fields group may share its attrs with the caller. + g := slices.Clone(a.Value.Group()) + ownStringValues(g) + attrs[i].Value = slog.GroupValue(g...) + } + } +} + // disableHandlerColors sets color=false on FastHandler or groupHandler. func disableHandlerColors(h slog.Handler) { switch v := h.(type) { diff --git a/middleware/logger/retained_attrs_linux_test.go b/middleware/logger/retained_attrs_linux_test.go new file mode 100644 index 00000000..c99ddbc6 --- /dev/null +++ b/middleware/logger/retained_attrs_linux_test.go @@ -0,0 +1,296 @@ +//go:build linux + +package logger_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "log/slog" + "net" + "os" + "sort" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/goceleris/celeris" + celerisengine "github.com/goceleris/celeris/engine" + "github.com/goceleris/celeris/middleware/logger" + "github.com/goceleris/celeris/probe" +) + +// keepHandler keeps every record, cloned as slog requires, and formats +// nothing until asked: what an asynchronous or batching handler does. +type keepHandler struct { + mu *sync.Mutex + recs *[]slog.Record +} + +func (h keepHandler) Enabled(context.Context, slog.Level) bool { return true } + +func (h keepHandler) Handle(_ context.Context, r slog.Record) error { + h.mu.Lock() + *h.recs = append(*h.recs, r.Clone()) + h.mu.Unlock() + return nil +} + +func (h keepHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h keepHandler) WithGroup(string) slog.Handler { return h } + +// TestKeptRecordSurvivesNextRequest pins the logger site of celeris#732. +// +// slog lets a Handler keep a Record after Handle returns by calling +// Record.Clone, which shares the strings. On epoll and io_uring the method +// (for one the H1 parser does not intern), path, query, Host, User-Agent, +// Referer and client IP the middleware logs, and the request ID, context +// value, response header and Fields value that code derives from request +// headers, are views of the connection's receive buffer, which the engine +// reuses for the connection's next request. Three requests with the same +// layout and different values go over one keep-alive connection; every kept +// record must still read its own request. +func TestKeptRecordSurvivesNextRequest(t *testing.T) { + methods := []string{"TRACE", "PURGE", "MKCOL"} + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + var mu sync.Mutex + var recs []slog.Record + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async}) + srv.Use(logger.New(logger.Config{ + Output: slog.New(keepHandler{mu: &mu, recs: &recs}), + LogHost: true, + LogUserAgent: true, + LogReferer: true, + LogQueryParams: true, + LogContextKeys: []string{"tenant"}, + LogResponseHeaders: []string{"x-echo"}, + Fields: func(c *celeris.Context, _ time.Duration) []slog.Attr { + return []slog.Attr{slog.Group("g", slog.String("tenant", c.Header("x-tenant")))} + }, + })) + h := func(c *celeris.Context) error { + c.SetRequestID(c.Header("x-request-id")) + c.Set("tenant", c.Header("x-tenant")) + c.SetHeader("x-echo", c.Header("x-id")) + // Not String or Blob: they reorder the response headers + // in place, and ResponseHeaders no longer lists x-echo. + return c.NoContent(204) + } + for _, m := range methods { + srv.Handle(m, "/l/:id", h) + } + return srv + }) + defer stop() + + conn, br := keptDial(t, addr) + defer func() { _ = conn.Close() }() + var want []map[string]string + for i, m := range methods { + v := strings.Repeat(string(rune('a'+i)), 4) + keptRoundTrip(t, conn, br, m+" /l/"+v+"?q="+v+" HTTP/1.1\r\nHost: h-"+v+".example\r\nUser-Agent: agent-"+v+ + "\r\nReferer: https://r-"+v+".example/\r\nX-Forwarded-For: 10.0.0."+strconv.Itoa(i+1)+ + "\r\nX-Request-Id: rid-"+v+"\r\nX-Tenant: tenant-"+v+"\r\nX-Id: id-"+v+"\r\n\r\n") + want = append(want, map[string]string{ + "method": m, "path": "/l/" + v, "query": "q=" + v, "host": "h-" + v + ".example", + "user_agent": "agent-" + v, "referer": "https://r-" + v + ".example/", "client_ip": "10.0.0." + strconv.Itoa(i+1), + "request_id": "rid-" + v, "ctx.tenant": "tenant-" + v, "resp_header.x-echo": "id-" + v, "g.tenant": "tenant-" + v, + }) + } + + // The middleware logs after the response reaches the client. + var kept []slog.Record + for i := 0; i < 200; i++ { + mu.Lock() + kept = append(kept[:0], recs...) + mu.Unlock() + if len(kept) >= len(methods) { + break + } + time.Sleep(10 * time.Millisecond) + } + if len(kept) != len(methods) { + t.Fatalf("got %d records, want %d", len(kept), len(methods)) + } + sort.Slice(kept, func(i, j int) bool { return kept[i].Time.Before(kept[j].Time) }) + var wrong []string + for i, r := range kept { + got := map[string]string{} + r.Attrs(func(a slog.Attr) bool { + flattenStrings(got, "", a) + return true + }) + keys := make([]string, 0, len(want[i])) + for k := range want[i] { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + if got[k] != want[i][k] { + wrong = append(wrong, fmt.Sprintf("record %d %s: %q (want %q)", i+1, k, got[k], want[i][k])) + } + } + } + t.Logf("KEPT732LOGGER arm=%s records=%d wrong=%d", a.name, len(kept), len(wrong)) + if len(wrong) > 0 { + t.Errorf("a kept record reads other bytes after the connection's later requests:\n %s", strings.Join(wrong, "\n ")) + } + }) + } +} + +func flattenStrings(dst map[string]string, prefix string, a slog.Attr) { + switch a.Value.Kind() { + case slog.KindString: + dst[prefix+a.Key] = a.Value.String() + case slog.KindGroup: + for _, g := range a.Value.Group() { + flattenStrings(dst, prefix+a.Key+".", g) + } + } +} + +type keptArm struct { + name string + engine celeris.EngineType + async bool +} + +// keptArms returns std, and epoll and io_uring with sync and async +// handlers. With CELERIS_REQUIRE_IOURING_WORKERS=1 a kernel with no usable +// io_uring fails the test instead of dropping the io_uring arms. +func keptArms(t *testing.T) []keptArm { + t.Helper() + arms := []keptArm{ + {"std", celeris.Std, false}, + {"epoll", celeris.Epoll, false}, + {"epoll-async", celeris.Epoll, true}, + } + if ok, p := keptProbeIOUring(); ok { + arms = append(arms, keptArm{"io_uring", celeris.IOUring, false}, keptArm{"io_uring-async", celeris.IOUring, true}) + } else if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") == "1" { + t.Fatalf("io_uring tier=%s kernel=%s, and CELERIS_REQUIRE_IOURING_WORKERS=1 forbids dropping the io_uring arms", p.IOUringTier, p.KernelVersion) + } else { + t.Logf("io_uring tier=%s kernel=%s: io_uring arms not run", p.IOUringTier, p.KernelVersion) + } + return arms +} + +// keptProbeIOUring probes the kernel's io_uring support. With +// CELERIS_REQUIRE_IOURING_WORKERS=1 a probe that finds no usable ring is +// retried for up to 10 s: the probe's ring can fail with ENOMEM against +// RLIMIT_MEMLOCK while the rings of engines stopped moments ago, or of +// another test binary of the same user, are still charged +// (engine/iouring/ring_budget_linux_test.go). +func keptProbeIOUring() (usable bool, p celerisengine.CapabilityProfile) { + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") != "1" { + return usable, p + } + for deadline := time.Now().Add(10 * time.Second); !usable && time.Now().Before(deadline); { + time.Sleep(10 * time.Millisecond) + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + } + return usable, p +} + +// startKeptServer starts the server mk builds on a fresh loopback listener +// and returns its address and a shutdown closure. A start that fails only +// with ENOMEM (io_uring ring memory still charged to RLIMIT_MEMLOCK, see +// keptProbeIOUring) is retried with a new server for up to 10 s. +func startKeptServer(t *testing.T, mk func() *celeris.Server) (string, func()) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + s := mk() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.StartWithListenerAndContext(ctx, ln) }() + addr, err := keptWaitReady(s, done) + if err == nil { + return addr, func() { cancel(); <-done } + } + cancel() + _ = ln.Close() + if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + continue + } + t.Fatalf("server did not start: %v", err) + } +} + +func keptWaitReady(s *celeris.Server, done <-chan error) (string, error) { + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + if err == nil { + err = errors.New("start returned before the server was ready") + } + return "", err + default: + } + if a := s.Addr(); a != nil { + if c, err := net.DialTimeout("tcp", a.String(), 100*time.Millisecond); err == nil { + _ = c.Close() + return a.String(), nil + } + } + time.Sleep(10 * time.Millisecond) + } + return "", errors.New("server not ready within 30s") +} + +func keptDial(t *testing.T, addr string) (net.Conn, *bufio.Reader) { + t.Helper() + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(20 * time.Second)) + return conn, bufio.NewReader(conn) +} + +// keptRoundTrip writes one request and reads its response; it fails the +// test unless the status is 2xx. +func keptRoundTrip(t *testing.T, conn net.Conn, br *bufio.Reader, req string) { + t.Helper() + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatal(err) + } + status, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read status: %v", err) + } + n := 0 + for { + line, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read header: %v", err) + } + if line == "\r\n" { + break + } + if k, v, ok := strings.Cut(line, ":"); ok && strings.EqualFold(k, "content-length") { + n, _ = strconv.Atoi(strings.TrimSpace(v)) + } + } + if _, err := br.Discard(n); err != nil { + t.Fatalf("read body: %v", err) + } + if f := strings.Fields(status); len(f) < 2 || f[1][0] != '2' { + t.Fatalf("status %q for %q", strings.TrimSpace(status), strings.SplitN(req, "\r\n", 2)[0]) + } +} diff --git a/middleware/metrics/config.go b/middleware/metrics/config.go index 7e3b567f..5cc55386 100644 --- a/middleware/metrics/config.go +++ b/middleware/metrics/config.go @@ -66,6 +66,8 @@ type Config struct { // label sets. Each map key becomes a label name and the function extracts // the label value from the request context. The functions are called // after c.Next() returns, so response-derived values are available. + // A function may return a request string such as c.Header(...): the + // middleware copies a label combination the first time it records it. LabelFuncs map[string]func(*celeris.Context) string // SizeBuckets defines histogram bucket boundaries for request and diff --git a/middleware/metrics/metrics.go b/middleware/metrics/metrics.go index 5eeb34a8..c7d7ea0f 100644 --- a/middleware/metrics/metrics.go +++ b/middleware/metrics/metrics.go @@ -2,10 +2,12 @@ package metrics import ( "bytes" + "encoding/binary" "sort" "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/prometheus/client_golang/prometheus" @@ -15,17 +17,6 @@ import ( "github.com/goceleris/celeris" ) -// labelValuesPool recycles the label-value slice that every recorded -// request builds for WithLabelValues. Prometheus consumes the values -// synchronously (lookup + Observe/Inc return before the slice is -// reused) so a pool is safe. Capacity 8 covers 3 base labels + up to -// 5 custom ones without re-grow; larger custom sets fall back to a -// fresh allocation (rare). -var labelValuesPool = sync.Pool{New: func() any { - s := make([]string, 0, 8) - return &s -}} - // New creates a Prometheus metrics middleware with the given config. func New(config ...Config) celeris.HandlerFunc { cfg := defaultConfig @@ -130,6 +121,12 @@ func New(config ...Config) celeris.HandlerFunc { metricsPath := cfg.Path authFunc := cfg.AuthFunc nCustom := len(customLabelNames) + nLabels := len(allLabels) + set := &seriesSet{ + requestsTotal: requestsTotal, + requestDuration: requestDuration, + m: make(map[string]*series), + } return func(c *celeris.Context) error { if c.Path() == metricsPath { @@ -187,41 +184,122 @@ func New(config ...Config) celeris.HandlerFunc { } path = strings.ToValidUTF8(path, "") - // Build label values: method, path, status + custom labels. - // Pool-backed slice: Prometheus consumes synchronously (no - // retention past WithLabelValues), so we can return the - // backing array to the pool. - lvPtr := labelValuesPool.Get().(*[]string) - lv := (*lvPtr)[:0] - lv = append(lv, c.Method(), path, statusStr) + // Label values: method, path, status + custom labels, as a lookup + // key. The key is built on the stack and the lookup copies + // nothing; see seriesSet. + var kb [256]byte + key := appendLabelValue(kb[:0], c.Method()) + key = appendLabelValue(key, path) + key = appendLabelValue(key, statusStr) for i := range nCustom { - lv = append(lv, customLabelFuncs[i](c)) + key = appendLabelValue(key, customLabelFuncs[i](c)) + } + s := set.get(key) + if s == nil { + s = set.add(key, nLabels) } - requestsTotal.WithLabelValues(lv...).Inc() - requestDuration.WithLabelValues(lv...).Observe(duration) + s.total.Inc() + s.duration.Observe(duration) if cl := c.ContentLength(); cl > 0 { - requestSize.WithLabelValues(lv...).Observe(float64(cl)) + s.observer(&s.reqSize, requestSize).Observe(float64(cl)) } if bw := c.BytesWritten(); bw > 0 { - responseSize.WithLabelValues(lv...).Observe(float64(bw)) + s.observer(&s.respSize, responseSize).Observe(float64(bw)) } - // Cap retained capacity so a pathological custom-label - // consumer doesn't hold a huge backing array in the pool. - if cap(lv) > 32 { - fresh := make([]string, 0, 8) - *lvPtr = fresh - } else { - *lvPtr = lv - } - labelValuesPool.Put(lvPtr) - return err } } +// seriesSet holds every label-value combination the middleware has +// recorded, keyed by the values, with copies of the values it owns and the +// series it resolved for them. +// +// The label values are request strings: c.Method() for a method the H1 +// parser does not intern, c.Path() when there is no route pattern, and +// whatever a LabelFuncs function reads from the request. On epoll and +// io_uring those are views of the connection's receive buffer, which the +// engine reuses for the connection's next request and, once the connection +// closes, for another connection. client_golang keeps the label values of +// every new series for the life of the registry and does not copy them, so a +// series created from views would change its labels to other request bytes, +// including another client's headers (celeris#732). Only Prometheus ever +// sees the owned copies, and they are made only when a combination is new: +// a request whose combination was seen before copies nothing and resolves +// its series with one map lookup instead of one WithLabelValues call per +// metric. +type seriesSet struct { + requestsTotal *prometheus.CounterVec + requestDuration *prometheus.HistogramVec + + mu sync.RWMutex + m map[string]*series +} + +// series is one label-value combination. +type series struct { + values []string // owned copies, cut from the seriesSet key + total prometheus.Counter + duration prometheus.Observer + // The size histograms are resolved on first use, so a combination that + // never carried a body has no request_size_bytes series, as before. + reqSize atomic.Pointer[prometheus.Observer] + respSize atomic.Pointer[prometheus.Observer] +} + +// appendLabelValue appends v to a lookup key, length-prefixed so that no +// two combinations share a key whatever bytes the values hold. +func appendLabelValue(key []byte, v string) []byte { + key = binary.AppendUvarint(key, uint64(len(v))) + return append(key, v...) +} + +func (set *seriesSet) get(key []byte) *series { + set.mu.RLock() + s := set.m[string(key)] + set.mu.RUnlock() + return s +} + +// add records a new combination of n label values. The map key is a copy of +// the lookup key, and the label values handed to Prometheus are cut from it. +func (set *seriesSet) add(key []byte, n int) *series { + owned := string(key) + values := make([]string, 0, n) + for off := 0; off < len(key); { + l, w := binary.Uvarint(key[off:]) + off += w + values = append(values, owned[off:off+int(l)]) + off += int(l) + } + // Resolved outside the lock: WithLabelValues panics on a label value + // that is not valid UTF-8, and the panic must not leave mu held. + s := &series{ + values: values, + total: set.requestsTotal.WithLabelValues(values...), + duration: set.requestDuration.WithLabelValues(values...), + } + set.mu.Lock() + defer set.mu.Unlock() + if prev := set.m[owned]; prev != nil { + return prev + } + set.m[owned] = s + return s +} + +// observer returns the series' observer in vec, resolving it on first use. +func (s *series) observer(p *atomic.Pointer[prometheus.Observer], vec *prometheus.HistogramVec) prometheus.Observer { + if o := p.Load(); o != nil { + return *o + } + o := vec.WithLabelValues(s.values...) + p.Store(&o) + return o +} + func serveMetrics(c *celeris.Context, gatherer prometheus.Gatherer) error { mfs, err := gatherer.Gather() if err != nil { diff --git a/middleware/metrics/retained_labels_linux_test.go b/middleware/metrics/retained_labels_linux_test.go new file mode 100644 index 00000000..e60118a6 --- /dev/null +++ b/middleware/metrics/retained_labels_linux_test.go @@ -0,0 +1,291 @@ +//go:build linux + +package metrics_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "os" + "sort" + "strconv" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + + "github.com/goceleris/celeris" + celerisengine "github.com/goceleris/celeris/engine" + "github.com/goceleris/celeris/middleware/metrics" + "github.com/goceleris/celeris/probe" +) + +// TestLabelValuesSurviveNextRequest pins the metrics site of celeris#732. +// +// client_golang keeps the label values of every new series for the life of +// the registry and does not copy them. On epoll and io_uring the method (for +// a method the H1 parser does not intern) and a LabelFuncs value read from a +// header are views of the connection's receive buffer, which the engine +// reuses for the connection's next request. Three requests with the same +// layout and different values go over one keep-alive connection; each +// creates a series in requests_total, request_duration_seconds and +// response_size_bytes, and every series must keep its own labels, with no +// Gather error. +func TestLabelValuesSurviveNextRequest(t *testing.T) { + methods := []string{"TRACE", "PURGE", "MKCOL"} + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + reg := prometheus.NewRegistry() + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async}) + srv.Use(metrics.New(metrics.Config{ + Registry: reg, + LabelFuncs: map[string]func(*celeris.Context) string{"tenant": func(c *celeris.Context) string { return c.Header("x-tenant") }}, + })) + for _, m := range methods { + srv.Handle(m, "/m", func(c *celeris.Context) error { return c.String(200, "ok") }) + } + return srv + }) + defer stop() + + conn, br := keptDial(t, addr) + defer func() { _ = conn.Close() }() + var want []string + for i, m := range methods { + tenant := "tenant-" + strings.Repeat(string(rune('a'+i)), 4) + keptRoundTrip(t, conn, br, m+" /m HTTP/1.1\r\nHost: x\r\nX-Tenant: "+tenant+"\r\n\r\n") + want = append(want, m+"|"+tenant) + } + sort.Strings(want) + for _, name := range []string{"celeris_requests_total", "celeris_request_duration_seconds", "celeris_response_size_bytes"} { + got, gerr := seriesOf(reg, name) + t.Logf("KEPT732METRICS arm=%s %s series=%q gather_err=%v", a.name, name, got, gerr) + if gerr != nil || strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("%s series (method|tenant) %q, Gather error %v; want %q and no error", name, got, gerr, want) + } + } + }) + } +} + +// TestLabelValuesSurviveConnectionReuse is the cross-connection form of +// TestLabelValuesSurviveNextRequest. When a connection closes, the engine +// pools its receive buffer and the next accepted connection reads into it, +// so a label kept as a view reads another client's request. Each round, +// connection A sends one labelled request and closes; connection B sends a +// request with an Authorization header laid out over the same offsets. No +// label may hold B's bytes. With async handlers the request is parsed from +// the dispatch input buffer, and in this layout no series read B's bytes +// before the fix either (0 of 20 rounds); those two arms are kept as +// coverage, and TestLabelValuesSurviveNextRequest covers async handlers. +func TestLabelValuesSurviveConnectionReuse(t *testing.T) { + const rounds = 20 + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + reg := prometheus.NewRegistry() + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async, Workers: 2}) + srv.Use(metrics.New(metrics.Config{ + Registry: reg, + LabelFuncs: map[string]func(*celeris.Context) string{"tenant": func(c *celeris.Context) string { return c.Header("x-tenant") }}, + })) + srv.GET("/m", func(c *celeris.Context) error { return c.String(200, "ok") }) + return srv + }) + defer stop() + + want := []string{"GET|"} // B's requests carry no X-Tenant + for i := range rounds { + tenant := "tenant-" + strconv.Itoa(1000+i) + want = append(want, "GET|"+tenant) + for _, req := range []string{ + "GET /m HTTP/1.1\r\nHost: x\r\nConnection: close\r\nX-Tenant: " + tenant + "\r\n\r\n", + "GET /m HTTP/1.1\r\nHost: x\r\nConnection: close\r\nAuthorization: SECRETSECRETSECRETSECRET\r\n\r\n", + } { + conn, br := keptDial(t, addr) + keptRoundTrip(t, conn, br, req) + _, _ = io.Copy(io.Discard, br) // until the server closes it + _ = conn.Close() + } + } + sort.Strings(want) + got, gerr := seriesOf(reg, "celeris_requests_total") + secret := 0 + for _, s := range got { + if strings.Contains(s, "SECRET") { + secret++ + } + } + t.Logf("KEPT732METRICSXCONN arm=%s rounds=%d series=%d holding another connection's Authorization bytes=%d gather_err=%v", a.name, rounds, len(got), secret, gerr) + if gerr != nil || secret > 0 || strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("requests_total series (method|tenant) %q (%d hold Authorization bytes), Gather error %v; want %q and no error", got, secret, gerr, want) + } + }) + } +} + +// seriesOf gathers reg and returns the series of the named metric as sorted +// "method|tenant" strings. +func seriesOf(reg *prometheus.Registry, name string) ([]string, error) { + mfs, err := reg.Gather() + var out []string + for _, mf := range mfs { + if mf.GetName() != name { + continue + } + for _, m := range mf.GetMetric() { + l := map[string]string{} + for _, lp := range m.GetLabel() { + l[lp.GetName()] = lp.GetValue() + } + out = append(out, fmt.Sprintf("%s|%s", l["method"], l["tenant"])) + } + } + sort.Strings(out) + return out, err +} + +type keptArm struct { + name string + engine celeris.EngineType + async bool +} + +// keptArms returns std, and epoll and io_uring with sync and async +// handlers. With CELERIS_REQUIRE_IOURING_WORKERS=1 a kernel with no usable +// io_uring fails the test instead of dropping the io_uring arms. +func keptArms(t *testing.T) []keptArm { + t.Helper() + arms := []keptArm{ + {"std", celeris.Std, false}, + {"epoll", celeris.Epoll, false}, + {"epoll-async", celeris.Epoll, true}, + } + if ok, p := keptProbeIOUring(); ok { + arms = append(arms, keptArm{"io_uring", celeris.IOUring, false}, keptArm{"io_uring-async", celeris.IOUring, true}) + } else if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") == "1" { + t.Fatalf("io_uring tier=%s kernel=%s, and CELERIS_REQUIRE_IOURING_WORKERS=1 forbids dropping the io_uring arms", p.IOUringTier, p.KernelVersion) + } else { + t.Logf("io_uring tier=%s kernel=%s: io_uring arms not run", p.IOUringTier, p.KernelVersion) + } + return arms +} + +// keptProbeIOUring probes the kernel's io_uring support. With +// CELERIS_REQUIRE_IOURING_WORKERS=1 a probe that finds no usable ring is +// retried for up to 10 s: the probe's ring can fail with ENOMEM against +// RLIMIT_MEMLOCK while the rings of engines stopped moments ago, or of +// another test binary of the same user, are still charged +// (engine/iouring/ring_budget_linux_test.go). +func keptProbeIOUring() (usable bool, p celerisengine.CapabilityProfile) { + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") != "1" { + return usable, p + } + for deadline := time.Now().Add(10 * time.Second); !usable && time.Now().Before(deadline); { + time.Sleep(10 * time.Millisecond) + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + } + return usable, p +} + +// startKeptServer starts the server mk builds on a fresh loopback listener +// and returns its address and a shutdown closure. A start that fails only +// with ENOMEM (io_uring ring memory still charged to RLIMIT_MEMLOCK, see +// keptProbeIOUring) is retried with a new server for up to 10 s. +func startKeptServer(t *testing.T, mk func() *celeris.Server) (string, func()) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + s := mk() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.StartWithListenerAndContext(ctx, ln) }() + addr, err := keptWaitReady(s, done) + if err == nil { + return addr, func() { cancel(); <-done } + } + cancel() + _ = ln.Close() + if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + continue + } + t.Fatalf("server did not start: %v", err) + } +} + +func keptWaitReady(s *celeris.Server, done <-chan error) (string, error) { + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + if err == nil { + err = errors.New("start returned before the server was ready") + } + return "", err + default: + } + if a := s.Addr(); a != nil { + if c, err := net.DialTimeout("tcp", a.String(), 100*time.Millisecond); err == nil { + _ = c.Close() + return a.String(), nil + } + } + time.Sleep(10 * time.Millisecond) + } + return "", errors.New("server not ready within 30s") +} + +func keptDial(t *testing.T, addr string) (net.Conn, *bufio.Reader) { + t.Helper() + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(20 * time.Second)) + return conn, bufio.NewReader(conn) +} + +// keptRoundTrip writes one request and reads its response; it fails the +// test unless the status is 2xx. +func keptRoundTrip(t *testing.T, conn net.Conn, br *bufio.Reader, req string) { + t.Helper() + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatal(err) + } + status, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read status: %v", err) + } + n := 0 + for { + line, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read header: %v", err) + } + if line == "\r\n" { + break + } + if k, v, ok := strings.Cut(line, ":"); ok && strings.EqualFold(k, "content-length") { + n, _ = strconv.Atoi(strings.TrimSpace(v)) + } + } + if _, err := br.Discard(n); err != nil { + t.Fatalf("read body: %v", err) + } + if f := strings.Fields(status); len(f) < 2 || f[1][0] != '2' { + t.Fatalf("status %q for %q", strings.TrimSpace(status), strings.SplitN(req, "\r\n", 2)[0]) + } +} diff --git a/middleware/otel/config.go b/middleware/otel/config.go index f18ce6d7..92b245bb 100644 --- a/middleware/otel/config.go +++ b/middleware/otel/config.go @@ -53,9 +53,13 @@ type Config struct { CollectUserAgent *bool // CustomAttributes is called per-request and appended to the span attributes. + // String and string-slice values are copied first (one allocation per + // string): the span keeps them, and a request string is only valid + // during the request on epoll and io_uring. Keys are not copied. CustomAttributes func(c *celeris.Context) []attribute.KeyValue // CustomMetricAttributes is called per-request and appended to the metric attributes. + // String values are copied as for CustomAttributes. CustomMetricAttributes func(c *celeris.Context) []attribute.KeyValue // ServerPort, when > 0, adds the "server.port" attribute to spans and metrics. diff --git a/middleware/otel/otel.go b/middleware/otel/otel.go index c9fa6037..288aa081 100644 --- a/middleware/otel/otel.go +++ b/middleware/otel/otel.go @@ -23,27 +23,82 @@ const ( // standardMethods is the set of HTTP methods recognized by the OTel semconv spec. // Non-standard methods are normalized to "_OTHER". -var standardMethods = map[string]struct{}{ - "GET": {}, - "HEAD": {}, - "POST": {}, - "PUT": {}, - "DELETE": {}, - "PATCH": {}, - "OPTIONS": {}, - "TRACE": {}, - "CONNECT": {}, +var standardMethods = map[string]string{ + "GET": "GET", + "HEAD": "HEAD", + "POST": "POST", + "PUT": "PUT", + "DELETE": "DELETE", + "PATCH": "PATCH", + "OPTIONS": "OPTIONS", + "TRACE": "TRACE", + "CONNECT": "CONNECT", } // normalizeMethod returns the method if it is a standard HTTP method, -// or "_OTHER" per the OTel semconv specification. +// or "_OTHER" per the OTel semconv specification. It returns the package's +// own constant, never method itself: the request method can be a view of the +// connection's receive buffer (see ownStrings), and the result goes into +// attributes the SDK keeps. func normalizeMethod(method string) string { - if _, ok := standardMethods[method]; ok { - return method + if m, ok := standardMethods[method]; ok { + return m } return "_OTHER" } +// ownStrings replaces each *p with a copy. The copies share one allocation, +// and nothing is allocated when every string is empty. +// +// On epoll and io_uring the request path, Host, headers and the strings +// derived from them are views of the connection's receive buffer, which the +// engine reuses for the connection's next request and, once the connection +// closes, for another connection. The attributes built from them outlive the +// request: a span processor keeps an ended span until it exports it, and the +// metric SDK keeps every attribute set it has seen as an aggregation key for +// the life of the provider. A kept view would read other request bytes, +// including another client's headers (celeris#732). +func ownStrings(ps ...*string) { + n := 0 + for _, p := range ps { + n += len(*p) + } + if n == 0 { + return + } + var b strings.Builder + b.Grow(n) + for _, p := range ps { + b.WriteString(*p) + } + rest := b.String() + for _, p := range ps { + l := len(*p) + *p, rest = rest[:l], rest[l:] + } +} + +// appendOwned appends attrs to dst with every string value copied. The +// attributes come from CustomAttributes or CustomMetricAttributes, which +// typically read request headers (views, see ownStrings). Keys are kept as +// they are: they are expected to be constants. +func appendOwned(dst, attrs []attribute.KeyValue) []attribute.KeyValue { + for _, kv := range attrs { + switch kv.Value.Type() { + case attribute.STRING: + kv = kv.Key.String(strings.Clone(kv.Value.AsString())) + case attribute.STRINGSLICE: + ss := kv.Value.AsStringSlice() + for i := range ss { + ss[i] = strings.Clone(ss[i]) + } + kv = kv.Key.StringSlice(ss) + } + dst = append(dst, kv) + } + return dst +} + // truncateString truncates s to maxLen bytes without splitting multi-byte // UTF-8 runes. After slicing at maxLen it backs up to the last valid rune // boundary using [utf8.DecodeLastRuneInString]. @@ -170,14 +225,6 @@ func New(config ...Config) celeris.HandlerFunc { carrier := headerCarrier{ctx: c} parentCtx := propagators.Extract(c.Context(), carrier) - spanName := c.Method() - if fp := c.FullPath(); fp != "" { - spanName += " " + fp - } - if spanNameFmt != nil { - spanName = spanNameFmt(c) - } - rawMethod := c.Method() method := normalizeMethod(rawMethod) @@ -187,41 +234,64 @@ func New(config ...Config) celeris.HandlerFunc { // pre-v1.2.4), the http.route attribute is simply omitted. route := c.FullPath() + // The request strings the span and the metric attribute sets keep, + // copied (see ownStrings). The route is the registered pattern and + // the protocol a constant. + spanName := rawMethod + if spanNameFmt != nil { + spanName = spanNameFmt(c) + } + var methodOrig, clientIP, userAgent string + if method != rawMethod { + methodOrig = rawMethod + } + if collectClientIP { + clientIP = c.ClientIP() + } + if collectUserAgent { + userAgent = c.Header("user-agent") + } + scheme, path, host, requestID := c.Scheme(), c.Path(), c.Host(), c.RequestID() + ownStrings(&spanName, &methodOrig, &scheme, &path, &clientIP, &host, &userAgent, &requestID) + if spanNameFmt == nil && route != "" { + spanName += " " + route + } + var spanBuf [14]attribute.KeyValue n := 0 spanBuf[n] = semconv.HTTPRequestMethodKey.String(method) n++ if method != rawMethod { - spanBuf[n] = attribute.String("http.request.method_original", rawMethod) + spanBuf[n] = attribute.String("http.request.method_original", methodOrig) n++ } if route != "" { spanBuf[n] = semconv.HTTPRoute(route) n++ } - spanBuf[n] = semconv.URLScheme(c.Scheme()) + spanBuf[n] = semconv.URLScheme(scheme) n++ - spanBuf[n] = semconv.URLPath(c.Path()) + spanBuf[n] = semconv.URLPath(path) n++ spanBuf[n] = semconv.NetworkProtocolVersion(c.Protocol()) n++ if collectClientIP { - spanBuf[n] = semconv.ClientAddress(c.ClientIP()) + spanBuf[n] = semconv.ClientAddress(clientIP) n++ } - spanBuf[n] = semconv.ServerAddress(c.Host()) + spanBuf[n] = semconv.ServerAddress(host) n++ if serverPort > 0 { spanBuf[n] = semconv.ServerPort(serverPort) n++ } if collectUserAgent { - spanBuf[n] = semconv.UserAgentOriginal(c.Header("user-agent")) + spanBuf[n] = semconv.UserAgentOriginal(userAgent) n++ } spanAttrs := spanBuf[:n] if customAttrs != nil { - spanAttrs = append(spanAttrs, customAttrs(c)...) + spanAttrs = appendOwned(spanAttrs, customAttrs(c)) } spanCtx, span := tracer.Start(parentCtx, spanName, @@ -232,8 +302,8 @@ func New(config ...Config) celeris.HandlerFunc { c.SetContext(spanCtx) - if s := c.RequestID(); s != "" { - span.SetAttributes(attribute.String("request.id", s)) + if requestID != "" { + span.SetAttributes(attribute.String("request.id", requestID)) } if metricsEnabled { @@ -245,9 +315,9 @@ func New(config ...Config) celeris.HandlerFunc { metricBuf[mn] = semconv.HTTPRoute(route) mn++ } - metricBuf[mn] = semconv.URLScheme(c.Scheme()) + metricBuf[mn] = semconv.URLScheme(scheme) mn++ - metricBuf[mn] = semconv.ServerAddress(c.Host()) + metricBuf[mn] = semconv.ServerAddress(host) mn++ if serverPort > 0 { metricBuf[mn] = semconv.ServerPort(serverPort) @@ -255,7 +325,7 @@ func New(config ...Config) celeris.HandlerFunc { } metricBaseAttrs := metricBuf[:mn:mn] if customMetricAttrs != nil { - metricBaseAttrs = append(metricBaseAttrs, customMetricAttrs(c)...) + metricBaseAttrs = appendOwned(metricBaseAttrs, customMetricAttrs(c)) } activeAttrSet := metric.WithAttributeSet(attribute.NewSet(metricBaseAttrs...)) if activeRequests != nil { diff --git a/middleware/otel/retained_attrs_linux_test.go b/middleware/otel/retained_attrs_linux_test.go new file mode 100644 index 00000000..06d124d5 --- /dev/null +++ b/middleware/otel/retained_attrs_linux_test.go @@ -0,0 +1,307 @@ +//go:build linux + +package otel_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "net" + "os" + "sort" + "strconv" + "strings" + "testing" + "time" + + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + + "github.com/goceleris/celeris" + celerisengine "github.com/goceleris/celeris/engine" + celotel "github.com/goceleris/celeris/middleware/otel" + "github.com/goceleris/celeris/probe" +) + +// TestAttributesSurviveNextRequest pins the otel site of celeris#732. +// +// A span processor keeps an ended span until it exports it, and the metric +// SDK keeps every attribute set it has seen as an aggregation key for the +// life of the provider; neither copies strings. On epoll and io_uring the +// request path, Host, User-Agent, the client IP and request ID and scheme +// that middleware derive from headers, a CustomAttributes value, a +// SpanNameFormatter result and a method the H1 parser does not intern are +// views of the connection's receive buffer, which the engine reuses for the +// connection's next request. Three requests with the same layout and +// different values go over one keep-alive connection; every span and every +// duration series must keep its own request's values. +func TestAttributesSurviveNextRequest(t *testing.T) { + methods := []string{"TRACE", "PURGE", "MKCOL"} // TRACE is standard, the others "_OTHER" + tenant := func(c *celeris.Context) []attribute.KeyValue { + return []attribute.KeyValue{attribute.String("tenant", c.Header("x-tenant"))} + } + for _, a := range keptArms(t) { + t.Run(a.name, func(t *testing.T) { + exp := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + addr, stop := startKeptServer(t, func() *celeris.Server { + srv := celeris.New(celeris.Config{Engine: a.engine, AsyncHandlers: a.async}) + // What requestid and proxy middleware store from request headers. + srv.Use(func(c *celeris.Context) error { + c.SetRequestID(c.Header("x-request-id")) + c.SetScheme(c.Header("x-forwarded-proto")) + return c.Next() + }) + srv.Use(celotel.New(celotel.Config{ + TracerProvider: tp, + MeterProvider: mp, + CollectClientIP: true, + SpanNameFormatter: func(c *celeris.Context) string { return c.Path() }, + CustomAttributes: tenant, + CustomMetricAttributes: tenant, + })) + for _, m := range methods { + srv.Handle(m, "/o/:id", func(c *celeris.Context) error { return c.String(200, "ok") }) + } + return srv + }) + defer stop() + + conn, br := keptDial(t, addr) + defer func() { _ = conn.Close() }() + var wantSpans []map[string]string + var wantSeries []string + for i, m := range methods { + l := string(rune('a' + i)) + v := strings.Repeat(l, 4) + keptRoundTrip(t, conn, br, m+" /o/"+v+" HTTP/1.1\r\nHost: host-"+v+".example\r\nUser-Agent: agent-"+v+ + "\r\nX-Forwarded-For: 10.0.0."+strconv.Itoa(i+1)+"\r\nX-Request-Id: rid-"+v+"\r\nX-Forwarded-Proto: sch"+l+ + "\r\nX-Tenant: tenant-"+v+"\r\n\r\n") + method, orig := "_OTHER", m + if m == "TRACE" { + method, orig = m, "" + } + wantSpans = append(wantSpans, map[string]string{ + "name": "/o/" + v, + "http.request.method": method, + "http.request.method_original": orig, + "url.scheme": "sch" + l, + "url.path": "/o/" + v, + "client.address": "10.0.0." + strconv.Itoa(i+1), + "server.address": "host-" + v + ".example", + "user_agent.original": "agent-" + v, + "request.id": "rid-" + v, + "tenant": "tenant-" + v, + }) + wantSeries = append(wantSeries, method+"|sch"+l+"|host-"+v+".example|tenant-"+v) + } + sort.Strings(wantSeries) + + // The middleware ends a span after the response reaches the client. + spans := exp.GetSpans() + for i := 0; len(spans) < len(methods) && i < 200; i++ { + time.Sleep(10 * time.Millisecond) + spans = exp.GetSpans() + } + if len(spans) != len(methods) { + t.Fatalf("got %d spans, want %d", len(spans), len(methods)) + } + sort.Slice(spans, func(i, j int) bool { return spans[i].StartTime.Before(spans[j].StartTime) }) + var wrong []string + for i, s := range spans { + got := map[string]string{"name": s.Name} + for _, kv := range s.Attributes { + if kv.Value.Type() == attribute.STRING { + got[string(kv.Key)] = kv.Value.AsString() + } + } + keys := make([]string, 0, len(wantSpans[i])) + for k := range wantSpans[i] { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + if got[k] != wantSpans[i][k] { + wrong = append(wrong, fmt.Sprintf("span %d %s: %q (want %q)", i+1, k, got[k], wantSpans[i][k])) + } + } + } + + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatal(err) + } + var series []string + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + h, ok := m.Data.(metricdata.Histogram[float64]) + if m.Name != "http.server.request.duration" || !ok { + continue + } + for _, dp := range h.DataPoints { + var parts []string + for _, k := range []attribute.Key{"http.request.method", "url.scheme", "server.address", "tenant"} { + v, _ := dp.Attributes.Value(k) + parts = append(parts, v.AsString()) + } + series = append(series, strings.Join(parts, "|")) + } + } + } + sort.Strings(series) + if strings.Join(series, ",") != strings.Join(wantSeries, ",") { + wrong = append(wrong, fmt.Sprintf("duration series (method|scheme|host|tenant) %q (want %q)", series, wantSeries)) + } + t.Logf("KEPT732OTEL arm=%s spans=%d series=%d wrong=%d", a.name, len(spans), len(series), len(wrong)) + if len(wrong) > 0 { + t.Errorf("attributes kept from a request read other bytes after the connection's later requests:\n %s", strings.Join(wrong, "\n ")) + } + }) + } +} + +type keptArm struct { + name string + engine celeris.EngineType + async bool +} + +// keptArms returns std, and epoll and io_uring with sync and async +// handlers. With CELERIS_REQUIRE_IOURING_WORKERS=1 a kernel with no usable +// io_uring fails the test instead of dropping the io_uring arms. +func keptArms(t *testing.T) []keptArm { + t.Helper() + arms := []keptArm{ + {"std", celeris.Std, false}, + {"epoll", celeris.Epoll, false}, + {"epoll-async", celeris.Epoll, true}, + } + if ok, p := keptProbeIOUring(); ok { + arms = append(arms, keptArm{"io_uring", celeris.IOUring, false}, keptArm{"io_uring-async", celeris.IOUring, true}) + } else if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") == "1" { + t.Fatalf("io_uring tier=%s kernel=%s, and CELERIS_REQUIRE_IOURING_WORKERS=1 forbids dropping the io_uring arms", p.IOUringTier, p.KernelVersion) + } else { + t.Logf("io_uring tier=%s kernel=%s: io_uring arms not run", p.IOUringTier, p.KernelVersion) + } + return arms +} + +// keptProbeIOUring probes the kernel's io_uring support. With +// CELERIS_REQUIRE_IOURING_WORKERS=1 a probe that finds no usable ring is +// retried for up to 10 s: the probe's ring can fail with ENOMEM against +// RLIMIT_MEMLOCK while the rings of engines stopped moments ago, or of +// another test binary of the same user, are still charged +// (engine/iouring/ring_budget_linux_test.go). +func keptProbeIOUring() (usable bool, p celerisengine.CapabilityProfile) { + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + if os.Getenv("CELERIS_REQUIRE_IOURING_WORKERS") != "1" { + return usable, p + } + for deadline := time.Now().Add(10 * time.Second); !usable && time.Now().Before(deadline); { + time.Sleep(10 * time.Millisecond) + p = probe.Probe() + usable = p.IOUringTier >= celerisengine.High && p.ProvidedBuffers + } + return usable, p +} + +// startKeptServer starts the server mk builds on a fresh loopback listener +// and returns its address and a shutdown closure. A start that fails only +// with ENOMEM (io_uring ring memory still charged to RLIMIT_MEMLOCK, see +// keptProbeIOUring) is retried with a new server for up to 10 s. +func startKeptServer(t *testing.T, mk func() *celeris.Server) (string, func()) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + s := mk() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.StartWithListenerAndContext(ctx, ln) }() + addr, err := keptWaitReady(s, done) + if err == nil { + return addr, func() { cancel(); <-done } + } + cancel() + _ = ln.Close() + if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + continue + } + t.Fatalf("server did not start: %v", err) + } +} + +func keptWaitReady(s *celeris.Server, done <-chan error) (string, error) { + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + if err == nil { + err = errors.New("start returned before the server was ready") + } + return "", err + default: + } + if a := s.Addr(); a != nil { + if c, err := net.DialTimeout("tcp", a.String(), 100*time.Millisecond); err == nil { + _ = c.Close() + return a.String(), nil + } + } + time.Sleep(10 * time.Millisecond) + } + return "", errors.New("server not ready within 30s") +} + +func keptDial(t *testing.T, addr string) (net.Conn, *bufio.Reader) { + t.Helper() + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + t.Fatal(err) + } + _ = conn.SetDeadline(time.Now().Add(20 * time.Second)) + return conn, bufio.NewReader(conn) +} + +// keptRoundTrip writes one request and reads its response; it fails the +// test unless the status is 2xx. +func keptRoundTrip(t *testing.T, conn net.Conn, br *bufio.Reader, req string) { + t.Helper() + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatal(err) + } + status, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read status: %v", err) + } + n := 0 + for { + line, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read header: %v", err) + } + if line == "\r\n" { + break + } + if k, v, ok := strings.Cut(line, ":"); ok && strings.EqualFold(k, "content-length") { + n, _ = strconv.Atoi(strings.TrimSpace(v)) + } + } + if _, err := br.Discard(n); err != nil { + t.Fatalf("read body: %v", err) + } + if f := strings.Fields(status); len(f) < 2 || f[1][0] != '2' { + t.Fatalf("status %q for %q", strings.TrimSpace(status), strings.SplitN(req, "\r\n", 2)[0]) + } +}