From e5537bc64d07ef22770a3dd07aeff8785ee0c0c4 Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Fri, 8 Nov 2024 15:47:14 +0100 Subject: [PATCH] lib/logstorage: properly take into account the `end` query arg when calculating time range for _time:duration filters --- app/vlselect/logsql/logsql.go | 29 +++++---- app/vmalert/config/types.go | 2 +- app/vmalert/rule/recording.go | 8 ++- docs/VictoriaLogs/CHANGELOG.md | 2 + lib/logstorage/parser.go | 111 +++++++++++++++++++++++++-------- lib/logstorage/parser_test.go | 19 ++++-- 6 files changed, 127 insertions(+), 44 deletions(-) diff --git a/app/vlselect/logsql/logsql.go b/app/vlselect/logsql/logsql.go index 0b4092e20..9cf15b7b7 100644 --- a/app/vlselect/logsql/logsql.go +++ b/app/vlselect/logsql/logsql.go @@ -962,14 +962,29 @@ func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID, } tenantIDs := []logstorage.TenantID{tenantID} + // Parse optional start and end args + start, okStart, err := getTimeNsec(r, "start") + if err != nil { + return nil, nil, err + } + end, okEnd, err := getTimeNsec(r, "end") + if err != nil { + return nil, nil, err + } + // Parse optional time arg timestamp, okTime, err := getTimeNsec(r, "time") if err != nil { return nil, nil, err } if !okTime { - // If time arg is missing, then evaluate query at the current timestamp - timestamp = time.Now().UnixNano() + // If time arg is missing, then evaluate query either at the end timestamp (if it is set) + // or at the current timestamp (if end query arg isn't set) + if okEnd { + timestamp = end + } else { + timestamp = time.Now().UnixNano() + } } // decrease timestamp by one nanosecond in order to avoid capturing logs belonging @@ -983,16 +998,8 @@ func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID, return nil, nil, fmt.Errorf("cannot parse query [%s]: %s", qStr, err) } - // Parse optional start and end args - start, okStart, err := getTimeNsec(r, "start") - if err != nil { - return nil, nil, err - } - end, okEnd, err := getTimeNsec(r, "end") - if err != nil { - return nil, nil, err - } if okStart || okEnd { + // Add _time:[start, end] filter if start or end args were set. if !okStart { start = math.MinInt64 } diff --git a/app/vmalert/config/types.go b/app/vmalert/config/types.go index b71325ec3..1cf286fe0 100644 --- a/app/vmalert/config/types.go +++ b/app/vmalert/config/types.go @@ -71,7 +71,7 @@ func (t *Type) ValidateExpr(expr string) error { return fmt.Errorf("bad prometheus expr: %q, err: %w", expr, err) } case "vlogs": - if _, err := logstorage.ParseStatsQuery(expr); err != nil { + if _, err := logstorage.ParseStatsQuery(expr, 0); err != nil { return fmt.Errorf("bad LogsQL expr: %q, err: %w", expr, err) } default: diff --git a/app/vmalert/rule/recording.go b/app/vmalert/rule/recording.go index 3de0524b5..7b2a7635e 100644 --- a/app/vmalert/rule/recording.go +++ b/app/vmalert/rule/recording.go @@ -10,6 +10,7 @@ import ( "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config" "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource" "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/logger" "github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage" "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal" ) @@ -221,6 +222,9 @@ func setIntervalAsTimeFilter(dType, expr string) bool { if dType != "vlogs" { return false } - q, _ := logstorage.ParseStatsQuery(expr) - return !q.ContainAnyTimeFilter() + q, err := logstorage.ParseStatsQuery(expr, 0) + if err != nil { + logger.Panicf("BUG: the LogsQL query must be valid here; got error: %s; query=[%s]", err, expr) + } + return !q.HasGlobalTimeFilter() } diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md index dd14ae036..5320b7fbb 100644 --- a/docs/VictoriaLogs/CHANGELOG.md +++ b/docs/VictoriaLogs/CHANGELOG.md @@ -18,6 +18,8 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta * FEATURE: [`join` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#join-pipe): add an ability to add prefix to all the log field names from the joined query, by using `| join by () () prefix "some_prefix"` syntax. * FEATURE: [`_time` filter](https://docs.victoriametrics.com/victorialogs/logsql/#time-filter): allow specifying offset without time range. For example, `_time:offset 1d` matches all the logs until `now-1d` in the [`_time` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field). This is useful when building graphs for time ranges with some offset in the past. +* BUGFIX: [HTTP querying APIs](https://docs.victoriametrics.com/victorialogs/querying/#http-api): properly take into account the `end` query arg when calculating time range for [`_time:duration` filter](https://docs.victoriametrics.com/victorialogs/logsql/#time-filter). Previously the `_time:duration` filter was treated as `_time:[now-duration, now)`, while it should be treated as `_time:[end-duration, end)`. + ## [v0.41.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.41.0-victorialogs) Released at 2024-11-06 diff --git a/lib/logstorage/parser.go b/lib/logstorage/parser.go index d753f835e..2f047c167 100644 --- a/lib/logstorage/parser.go +++ b/lib/logstorage/parser.go @@ -531,7 +531,7 @@ func (q *Query) AddPipeLimit(n uint64) { }) } -// optimize tries optimizing the query. +// optimize applies various optimations to q. func (q *Query) optimize() { q.pipes = optimizeSortOffsetPipes(q.pipes) q.pipes = optimizeSortLimitPipes(q.pipes) @@ -555,6 +555,12 @@ func (q *Query) optimize() { } } + // flatten nested AND filters + q.f = flattenFiltersAnd(q.f) + + // flatten nested OR filters + q.f = flattenFiltersOr(q.f) + // Substitute '*' prefixFilter with filterNoop in order to avoid reading _msg data. q.f = removeStarFilters(q.f) } @@ -734,6 +740,78 @@ func addByTimeField(byFields []*byStatsField, step int64) []*byStatsField { return dstFields } +func flattenFiltersAnd(f filter) filter { + visitFunc := func(f filter) bool { + fa, ok := f.(*filterAnd) + if !ok { + return false + } + for _, f := range fa.filters { + if _, ok := f.(*filterAnd); ok { + return true + } + } + return false + } + copyFunc := func(f filter) (filter, error) { + fa := f.(*filterAnd) + + var resultFilters []filter + for _, f := range fa.filters { + child, ok := f.(*filterAnd) + if !ok { + resultFilters = append(resultFilters, f) + continue + } + resultFilters = append(resultFilters, child.filters...) + } + return &filterAnd{ + filters: resultFilters, + }, nil + } + f, err := copyFilter(f, visitFunc, copyFunc) + if err != nil { + logger.Fatalf("BUG: unexpected error: %s", err) + } + return f +} + +func flattenFiltersOr(f filter) filter { + visitFunc := func(f filter) bool { + fo, ok := f.(*filterOr) + if !ok { + return false + } + for _, f := range fo.filters { + if _, ok := f.(*filterOr); ok { + return true + } + } + return false + } + copyFunc := func(f filter) (filter, error) { + fo := f.(*filterOr) + + var resultFilters []filter + for _, f := range fo.filters { + child, ok := f.(*filterOr) + if !ok { + resultFilters = append(resultFilters, f) + continue + } + resultFilters = append(resultFilters, child.filters...) + } + return &filterOr{ + filters: resultFilters, + }, nil + } + f, err := copyFilter(f, visitFunc, copyFunc) + if err != nil { + logger.Fatalf("BUG: unexpected error: %s", err) + } + return f +} + func removeStarFilters(f filter) filter { // Substitute `*` filterPrefix with filterNoop visitFunc := func(f filter) bool { @@ -915,9 +993,9 @@ func ParseQuery(s string) (*Query, error) { return ParseQueryAtTimestamp(s, timestamp) } -// ParseStatsQuery parses s with needed stats query checks. -func ParseStatsQuery(s string) (*Query, error) { - q, err := ParseQuery(s) +// ParseStatsQuery parses LogsQL query s at the given timestamp with the needed stats query checks. +func ParseStatsQuery(s string, timestamp int64) (*Query, error) { + q, err := ParseQueryAtTimestamp(s, timestamp) if err != nil { return nil, err } @@ -927,29 +1005,12 @@ func ParseStatsQuery(s string) (*Query, error) { return q, nil } -// ContainAnyTimeFilter returns true when query contains a global time filter. -func (q *Query) ContainAnyTimeFilter() bool { - if hasTimeFilter(q.f) { - return true - } - for _, p := range q.pipes { - if pf, ok := p.(*pipeFilter); ok { - if hasTimeFilter(pf.f) { - return true - } - } - } - return false -} - -func hasTimeFilter(f filter) bool { - if f == nil { - return false - } - switch t := f.(type) { +// HasGlobalTimeFilter returns true when query contains a global time filter. +func (q *Query) HasGlobalTimeFilter() bool { + switch t := q.f.(type) { case *filterAnd: for _, subF := range t.filters { - if hasTimeFilter(subF) { + if _, ok := subF.(*filterTime); ok { return true } } diff --git a/lib/logstorage/parser_test.go b/lib/logstorage/parser_test.go index 5f13fdfff..37b67e777 100644 --- a/lib/logstorage/parser_test.go +++ b/lib/logstorage/parser_test.go @@ -700,6 +700,14 @@ func TestParseQuerySuccess(t *testing.T) { f("level: ( ((error or warn*) and re(foo))) (not (bar))", `(level:error or level:warn*) level:~foo !bar`) f("!(foo bar or baz and not aa*)", `!(foo bar or baz !aa*)`) + // nested AND filters + f(`(foo AND bar) AND (baz AND x:y)`, `foo bar baz x:y`) + f(`(foo AND bar) OR (baz AND x:y)`, `foo bar or baz x:y`) + + // nested OR filters + f(`(foo OR bar) OR (baz OR x:y)`, `foo or bar or baz or x:y`) + f(`(foo OR bar) AND (baz OR x:y)`, `(foo or bar) (baz or x:y)`) + // prefix search f(`'foo'* and (a:x* and x:* or y:i(""*)) and i("abc def"*)`, `foo* (a:x* x:* or y:i(*)) i("abc def"*)`) @@ -2421,16 +2429,17 @@ func TestQueryGetStatsByFields_Failure(t *testing.T) { f(`* | by (x) count() y | format 'foo' as y`) } -func TestHasTimeFilter(t *testing.T) { - f := func(qStr string, expected bool) { +func TestQueryHasGlobalTimeFilter(t *testing.T) { + f := func(qStr string, resultExpected bool) { t.Helper() - q, err := ParseStatsQuery(qStr) + q, err := ParseStatsQuery(qStr, 0) if err != nil { t.Fatalf("cannot parse [%s]: %s", qStr, err) } - if q.ContainAnyTimeFilter() != expected { - t.Fatalf("unexpected result for hasTimeFilter(%q); want %v", qStr, expected) + result := q.HasGlobalTimeFilter() + if result != resultExpected { + t.Fatalf("unexpected result for hasTimeFilter(%q); got %v; want %v", qStr, result, resultExpected) } }