diff --git a/internal/observabilityproxy/server.go b/internal/observabilityproxy/server.go new file mode 100644 index 000000000..a0e2757c5 --- /dev/null +++ b/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 + } + } +} diff --git a/internal/observabilityproxy/server_test.go b/internal/observabilityproxy/server_test.go new file mode 100644 index 000000000..ed50f72a8 --- /dev/null +++ b/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) + } +}