feat(observability-proxy): serve the namespace logs root

Tenants get a way to read their namespace's logs without Grafana and
without reaching vlselect directly, which would give them neither an
RBAC check nor a forced scope. The handler authenticates, authorizes
from the path, rebuilds the upstream request with fixed tenant headers
and no client headers, and returns VictoriaLogs' native response.

A live tail is flushed after every line, is not subject to the request
timeout, and is closed once idle, so it survives far past the apiserver
request timeout without leaving forgotten browser tabs holding streams
open. An unreachable store answers 502 so the dashboard can tell an
outage from an empty result.

Assisted-by: LLM
Signed-off-by: ZverGuy <maximbel2003@gmail.com>

internal/observabilityproxy/server.go

@@ -0,0 +1,270 @@
+/*
+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"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// maxFormBytes bounds a POST body; LogsQL queries are a few KiB at most.
+const maxFormBytes = 1 << 20
+
+// Config is the runtime configuration of the proxy handler.
+type Config struct {
+	ClusterDomain string
+	// UpstreamURL maps a store namespace to the vlselect base URL; nil
+	// means the in-cluster Service of the "generic" logs storage.
+	UpstreamURL     func(store string) string
+	Logs            LogsLimits
+	Limiter         LimiterConfig
+	RequestTimeout  time.Duration
+	TailIdleTimeout time.Duration
+}
+
+// StoreLookup resolves a request namespace to the namespace of its store.
+type StoreLookup interface {
+	Resolve(namespace string) (string, error)
+}
+
+// Server serves the tenant observability roots.
+type Server struct {
+	cfg      Config
+	auth     *Authorizer
+	stores   StoreLookup
+	client   *http.Client
+	limits   *limiter
+	now      func() time.Time
+	requests *prometheus.CounterVec
+}
+
+func NewServer(cfg Config, auth *Authorizer, stores StoreLookup, reg prometheus.Registerer) *Server {
+	if cfg.UpstreamURL == nil {
+		domain := cfg.ClusterDomain
+		cfg.UpstreamURL = func(store string) string {
+			return fmt.Sprintf("http://vlselect-generic.%s.svc.%s:9471", store, domain)
+		}
+	}
+	s := &Server{
+		cfg:    cfg,
+		auth:   auth,
+		stores: stores,
+		// No client timeout: it would cut live tails. Non-tail requests are
+		// bounded by a context deadline instead.
+		client: &http.Client{Transport: http.DefaultTransport.(*http.Transport).Clone()},
+		limits: newLimiter(cfg.Limiter),
+		now:    time.Now,
+		requests: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "observability_proxy_requests_total",
+			Help: "Requests served, by route kind and status code.",
+		}, []string{"route", "code"}),
+	}
+	if reg != nil {
+		reg.MustRegister(s.requests, prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+			Name: "observability_proxy_open_tails",
+			Help: "Live tails currently open.",
+		}, func() float64 { return float64(s.limits.openTails()) }))
+	}
+	return s
+}
+
+type statusRecorder struct {
+	http.ResponseWriter
+	code int
+}
+
+func (r *statusRecorder) WriteHeader(code int) {
+	if r.code == 0 {
+		r.code = code
+	}
+	r.ResponseWriter.WriteHeader(code)
+}
+
+func (r *statusRecorder) Write(b []byte) (int, error) {
+	if r.code == 0 {
+		r.code = http.StatusOK
+	}
+	return r.ResponseWriter.Write(b)
+}
+
+func (r *statusRecorder) Unwrap() http.ResponseWriter { return r.ResponseWriter }
+
+func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	rec := &statusRecorder{ResponseWriter: w}
+	route, herr := ParseRoute(r.URL.EscapedPath())
+	if herr == nil {
+		herr = s.serve(rec, r, route)
+	}
+	if herr != nil {
+		writeError(rec, herr)
+	}
+	s.requests.WithLabelValues(route.Kind.String(), strconv.Itoa(rec.code)).Inc()
+}
+
+func writeError(w http.ResponseWriter, e *HTTPError) {
+	if e.Code == http.StatusUnauthorized {
+		w.Header().Set("WWW-Authenticate", `Bearer realm="cozystack"`)
+	}
+	http.Error(w, e.Message, e.Code)
+}
+
+// serve answers an unimplemented root before authenticating, so later
+// waves only add a case here.
+func (s *Server) serve(w http.ResponseWriter, r *http.Request, route Route) *HTTPError {
+	if route.Kind != NamespaceLogs {
+		return errorf(http.StatusNotFound, "not found")
+	}
+	user, herr := s.authorize(r, route)
+	if herr != nil {
+		return herr
+	}
+	if r.Method != http.MethodGet && r.Method != http.MethodPost {
+		w.Header().Set("Allow", "GET, POST")
+		return errorf(http.StatusMethodNotAllowed, "method not allowed")
+	}
+	if !s.limits.allowRate(user) {
+		return errorf(http.StatusUnprocessableEntity, "request rate limit exceeded")
+	}
+	endpoint, ok := strings.CutPrefix(route.Endpoint, "select/logsql/")
+	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 {
+		acquire = s.limits.acquireTail
+	}
+	release, ok := acquire(user)
+	if !ok {
+		return errorf(http.StatusUnprocessableEntity, "too many concurrent requests")
+	}
+	defer release()
+
+	store, err := s.stores.Resolve(route.Namespace)
+	if err != nil {
+		slog.Error("resolving log store", "namespace", route.Namespace, "error", err)
+		return errorf(http.StatusServiceUnavailable, "cannot resolve the log store")
+	}
+
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
+	if !tail && s.cfg.RequestTimeout > 0 {
+		ctx, cancel = context.WithTimeout(ctx, s.cfg.RequestTimeout)
+		defer cancel()
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
+		s.cfg.UpstreamURL(store)+"/select/logsql/"+endpoint, strings.NewReader(params.Encode()))
+	if err != nil {
+		return errorf(http.StatusInternalServerError, "building upstream request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Set("AccountID", "0")
+	req.Header.Set("ProjectID", "0")
+	resp, err := s.client.Do(req)
+	if err != nil {
+		slog.Warn("log store unreachable", "store", store, "error", err)
+		return errorf(http.StatusBadGateway, "log store unreachable")
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if ct := resp.Header.Get("Content-Type"); ct != "" {
+		w.Header().Set("Content-Type", ct)
+	}
+	w.WriteHeader(resp.StatusCode)
+	rc := http.NewResponseController(w)
+	_ = rc.Flush()
+	if tail {
+		s.streamLines(w, rc, resp.Body, cancel)
+		return nil
+	}
+	if _, err := io.Copy(w, resp.Body); err != nil && !errors.Is(err, context.Canceled) {
+		slog.Debug("copying upstream response", "error", err)
+	}
+	return nil
+}
+
+func (s *Server) authorize(r *http.Request, route Route) (string, *HTTPError) {
+	token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ")
+	if !ok {
+		return "", errorf(http.StatusUnauthorized, "missing bearer token")
+	}
+	user, err := s.auth.Authenticate(r.Context(), strings.TrimSpace(token))
+	if errors.Is(err, ErrUnauthenticated) {
+		return "", errorf(http.StatusUnauthorized, "invalid bearer token")
+	}
+	if err != nil {
+		slog.Error("token review failed", "error", err)
+		return "", errorf(http.StatusServiceUnavailable, "authentication unavailable")
+	}
+	allowed, err := s.auth.Authorize(r.Context(), user, route.ResourceAttributes())
+	if err != nil {
+		slog.Error("subject access review failed", "error", err)
+		return "", errorf(http.StatusServiceUnavailable, "authorization unavailable")
+	}
+	if !allowed {
+		return "", errorf(http.StatusForbidden, "forbidden")
+	}
+	return user.Username, nil
+}
+
+// streamLines flushes after every line so a live tail reaches the browser
+// as it happens, and cancels the upstream stream once no line has arrived
+// for the idle timeout.
+func (s *Server) streamLines(w http.ResponseWriter, rc *http.ResponseController, body io.Reader, cancel context.CancelFunc) {
+	var idle *time.Timer
+	if s.cfg.TailIdleTimeout > 0 {
+		idle = time.AfterFunc(s.cfg.TailIdleTimeout, cancel)
+		defer idle.Stop()
+	}
+	rd := bufio.NewReader(body)
+	for {
+		line, err := rd.ReadBytes('\n')
+		if len(line) > 0 {
+			if idle != nil {
+				idle.Reset(s.cfg.TailIdleTimeout)
+			}
+			if _, werr := w.Write(line); werr != nil {
+				return
+			}
+			if ferr := rc.Flush(); ferr != nil {
+				return
+			}
+		}
+		if err != nil {
+			return
+		}
+	}
+}

internal/observabilityproxy/server_test.go

@@ -0,0 +1,364 @@
+/*
+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"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	authorizationv1 "k8s.io/api/authorization/v1"
+)
+
+type upstreamRequest struct {
+	path    string
+	form    url.Values
+	headers http.Header
+}
+
+type harness struct {
+	t        *testing.T
+	kube     *fakeKube
+	proxy    *httptest.Server
+	server   *Server
+	mu       sync.Mutex
+	got      []upstreamRequest
+	stores   []string
+	upstream *httptest.Server
+}
+
+type mapStores map[string]string
+
+func (m mapStores) Resolve(ns string) (string, error) { return m[ns], nil }
+
+// newHarness wires the proxy to a fake apiserver that knows tokens "good"
+// (alice) and "bob-token" (bob), both allowed list logs in tenant-a only,
+// and to an upstream served by handler.
+func newHarness(t *testing.T, cfg Config, handler http.HandlerFunc) *harness {
+	h := &harness{t: t}
+	h.kube = newFakeKube(map[string]string{"good": "alice", "bob-token": "bob"}, func(_ string, a *authorizationv1.ResourceAttributes) bool {
+		return a.Namespace == "tenant-a" && a.Resource == "logs"
+	})
+	h.upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if err := r.ParseForm(); err != nil {
+			t.Errorf("upstream ParseForm: %v", err)
+		}
+		h.mu.Lock()
+		h.got = append(h.got, upstreamRequest{path: r.URL.Path, form: r.PostForm, headers: r.Header.Clone()})
+		h.mu.Unlock()
+		if handler != nil {
+			handler(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "application/stream+json")
+		_, _ = io.WriteString(w, `{"_msg":"hello"}`+"\n")
+	}))
+	t.Cleanup(h.upstream.Close)
+	cfg.UpstreamURL = func(store string) string {
+		h.mu.Lock()
+		h.stores = append(h.stores, store)
+		h.mu.Unlock()
+		return h.upstream.URL
+	}
+	if cfg.Logs == (LogsLimits{}) {
+		cfg.Logs = testLimits()
+	}
+	h.server = NewServer(cfg, NewAuthorizer(h.kube.client, 10*time.Second), mapStores{"tenant-a": "tenant-a-store"}, nil)
+	h.server.now = func() time.Time { return testNow }
+	h.proxy = httptest.NewServer(h.server)
+	t.Cleanup(h.proxy.Close)
+	return h
+}
+
+const logsRoot = "/apis/cozystack.io/v1alpha1/namespaces/tenant-a/logs/select/logsql/"
+
+func (h *harness) do(method, path, token string, body url.Values, headers map[string]string) *http.Response {
+	h.t.Helper()
+	var rd io.Reader
+	if body != nil {
+		rd = strings.NewReader(body.Encode())
+	}
+	req, err := http.NewRequest(method, h.proxy.URL+path, rd)
+	if err != nil {
+		h.t.Fatal(err)
+	}
+	if body != nil {
+		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	}
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	for k, v := range headers {
+		req.Header.Set(k, v)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		h.t.Fatal(err)
+	}
+	return resp
+}
+
+func (h *harness) status(method, path, token string) int {
+	h.t.Helper()
+	resp := h.do(method, path, token, nil, nil)
+	_, _ = io.Copy(io.Discard, resp.Body)
+	_ = resp.Body.Close()
+	return resp.StatusCode
+}
+
+func (h *harness) requests() []upstreamRequest {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return append([]upstreamRequest(nil), h.got...)
+}
+
+func TestServerStatusCodes(t *testing.T) {
+	h := newHarness(t, Config{}, nil)
+	tests := []struct {
+		name, method, path, token string
+		want                      int
+	}{
+		{"ok", http.MethodGet, logsRoot + "query?query=error", "good", http.StatusOK},
+		{"missing token", http.MethodGet, logsRoot + "query", "", http.StatusUnauthorized},
+		{"invalid token", http.MethodGet, logsRoot + "query", "bad", http.StatusUnauthorized},
+		{"other namespace", http.MethodGet, "/apis/cozystack.io/v1alpha1/namespaces/tenant-b/logs/select/logsql/query", "good", http.StatusForbidden},
+		{"invalid namespace", http.MethodGet, "/apis/cozystack.io/v1alpha1/namespaces/Tenant/logs/select/logsql/query", "good", http.StatusBadRequest},
+		{"unknown root", http.MethodGet, "/api/v1/namespaces/tenant-a/pods", "good", http.StatusNotFound},
+		{"namespace metrics not served yet", http.MethodGet, "/apis/cozystack.io/v1alpha1/namespaces/tenant-a/metrics/api/v1/query", "good", http.StatusNotFound},
+		{"application log not served yet", http.MethodGet, "/apis/apps.cozystack.io/v1alpha1/namespaces/tenant-a/postgreses/db/log/select/logsql/query", "good", http.StatusNotFound},
+		{"application metrics not served yet", http.MethodGet, "/apis/apps.cozystack.io/v1alpha1/namespaces/tenant-a/postgreses/db/metrics/api/v1/query", "good", http.StatusNotFound},
+		{"tenant_ids not proxied", http.MethodGet, "/apis/cozystack.io/v1alpha1/namespaces/tenant-a/logs/select/tenant_ids", "good", http.StatusNotFound},
+		{"unknown logsql endpoint", http.MethodGet, logsRoot + "stats_query", "good", http.StatusNotFound},
+		{"logs root itself", http.MethodGet, "/apis/cozystack.io/v1alpha1/namespaces/tenant-a/logs/", "good", http.StatusNotFound},
+		{"ignore_global_time_filter", http.MethodGet, logsRoot + "query?query=" + url.QueryEscape("options(ignore_global_time_filter=true) *"), "good", http.StatusBadRequest},
+		{"inverted range", http.MethodGet, logsRoot + "query?start=2026-11-04T11:00:00Z&end=2026-11-04T10:00:00Z", "good", http.StatusBadRequest},
+		{"delete", http.MethodDelete, logsRoot + "query", "good", http.StatusMethodNotAllowed},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := h.status(tt.method, tt.path, tt.token); got != tt.want {
+				t.Fatalf("status = %d, want %d", got, tt.want)
+			}
+		})
+	}
+	if reqs := h.requests(); len(reqs) != 1 {
+		t.Fatalf("upstream saw %d requests, want only the allowed one", len(reqs))
+	}
+}
+
+func TestServerRebuildsUpstreamRequest(t *testing.T) {
+	h := newHarness(t, Config{}, nil)
+	hostile := url.Values{
+		"query":                {"error"},
+		"extra_filters":        {`{kubernetes_namespace_name="tenant-b"}`},
+		"extra_stream_filters": {`{kubernetes_namespace_name=~".*"}`},
+		"AccountID":            {"9"},
+	}
+	headers := map[string]string{"AccountID": "9", "ProjectID": "9", "X-Forwarded-For": "1.2.3.4", "Cookie": "s=1"}
+
+	resp := h.do(http.MethodGet, logsRoot+"query?"+hostile.Encode(), "good", nil, headers)
+	_ = resp.Body.Close()
+	resp = h.do(http.MethodPost, logsRoot+"query?extra_filters="+url.QueryEscape(`{x="y"}`), "good", hostile, headers)
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK || string(body) != `{"_msg":"hello"}`+"\n" {
+		t.Fatalf("POST answered %d %q", resp.StatusCode, body)
+	}
+	if ct := resp.Header.Get("Content-Type"); ct != "application/stream+json" {
+		t.Fatalf("Content-Type = %q, want the upstream's", ct)
+	}
+
+	reqs := h.requests()
+	if len(reqs) != 2 {
+		t.Fatalf("upstream saw %d requests", len(reqs))
+	}
+	if reqs[0].form.Encode() != reqs[1].form.Encode() {
+		t.Fatalf("GET and POST built different upstream requests:\n%v\n%v", reqs[0].form, reqs[1].form)
+	}
+	if n := h.kube.sars.Load(); n != 1 {
+		t.Fatalf("SubjectAccessReviews = %d, want 1: POST must reuse the GET decision", n)
+	}
+	for _, r := range reqs {
+		if r.path != "/select/logsql/query" {
+			t.Errorf("upstream path = %s", r.path)
+		}
+		if got := r.form["extra_stream_filters"]; len(got) != 1 || got[0] != wantScope {
+			t.Errorf("extra_stream_filters = %q", got)
+		}
+		if r.form.Has("extra_filters") || r.form.Has("AccountID") {
+			t.Errorf("client scope parameter forwarded: %v", r.form)
+		}
+		if r.headers.Get("AccountID") != "0" || r.headers.Get("ProjectID") != "0" {
+			t.Errorf("tenant headers = %q/%q, want 0/0", r.headers.Get("AccountID"), r.headers.Get("ProjectID"))
+		}
+		for _, k := range []string{"Authorization", "X-Forwarded-For", "Cookie"} {
+			if r.headers.Get(k) != "" {
+				t.Errorf("client header %s forwarded", k)
+			}
+		}
+	}
+	for _, s := range h.stores {
+		if s != "tenant-a-store" {
+			t.Errorf("store = %s, want the resolved one", s)
+		}
+	}
+}
+
+func TestServerUpstreamErrors(t *testing.T) {
+	h := newHarness(t, Config{}, func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "cannot parse query", http.StatusBadRequest)
+	})
+	if got := h.status(http.MethodGet, logsRoot+"query?query=|", "good"); got != http.StatusBadRequest {
+		t.Fatalf("upstream 400 answered as %d", got)
+	}
+	h.upstream.Close()
+	if got := h.status(http.MethodGet, logsRoot+"query", "good"); got != http.StatusBadGateway {
+		t.Fatalf("unreachable upstream answered %d, want 502", got)
+	}
+}
+
+func TestServerLimits(t *testing.T) {
+	release := make(chan struct{})
+	var once sync.Once
+	t.Cleanup(func() { once.Do(func() { close(release) }) })
+	h := newHarness(t, Config{Limiter: LimiterConfig{MaxConcurrentPerUser: 1}}, func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		w.(http.Flusher).Flush()
+		<-release
+	})
+	first := h.do(http.MethodGet, logsRoot+"query", "good", nil, nil)
+	defer func() { _ = first.Body.Close() }()
+	if got := h.status(http.MethodGet, logsRoot+"query", "good"); got != http.StatusUnprocessableEntity {
+		t.Fatalf("concurrent request answered %d, want 422", got)
+	}
+	once.Do(func() { close(release) })
+
+	h2 := newHarness(t, Config{Limiter: LimiterConfig{RatePerUser: 0.001, RateBurst: 1}}, nil)
+	if got := h2.status(http.MethodGet, logsRoot+"query", "good"); got != http.StatusOK {
+		t.Fatalf("first request answered %d", got)
+	}
+	if got := h2.status(http.MethodGet, logsRoot+"query", "good"); got != http.StatusUnprocessableEntity {
+		t.Fatalf("rate-limited request answered %d, want 422", got)
+	}
+}
+
+// tailUpstream sends one line per value received on lines and holds the
+// stream open until done is closed.
+func tailUpstream(lines <-chan string, done <-chan struct{}) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/stream+json")
+		w.WriteHeader(http.StatusOK)
+		w.(http.Flusher).Flush()
+		for {
+			select {
+			case l := <-lines:
+				_, _ = io.WriteString(w, l+"\n")
+				w.(http.Flusher).Flush()
+			case <-done:
+				return
+			case <-r.Context().Done():
+				return
+			}
+		}
+	}
+}
+
+func TestServerTailFlushesEveryLine(t *testing.T) {
+	lines := make(chan string)
+	done := make(chan struct{})
+	t.Cleanup(func() { close(done) })
+	h := newHarness(t, Config{TailIdleTimeout: time.Minute, RequestTimeout: 50 * time.Millisecond}, tailUpstream(lines, done))
+
+	resp := h.do(http.MethodGet, logsRoot+"tail?query=*&since=2026-11-04T08:00:00Z", "good", nil, nil)
+	defer func() { _ = resp.Body.Close() }()
+	rd := bufio.NewReader(resp.Body)
+	for i, l := range []string{`{"_msg":"one"}`, `{"_msg":"two"}`} {
+		select {
+		case lines <- l:
+		case <-time.After(2 * time.Second):
+			t.Fatalf("upstream tail closed before line %d", i)
+		}
+		got := make(chan string, 1)
+		go func() {
+			s, _ := rd.ReadString('\n')
+			got <- s
+		}()
+		select {
+		case s := <-got:
+			if s != l+"\n" {
+				t.Fatalf("line %d = %q", i, s)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("line %d not flushed to the client", i)
+		}
+		// Past the request timeout: a tail must not be cut by it.
+		time.Sleep(60 * time.Millisecond)
+	}
+	if r := h.requests()[0]; r.path != "/select/logsql/tail" || r.form.Get("start_offset") != "900000ms" {
+		t.Fatalf("tail upstream = %s %v, want start_offset capped at 15m", r.path, r.form)
+	}
+}
+
+func TestServerTailCapsAndIdle(t *testing.T) {
+	done := make(chan struct{})
+	t.Cleanup(func() { close(done) })
+	h := newHarness(t, Config{
+		Limiter:         LimiterConfig{MaxTailsPerUser: 1, MaxTails: 2},
+		TailIdleTimeout: 500 * time.Millisecond,
+	}, tailUpstream(make(chan string), done))
+
+	a := h.do(http.MethodGet, logsRoot+"tail", "good", nil, nil)
+	defer func() { _ = a.Body.Close() }()
+	if a.StatusCode != http.StatusOK {
+		t.Fatalf("first tail answered %d", a.StatusCode)
+	}
+	if got := h.status(http.MethodGet, logsRoot+"tail", "good"); got != http.StatusUnprocessableEntity {
+		t.Fatalf("second tail of one user answered %d, want 422", got)
+	}
+	b := h.do(http.MethodGet, logsRoot+"tail", "bob-token", nil, nil)
+	defer func() { _ = b.Body.Close() }()
+	if b.StatusCode != http.StatusOK {
+		t.Fatalf("bob's tail answered %d", b.StatusCode)
+	}
+	if got := h.server.limits.openTails(); got != 2 {
+		t.Fatalf("open tails = %d, want 2", got)
+	}
+
+	closed := make(chan error, 1)
+	go func() {
+		_, err := io.Copy(io.Discard, a.Body)
+		closed <- err
+	}()
+	select {
+	case <-closed:
+	case <-time.After(3 * time.Second):
+		t.Fatal("idle tail not closed")
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for h.server.limits.openTails() > 1 {
+		if time.Now().After(deadline) {
+			t.Fatal("closed tail still holds its slot")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}