diff --git a/internal/observabilityproxy/logsparams.go b/internal/observabilityproxy/logsparams.go new file mode 100644 index 000000000..cca53e62d --- /dev/null +++ b/internal/observabilityproxy/logsparams.go @@ -0,0 +1,314 @@ +/* +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 ( + "fmt" + "net/http" + "net/url" + "regexp" + "strconv" + "strings" + "time" +) + +// LogsLimits bound what one logs request may make vlselect do. +type LogsLimits struct { + DefaultRange time.Duration + MaxRange time.Duration + MaxBuckets int + DefaultLimit int + MaxLimit int + MaxTailResume time.Duration +} + +// tailOffset is vlselect's default tail offset. The proxy never forwards a +// client offset, so the resume arithmetic can rely on this value. +const tailOffset = 5 * time.Second + +const maxHitsFields = 10 + +var logsEndpoints = map[string]bool{ + "query": true, "hits": true, "stats_query_range": true, "tail": true, + "field_values": true, "stream_field_values": true, "facets": true, +} + +// BuildLogsParams rebuilds the upstream parameters from scratch: only the +// allowlisted parameters of the endpoint are read from the client, and the +// namespace scope is added as the only extra_stream_filters value. +// VictoriaLogs ANDs extra_stream_filters into the query and every subquery +// (union, join, in()), which is what lets the query itself pass through +// unparsed. +func BuildLogsParams(endpoint string, in url.Values, namespace string, now time.Time, lim LogsLimits) (url.Values, *HTTPError) { + if !logsEndpoints[endpoint] { + return nil, errorf(http.StatusNotFound, "not found") + } + query := in.Get("query") + if query == "" { + query = "*" + } + // The option lifts the time window this function enforces, and on tail + // it rescans the whole retention every tick. + if strings.Contains(strings.ToLower(query), "ignore_global_time_filter") { + return nil, errorf(http.StatusBadRequest, "ignore_global_time_filter is not allowed") + } + out := url.Values{} + out.Set("query", query) + out.Set("extra_stream_filters", "{kubernetes_namespace_name="+strconv.Quote(namespace)+"}") + + if endpoint == "tail" { + return out, tailParams(in, out, now, lim) + } + + start, end, herr := window(in, now, lim) + if herr != nil { + return nil, herr + } + out.Set("start", start.Format(time.RFC3339Nano)) + out.Set("end", end.Format(time.RFC3339Nano)) + + switch endpoint { + case "query": + herr = setLimit(out, in, "limit", lim.DefaultLimit, lim.MaxLimit) + case "hits": + if herr = setStep(out, in, end.Sub(start), lim); herr != nil { + return nil, herr + } + fields := in["field"] + if len(fields) > maxHitsFields { + return nil, errorf(http.StatusBadRequest, "at most %d field parameters are allowed", maxHitsFields) + } + for _, f := range fields { + if f != "" { + out.Add("field", f) + } + } + if in.Has("fields_limit") { + herr = setLimit(out, in, "fields_limit", 0, lim.MaxLimit) + } + case "stats_query_range": + herr = setStep(out, in, end.Sub(start), lim) + case "field_values", "stream_field_values": + if in.Get("field") == "" { + return nil, errorf(http.StatusBadRequest, "field is required") + } + out.Set("field", in.Get("field")) + if f := in.Get("filter"); f != "" { + out.Set("filter", f) + } + herr = setLimit(out, in, "limit", lim.DefaultLimit, lim.MaxLimit) + case "facets": + for _, k := range []string{"limit", "max_values_per_field", "max_value_len"} { + if k == "limit" || in.Has(k) { + if herr = setLimit(out, in, k, lim.DefaultLimit, lim.MaxLimit); herr != nil { + return nil, herr + } + } + } + if v := in.Get("keep_const_fields"); v == "1" || v == "true" { + out.Set("keep_const_fields", "1") + } + } + if herr != nil { + return nil, herr + } + return out, nil +} + +// window derives a finite [start, end]: VictoriaLogs otherwise starts at +// the oldest stored entry and scans the whole retention. +func window(in url.Values, now time.Time, lim LogsLimits) (time.Time, time.Time, *HTTPError) { + end := now + if v := in.Get("end"); v != "" { + t, err := parseTime(v, now) + if err != nil { + return time.Time{}, time.Time{}, errorf(http.StatusBadRequest, "invalid end: %v", err) + } + end = t + } + start := end.Add(-lim.DefaultRange) + if v := in.Get("start"); v != "" { + t, err := parseTime(v, now) + if err != nil { + return time.Time{}, time.Time{}, errorf(http.StatusBadRequest, "invalid start: %v", err) + } + start = t + } + if start.After(end) { + return time.Time{}, time.Time{}, errorf(http.StatusBadRequest, "start is after end") + } + if end.Sub(start) > lim.MaxRange { + start = end.Add(-lim.MaxRange) + } + return start.UTC(), end.UTC(), nil +} + +// setStep raises step so that hits and stats_query_range, which return one +// bucket per step, return at most MaxBuckets buckets over the window. +func setStep(out, in url.Values, span time.Duration, lim LogsLimits) *HTTPError { + minStep := max((span+time.Duration(lim.MaxBuckets)-1)/time.Duration(lim.MaxBuckets), time.Second) + step := minStep + if v := in.Get("step"); v != "" { + d, err := parseDuration(v) + if err != nil || d <= 0 { + return errorf(http.StatusBadRequest, "invalid step %q", v) + } + step = max(d, minStep) + } + out.Set("step", formatDuration(step)) + return nil +} + +func setLimit(out, in url.Values, key string, def, maxv int) *HTTPError { + n := def + if v := in.Get(key); v != "" { + parsed, err := strconv.Atoi(v) + if err != nil || parsed <= 0 { + return errorf(http.StatusBadRequest, "invalid %s %q", key, v) + } + n = parsed + } + out.Set(key, strconv.Itoa(min(n, maxv))) + return nil +} + +// tailParams turns a resume point into a capped start_offset. A reconnect +// older than the cap resumes at the cap; the gap needs a historical query. +func tailParams(in, out url.Values, now time.Time, lim LogsLimits) *HTTPError { + var start, end time.Time + for _, k := range []string{"start", "end"} { + v := in.Get(k) + if v == "" { + continue + } + t, err := parseTime(v, now) + if err != nil { + return errorf(http.StatusBadRequest, "invalid %s: %v", k, err) + } + out.Set(k, t.UTC().Format(time.RFC3339Nano)) + if k == "start" { + start = t + } else { + end = t + } + } + if !start.IsZero() && !end.IsZero() && start.After(end) { + return errorf(http.StatusBadRequest, "start is after end") + } + + since, offset := in.Get("since"), in.Get("start_offset") + var d time.Duration + switch { + case since != "" && offset != "": + return errorf(http.StatusBadRequest, "since and start_offset are mutually exclusive") + case since != "": + t, err := parseTime(since, now) + if err != nil { + return errorf(http.StatusBadRequest, "invalid since: %v", err) + } + d = now.Add(-tailOffset).Sub(t) + case offset != "": + parsed, err := parseDuration(offset) + if err != nil { + return errorf(http.StatusBadRequest, "invalid start_offset %q", offset) + } + d = parsed + default: + return nil + } + out.Set("start_offset", formatDuration(min(max(d, 0), lim.MaxTailResume))) + return nil +} + +// formatDuration writes whole milliseconds, a form every VictoriaLogs +// duration parser accepts. +func formatDuration(d time.Duration) string { + return fmt.Sprintf("%dms", d.Milliseconds()) +} + +var durationToken = regexp.MustCompile(`^(\d+(?:\.\d+)?)(ms|s|m|h|d|w|y)`) + +var durationUnits = map[string]time.Duration{ + "ms": time.Millisecond, "s": time.Second, "m": time.Minute, "h": time.Hour, + "d": 24 * time.Hour, "w": 7 * 24 * time.Hour, "y": 365 * 24 * time.Hour, +} + +// parseDuration accepts the LogsQL duration forms (1h30m, 1d, 2w) and bare +// seconds; Go's parser knows neither d nor w. +func parseDuration(s string) (time.Duration, error) { + if f, err := strconv.ParseFloat(s, 64); err == nil { + if f < 0 { + return 0, fmt.Errorf("negative duration %q", s) + } + return time.Duration(f * float64(time.Second)), nil + } + if s == "" { + return 0, fmt.Errorf("empty duration") + } + var total time.Duration + for rest := s; rest != ""; { + m := durationToken.FindStringSubmatch(rest) + if m == nil { + return 0, fmt.Errorf("invalid duration %q", s) + } + f, _ := strconv.ParseFloat(m[1], 64) + total += time.Duration(f * float64(durationUnits[m[2]])) + rest = rest[len(m[0]):] + } + return total, nil +} + +// parseTime accepts "now", unix seconds with an optional fraction, RFC3339, +// and a duration meaning that long before now (VictoriaLogs' "start=3h"). +// Anything else is refused rather than guessed, since a misread timestamp +// silently moves the window. +func parseTime(s string, now time.Time) (time.Time, error) { + if s == "now" { + return now, nil + } + if sec, frac, ok := strings.Cut(s, "."); isDigits(sec) && (!ok || isDigits(frac)) { + secs, err := strconv.ParseInt(sec, 10, 64) + if err != nil { + return time.Time{}, err + } + var nanos int64 + if ok { + frac = (frac + "000000000")[:9] + nanos, _ = strconv.ParseInt(frac, 10, 64) + } + return time.Unix(secs, nanos), nil + } + if t, err := time.Parse(time.RFC3339Nano, s); err == nil { + return t, nil + } + if d, err := parseDuration(s); err == nil { + return now.Add(-d), nil + } + return time.Time{}, fmt.Errorf("unsupported time %q", s) +} + +func isDigits(s string) bool { + if s == "" { + return false + } + for _, r := range s { + if r < '0' || r > '9' { + return false + } + } + return true +} diff --git a/internal/observabilityproxy/logsparams_test.go b/internal/observabilityproxy/logsparams_test.go new file mode 100644 index 000000000..1ebc03fd9 --- /dev/null +++ b/internal/observabilityproxy/logsparams_test.go @@ -0,0 +1,272 @@ +/* +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 ( + "net/http" + "net/url" + "sort" + "strings" + "testing" + "time" +) + +var testNow = time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) + +func testLimits() LogsLimits { + return LogsLimits{ + DefaultRange: time.Hour, + MaxRange: 7 * 24 * time.Hour, + MaxBuckets: 1000, + DefaultLimit: 1000, + MaxLimit: 5000, + MaxTailResume: 15 * time.Minute, + } +} + +const wantScope = `{kubernetes_namespace_name="tenant-a"}` + +func build(t *testing.T, endpoint string, in url.Values) url.Values { + t.Helper() + out, err := BuildLogsParams(endpoint, in, "tenant-a", testNow, testLimits()) + if err != nil { + t.Fatalf("BuildLogsParams(%s, %v) error: %v", endpoint, in, err) + } + return out +} + +func buildErr(t *testing.T, endpoint string, in url.Values) *HTTPError { + t.Helper() + out, err := BuildLogsParams(endpoint, in, "tenant-a", testNow, testLimits()) + if err == nil { + t.Fatalf("BuildLogsParams(%s, %v) = %v, want error", endpoint, in, out) + } + return err +} + +func TestBuildLogsParamsAllowlist(t *testing.T) { + allowed := map[string][]string{ + "query": {"query", "start", "end", "limit"}, + "hits": {"query", "start", "end", "step", "field", "fields_limit"}, + "stats_query_range": {"query", "start", "end", "step"}, + "field_values": {"query", "start", "end", "field", "limit", "filter"}, + "stream_field_values": {"query", "start", "end", "field", "limit", "filter"}, + "facets": {"query", "start", "end", "limit", "max_values_per_field", "max_value_len", "keep_const_fields"}, + "tail": {"query", "start", "end", "start_offset"}, + } + hostile := url.Values{ + "extra_filters": {`{kubernetes_namespace_name="tenant-b"}`}, + "extra_filters[]": {`kubernetes_namespace_name:~".*"`}, + "extra_stream_filters": {`{kubernetes_namespace_name=~".*"}`}, + "AccountID": {"7"}, + "ProjectID": {"7"}, + "tenant": {"tenant-b"}, + "offset": {"1h"}, + "refresh_interval": {"1ms"}, + "ignore_pipes": {"1"}, + "format": {"csv"}, + "timeout": {"1h"}, + "query": {"error"}, + "field": {"kubernetes_pod_name"}, + "filter": {"api"}, + "keep_const_fields": {"1"}, + "since": {"2026-11-04T11:59:00Z"}, + } + for endpoint, keys := range allowed { + t.Run(endpoint, func(t *testing.T) { + out := build(t, endpoint, hostile) + want := map[string]bool{"extra_stream_filters": true} + for _, k := range keys { + want[k] = true + } + for k := range out { + if !want[k] { + t.Errorf("forwarded non-allowlisted parameter %q", k) + } + } + if got := out["extra_stream_filters"]; len(got) != 1 || got[0] != wantScope { + t.Errorf("extra_stream_filters = %q, want exactly [%s]", got, wantScope) + } + if _, ok := out["extra_filters"]; ok { + t.Error("extra_filters forwarded") + } + }) + } +} + +func TestBuildLogsParamsRejects(t *testing.T) { + tests := []struct { + name string + endpoint string + in url.Values + code int + }{ + {"unknown endpoint", "tenant_ids", url.Values{}, http.StatusNotFound}, + {"stats_query is not proxied", "stats_query", url.Values{}, http.StatusNotFound}, + {"ignore_global_time_filter", "query", url.Values{"query": {`* options(ignore_global_time_filter=true)`}}, http.StatusBadRequest}, + {"ignore_global_time_filter in subquery", "hits", url.Values{"query": {`* | union (options(ignore_global_time_filter=true) error)`}}, http.StatusBadRequest}, + {"ignore_global_time_filter on tail", "tail", url.Values{"query": {`options(IGNORE_GLOBAL_TIME_FILTER=true) *`}}, http.StatusBadRequest}, + {"inverted range", "query", url.Values{"start": {"2026-11-04T11:00:00Z"}, "end": {"2026-11-04T10:00:00Z"}}, http.StatusBadRequest}, + {"inverted range on tail", "tail", url.Values{"start": {"2026-11-04T11:00:00Z"}, "end": {"2026-11-04T10:00:00Z"}}, http.StatusBadRequest}, + {"bad start", "query", url.Values{"start": {"yesterday"}}, http.StatusBadRequest}, + {"bad limit", "query", url.Values{"limit": {"-1"}}, http.StatusBadRequest}, + {"zero limit", "query", url.Values{"limit": {"0"}}, http.StatusBadRequest}, + {"bad step", "hits", url.Values{"step": {"0s"}}, http.StatusBadRequest}, + {"field required", "field_values", url.Values{}, http.StatusBadRequest}, + {"too many hits fields", "hits", url.Values{"field": strings.Fields("a b c d e f g h i j k")}, http.StatusBadRequest}, + {"since and start_offset", "tail", url.Values{"since": {"2026-11-04T11:59:00Z"}, "start_offset": {"1m"}}, http.StatusBadRequest}, + {"bad since", "tail", url.Values{"since": {"soon"}}, http.StatusBadRequest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if err := buildErr(t, tt.endpoint, tt.in); err.Code != tt.code { + t.Fatalf("code = %d (%s), want %d", err.Code, err.Message, tt.code) + } + }) + } +} + +// VictoriaLogs propagates extra_stream_filters into every subquery +// (docs.victoriametrics.com/victorialogs/querying/), so union and in() +// queries are forwarded verbatim with the scope beside them. +func TestBuildLogsParamsSubqueriesKeepScope(t *testing.T) { + for _, q := range []string{ + `* | union ({kubernetes_namespace_name="tenant-b"})`, + `_stream_id:in({kubernetes_namespace_name="tenant-b"} | fields _stream_id)`, + } { + out := build(t, "query", url.Values{"query": {q}}) + if out.Get("query") != q || out.Get("extra_stream_filters") != wantScope { + t.Errorf("query %q forwarded as %v", q, out) + } + } +} + +func TestBuildLogsParamsWindow(t *testing.T) { + ts := func(s string) string { + v, err := time.Parse(time.RFC3339, s) + if err != nil { + t.Fatal(err) + } + return v.UTC().Format(time.RFC3339Nano) + } + tests := []struct { + name string + in url.Values + wantStart, wantEnd string + }{ + {"defaults", url.Values{}, ts("2026-11-04T11:00:00Z"), ts("2026-11-04T12:00:00Z")}, + {"start only", url.Values{"start": {"2026-11-04T09:00:00Z"}}, ts("2026-11-04T09:00:00Z"), ts("2026-11-04T12:00:00Z")}, + {"end only", url.Values{"end": {"2026-11-03T00:00:00Z"}}, ts("2026-11-02T23:00:00Z"), ts("2026-11-03T00:00:00Z")}, + {"unix seconds", url.Values{"start": {"1793786400"}, "end": {"1793788200.5"}}, "2026-11-04T10:00:00Z", "2026-11-04T10:30:00.5Z"}, + {"relative", url.Values{"start": {"3h"}, "end": {"now"}}, ts("2026-11-04T09:00:00Z"), ts("2026-11-04T12:00:00Z")}, + {"relative days", url.Values{"start": {"1d"}}, ts("2026-11-03T12:00:00Z"), ts("2026-11-04T12:00:00Z")}, + {"over-wide clamped", url.Values{"start": {"2026-01-01T00:00:00Z"}}, ts("2026-10-28T12:00:00Z"), ts("2026-11-04T12:00:00Z")}, + {"empty range allowed", url.Values{"start": {"2026-11-04T10:00:00Z"}, "end": {"2026-11-04T10:00:00Z"}}, ts("2026-11-04T10:00:00Z"), ts("2026-11-04T10:00:00Z")}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + for _, endpoint := range []string{"query", "hits", "stats_query_range", "facets"} { + out := build(t, endpoint, tt.in) + if out.Get("start") != tt.wantStart || out.Get("end") != tt.wantEnd { + t.Errorf("%s: window = [%s, %s], want [%s, %s]", endpoint, out.Get("start"), out.Get("end"), tt.wantStart, tt.wantEnd) + } + } + }) + } +} + +func TestBuildLogsParamsStepAndLimit(t *testing.T) { + day := url.Values{"start": {"2026-11-03T12:00:00Z"}} + with := func(base url.Values, k, v string) url.Values { + c := url.Values{} + for key, vals := range base { + c[key] = vals + } + c.Set(k, v) + return c + } + // 24h over at most 1000 buckets is 86.4s. + if got := build(t, "hits", with(day, "step", "1s")).Get("step"); got != "86400ms" { + t.Errorf("tiny step = %s, want clamped to 86400ms", got) + } + if got := build(t, "stats_query_range", with(day, "step", "1h")).Get("step"); got != "3600000ms" { + t.Errorf("1h step = %s", got) + } + if got := build(t, "stats_query_range", with(day, "step", "1w")).Get("step"); got != "604800000ms" { + t.Errorf("1w step = %s", got) + } + if got := build(t, "hits", day).Get("step"); got != "86400ms" { + t.Errorf("omitted step = %s, want the bucket-bounded minimum", got) + } + if got := build(t, "hits", url.Values{"start": {"2026-11-04T11:59:59Z"}}).Get("step"); got != "1000ms" { + t.Errorf("omitted step on a tiny window = %s, want 1000ms floor", got) + } + + if got := build(t, "query", url.Values{}).Get("limit"); got != "1000" { + t.Errorf("default limit = %s", got) + } + if got := build(t, "query", url.Values{"limit": {"10"}}).Get("limit"); got != "10" { + t.Errorf("limit = %s", got) + } + for _, endpoint := range []string{"query", "field_values", "facets"} { + in := url.Values{"limit": {"1000000"}, "field": {"f"}} + if got := build(t, endpoint, in).Get("limit"); got != "5000" { + t.Errorf("%s: limit = %s, want clamped to 5000", endpoint, got) + } + } + facets := build(t, "facets", url.Values{"max_values_per_field": {"99999"}, "max_value_len": {"99999"}}) + if facets.Get("max_values_per_field") != "5000" || facets.Get("max_value_len") != "5000" { + t.Errorf("facets caps not clamped: %v", facets) + } + hits := build(t, "hits", url.Values{"field": {"b", "a"}, "fields_limit": {"99999"}}) + fields := hits["field"] + sort.Strings(fields) + if strings.Join(fields, ",") != "a,b" || hits.Get("fields_limit") != "5000" { + t.Errorf("hits fields = %v", hits) + } +} + +func TestBuildLogsParamsTail(t *testing.T) { + tests := []struct { + name string + in url.Values + want string + }{ + {"live only", url.Values{}, ""}, + // now - 5s tail offset - since = 55s + {"since within cap", url.Values{"since": {"2026-11-04T11:59:00Z"}}, "55000ms"}, + {"since older than cap", url.Values{"since": {"2026-11-04T08:00:00Z"}}, "900000ms"}, + {"since in the future", url.Values{"since": {"2026-11-04T13:00:00Z"}}, "0ms"}, + {"start_offset within cap", url.Values{"start_offset": {"2m"}}, "120000ms"}, + {"start_offset over cap", url.Values{"start_offset": {"30d"}}, "900000ms"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + out := build(t, "tail", tt.in) + if got := out.Get("start_offset"); got != tt.want { + t.Errorf("start_offset = %q, want %q", got, tt.want) + } + if out.Has("start") || out.Has("end") || out.Has("limit") { + t.Errorf("tail got a derived window or limit: %v", out) + } + }) + } + out := build(t, "tail", url.Values{"start": {"2026-11-04T11:00:00Z"}}) + if out.Get("start") != "2026-11-04T11:00:00Z" { + t.Errorf("tail start = %q, want the client value ANDed into each tick", out.Get("start")) + } +}