From 2c9087f5f3327e3e4bcdc516e8b297b89bdc982a Mon Sep 17 00:00:00 2001 From: Vadim Alekseev Date: Mon, 1 Jun 2026 18:40:54 +0400 Subject: [PATCH] lib/logstorage: introduce VictoriaLogs transformations language Updates #858 --- app/vlagent/remotewrite/remotewrite.go | 139 ++- docs/victorialogs/CHANGELOG.md | 1 + .../{vlagent.md => vlagent/README.md} | 19 +- docs/victorialogs/vlagent/_index.md | 16 + docs/victorialogs/vlagent/transformations.md | 530 ++++++++ lib/logstorage/block_result.go | 79 +- lib/logstorage/log_rows.go | 82 ++ lib/logstorage/transforms.go | 1090 +++++++++++++++++ lib/logstorage/transforms_test.go | 785 ++++++++++++ lib/logstorage/transforms_timing_test.go | 187 +++ 10 files changed, 2906 insertions(+), 22 deletions(-) rename docs/victorialogs/{vlagent.md => vlagent/README.md} (99%) create mode 100644 docs/victorialogs/vlagent/_index.md create mode 100644 docs/victorialogs/vlagent/transformations.md create mode 100644 lib/logstorage/transforms.go create mode 100644 lib/logstorage/transforms_test.go create mode 100644 lib/logstorage/transforms_timing_test.go diff --git a/app/vlagent/remotewrite/remotewrite.go b/app/vlagent/remotewrite/remotewrite.go index 1a733b4b64..6a48079d8e 100644 --- a/app/vlagent/remotewrite/remotewrite.go +++ b/app/vlagent/remotewrite/remotewrite.go @@ -4,17 +4,23 @@ import ( "flag" "fmt" "net/url" + "os" "path/filepath" + "sort" + "strings" "sync" "sync/atomic" "github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/envtemplate" "github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil" "github.com/VictoriaMetrics/VictoriaMetrics/lib/fs" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/fs/fscore" "github.com/VictoriaMetrics/VictoriaMetrics/lib/logger" "github.com/VictoriaMetrics/VictoriaMetrics/lib/memory" "github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue" "github.com/VictoriaMetrics/metrics" + "github.com/bmatcuk/doublestar/v4" "github.com/cespare/xxhash/v2" "github.com/VictoriaMetrics/VictoriaLogs/app/vlstorage/netinsert" @@ -33,6 +39,14 @@ var ( format = flagutil.NewArrayString("remoteWrite.format", "The data format to use for sending data to the corresponding -remoteWrite.url. "+ "Available formats: native, jsonline. Default is native. See https://docs.victoriametrics.com/victorialogs/vlagent/#remote-write-format") + transforms = flag.String("remoteWrite.transforms", "", "Path to transformations program, which is applied "+ + "to all the logs before sending them to -remoteWrite.url. See also -remoteWrite.urlTransforms. "+ + "The path can be a glob pattern pointing to multiple files, http url or inline content with prefix 'inline:'. "+ + "See https://docs.victoriametrics.com/victorialogs/vlagent/transformations/") + urlTransforms = flagutil.NewArrayString("remoteWrite.urlTransforms", "Path to transformations program for the corresponding -remoteWrite.url. "+ + "See also -remoteWrite.transforms. The path can be a glob pattern pointing to multiple files, http url or inline content with prefix 'inline:'. "+ + "See https://docs.victoriametrics.com/victorialogs/vlagent/transformations/") + remoteWriteTmpDataPath = flag.String("remoteWrite.tmpDataPath", "", "Path to directory for storing pending data, which isn't sent to the configured -remoteWrite.url . "+ "if this flag isn't set, then pending data is stored in the vlagent-remotewrite-data subdirectory under the -tmpDataPath directory; "+ "see also -remoteWrite.maxDiskUsagePerURL") @@ -47,12 +61,19 @@ var ( // rwctxsGlobal contains statically populated entries when -remoteWrite.url is specified. var rwctxsGlobal []*remoteWriteCtx +var globalTransformer *logstorage.Transformer + // Storage implements insertutil.LogRowsStorage interface type Storage struct{} // MustAddRows implements insertutil.LogRowsStorage interface func (*Storage) MustAddRows(lr *logstorage.LogRows) { - pushToRemoteStorages(lr) + if tr := globalTransformer; tr != nil { + // globalTransformer will flush the result to pushToRemoteStorages. + tr.Transform(lr) + } else { + pushToRemoteStorages(lr) + } } // CanWriteData implements insertutil.LogRowsStorage interface @@ -98,6 +119,7 @@ func Init(tmpDataPath string) { if len(path) == 0 { path = filepath.Join(tmpDataPath, "vlagent-remotewrite-data") } + initGlobalTransformer() initRemoteWriteCtxs(path, *remoteWriteURLs) dropDanglingQueues(path) } @@ -112,6 +134,93 @@ func Stop() { rwctxsGlobal = nil } +func initGlobalTransformer() { + if s := *transforms; s != "" { + globalTransformer = loadTransforms(s, pushToRemoteStorages) + } +} + +func loadTransforms(s string, flush func(lr *logstorage.LogRows)) *logstorage.Transformer { + switch { + case strings.HasPrefix(s, "inline:"): + content := strings.TrimPrefix(s, "inline:") + content = envtemplate.ReplaceString(content) + prog, err := logstorage.ParseTransformsProgram(content) + if err != nil { + logger.Fatalf("FATAL: failed to parse inline transformations: %s", err) + } + tr := prog.NewTransformer(flush) + return tr + case isHTTPURL(s): + content, err := fscore.ReadFileOrHTTP(s) + if err != nil { + logger.Fatalf("FATAL: cannot read transformations: %s", err) + } + content = envtemplate.ReplaceBytes(content) + prog, err := logstorage.ParseTransformsProgram(string(content)) + if err != nil { + if len(content) > 4*1024 { + content = content[:4*1024] + content = append(content, "..."...) + } + logger.Fatalf("FATAL: failed to parse transformations by URL %q: %s; content: %q", s, err, content) + } + tr := prog.NewTransformer(flush) + return tr + default: + var globOpts = []doublestar.GlobOption{ + // Follow traditional shell glob behavior where `*` or a `?` at the start will not match dotfiles by default. + // Users can explicitly use `.*` or `.?` syntax to collect logs from the hidden files. + doublestar.WithNoHidden(), + } + files, err := doublestar.FilepathGlob(s, globOpts...) + if err != nil { + logger.Fatalf("FATAL: cannot process glob pattern %q: %s", s, err) + } + if len(files) == 0 { + logger.Fatalf("FATAL: no files found by glob pattern %q", s) + } + sort.Strings(files) + filesContent := make([]string, len(files)) + for i, file := range files { + data, err := os.ReadFile(file) + if err != nil { + logger.Fatalf("FATAL: cannot read file with transformations: %s", err) + } + data = envtemplate.ReplaceBytes(data) + filesContent[i] = string(data) + } + + // Handle the first file. + firstFile := files[0] + files = files[1:] + firstFileContent := filesContent[0] + filesContent = filesContent[1:] + prog, err := logstorage.ParseTransformsProgram(firstFileContent) + if err != nil { + logger.Fatalf("FATAL: cannot parse transformations program by path %q: %s", firstFile, err) + } + + // Parse additional content. + for i, file := range files { + content := filesContent[i] + if err := prog.ParseAdditional(content); err != nil { + logger.Fatalf("FATAL: cannot parse transformations program by path %q: %s", file, err) + } + } + + tr := prog.NewTransformer(flush) + return tr + } +} + +// isHTTPURL checks if a given targetURL is valid and contains a valid http scheme. +// Copied from fscore.ReadFileOrHTTP. +func isHTTPURL(targetURL string) bool { + parsed, err := url.Parse(targetURL) + return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") && parsed.Host != "" +} + func dropDanglingQueues(tmpDataPath string) { // Remove dangling persistent queues, if any. // This is required for the case when the number of queues has been changed or URL have been changed. @@ -146,6 +255,10 @@ func initRemoteWriteCtxs(tmpDataPath string, urls []string) { if len(urls) == 0 { logger.Panicf("BUG: urls must be non-empty") } + if len(urls) < len(*urlTransforms) { + logger.Fatalf("FATAL: the number of specified -remoteWrite.urlTransforms flags (%d) exceeds the number of specified -remoteWrite.url flags (%d); "+ + "use glob patterns to specify multiple transformation files for a single -remoteWrite.url", len(*urlTransforms), len(urls)) + } maxInmemoryBlocks := memory.Allowed() / len(urls) / 10000 if maxInmemoryBlocks / *queues > 100 { @@ -179,12 +292,11 @@ func initRemoteWriteCtxs(tmpDataPath string, urls []string) { func pushToRemoteStorages(lr *logstorage.LogRows) { rwctxs := rwctxsGlobal if len(rwctxs) == 1 { - // fast path + // Fast path: there is only one remote storage system. rwctxs[0].push(lr) return } - // Push samples to remote storage systems in parallel in order to reduce - // the time needed for sending the data to multiple remote storage systems. + // Slow path: push lr to remote storage systems in parallel. var wg sync.WaitGroup for _, rwctx := range rwctxs { wg.Go(func() { @@ -199,6 +311,8 @@ type remoteWriteCtx struct { fq *persistentqueue.FastQueue c *client + transformer *logstorage.Transformer + pls []*pendingLogs pssNextIdx atomic.Uint64 } @@ -276,10 +390,27 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in pls: pls, } + // Do not use the GetOptionalArg method here + // because it returns the same value regardless of argIdx if the flag is set once. + urlTrs := *urlTransforms + if argIdx < len(urlTrs) && urlTrs[argIdx] != "" { + tr := loadTransforms(urlTrs[argIdx], rwctx.pushInternal) + rwctx.transformer = tr + } + return rwctx } func (rwctx *remoteWriteCtx) push(lr *logstorage.LogRows) { + if tr := rwctx.transformer; tr != nil { + // tr will flush the result to pushInternal. + tr.Transform(lr) + } else { + rwctx.pushInternal(lr) + } +} + +func (rwctx *remoteWriteCtx) pushInternal(lr *logstorage.LogRows) { pls := rwctx.pls idx := rwctx.pssNextIdx.Add(1) % uint64(len(pls)) pls[idx].add(lr) diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 0a31708495..f3e05635c5 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/docs/victorialogs/CHANGELOG.md @@ -33,6 +33,7 @@ according to the following docs: * FEATURE: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): add [`json_array_concat` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#json_array_concat-pipe) for joining JSON array items stored in the given [log field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) into a string with the given delimiter. See [#712](https://github.com/VictoriaMetrics/VictoriaLogs/issues/712). * FEATURE: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/): adjust [`_stream` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) value according to the actual log fields. This simplifies importing of previously modified VictoriaLogs-exported data. See [#1122](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1122). * FEATURE: [vlagent](https://docs.victoriametrics.com/victorialogs/vlagent/): add `-remoteWrite.basicAuth.usernameFile` command-line flag for dynamically reloading basic auth username for the corresponding `-remoteWrite.url` from the given file. +* FEATURE: [vlagent](https://docs.victoriametrics.com/victorialogs/vlagent/transforms/): add ability to transform incoming logs via VictoriaLogs Transformations language. The language is based on [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/) which supports conditional execution, dropping and routing of logs. See [#858](https://github.com/VictoriaMetrics/VictoriaLogs/issues/858) and [these docs](https://docs.victoriametrics.com/victorialogs/vlagent/transformations/) for details. * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): properly select the top `N` log entries in the [`json_values`](https://docs.victoriametrics.com/victorialogs/logsql/#json_values-stats) stats function with `sort by (...) limit N` when more than `N` logs match per group. Previously an arbitrary subset of the matching logs could be returned instead of the top `N` ones. The returned entries were correctly sorted, but they could be the wrong entries. See [#1311](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1311). * BUGFIX: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): avoid stamping [RFC3164](https://datatracker.ietf.org/doc/html/rfc3164) messages received right after the new year with the previous year (which could drop them under a short `-retentionPeriod`), and compute the year in `-syslog.timezone` instead of the server local timezone. See [#1556](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1556). diff --git a/docs/victorialogs/vlagent.md b/docs/victorialogs/vlagent/README.md similarity index 99% rename from docs/victorialogs/vlagent.md rename to docs/victorialogs/vlagent/README.md index 347a66ef47..1ea7b6c16f 100644 --- a/docs/victorialogs/vlagent.md +++ b/docs/victorialogs/vlagent/README.md @@ -1,17 +1,10 @@ --- -title: vlagent -weight: 4 -menu: - docs: - weight: 4 - parent: "victorialogs" - identifier: vlagent -tags: - - logs -aliases: - - /vlagent.html - - /vlagent/index.html - - /vlagent/ +build: + list: never + publishResources: false + render: never +sitemap: + disable: true --- `vlagent` is an agent for collecting logs from various sources and storing them in [VictoriaLogs](https://docs.victoriametrics.com/victorialogs/). diff --git a/docs/victorialogs/vlagent/_index.md b/docs/victorialogs/vlagent/_index.md new file mode 100644 index 0000000000..ff130fcbbc --- /dev/null +++ b/docs/victorialogs/vlagent/_index.md @@ -0,0 +1,16 @@ +--- +title: vlagent +weight: 4 +menu: + docs: + weight: 4 + parent: "victorialogs" + identifier: vlagent +tags: + - logs +aliases: + - /vlagent.html + - /vlagent/index.html + - /vlagent/ +--- +{{% content "README.md" %}} diff --git a/docs/victorialogs/vlagent/transformations.md b/docs/victorialogs/vlagent/transformations.md new file mode 100644 index 0000000000..e1069bc7e2 --- /dev/null +++ b/docs/victorialogs/vlagent/transformations.md @@ -0,0 +1,530 @@ +--- +weight: 1 +title: Transformations +menu: + docs: + parent: "vlagent" + weight: 1 +tags: + - logs +aliases: + - /victorialogs/vlagent/transformations/ + - /victorialogs/vlagent/transformations.html +--- + +`vlagent` can transform logs before sending them to remote storage. +Use it to normalize fields, parse unstructured messages, enrich records, and route logs based on their content. + +Transformations are written in the VictoriaLogs transformations language, which is built on top of [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/). +LogsQL queries data that is already in storage, while transformations process logs line-by-line before they reach remote storage. + +The transformation language adds control flow on top of LogsQL. +You can declare reusable transformation blocks, process logs conditionally, and drop unwanted data independently for each remote storage destination. +You can also override the [`_stream`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) field or change the target tenant based on log content. + +## Quick start + +To start transforming logs, pass the `-remoteWrite.transforms` flag with one +or more [supported](https://docs.victoriametrics.com/victorialogs/vlagent/transformations/#supported-pipes) LogsQL pipes. +For example: + +```sh +./vlagent -remoteWrite.url=http://victorialogs:9428/insert/native \ + -remoteWrite.transforms="inline: extract 'my service name is: ' from _msg | set_stream_fields service;" +``` + +This extracts the service name from the `_msg` field and uses it as the [`_stream`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) field. + +An incoming log like this: + +```json +{ + "_msg": "my service name is: payment-api" +} +``` + +is sent to VictoriaLogs as: + +```json +{ + "_msg": "my service name is payment-api", + "_stream": "{service=\"payment-api\"}", + "_time": "...", + "service": "payment-api" +} +``` + +To load transformations from a file instead, pass its path to `-remoteWrite.transforms`. +For example, create `/tmp/transforms.vlt` with the following content: + +```vlt +coalesce(level, log.level, LEVEL) as level; + +if (level:"") { + # The 'level' field is empty or not set - try to analyze the message content. + if (_msg:i(error)) { + # '_msg' field contains the word 'error' + format error as level; + } else { + format unknown as level; + } +} +``` + +Then run `vlagent` with the path to the file: + +```sh +./vlagent -remoteWrite.url=http://victorialogs:9428/insert/native \ + -remoteWrite.transforms="/tmp/transforms.vlt" +``` + +`vlagent` sets the `level` field for each log based on its content. +First it takes the first non-empty value among `level`, `log.level`, and `LEVEL` and writes it to `level`. +Then, if `level` is still empty and the message contains the word `error`, it sets `level` to `error`, otherwise it sets it to `unknown`. + +For more detail, see the [syntax description](https://docs.victoriametrics.com/victorialogs/vlagent/transformations/#syntax) +and ready-to-use [log transformation examples](https://docs.victoriametrics.com/victorialogs/vlagent/transformations/#examples). + +## Syntax + +A transformation program is a sequence of statements, one per line. +A statement can be: + +* a pipes line: one or more LogsQL pipes connected via `|`. +* an `if` condition. +* a `block` declaration or its invocation via `do`. +* a control statement: `return`, `send`, or `drop`. + +A pipes line, `return`, `send`, and `drop` must end with `;`. + +Each statement is described below. + +### Pipes line + +Multiple pipes in a single line are applied left to right: + +```vlt +unpack_json | rename log.level as level | delete _msg | rename message as _msg; +``` + +This is equivalent to a line-by-line definition: + +```vlt +unpack_json; +rename log.level as level; +delete _msg; +rename message as _msg; +``` + +Any supported [LogsQL](https://docs.victoriametrics.com/victorialogs/vlagent/transformations/#supported-pipes) pipe can be used here. + +### Conditional execution + +`if` applies nested statements only to logs that match the filter. +The filter can be any [LogsQL filter](https://docs.victoriametrics.com/victorialogs/logsql/#filters): +an [exact filter](https://docs.victoriametrics.com/victorialogs/logsql/#exact-filter), +a [word match filter](https://docs.victoriametrics.com/victorialogs/logsql/#word-filter), +a [prefix filter](https://docs.victoriametrics.com/victorialogs/logsql/#prefix-filter), +an [IP filter](https://docs.victoriametrics.com/victorialogs/logsql/#ipv4-range-filter), and others. +Logs that do not match pass through unchanged. + +```vlt +if (level:=error) { + unpack_json; +} +``` + +`else if` chains and a trailing `else` are supported: + +```vlt +if (level:=error) { + format error as kind; +} else if (level:=warn) { + format warning as kind; +} else { + format other as kind; +} +``` + +### Named blocks + +A reusable set of statements can be extracted into a named block and invoked with `do`: + +```vlt +block normalize { + unpack_json; + rename level as severity; +} + +do normalize; +``` + +Blocks are declared at the top level of the program. +A block cannot be declared inside another block or inside an `if` statement. +Recursive block invocations are forbidden, even when executed conditionally. + +A block is only visible within the file where it is declared. +You cannot invoke a block declared in another file. + +### Control statements + +#### `send` + +`send` sends the current log downstream - to the per-URL transformations (if any) +and then to each `-remoteWrite.url` - and stops further processing by the remaining statements. +Use it to handle transformations separately for each service: + +```vlt +if (service:=payment) { + unpack_logfmt; + rename MESSAGE as _msg; + rename log.level as level; + send; +} +if (service:=checkout) { + unpack_json; + send; +} + +format 'unknown service' as transforms_status; +``` + +In the program above, logs from the `payment` and `checkout` services are sent downstream right away +and never reach the `format 'unknown service' as transforms_status` statement. +You can call `send` inside a `block` too, where it works the same way. + +#### `return` + +`return` stops the current block and returns log processing to the statements following `do`. +Use it for an early exit from a block: + +```vlt +block normalize_errors { + if (not level:=error) { + return; + } + format critical as severity; +} + +do normalize_errors; +format processed as status; +``` + +A log whose `level` is not `error` exits the `normalize_errors` block right away, +so the `format critical as severity` pipe is not applied to it. +Every log still reaches the `format processed as status` pipe, regardless of its `level` field. + +#### `drop` + +`drop` discards the current log, and subsequent pipes do not process it. +For example, to ignore all logs with the `debug` level: + +```vlt +if (level:=debug) { + # Drop all debug logs. + drop; +} +``` + +## Supported pipes + +VictoriaLogs transformations support a subset of LogsQL pipes. +Aggregating pipes such as `stats`, `sort`, `top`, `uniq`, `last`, and `facets` are not supported, because each log is processed on its own. +The same goes for pipes like `limit`, `offset`, `first`, and `filter`, which only make sense inside a query. + +The supported pipes are listed below: + +| Pipe | Purpose | Documentation | +|-----------------------|----------------------------------------------|---------------------------------------------------------------------------------------------------| +| `coalesce` | first non-empty value from a list of fields | [coalesce](https://docs.victoriametrics.com/victorialogs/logsql/#coalesce-pipe) | +| `collapse_nums` | collapsing numeric sequences into a template | [collapse_nums](https://docs.victoriametrics.com/victorialogs/logsql/#collapse_nums-pipe) | +| `copy`, `cp` | copying fields | [copy](https://docs.victoriametrics.com/victorialogs/logsql/#copy-pipe) | +| `decolorize` | removing ANSI color codes | [decolorize](https://docs.victoriametrics.com/victorialogs/logsql/#decolorize-pipe) | +| `delete`, `del`, `rm` | deleting fields | [delete](https://docs.victoriametrics.com/victorialogs/logsql/#delete-pipe) | +| `drop_empty_fields` | removing empty fields | [drop_empty_fields](https://docs.victoriametrics.com/victorialogs/logsql/#drop_empty_fields-pipe) | +| `extract` | extracting fields by template | [extract](https://docs.victoriametrics.com/victorialogs/logsql/#extract-pipe) | +| `extract_regexp` | extracting fields by regular expression | [extract_regexp](https://docs.victoriametrics.com/victorialogs/logsql/#extract_regexp-pipe) | +| `math`, `eval` | arithmetic calculations on fields | [math](https://docs.victoriametrics.com/victorialogs/logsql/#math-pipe) | +| `fields`, `keep` | keep only specified fields | [fields](https://docs.victoriametrics.com/victorialogs/logsql/#fields-pipe) | +| `format` | formatting a field by template | [format](https://docs.victoriametrics.com/victorialogs/logsql/#format-pipe) | +| `hash` | hashing field values | [hash](https://docs.victoriametrics.com/victorialogs/logsql/#hash-pipe) | +| `json_array_len` | JSON array length in a field | [json_array_len](https://docs.victoriametrics.com/victorialogs/logsql/#json_array_len-pipe) | +| `len` | field value length | [len](https://docs.victoriametrics.com/victorialogs/logsql/#len-pipe) | +| `rename`, `mv` | renaming fields | [rename](https://docs.victoriametrics.com/victorialogs/logsql/#rename-pipe) | +| `pack_json` | packing fields into JSON | [pack_json](https://docs.victoriametrics.com/victorialogs/logsql/#pack_json-pipe) | +| `pack_logfmt` | packing fields into logfmt | [pack_logfmt](https://docs.victoriametrics.com/victorialogs/logsql/#pack_logfmt-pipe) | +| `replace` | replacing substrings in fields | [replace](https://docs.victoriametrics.com/victorialogs/logsql/#replace-pipe) | +| `replace_regexp` | replacing by regular expression | [replace_regexp](https://docs.victoriametrics.com/victorialogs/logsql/#replace_regexp-pipe) | +| `sample` | log stream downsampling (sampling) | [sample](https://docs.victoriametrics.com/victorialogs/logsql/#sample-pipe) | +| `set_stream_fields` | overriding the `_stream` field | [set_stream_fields](https://docs.victoriametrics.com/victorialogs/logsql/#set_stream_fields-pipe) | +| `split` | splitting a field value | [split](https://docs.victoriametrics.com/victorialogs/logsql/#split-pipe) | +| `time_add` | shifting log time | [time_add](https://docs.victoriametrics.com/victorialogs/logsql/#time_add-pipe) | +| `unpack_json` | parsing JSON into fields | [unpack_json](https://docs.victoriametrics.com/victorialogs/logsql/#unpack_json-pipe) | +| `unpack_logfmt` | parsing logfmt into fields | [unpack_logfmt](https://docs.victoriametrics.com/victorialogs/logsql/#unpack_logfmt-pipe) | +| `unpack_syslog` | parsing syslog into fields | [unpack_syslog](https://docs.victoriametrics.com/victorialogs/logsql/#unpack_syslog-pipe) | +| `unpack_words` | parsing a value into words | [unpack_words](https://docs.victoriametrics.com/victorialogs/logsql/#unpack_words-pipe) | +| `unroll` | unrolling an array into separate logs | [unroll](https://docs.victoriametrics.com/victorialogs/logsql/#unroll-pipe) | + +## Configuration + +`-remoteWrite.transforms` applies global transformations to all logs before they are sent to every `-remoteWrite.url`: + +```sh +./vlagent -remoteWrite.transforms=transforms.vlt +``` + +`-remoteWrite.urlTransforms` applies transformations to a single `-remoteWrite.url`. +The flag is matched to `-remoteWrite.url` by position. +The first `-remoteWrite.urlTransforms` goes with the first `-remoteWrite.url`, the second with the second, and so on: + +```sh +./vlagent \ + -remoteWrite.transforms=global.vlt \ + -remoteWrite.url=http://victoria-logs-hot:9428/insert/native \ + -remoteWrite.urlTransforms=transforms-hot.vlt \ + -remoteWrite.url=http://victoria-logs-cold:9428/insert/native \ + -remoteWrite.urlTransforms=transforms-cold.vlt +``` + +To skip per-URL transformations for a specific `-remoteWrite.url`, pass an empty value: `-remoteWrite.urlTransforms=""`. + +When both global and per-URL transformations are set, the global one runs first, +and its result is passed to the per-URL transformations of each matching `-remoteWrite.url`. + +### Transformation sources + +The value of `-remoteWrite.transforms` and `-remoteWrite.urlTransforms` can be one of the following: + +- A path to a single file, for example `-remoteWrite.transforms=/etc/vlagent/transforms.vlt`. +- A glob pattern matching one or more files, for example `-remoteWrite.transforms=/etc/vlagent/*.vlt`. + Matched files are sorted lexicographically and applied in that order. + For example, files `01.vlt`, `03.vlt`, `02.vlt` are applied as `01.vlt`, `02.vlt`, `03.vlt`. +- An `http` or `https` URL, for example `-remoteWrite.transforms=http://config-server/transforms.vlt`. +- An inline program prefixed with `inline:`, for example `-remoteWrite.transforms="inline: unpack_json | delete _msg;"`. + +### Environment variables substitution + +A transformation program can reference environment variables with the `%{VAR_NAME}` syntax. +Values are substituted into the program text as-is, without escaping, so they can include transformations language syntax. + +For example: + +```vlt +if (service:in(%{IGNORED_SERVICES})) { + drop; +} +``` + +With this environment variable: + +```sh +IGNORED_SERVICES="payment,checkout" +``` + +the program expands into: + +```vlt +if (service:in(payment,checkout)) { + drop; +} +``` + +Environment variables cannot be changed while `vlagent` is running - restart `vlagent` to apply changes. + +See [these docs](https://docs.victoriametrics.com/victoriametrics/#environment-variables) for more details. + +## Examples + +### Dynamic setting of the `_stream` field + +```vlt +unpack_logfmt from _msg; +set_stream_fields service, level, host; +``` + +### Parsing unstructured logs + +Suppose an application writes logs in this format: + +```json +{ + "_msg": "User login successful for john.doe@example.com at 2025-05-30T15:10:22Z" +} +``` + +To turn it into a structured record, use the `extract` pipe: + +```vlt +extract "User login successful for at " from _msg; + +if (email:* and login_timestamp:*) { + # Content was extracted - override the _msg field. + format "User login successful" as _msg; +} +``` + +This produces the following log: + +```json +{ + "_msg": "User login successful", + "email": "john.doe@example.com", + "login_timestamp": "2025-05-30T15:10:22Z" +} +``` + +### Log level normalization + +The program below stores the log level in a single `level` field. +If the level is missing, it infers it from the log message content. + +```vlt +do normalize_log_level; + +block normalize_log_level { + coalesce(level, LEVEL, lvl, log.level, severity, severity_text, SeverityText) as level; + delete LEVEL, lvl, log.level, severity, severity_text, SeverityText; + + if (level:"") { + # The 'level' field is empty or not set - try to analyze the message content. + if (i(error)) { + # '_msg' field contains the word 'error' + format error as level; + } else { + format unknown as level; + } + } +} +``` + +### Routing logs per destination + +Each `-remoteWrite.urlTransforms` filters its destination independently, +so you can split logs between storages by dropping the unwanted ones per URL. + +For example, send audit logs to a dedicated `audit` storage and everything else to the main `hot` storage. +The two programs are mirror images of each other: + +```vlt +# hot.vlt - everything except audit logs +if (event_type:=audit) { + drop; +} +``` + +```vlt +# audit.vlt - audit logs only +if (not event_type:=audit) { + drop; +} +``` + +```sh +./vlagent \ + -remoteWrite.url=http://victorialogs-hot:9428/insert/native \ + -remoteWrite.urlTransforms="hot.vlt" \ + -remoteWrite.url=http://victorialogs-audit:9428/insert/native \ + -remoteWrite.urlTransforms="audit.vlt" +``` + +`hot` drops audit logs and keeps the rest, while `audit` drops everything that is not an audit log. +Because the two filters are opposite, each log lands in exactly one storage. + +### Dropping logs for a single destination + +You can send the same logs to several storages and drop only a part of them for one destination. +This is useful when one storage does not need the full volume. + +For example, keep all logs in `hot`, but skip debug logs in `cold` to save space: + +```vlt +# cold.vlt +if (level:=debug) { + drop; +} +``` + +```sh +./vlagent \ + -remoteWrite.url=http://victorialogs-hot:9428/insert/native \ + -remoteWrite.urlTransforms="" \ + -remoteWrite.url=http://victorialogs-cold:9428/insert/native \ + -remoteWrite.urlTransforms="cold.vlt" +``` + +`hot` has no per-URL program, so it gets every log, including debug ones. +`cold` drops debug logs and keeps everything else. +A non-debug log goes to both storages - unlike [logs routing](https://docs.victoriametrics.com/victorialogs/vlagent/transformations/#routing-logs-per-destination), the destinations overlap. + +### Dynamic tenant assignment + +During transformations, `vlagent` provides the `vl_account_id` and `vl_project_id` fields. +They set the [tenant](https://docs.victoriametrics.com/victorialogs/#multitenancy) that the log is written to in VictoriaLogs. + +To set the tenant dynamically, use the `/insert/multitenant/native` endpoint instead of `/insert/native`, +as described in the [vlagent multitenancy](https://docs.victoriametrics.com/victorialogs/vlagent/#multitenancy) docs. + +Override `vl_account_id` and `vl_project_id` based on log content to change the target tenant. +For example: + +```vlt +if (kubernetes.pod_namespace:=kube-system) { + format 1 as vl_account_id; + format 0 as vl_project_id; +} +if (kubernetes.pod_namespace:=prod) { + format 2 as vl_account_id; + format 0 as vl_project_id; +} +if (kubernetes.pod_namespace:=stg) { + format 3 as vl_account_id; + format 0 as vl_project_id; +} +# ... +``` + +The tenant value can be read directly from the log content. +To do this, produce the `vl_account_id` or `vl_project_id` field in a transformation with `rename`, `format`, +or any pipe that assigns a value to the field: + +```vlt +if (kubernetes.pod_annotations.account_id:* and kubernetes.pod_annotations.project_id:*) { + # both 'account_id' and 'project_id' annotations are set, use them to override the target tenant. + rename kubernetes.pod_annotations.account_id as vl_account_id; + rename kubernetes.pod_annotations.project_id as vl_project_id; +} +``` + +This sets the tenant from the `kubernetes.pod_annotations.account_id` and `kubernetes.pod_annotations.project_id` fields, +so the target tenant is controlled through Kubernetes annotations. + +If `vl_account_id` and `vl_project_id` fields arrive in the incoming log, +`vlagent` treats them as ordinary fields and does not change the tenant. + +### Head sampling based on trace_id + +The program below keeps only 20% of logs for the `payments` service, based on the `trace_id` field. + +```vlt +if (service:=payments) { + do sample_by_trace_id; +} + +block sample_by_trace_id { + if (trace_id:"") { + # The 'trace_id' field is not set. + return; + } + + # The 'hash_remainder' field will contain a hash remainder from 0 to 99. + hash(trace_id) as trace_id_hash | math trace_id_hash % 100 as hash_remainder; + + # Keep remainders 0-19 (20%), drop the rest. + if (hash_remainder:>=20) { + drop; + } + delete trace_id_hash, hash_remainder; +} +``` + +Sampling is applied per log, so it gives consistent results even across nodes that share the same `vlagent` configuration. diff --git a/lib/logstorage/block_result.go b/lib/logstorage/block_result.go index 1c9bb6369c..d2a81599cd 100644 --- a/lib/logstorage/block_result.go +++ b/lib/logstorage/block_result.go @@ -268,7 +268,8 @@ func (br *blockResult) mustInitFromRows(rows [][]Field) { // Fast path - all the rows have the same fields fields := rows[0] for i := range fields { - name := br.addValue(fields[i].Name) + name := getCanonicalColumnName(fields[i].Name) + name = br.addValue(name) valuesBufLen := len(br.valuesBuf) for _, row := range rows { @@ -289,7 +290,8 @@ func (br *blockResult) mustInitFromRows(rows [][]Field) { columnIdxs := getColumnIdxs() for _, fields := range rows { for j := range fields { - name := br.addValue(fields[j].Name) + name := getCanonicalColumnName(fields[j].Name) + name = br.addValue(name) if _, ok := columnIdxs[name]; !ok { columnIdxs[name] = len(columnIdxs) } @@ -315,7 +317,8 @@ func (br *blockResult) mustInitFromRows(rows [][]Field) { // Add values to columns for i := range rows { for _, f := range rows[i] { - idx := columnIdxs[f.Name] + name := getCanonicalColumnName(f.Name) + idx := columnIdxs[name] value := br.addValue(f.Value) cs[idx].valuesEncoded[i] = value } @@ -323,6 +326,57 @@ func (br *blockResult) mustInitFromRows(rows [][]Field) { putColumnIdxs(columnIdxs) } +func (br *blockResult) mustInitFromLogRows(lr *LogRows) { + br.mustInitFromRows(lr.rows) + if br.rowsLen == 0 { + return + } + + br.timestampsBuf = append(br.timestampsBuf[:0], lr.timestamps...) + br.addTimeColumn() + + bb := bbPool.Get() + defer bbPool.Put(bb) + + // Initialize the '_stream' column from the canonical form. + st := GetStreamTags() + valuesBufLen := len(br.valuesBuf) + for _, stc := range lr.streamTagsCanonicals { + mustUnmarshalStreamTagsInplace(st, stc) + bb.Reset() + bb.B = st.marshalString(bb.B) + stream := bytesutil.ToUnsafeString(bb.B) + br.addValue(stream) + } + streamValues := br.valuesBuf[valuesBufLen:] + br.addResultColumn(resultColumn{name: "_stream", values: streamValues}) + PutStreamTags(st) + + // Initialize the 'vl_account_id' column. + valuesBufLen = len(br.valuesBuf) + for i := range lr.streamIDs { + accountID := lr.streamIDs[i].tenantID.AccountID + bb.Reset() + bb.B = strconv.AppendInt(bb.B, int64(accountID), 10) + accountIDStr := bytesutil.ToUnsafeString(bb.B) + br.addValue(accountIDStr) + } + accountIDs := br.valuesBuf[valuesBufLen:] + br.addResultColumn(resultColumn{name: "vl_account_id", values: accountIDs}) + + // Initialize the 'vl_project_id' column. + valuesBufLen = len(br.valuesBuf) + for i := range lr.streamIDs { + projectID := lr.streamIDs[i].tenantID.ProjectID + bb.Reset() + bb.B = strconv.AppendInt(bb.B, int64(projectID), 10) + projectIDStr := bytesutil.ToUnsafeString(bb.B) + br.addValue(projectIDStr) + } + projectIDs := br.valuesBuf[valuesBufLen:] + br.addResultColumn(resultColumn{name: "vl_project_id", values: projectIDs}) +} + // setResultColumns sets the given rcs as br columns. // // The br is valid only until rcs are modified. @@ -570,9 +624,15 @@ func (br *blockResult) getMaxTimestamp(maxTimestamp int64) int64 { } func (br *blockResult) getTimestamps() []int64 { - if br.rowsLen > 0 && len(br.timestampsBuf) == 0 { + if br.rowsLen == 0 { + return nil + } + if len(br.timestampsBuf) == 0 { + // Timestamps aren't initialized yet. br.initTimestamps() + return br.timestampsBuf } + br.ensureTimestampsParsed() return br.timestampsBuf } @@ -587,9 +647,18 @@ func (br *blockResult) initTimestamps() { br.initTimestampsInternal(srcTimestamps) return } + br.ensureTimestampsParsed() +} - // Try decoding timestamps from _time field +// ensureTimestampsParsed initializes br.timestampsBuf with parsed timestamps from the _time column. +// If the column cannot be parsed, the buffer is filled with zeros. +// It is a no-op if the timestamps are already parsed. +func (br *blockResult) ensureTimestampsParsed() { c := br.getColumnByName("_time") + if c.isTime { + // Already parsed. + return + } timestampValues := c.getValues(br) var ok bool br.timestampsBuf, ok = tryParseTimestamps(br.timestampsBuf[:0], timestampValues) diff --git a/lib/logstorage/log_rows.go b/lib/logstorage/log_rows.go index b0db16ca68..d9bc45b057 100644 --- a/lib/logstorage/log_rows.go +++ b/lib/logstorage/log_rows.go @@ -823,3 +823,85 @@ func (r *InsertRow) UnmarshalInplace(src []byte) ([]byte, error) { return src, nil } + +func (lr *LogRows) initFromBlockResult(br *blockResult) { + lr.ResetKeepSettings() + if br.rowsLen == 0 { + return + } + + ir := GetInsertRow() + defer PutInsertRow(ir) + st := GetStreamTags() + defer PutStreamTags(st) + stcBuf := bbPool.Get() + defer bbPool.Put(stcBuf) + + timestamps := br.getTimestamps() + + cs := br.getColumns() + for i := 0; i < br.rowsLen; i++ { + ir.Reset() + var ( + stream string + accountID string + projectID string + ) + fields := ir.Fields + for _, c := range cs { + if c.name == "_time" { + // Timestamps initialized above. + continue + } + s := c.getValueAtRow(br, i) + switch c.name { + case "_stream": + stream = s + case "vl_account_id": + accountID = s + case "vl_project_id": + projectID = s + default: + fields = append(fields, Field{ + Name: c.name, + Value: s, + }) + } + } + ir.Fields = fields + ir.Timestamp = timestamps[i] + + // Initialize StreamTagsCanonical. + st.Reset() + stcBuf.Reset() + if stream != "" { + err := st.unmarshalStringInplace(stream) + if err != nil { + // Fallback to an empty stream. + st.Reset() + } + } + stcBuf.B = st.MarshalCanonical(stcBuf.B) + ir.StreamTagsCanonical = bytesutil.ToUnsafeString(stcBuf.B) + + // Restore AccountID from vl_account_id. + v, err := getUint32FromString(accountID) + if err != nil { + skipMalformedLogEntryLogger.Errorf("skipping log entry with invalid 'vl_account_id': %s", err) + continue + } + ir.TenantID.AccountID = v + + // Restore ProjectID from vl_project_id. + v, err = getUint32FromString(projectID) + if err != nil { + skipMalformedLogEntryLogger.Errorf("skipping log entry with invalid 'vl_project_id': %s", err) + continue + } + ir.TenantID.ProjectID = v + + lr.MustAddInsertRow(ir) + } +} + +var skipMalformedLogEntryLogger = logger.WithThrottler("skip_malformed_log_entry", time.Second*5) diff --git a/lib/logstorage/transforms.go b/lib/logstorage/transforms.go new file mode 100644 index 0000000000..92b71295a6 --- /dev/null +++ b/lib/logstorage/transforms.go @@ -0,0 +1,1090 @@ +package logstorage + +import ( + "fmt" + "slices" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + "unicode" + + "github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup" +) + +// TransformsProgram is a parsed VictoriaLogs transformations program, +// assembled from one or more content fragments via ParseAdditional and can be compiled into a Transformer. +type TransformsProgram struct { + tps []*transformsProgram +} + +// ParseTransformsProgram creates a new instance of TransformsProgram. +// Use ParseAdditional to parse additional programs. +func ParseTransformsProgram(s string) (*TransformsProgram, error) { + tp := &TransformsProgram{} + if err := tp.ParseAdditional(s); err != nil { + return nil, err + } + return tp, nil +} + +// ParseAdditional parses s as an additional program appended to the transformation pipeline. +// Note that named blocks are scoped to the program that declares them and are not visible to other programs. +func (tp *TransformsProgram) ParseAdditional(s string) error { + tpNew, err := parseTransformsProgram(s) + if err != nil { + return err + } + tp.tps = append(tp.tps, tpNew) + return nil +} + +// Transformer can transform data according to the compiled TransformsProgram. +type Transformer struct { + nextShard atomic.Uint64 + shards []transformsProcessorShard + pp pipeProcessor +} + +type transformsProcessorShard struct { + br blockResult + lr LogRows + mu sync.Mutex + _ [atomicutil.CacheLineSize]byte +} + +// NewTransformer creates a Transformer from tp that can transform incoming data according to the transformations program. +// +// flush is called for each batch of transformed log rows. +// The lr passed to flush is owned by the Transformer and is reused after flush returns, +// so flush must fully consume or copy it before returning. +func (tp *TransformsProgram) NewTransformer(flush func(lr *LogRows)) *Transformer { + // Use twice CPU since workers have variable execution times, + // causing faster goroutines to sit idle waiting for others. + // This is necessary because the Transformer struct simulates workers that don't actually exist yet. + // + // TODO: add workers + concurrency := cgroup.AvailableCPUs() * 2 + + shards := make([]transformsProcessorShard, concurrency) + storeResult := func(workerID uint, br *blockResult) { + if br.rowsLen == 0 { + return + } + shard := &shards[workerID] + lr := &shard.lr + lr.initFromBlockResult(br) + flush(lr) + } + neverStopCh := make(chan struct{}) + noop := newNoopPipeProcessor(neverStopCh, storeResult) + + ppSend := noop + ppNext := noop + for i := len(tp.tps) - 1; i >= 0; i-- { + prog := tp.tps[i] + ppNext = prog.newProcessor(concurrency, ppSend, ppNext) + } + + return &Transformer{ + shards: shards, + pp: ppNext, + } +} + +// Transform runs the compiled transformations over lr using the given workerID, +// and calls the flush callback for the result. +// +// It is safe to call Transform concurrently from worker goroutines. +func (t *Transformer) Transform(lr *LogRows) { + // Emulate workers, as the writeBlock requires a workerID. + workerID := t.nextShard.Add(1) % uint64(len(t.shards)) + shard := &t.shards[workerID] + shard.mu.Lock() + defer shard.mu.Unlock() + + br := &shard.br + br.mustInitFromLogRows(lr) + t.pp.writeBlock(uint(workerID), br) +} + +// transformsProgram is a parsed VictoriaLogs transformations program. +type transformsProgram struct { + // namedBlocks contains use-defined blocks to use via "do" keyword. + namedBlocks []namedTransformBlock + // root is the top-level transform executed for every log. + root transform +} + +type transform interface { + // String return string representation of transform. + String() string + + // newTransformProcessor builds the processor for this transform, chaining into ppNext. + // ppSend is used to interrupt the execution of subsequent ppNext steps, redirecting logs directly to the final destination. + // ppReturn is used to exit the execution of ppNext and resumes the main program flow. + newTransformProcessor(concurrency int, ppSend, ppReturn, ppNext pipeProcessor) pipeProcessor +} + +// parseTransformsProgram parses s into a transformsProgram. +func parseTransformsProgram(s string) (*transformsProgram, error) { + lex := newLexer(s, time.Now().UnixNano()) + + prog, err := parseTransformsProgramInternal(lex) + if err != nil { + return nil, fmt.Errorf("cannot parse transforms program: %w; context: [%s]", err, lex.context()) + } + if err := prog.checkRecursion(); err != nil { + return nil, err + } + if err := prog.initTransformsDo(); err != nil { + return nil, err + } + if err := prog.checkTimeFilter(); err != nil { + return nil, err + } + if err := prog.checkSubqueries(); err != nil { + return nil, err + } + + return prog, nil +} + +func parseTransformsProgramInternal(lex *lexer) (*transformsProgram, error) { + if lex.isKeyword("") { + return nil, fmt.Errorf("missing transformations") + } + + var namedBlocks []namedTransformBlock + uniqNamedBlocks := make(map[string]struct{}) + + var trs []transform + for !lex.isKeyword("") { + if lex.isKeyword("block") { + tr, name, err := parseNamedTransformBlock(lex) + if err != nil { + return nil, err + } + if _, ok := uniqNamedBlocks[name]; ok { + return nil, fmt.Errorf("duplicate named transformation block %q", name) + } + + uniqNamedBlocks[name] = struct{}{} + namedBlocks = append(namedBlocks, namedTransformBlock{ + name: name, + body: tr, + }) + continue + } + + tr, err := parseGenericTransformBlockLine(lex) + if err != nil { + return nil, err + } + trs = append(trs, tr) + } + + return &transformsProgram{ + namedBlocks: namedBlocks, + root: &transformBlock{ + transforms: trs, + }, + }, nil +} + +func (tp *transformsProgram) String() string { + var s string + for _, nb := range tp.namedBlocks { + s += fmt.Sprintf("block %s ", nb.name) + s += nb.body.String() + s += "\n" + } + + // Override the top-level block implementation in order to remove curly braces and indentation. + if tb, ok := tp.root.(*transformBlock); ok { + for i, p := range tb.transforms { + if i > 0 { + s += "\n" + } + s += p.String() + } + return s + } + + s += tp.root.String() + return s +} + +// newProcessor builds the processor chain for the whole program. +// +// At the root, ppSend, ppReturn and ppNext all point to the same downstream sink, +// so 'send' and 'return' at the top level are equivalent to falling through. +func (tp *transformsProgram) newProcessor(concurrency int, ppSend, ppNext pipeProcessor) pipeProcessor { + return tp.root.newTransformProcessor(concurrency, ppSend, ppNext, ppNext) +} + +// namedTransformBlock is a reusable block declared via 'block name {...}'. +type namedTransformBlock struct { + name string + body *transformBlock +} + +func parseNamedTransformBlock(lex *lexer) (*transformBlock, string, error) { + if !lex.isKeyword("block") { + return nil, "", fmt.Errorf("expected keyword 'block', got %q", lex.token) + } + lex.nextToken() + + name, err := parseTransformBlockName(lex) + if err != nil { + return nil, "", err + } + + tr, err := parseGenericTransformBlock(lex) + if err != nil { + return nil, "", err + } + return tr, name, nil +} + +// transformsKeywords are reserved words that cannot be used as block names. +var transformsKeywords = []string{ + "if", + "else", + "send", + "return", + "drop", + "block", + "do", +} + +func parseTransformBlockName(lex *lexer) (string, error) { + if lex.isKeyword("") { + return "", fmt.Errorf("unexpected end of block name") + } + if lex.isQuotedToken() { + return "", fmt.Errorf("unexpected quotation mark in block declaration") + } + name := lex.token + if slices.Contains(transformsKeywords, name) || isPipeName(name) { + return "", fmt.Errorf("cannot declare block with name %q as it is a keyword", name) + } + for _, r := range name { + if !unicode.IsLetter(r) && !unicode.IsDigit(r) && r != '_' { + return "", fmt.Errorf("invalid character %q in block declaration: %q", r, name) + } + } + lex.nextToken() + return name, nil +} + +// transformBlock is a sequence of transforms wrapped in '{ }'. +type transformBlock struct { + transforms []transform +} + +// parseGenericTransformBlock parses a transformBlock wrapped in '{ }'. +func parseGenericTransformBlock(lex *lexer) (*transformBlock, error) { + if !lex.isKeyword("{") { + return nil, fmt.Errorf("unexpected token: %q; want %q", lex.token, "{") + } + lex.nextToken() + + var trs []transform + for !lex.isKeyword("}", "") { + tr, err := parseGenericTransformBlockLine(lex) + if err != nil { + return nil, err + } + trs = append(trs, tr) + } + if !lex.isKeyword("}") { + return nil, fmt.Errorf("unexpected end of the transforms block") + } + lex.nextToken() + + return &transformBlock{ + transforms: trs, + }, nil +} + +// parseGenericTransformBlockLine parses a single line inside a transformBlock. +func parseGenericTransformBlockLine(lex *lexer) (transform, error) { + if lex.isKeyword("block") { + return nil, fmt.Errorf("declaring a named block inside another block is not allowed; declare the block at the top level of the file") + } + switch { + case lex.isKeyword("do"): + return parseTransformDo(lex) + case lex.isKeyword("if"): + return parseTransformIf(lex) + case lex.isKeyword("send"): + return parseTransformSend(lex) + case lex.isKeyword("return"): + return parseTransformReturn(lex) + case lex.isKeyword("drop"): + return parseTransformDrop(lex) + default: + return parseTransformPipes(lex) + } +} + +func (tb *transformBlock) String() string { + if len(tb.transforms) == 0 { + return "{}" + } + var s string + s += "{" + for _, p := range tb.transforms { + s += "\n" + s += addPrefixToLines(p.String(), " ") + } + s += "\n" + s += "}" + return s +} + +func (tb *transformBlock) newTransformProcessor(concurrency int, ppSend, ppReturn, ppNext pipeProcessor) pipeProcessor { + for i := len(tb.transforms) - 1; i >= 0; i-- { + tr := tb.transforms[i] + ppNext = tr.newTransformProcessor(concurrency, ppSend, ppReturn, ppNext) + } + return &transformBlockProcessor{ + ppNext: ppNext, + } +} + +type transformBlockProcessor struct { + ppNext pipeProcessor +} + +func (tbp *transformBlockProcessor) writeBlock(workerID uint, br *blockResult) { + tbp.ppNext.writeBlock(workerID, br) +} + +func (tbp *transformBlockProcessor) flush() error { + return tbp.ppNext.flush() +} + +// transformDo references a named block declared elsewhere via 'do '. +type transformDo struct { + // blockName is the referenced block's blockName. + blockName string + + // body contains initialized transformBlock to execute. + // This field initializes after program is parsed. + body *transformBlock +} + +func parseTransformDo(lex *lexer) (*transformDo, error) { + if !lex.isKeyword("do") { + return nil, fmt.Errorf("unexpected token: %q; want %q", lex.token, "do") + } + lex.nextToken() + + name, err := parseTransformBlockName(lex) + if err != nil { + return nil, err + } + + if !lex.isKeyword(";") { + return nil, fmt.Errorf("expected 'do' to be ended with ';', got %q", lex.token) + } + lex.nextToken() + + return &transformDo{ + blockName: name, + }, nil +} + +func (td *transformDo) String() string { + return fmt.Sprintf("do %s;", td.blockName) +} + +func (td *transformDo) newTransformProcessor(concurrency int, ppSend, _, ppNext pipeProcessor) pipeProcessor { + if td.body == nil { + panic(fmt.Errorf("BUG: transform %q is not initialized", td.String())) + } + // Override ppReturn with ppNext, so the next 'return' call inside a named block will process ppNext. + return td.body.newTransformProcessor(concurrency, ppSend, ppNext, ppNext) +} + +// transformIf represents a single 'if ... else if ... else ...' chain. +type transformIf struct { + // branches holds the 'if' branch followed by any 'else if' branches, in order. + branches []*ifBranch + // elseBody is an 'else' block, or nil if there is no 'else'. + elseBody *transformBlock +} + +// parseTransformIf parses transformIf. +// +// The chain is extended only through 'else if'. +// Two adjacent 'if' blocks with no 'else' between them are independent chains, +// so parsing stops after the first 'if', and the second one is parsed later as a separate transformIf. +func parseTransformIf(lex *lexer) (*transformIf, error) { + if !lex.isKeyword("if") { + return nil, fmt.Errorf("unexpected token: %q; want %q", lex.token, "if") + } + + // Parse the start of the chain. + var ifs []*ifBranch + b, err := parseIfBranch(lex) + if err != nil { + return nil, err + } + ifs = append(ifs, b) + + var elseBody *transformBlock + // Parse 'else if' chain + for lex.isKeyword("else") { + lex.nextToken() + // Parse 'else if' block + if lex.isKeyword("if") { + b, err := parseIfBranch(lex) + if err != nil { + return nil, err + } + ifs = append(ifs, b) + continue + } + // Parse 'else' block. + b, err := parseGenericTransformBlock(lex) + if err != nil { + return nil, err + } + elseBody = b + break + } + return &transformIf{ + branches: ifs, + elseBody: elseBody, + }, nil +} + +func (ti *transformIf) String() string { + var s string + for i, ib := range ti.branches { + if i > 0 { + s += " else " + } + s += ib.String() + } + if ti.elseBody != nil { + s += " else " + s += ti.elseBody.String() + } + return s +} + +func (ti *transformIf) newTransformProcessor(concurrency int, ppSend, ppReturn, ppNext pipeProcessor) pipeProcessor { + // ppMatched is a next processor to use if a branch is matched. + ppMatched := ppNext + + pp := ppNext + if ti.elseBody != nil { + pp = ti.elseBody.newTransformProcessor(concurrency, ppSend, ppReturn, pp) + } + for i := len(ti.branches) - 1; i >= 0; i-- { + ib := ti.branches[i] + pp = ib.newIfBranchProcessor(concurrency, ppSend, ppReturn, ppMatched, pp) + } + return pp +} + +// ifBranch is a single 'if (filter) { body }' branch of a transformIf. +type ifBranch struct { + // f is the condition the log must match for body to run. + f filter + body *transformBlock +} + +func (ib *ifBranch) String() string { + return fmt.Sprintf("if (%s) %s", ib.f.String(), ib.body.String()) +} + +func (ib *ifBranch) newIfBranchProcessor(concurrency int, ppSend, ppReturn, ppNext, ppUnmatched pipeProcessor) pipeProcessor { + ppMatched := ib.body.newTransformProcessor(concurrency, ppSend, ppReturn, ppNext) + return &ifBranchProcessor{ + f: ib.f, + ppMatched: ppMatched, + ppUnmatched: ppUnmatched, + } +} + +type ifBranchProcessor struct { + f filter + ppMatched pipeProcessor + ppUnmatched pipeProcessor + + shards atomicutil.Slice[ifBranchProcessorShard] +} + +type ifBranchProcessorShard struct { + bmMatched bitmap + bmUnmatched bitmap + brMatched blockResult + brUnmatched blockResult +} + +func (ibp *ifBranchProcessor) writeBlock(workerID uint, br *blockResult) { + if br.rowsLen == 0 { + return + } + shard := ibp.shards.Get(workerID) + bmMatched := &shard.bmMatched + bmMatched.init(br.rowsLen) + bmMatched.setBits() + ibp.f.applyToBlockResult(br, bmMatched) + // Fast path: all rows matched or unmatched. + if bmMatched.isZero() { + ibp.ppUnmatched.writeBlock(workerID, br) + return + } + if bmMatched.areAllBitsSet() { + ibp.ppMatched.writeBlock(workerID, br) + return + } + + // Slow path: split br by matched and unmatched. + bmUnmatched := &shard.bmUnmatched + bmUnmatched.init(bmMatched.bitsLen) + bmUnmatched.setBits() + bmUnmatched.andNot(bmMatched) + + brMatched := &shard.brMatched + brUnmatched := &shard.brUnmatched + brMatched.initFromFilterAllColumns(br, bmMatched) + brUnmatched.initFromFilterAllColumns(br, bmUnmatched) + + ibp.ppMatched.writeBlock(workerID, brMatched) + ibp.ppUnmatched.writeBlock(workerID, brUnmatched) +} + +func (ibp *ifBranchProcessor) flush() error { + if err := ibp.ppMatched.flush(); err != nil { + return err + } + return ibp.ppUnmatched.flush() +} + +func parseIfBranch(lex *lexer) (*ifBranch, error) { + if !lex.isKeyword("if") { + return nil, fmt.Errorf("expected keyword 'if', got %q", lex.token) + } + lex.nextToken() + + if !lex.isKeyword("(") { + return nil, fmt.Errorf("condition in the 'if' statement should be wrapped with parentheses, got %s", lex.token) + } + lex.nextToken() + + f, err := parseFilter(lex) + if err != nil { + return nil, err + } + + if !lex.isKeyword(")") { + return nil, fmt.Errorf("condition in the 'if' statement should be wrapped with parentheses, got %s", lex.token) + } + lex.nextToken() + + tb, err := parseGenericTransformBlock(lex) + if err != nil { + return nil, err + } + + return &ifBranch{ + f: f, + body: tb, + }, nil +} + +// transformSend stops processing of the current log and emits it downstream. +type transformSend struct { +} + +func parseTransformSend(lex *lexer) (*transformSend, error) { + if !lex.isKeyword("send") { + return nil, fmt.Errorf("unexpected token: %q; want %q", lex.token, "send") + } + lex.nextToken() + if !lex.isKeyword(";") { + return nil, fmt.Errorf("expected 'send' to be ended with ';', got %q", lex.token) + } + lex.nextToken() + return &transformSend{}, nil +} + +func (ts *transformSend) String() string { + return "send;" +} + +func (ts *transformSend) newTransformProcessor(_ int, ppSend, _, _ pipeProcessor) pipeProcessor { + return ppSend +} + +type transformReturn struct { +} + +func parseTransformReturn(lex *lexer) (*transformReturn, error) { + if !lex.isKeyword("return") { + return nil, fmt.Errorf("unexpected token: %q; want %q", lex.token, "return") + } + lex.nextToken() + if !lex.isKeyword(";") { + return nil, fmt.Errorf("expected 'return' to be ended with ';', got %q", lex.token) + } + lex.nextToken() + return &transformReturn{}, nil +} + +func (tr *transformReturn) String() string { + return "return;" +} + +func (tr *transformReturn) newTransformProcessor(_ int, _, ppReturn, _ pipeProcessor) pipeProcessor { + return ppReturn +} + +type transformDrop struct { +} + +func parseTransformDrop(lex *lexer) (*transformDrop, error) { + if !lex.isKeyword("drop") { + return nil, fmt.Errorf("unexpected token: %q; want %q", lex.token, "drop") + } + lex.nextToken() + if !lex.isKeyword(";") { + return nil, fmt.Errorf("expected 'drop' to be ended with ';', got %q", lex.token) + } + lex.nextToken() + return &transformDrop{}, nil +} + +func (td *transformDrop) String() string { + return "drop;" +} + +func (td *transformDrop) newTransformProcessor(_ int, _, _, _ pipeProcessor) pipeProcessor { + return &transformDropProcessor{} +} + +type transformDropProcessor struct { +} + +func (t *transformDropProcessor) writeBlock(_ uint, _ *blockResult) { + // Drop the block. +} + +func (t *transformDropProcessor) flush() error { + return nil +} + +// transformPipes is a single line of '|' separated pipes. +type transformPipes struct { + pipes []pipe +} + +func parseTransformPipes(lex *lexer) (*transformPipes, error) { + pipes, err := parseTransformPipesAtLine(lex) + if err != nil { + return nil, err + } + return &transformPipes{ + pipes: pipes, + }, nil +} + +var transformsPipeParsers map[string]pipeParseFunc +var transformsPipeParsersOnce sync.Once + +func getTransformsPipeParsers() map[string]pipeParseFunc { + transformsPipeParsersOnce.Do(initTransformsPipeParsers) + return transformsPipeParsers +} + +// initTransformsPipeParsers registers the pipes allowed inside VictoriaLogs transformations. +func initTransformsPipeParsers() { + transformsPipeParsers = map[string]pipeParseFunc{ + "coalesce": parsePipeCoalesce, + "collapse_nums": parsePipeCollapseNums, + "copy": parsePipeCopy, + "cp": parsePipeCopy, + "decolorize": parsePipeDecolorize, + "del": parsePipeDelete, + "delete": parsePipeDelete, + "drop_empty_fields": parsePipeDropEmptyFields, + "extract": parsePipeExtract, + "extract_regexp": parsePipeExtractRegexp, + "eval": parsePipeMath, + "fields": parsePipeFields, + "format": parsePipeFormat, + "hash": parsePipeHash, + "json_array_len": parsePipeJSONArrayLen, + "keep": parsePipeFields, + "len": parsePipeLen, + "math": parsePipeMath, + "mv": parsePipeRename, + "pack_json": parsePipePackJSON, + "pack_logfmt": parsePipePackLogfmt, + "rename": parsePipeRename, + "replace": parsePipeReplace, + "replace_regexp": parsePipeReplaceRegexp, + "rm": parsePipeDelete, + "sample": parsePipeSample, + "set_stream_fields": parsePipeSetStreamFields, + "split": parsePipeSplit, + "time_add": parsePipeTimeAdd, + "unpack_json": parsePipeUnpackJSON, + "unpack_logfmt": parsePipeUnpackLogfmt, + "unpack_syslog": parsePipeUnpackSyslog, + "unpack_words": parsePipeUnpackWords, + "unroll": parsePipeUnroll, + } +} + +// parseTransformPipesAtLine parses one line of '|' separated pipes. +func parseTransformPipesAtLine(lex *lexer) ([]pipe, error) { + var pipes []pipe + for { + p, err := parseTransformPipe(lex) + if err != nil { + return nil, err + } + pipes = append(pipes, p) + + switch { + case lex.isKeyword(";"): + // The end of the line. + lex.nextToken() + return pipes, nil + case lex.isKeyword("|"): + // Pipe delimiter, consume next pipe. + lex.nextToken() + default: + return nil, fmt.Errorf("unexpected token: %q; want pipe '|' or line ';' delimiter", lex.token) + } + } +} + +func parseTransformPipe(lex *lexer) (pipe, error) { + pps := getTransformsPipeParsers() + for pipeName, parseFunc := range pps { + if !lex.isKeyword(pipeName) { + continue + } + p, err := parseFunc(lex) + if err != nil { + return nil, fmt.Errorf("cannot parse %q pipe: %w", pipeName, err) + } + return p, nil + } + + if isPipeKeyword(lex) { + return nil, fmt.Errorf("pipe %q is not allowed in transformations", lex.token) + } + + return nil, fmt.Errorf("unknown pipe with name %q", lex.token) +} + +// isPipeKeyword reports whether the current token is any known pipe name. +func isPipeKeyword(lex *lexer) bool { + pps := getPipeParsers() + for pipeName := range pps { + if lex.isKeyword(pipeName) { + return true + } + } + return false +} + +func (tp *transformPipes) String() string { + var s string + for i, p := range tp.pipes { + if i > 0 { + s += " | " + } + s += p.String() + } + s += ";" + return s +} + +func (tp *transformPipes) newTransformProcessor(concurrency int, _, _, ppNext pipeProcessor) pipeProcessor { + neverStopCh := make(chan struct{}) + cancel := func() { + // The cancel function is usually required for stateful streaming pipes. + // These types of pipes must not be used in transformations. + panic(fmt.Errorf("BUG: 'cancel' function should not be called in transformations")) + } + pp := ppNext + for i := len(tp.pipes) - 1; i >= 0; i-- { + p := tp.pipes[i] + pp = p.newPipeProcessor(concurrency, neverStopCh, cancel, pp) + } + return pp +} + +// addPrefixToLines prepends prefix to every line of s. +func addPrefixToLines(s, prefix string) string { + lines := strings.Split(s, "\n") + for i := range lines { + lines[i] = prefix + lines[i] + } + return strings.Join(lines, "\n") +} + +// checkRecursion scans named blocks and returns an error on any recursive call. +// It does not analyze execution paths and ignores conditional guards, +// so it returns an error even if the condition is always false. +func (tp *transformsProgram) checkRecursion() error { + namedBlockCalls := make(map[string][]string, len(tp.namedBlocks)) + for _, nb := range tp.namedBlocks { + namedBlockCalls[nb.name] = collectNamedBlockCalls(nb.body) + } + + visited := make(map[string]struct{}) + var stacktrace []string + var visitFunc func(name string) error + visitFunc = func(name string) error { + if _, ok := visited[name]; ok { + return nil + } + + if slices.Contains(stacktrace, name) { + // Recursion found, build call path and return an error. + cycle := stacktrace + cycle = append(cycle, name) + for i := range cycle { + cycle[i] = strconv.Quote(cycle[i]) + } + return fmt.Errorf("recursive block call: %s", strings.Join(cycle, " calls ")) + } + + if _, ok := namedBlockCalls[name]; !ok { + return fmt.Errorf("cannot find block with name %q", name) + } + + stacktrace = append(stacktrace, name) + for _, ref := range namedBlockCalls[name] { + if err := visitFunc(ref); err != nil { + return err + } + } + stacktrace = stacktrace[:len(stacktrace)-1] + visited[name] = struct{}{} + + return nil + } + + for _, nb := range tp.namedBlocks { + if err := visitFunc(nb.name); err != nil { + return err + } + } + + return nil +} + +func collectNamedBlockCalls(body *transformBlock) []string { + var calls []string + uniq := make(map[string]struct{}) + visitFunc := func(tr transform) bool { + v, ok := tr.(*transformDo) + if !ok { + return false + } + if _, ok := uniq[v.blockName]; ok { + return false + } + uniq[v.blockName] = struct{}{} + calls = append(calls, v.blockName) + return false + } + visitTransformRecursive(body, visitFunc) + return calls +} + +// initTransformsDo initializes transformDo with corresponding named block body. +func (tp *transformsProgram) initTransformsDo() error { + var errGlobal error + visitFunc := func(tr transform) bool { + v, ok := tr.(*transformDo) + if !ok { + return false + } + n := slices.IndexFunc(tp.namedBlocks, func(nb namedTransformBlock) bool { + return nb.name == v.blockName + }) + if n < 0 { + errGlobal = fmt.Errorf("cannot find block with name %q", v.blockName) + return false + } + v.body = tp.namedBlocks[n].body + return false + } + for _, nb := range tp.namedBlocks { + visitTransformRecursive(nb.body, visitFunc) + } + visitTransformRecursive(tp.root, visitFunc) + return errGlobal +} + +// checkTimeFilter returns an error if tp has filterTime that is not allowed in log transformations. +func (tp *transformsProgram) checkTimeFilter() error { + var errGlobal error + visitFunc := func(f filter) bool { + _, ok := f.(*filterTime) + if !ok { + return false + } + if errGlobal == nil { + errGlobal = fmt.Errorf("time filters are not allowed in log transformations; got %q", f.String()) + } + return true + } + visitFilterRecursiveForTransforms(tp, visitFunc) + return errGlobal +} + +// checkSubqueries returns an error if tp has subqueries that are not allowed in log transformations. +func (tp *transformsProgram) checkSubqueries() error { + var errGlobal error + visitFunc := func(f filter) bool { + if !hasFilterInWithQueryForFilter(f) { + return false + } + if errGlobal == nil { + errGlobal = fmt.Errorf("subqueries are not allowed in log transformations; got %q", f.String()) + } + return true + } + visitFilterRecursiveForTransforms(tp, visitFunc) + return errGlobal +} + +// visitFilterRecursiveForTransforms recursively calls visitFunc for filters inside tp. +// +// It stops calling visitFunc on the remaining transforms as soon as visitFunc returns true. +// It returns the result of the last visitFunc call. +func visitFilterRecursiveForTransforms(tp *transformsProgram, visitFunc func(f filter) bool) bool { + visitTransformFunc := func(tr transform) bool { + switch tr := tr.(type) { + case *transformIf: + for _, ib := range tr.branches { + if visitFilterRecursive(ib.f, visitFunc) { + return true + } + } + case *transformPipes: + for _, p := range tr.pipes { + if visitFilterRecursiveForPipe(p, visitFunc) { + return true + } + } + } + return false + } + for _, nb := range tp.namedBlocks { + if visitTransformRecursive(nb.body, visitTransformFunc) { + return true + } + } + return visitTransformRecursive(tp.root, visitTransformFunc) +} + +// visitTransformRecursive recursively calls visitFunc for transforms inside tr. +// +// It stops calling visitFunc on the remaining transforms as soon as visitFunc returns true. +// It returns the result of the last visitFunc call. +func visitTransformRecursive(tr transform, visitFunc func(tr transform) bool) bool { + // Visit the tr itself. + if visitFunc(tr) { + return true + } + + // Visit nested transforms. + switch tr := tr.(type) { + case *transformBlock: + for _, subTr := range tr.transforms { + if visitTransformRecursive(subTr, visitFunc) { + return true + } + } + case *transformIf: + for _, ib := range tr.branches { + if visitTransformRecursive(ib.body, visitFunc) { + return true + } + } + if tr.elseBody != nil { + if visitTransformRecursive(tr.elseBody, visitFunc) { + return true + } + } + } + return false +} + +// visitFilterRecursiveForPipe recursively calls visitFunc for filters inside p. +// +// It stops calling visitFunc on the remaining filters as soon as visitFunc returns true. +// It returns the result of the last visitFunc call. +func visitFilterRecursiveForPipe(p pipe, visitFunc func(f filter) bool) bool { + // All possible pipes that may contain a nested filter. + switch p := p.(type) { + case *pipeCollapseNums: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeExtract: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeExtractRegexp: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeFormat: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeReplace: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeReplaceRegexp: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeSetStreamFields: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeUnpackJSON: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeUnpackLogfmt: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeUnpackSyslog: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeUnroll: + if p.iff != nil { + return visitFilterRecursive(p.iff.f, visitFunc) + } + case *pipeFilter: + if p.f != nil { + return visitFilterRecursive(p.f, visitFunc) + } + } + // Does not have filter. + return false +} diff --git a/lib/logstorage/transforms_test.go b/lib/logstorage/transforms_test.go new file mode 100644 index 0000000000..4758fcded5 --- /dev/null +++ b/lib/logstorage/transforms_test.go @@ -0,0 +1,785 @@ +package logstorage + +import ( + "reflect" + "testing" + "time" +) + +func TestParseTransforms(t *testing.T) { + f := func(s, want string) { + t.Helper() + + tr, err := parseTransformsProgram(s) + if err != nil { + t.Fatalf("cannot parse transforms\nconfig:\n%s\nerror: %s", s, err) + } + got := tr.String() + if got != want { + t.Fatalf("unexpected parse result\nwant:\n%s\ngot:\n%s", want, got) + } + + // Re-parse the String() output and ensure it round-trips to the same result. + tr2, err := parseTransformsProgram(got) + if err != nil { + t.Fatalf("cannot reparse transformations\nconfig:\n%s\nerror: %s", got, err) + } + got2 := tr2.String() + if got2 != want { + t.Fatalf("transformations are not the same after round-trip parsing\nwant:\n%s\ngot:\n%s", want, got2) + } + } + + // A single pipe at the top level. + from := "format foo as bar;" + into := "format foo as bar;" + f(from, into) + + // A single pipe at the top level. + from = "unpack_json;" + into = "unpack_json;" + f(from, into) + + // Pipes chain in a line. + from = "unpack_json | format foo as bar | unpack_words | pack_json;" + into = "unpack_json | format foo as bar | unpack_words | pack_json;" + f(from, into) + + // Pipes chain delimited by ';'. + from = `unpack_json; +format foo as bar; +unpack_words | pack_json; +pack_json;` + into = `unpack_json; +format foo as bar; +unpack_words | pack_json; +pack_json;` + f(from, into) + + // Trailing '|' continues a single pipe chain across newlines. + from = `unpack_json | +format foo as bar | +unpack_words | # foo bar +pack_json;` + into = `unpack_json | format foo as bar | unpack_words | pack_json;` + f(from, into) + + // Empty conditional block content. + from = `if (foo:=bar) {} else if (bar:=foo) {} else {}` + into = `if (foo:=bar) {} else if (bar:=foo) {} else {}` + f(from, into) + + // Conditional block. + from = `if (foo:=bar) { # foo bar + unpack_json; +}` + into = `if (foo:=bar) { + unpack_json; +}` + f(from, into) + + // Two adjacent 'if' blocks with no 'else' are independent chains. + from = `if (foo:=bar or '()') { + unpack_json; +} +if (a:=b or v:'(' or v:')') { + pack_json; +}` + into = `if (foo:=bar or "()") { + unpack_json; +} +if (a:=b or v:"(" or v:")") { + pack_json; +}` + f(from, into) + + // A full 'if ... else if ... else' chain is one transformIf. + from = `if (level:="") { + format unknown as level; +} else if (level:=error) { # foo bar + format critical as severity; +} else { + format normal as severity; +}` + into = `if (level:="") { + format unknown as level; +} else if (level:=error) { + format critical as severity; +} else { + format normal as severity; +}` + f(from, into) + + // An empty named block. + from = `block foobar {} +do foobar;` + into = `block foobar {} +do foobar;` + f(from, into) + + // Named block with content. + from = `block normalize { + unpack_json; + pack_json; + if (foo:=bar) { + unpack_words; + } +} +do normalize;` + into = `block normalize { + unpack_json; + pack_json; + if (foo:=bar) { + unpack_words; + } +} +do normalize;` + f(from, into) + + // Blocks declared at the end of the program. + from = `do a; +do b; +do a; +do b; +block a { + unpack_json; +} +block b { + pack_json; +}` + into = `block a { + unpack_json; +} +block b { + pack_json; +} +do a; +do b; +do a; +do b;` + f(from, into) + + // 'send' at the top level of program. + from = `send;` + into = `send;` + f(from, into) + + // 'send' inside a named block. + from = `block enrich { + if (foo:=bar) { + send; + } + pack_json; +} +do enrich;` + into = `block enrich { + if (foo:=bar) { + send; + } + pack_json; +} +do enrich;` + f(from, into) + + // 'return' at the top level of program. + from = `return;` + into = `return;` + f(from, into) + + // 'return' inside named block. + from = `block foobar { return; }` + into = `block foobar { + return; +} +` + f(from, into) + + // 'drop' at the top level of program. + from = `drop;` + into = `drop;` + f(from, into) + + // 'drop' inside a named block. + from = `block enrich { + if (foo:=bar) { + drop; + } + pack_json; +} +do enrich;` + into = `block enrich { + if (foo:=bar) { + drop; + } + pack_json; +} +do enrich;` + f(from, into) + + // Ensure every supported pipe parses without errors with semicolon at the end. + from = `coalesce (user_id, username, email) default "anonymous" as user; +collapse_nums at _msg; +copy host as server; +cp host as server; +decolorize; +del foo, bar; +delete foo, bar; +drop_empty_fields; +extract 'email: ,' from foo; +extract_regexp "(?P([0-9]+[.]){3}[0-9]+)" from _msg; +eval x+y; +fields foo, bar; +format foo as bar; +hash foo; +json_array_len (foo); +keep foo, bar; +len foo; +math x+y; +mv a as b; +pack_json; +pack_logfmt; +rename a as b; +replace ("secret-password", "***") at _msg; +replace_regexp ("host-(.+?)-foo", "$1") at _msg; +rm foo; +sample 100; +set_stream_fields host, path; +split "," from _msg; +time_add 1h; +unpack_json; +unpack_logfmt; +unpack_syslog; +unpack_words; +unroll by foo;` + into = `coalesce(user_id, username, email) default anonymous as user; +collapse_nums; +copy host as server; +copy host as server; +decolorize; +delete foo, bar; +delete foo, bar; +drop_empty_fields; +extract "email: ," from foo; +extract_regexp "(?P([0-9]+[.]){3}[0-9]+)"; +math (x + y) as "x + y"; +fields foo, bar; +format foo as bar; +hash(foo); +json_array_len(foo); +fields foo, bar; +len(foo); +math (x + y) as "x + y"; +rename a as b; +pack_json; +pack_logfmt; +rename a as b; +replace ("secret-password", "***"); +replace_regexp ("host-(.+?)-foo", "$1"); +delete foo; +sample 100; +set_stream_fields host, path; +split ","; +time_add 1h; +unpack_json; +unpack_logfmt; +unpack_syslog; +unpack_words; +unroll by (foo);` + f(from, into) +} + +func TestParseTransformsFailure(t *testing.T) { + f := func(s string) { + t.Helper() + tr, err := parseTransformsProgram(s) + if err == nil { + t.Fatalf("expected error when parsing\nconfig:\n%s\ngot:\n%s", s, tr.String()) + } + } + + // Empty program is not valid. + f("") + + // Line is not ended with ';'. + f(`unpack_json`) + + // Valid LogsQL, but not transforms config. + f("*;") + + // Curly braces at the top level. + f(`{}`) + f(`{ + format foo as bar; + pack_json; +}`) + f("delete x };") + f("delete x {;") + f("delete x | { delete y;") + f("delete x } delete y;") + + // Unknown pipe name. + f("foobar;") + + // Block declared with a reserved keyword as its name. + f(`block if {}`) + + // Two blocks declared with the same name. + f(`block a {} +block a {}`) + + // Block is declared with special symbols. + f(`block abc^$ {}`) + + // Quoted block name. + f(`block "abc" {}`) + + // Unterminated named block. + f(`block`) + f(`block foo;`) + f(`block foo {`) + f(`block foo { + unpack_words;`) + + // Unterminated conditional block. + f(`if (foo:=bar) { + unpack_json; `) + f(`if`) + f(`if (`) + f(`if (foo:=bar`) + f(`if (foo:=bar)`) + f(`if (foo:=bar) {`) + f(`if (foo:=bar) { +unpack_words;`) + + // 'if' condition without parentheses. + f(`if foo:=bar { + unpack_json; +}`) + + // Execute undeclared block. + f(`do foobar;`) + + // Recursive call. + f(`block foobar { + do foobar; +}`) + + // Nested recursive call. + f(`block foo { + do bar; +} +block bar { + do foo; +}`) + + // Deeply nested recursive call. + f(`block foo { + do bar; +} +block bar { + do baz; +} +block baz { + do foo; +}`) + + // Conditional recursive call. + f(`block foo { + do bar; +} +block bar { + if (abc) { + do baz; + } +} +block baz { + if (def) { + do foo; + } +}`) + + // Time filter. + f(`if (_time:5m) {}`) + + // Time filter in named blocks. + f(`block foo { + if (_time:offset 5m) {} +} +do foo;`) + + // Deeply nested time filter. + f(`block foo { do bar; } +block bar { do baz; } +block baz { if (_time:[5m, 1m]) {} } +do foo;`) + + // Time filter in a pipe. + f(`unpack_json if (foo:=bar and _time:7m);`) + + // Subquery. + f(`if (foo:=bar and level:in(* | fields level)) {}`) + + // Subquery in a pipe. + f(`unpack_json if (foo:=bar and level:in(* | fields level));`) +} + +func TestTransformsProgram(t *testing.T) { + f := func(program string, rows, rowsExpected [][]Field) { + t.Helper() + tr, err := parseTransformsProgram(program) + if err != nil { + t.Fatalf("cannot parse transforms config:\n%s\nerror: %s", program, err) + } + + workersCount := 5 + ppTest := newTestPipeProcessor() + pp := tr.newProcessor(workersCount, ppTest, ppTest) + + brw := newTestBlockResultWriter(workersCount, pp) + for _, row := range rows { + brw.writeRow(row) + } + if err := pp.flush(); err != nil { + t.Fatal(err) + } + brw.flush() + + ppTest.expectRows(t, rowsExpected) + } + + // Single 'unpack_json' pipe. + f("unpack_json;", [][]Field{ + {{"_msg", `{"a":"b","foo":"bar","ping":"pong"}`}}, + }, [][]Field{ + {{"_msg", `{"a":"b","foo":"bar","ping":"pong"}`}, {"a", "b"}, {"foo", "bar"}, {"ping", "pong"}}, + }) + + // Single 'copy' pipe. + f(`copy _msg as copy_msg;`, [][]Field{ + {{"_msg", "hello"}}, + }, [][]Field{ + {{"_msg", "hello"}, {"copy_msg", "hello"}}, + }) + + // Multiple pipes in a single line. + f(`unpack_json | delete a;`, [][]Field{ + {{"_msg", `{"a":"b","foo":"bar"}`}}, + }, [][]Field{ + {{"_msg", `{"a":"b","foo":"bar"}`}, {"foo", "bar"}}, + }) + + // Single 'if'. + f(`if (level:=error) { + unpack_json; +}`, [][]Field{ + {{"_msg", `{"a":"b","foo":"bar","ping":"pong"}`}}, + {{"_msg", `{"a":"b","foo":"bar","ping":"pong"}`}, {"level", "error"}}, + }, [][]Field{ + {{"_msg", `{"a":"b","foo":"bar","ping":"pong"}`}}, + {{"_msg", `{"a":"b","foo":"bar","ping":"pong"}`}, {"level", "error"}, {"a", "b"}, {"foo", "bar"}, {"ping", "pong"}}, + }) + + // if-else chain. + f(`if (level:=error) { + format first as branch; +} else if (level:=warn) { + format second as branch; +} else { + format third as branch; +}`, [][]Field{ + {{"_msg", "1"}, {"level", "error"}}, + {{"_msg", "2"}, {"level", "warn"}}, + {{"_msg", "3"}, {"level", "info"}}, + }, [][]Field{ + {{"_msg", "1"}, {"level", "error"}, {"branch", "first"}}, + {{"_msg", "2"}, {"level", "warn"}, {"branch", "second"}}, + {{"_msg", "3"}, {"level", "info"}, {"branch", "third"}}, + }) + + // Nested 'if'. + f(`if (level:=error) { + if (code:=500) { + format crit as severity; + } +}`, [][]Field{ + {{"_msg", "1"}, {"level", "error"}, {"code", "500"}}, + {{"_msg", "2"}, {"level", "error"}, {"code", "200"}}, + {{"_msg", "3"}, {"level", "info"}}, + }, [][]Field{ + {{"_msg", "1"}, {"level", "error"}, {"code", "500"}, {"severity", "crit"}}, + {{"_msg", "2"}, {"level", "error"}, {"code", "200"}}, + {{"_msg", "3"}, {"level", "info"}}, + }) + + // 'return' at the top level passes rows through; transforms after it are dead code. + f(`delete x; +return; +delete y;`, [][]Field{ + {{"_msg", "m"}, {"x", "abc"}, {"y", "def"}}, + }, [][]Field{ + {{"_msg", "m"}, {"y", "def"}}, + }) + + // 'send' at the top level passes rows through; transforms after it are dead code. + f(`delete x; +send; +delete y;`, [][]Field{ + {{"_msg", "m"}, {"x", "abc"}, {"y", "def"}}, + }, [][]Field{ + {{"_msg", "m"}, {"y", "def"}}, + }) + + // Named block call. + f(`block set_foo_bar { + format bar as foo; +} +do set_foo_bar;`, [][]Field{ + {{"_msg", "abc"}}, + }, [][]Field{ + {{"_msg", "abc"}, {"foo", "bar"}}, + }) + + // Early return inside a named block. + f(`block normalize_errors { + if (!level:=error) { + return; + } + unpack_json; +} +do normalize_errors; +if (_msg:*) { + format processed as status; +}`, [][]Field{ + {{"_msg", `{"parsed":false}`}, {"level", "info"}}, + {{"_msg", `{"parsed":true}`}, {"level", "error"}}, + }, [][]Field{ + {{"_msg", `{"parsed":false}`}, {"level", "info"}, {"status", "processed"}}, + {{"_msg", `{"parsed":true}`}, {"level", "error"}, {"parsed", "true"}, {"status", "processed"}}, + }) + + // Nested named block call. + f(`block c { delete x; } +block b { do c; } +block a { do b; } +do a;`, [][]Field{ + {{"_msg", "foobar"}, {"x", "abc"}}, + }, [][]Field{ + {{"_msg", "foobar"}}, + }) + + // 'do' inside an 'if' branch. + f(`block enrich { + format yes as enriched; +} +if (level:=error) { + do enrich; +}`, [][]Field{ + {{"_msg", "1"}, {"level", "error"}}, + {{"_msg", "2"}, {"level", "info"}}, + }, [][]Field{ + {{"_msg", "1"}, {"level", "error"}, {"enriched", "yes"}}, + {{"_msg", "2"}, {"level", "info"}}, + }) + + // 'do' with no 'return'. + f(`block enrich { + format yes as enriched; +} +do enrich; +format done as status;`, [][]Field{ + {{"_msg", "x"}}, + }, [][]Field{ + {{"_msg", "x"}, {"enriched", "yes"}, {"status", "done"}}, + }) + + // 'return' inside a 'do'. + f(`block maybe_skip { + if ("skip":=yes) { + return; + } + format bar as foo; +} +do maybe_skip; +format finished as status;`, [][]Field{ + {{"_msg", "a"}, {"skip", "yes"}}, + {{"_msg", "b"}, {"skip", "no"}}, + }, [][]Field{ + {{"_msg", "a"}, {"skip", "yes"}, {"status", "finished"}}, + {{"_msg", "b"}, {"skip", "no"}, {"foo", "bar"}, {"status", "finished"}}, + }) + + // Drop all the logs. + f(`drop;`, [][]Field{ + {{"_msg", "a"}, {"skip", "yes"}}, + {{"_msg", "b"}, {"skip", "no"}}, + }, [][]Field{}) + + // Drop based on condition. + f(`if (level:=debug) { drop; }`, [][]Field{ + {{"_msg", "a"}, {"level", "debug"}}, + {{"_msg", "b"}, {"level", "info"}}, + {{"_msg", "c"}}, + }, [][]Field{ + {{"_msg", "b"}, {"level", "info"}}, + {{"_msg", "c"}}, + }) + + // Drop based on condition inside a named block. + f(`block foo { + do bar; +} +block bar { + do baz; +} +block baz { + if (!drop:=true) { + return; + } + drop; +} +do baz;`, [][]Field{ + {{"_msg", "a"}, {"drop", "true"}}, + {{"_msg", "b"}, {"drop", "false"}}, + {{"_msg", "c"}}, + }, [][]Field{ + {{"_msg", "b"}, {"drop", "false"}}, + {{"_msg", "c"}}, + }) +} + +func TestTransformsProgramSetTime(t *testing.T) { + f := func(program string, timestamp, timestampExpected int64) { + t.Helper() + + var timestampsGot []int64 + storeTimestamp := func(lr *LogRows) { + lr.ForEachRow(func(_ uint64, r *InsertRow) { + timestampsGot = append(timestampsGot, r.Timestamp) + }) + } + + lr := GetLogRows(nil, nil, nil, nil, "") + defer PutLogRows(lr) + fields := []Field{ + {"_msg", "foobar"}, + } + lr.MustAdd(TenantID{}, timestamp, fields, -1) + + prog, err := ParseTransformsProgram(program) + if err != nil { + t.Fatal(err) + } + tr := prog.NewTransformer(storeTimestamp) + tr.Transform(lr) + + if len(timestampsGot) != 1 { + t.Fatalf("expected 1 timestamp, got %d", len(timestampsGot)) + } + tsGot := timestampsGot[0] + if tsGot != timestampExpected { + t.Fatalf("unexpected timestamp; got %v; want %v", tsGot, timestampExpected) + } + } + + // Modify original _time field. + f(`time_add 0h;`, 1234, 1234) + f(`time_add 1h;`, 1, 1+nsecsPerHour) + f(`time_add 2h;`, 1234, 1234+2*nsecsPerHour) + + // Change the type of the '_time' field to string and then modify it. + f(`format 1970-01-01T12:00:00Z as _time | time_add 1h;`, 1234, 13*nsecsPerHour) + + // Drop the _time field and then modify it. + f(`delete _time | time_add 1h;`, 1234, 0) +} + +// TestTransformsProgramSetTimeWithBlockSource ensures blockResult with brSrc correctly restores modified _time column. +// See https://github.com/VictoriaMetrics/VictoriaLogs/pull/1508#discussion_r3523272574 +func TestTransformsProgramSetTimeWithBlockSource(t *testing.T) { + var timestampsGot []int64 + storeTimestamp := func(lr *LogRows) { + lr.ForEachRow(func(_ uint64, r *InsertRow) { + timestampsGot = append(timestampsGot, r.Timestamp) + }) + } + + lr := GetLogRows(nil, nil, nil, nil, "") + defer PutLogRows(lr) + lr.MustAdd(TenantID{}, 1234, []Field{{"_msg", "match"}}, -1) + lr.MustAdd(TenantID{}, 1234, []Field{{"_msg", "does-not-match"}}, -1) + + prog, err := ParseTransformsProgram(`if (="match") { time_add 1h; }`) + if err != nil { + t.Fatal(err) + } + tr := prog.NewTransformer(storeTimestamp) + tr.Transform(lr) + + timestampsExpected := []int64{1234 + nsecsPerHour, 1234} + if len(timestampsGot) != len(timestampsExpected) { + t.Fatalf("expected %d timestamp, got %d", len(timestampsExpected), len(timestampsGot)) + } + if !reflect.DeepEqual(timestampsGot, timestampsExpected) { + t.Fatalf("unexpected timestamps\ngot:\n%v\nwant:\n%v", timestampsGot, timestampsExpected) + } +} + +func TestTransformsProgramSetTenantID(t *testing.T) { + defaultTenantID := TenantID{ + AccountID: 1234, + ProjectID: 5678, + } + f := func(program string, row []Field, tenantsExpected []TenantID) { + t.Helper() + + var tenantsGot []TenantID + storeTenant := func(lr *LogRows) { + lr.ForEachRow(func(_ uint64, r *InsertRow) { + tenantsGot = append(tenantsGot, r.TenantID) + }) + } + + lr := GetLogRows(nil, nil, nil, nil, "") + defer PutLogRows(lr) + lr.MustAdd(defaultTenantID, time.Now().UnixNano(), row, -1) + + prog, err := ParseTransformsProgram(program) + if err != nil { + t.Fatal(err) + } + tr := prog.NewTransformer(storeTenant) + tr.Transform(lr) + + if !reflect.DeepEqual(tenantsGot, tenantsExpected) { + t.Fatalf("unexpected tenant are set\ngot\n%v\nwant\n%v", tenantsGot, tenantsExpected) + } + } + + // vl_account_id and vl_project_id fields are not touched. + f(`unpack_json;`, []Field{}, []TenantID{defaultTenantID}) + + // Ignore vl_account_id and vl_project_id fields when they come from log rows. + f(`unpack_json;`, []Field{ + {"_msg", "foobar"}, + {"vl_account_id", "42"}, + {"vl_project_id", "42"}, + }, []TenantID{defaultTenantID}) + + // vl_account_id and vl_project_id fields come from the unpack_json pipe. + f(`unpack_json;`, []Field{ + {"_msg", `{"foo":"bar","vl_account_id":11,"vl_project_id":"12"}`}, + }, []TenantID{{11, 12}}) + + // vl_account_id and vl_project_id fields are set via the format pipe. + f(`format 3 as vl_account_id | format 4 as vl_project_id;`, []Field{ + {"_msg", `foobar`}, + {"vl_account_id", "1"}, + {"vl_project_id", "2"}, + }, []TenantID{{3, 4}}) + + // vl_account_id and vl_project_id fields come from the unroll pipe. + f(`unroll by (_msg) | unpack_json;`, []Field{ + {"_msg", `[{"foo":"bar","vl_account_id":1,"vl_project_id":2}, {"bar":"baz","vl_account_id":3,"vl_project_id":4}, {"ping":"pong"}]`}, + {"vl_account_id", "3"}, + }, []TenantID{{1, 2}, {3, 4}, defaultTenantID}) +} diff --git a/lib/logstorage/transforms_timing_test.go b/lib/logstorage/transforms_timing_test.go new file mode 100644 index 0000000000..66e14b6c7b --- /dev/null +++ b/lib/logstorage/transforms_timing_test.go @@ -0,0 +1,187 @@ +package logstorage + +import ( + "testing" + "time" +) + +func BenchmarkTransforms(b *testing.B) { + b.Run("set-tenant", func(b *testing.B) { + program := `if (source:=Kubernetes) { + format 1 as vl_account_id; format 2 as vl_project_id; + } else if (source:=frontend) { + format 3 as vl_account_id; format 4 as vl_project_id; + } else if (source:=iOS) { + format 5 as vl_account_id; format 6 as vl_project_id; + } else if (source:=Android) { + format 7 as vl_account_id; format 8 as vl_project_id; + } else { + format 0 as vl_account_id; format 0 as vl_project_id; + }` + content := []string{ + `{"_msg":"a message from Kubernetes", "source": "Kubernetes", "kubernetes.pod_labels.app":"VictoriaStack"}`, + `{"_msg":"a message from front-end", "source": "fe", "page":"VMUI-web"}`, + `{"_msg":"a message from iOS", "source": "iOS", "service":"VMUI-iOS"}`, + `{"_msg":"a message from Android", "source": "Android", "app":"VMUI-Android"}`, + } + benchmarkTransforms(b, program, content) + }) + + b.Run("normalize-logfmt", func(b *testing.B) { + program := `unpack_logfmt | rename msg as _msg, time as _time;` + content := []string{ + `{"_msg":"level=error msg=\"context canceled\" db_host=127.0.0.1 retry_count=5 time=2026-06-10T11:02:02.000Z"}`, + `{"_msg":"not a logfmt-encoded message"}`, + `{"_msg":"msg=\"new request\" method=GET path=/login host=example.com status=200 duration=12ms"}`, + } + benchmarkTransforms(b, program, content) + }) + + b.Run("normalize-json", func(b *testing.B) { + program := `unpack_json from payload | rename msg as _msg, time as _time | delete payload;` + content := []string{ + `{"payload":"{\"level\":\"error\",\"msg\":\"context canceled\",\"db_host\":\"127.0.0.1\",\"retry_count\":5,\"time\":\"2026-06-10T11:02:02.000Z\"}"}`, + `{"payload":"not a json-encoded message"}`, + `{"payload":"{\"msg\":\"new request\",\"method\":\"GET\",\"path\":\"/login\",\"host\":\"example.com\",\"status\":200,\"duration\":\"12ms\"}"}`, + } + benchmarkTransforms(b, program, content) + }) + + b.Run("simplify-k8s-labels", func(b *testing.B) { + program := `rename kubernetes.pod_labels.* as pod.*, kubernetes.pod_annotations.* as pod.*, kubernetes.* as k8s.*, k8s.pod_ip as ip, k8s.pod_name as pod, k8s.pod_node_name as node;` + content := []string{ + `{ + "_msg": "Order confirmation email sent to: reed@example.com", + "collector": "vlagent", + "kubernetes.container_id": "containerd://da795e859cba3854e0f5e31ea5ae17279380499fdea05fc34ebc847bce4e31a5", + "kubernetes.container_name": "email", + "kubernetes.pod_ip": "10.71.1.150", + "kubernetes.pod_labels.app.kubernetes.io/component": "email", + "kubernetes.pod_labels.app.kubernetes.io/name": "email", + "kubernetes.pod_labels.opentelemetry.io/name": "email", + "kubernetes.pod_labels.pod-template-hash": "686bdfd59b", + "kubernetes.pod_labels.topology.kubernetes.io/region": "us-east1", + "kubernetes.pod_labels.topology.kubernetes.io/zone": "us-east1-b", + "kubernetes.pod_name": "email-686bdfd59b-dd79g", + "kubernetes.pod_namespace": "play-otel", + "kubernetes.pod_node_name": "gke-sandbox-n2d-std-8-202603301026051-852214e6-v88y" + }`, + `{ + "_msg": "Order confirmation email sent to: jack@example.com", + "collector": "vector", + "file": "/var/log/pods/play-otel_email-686bdfd59b-dd79g_745f66b9-2e21-4ee9-a053-a7cc849a0083/email/0.log", + "kubernetes.container_id": "containerd://da795e859cba3854e0f5e31ea5ae17279380499fdea05fc34ebc847bce4e31a5", + "kubernetes.container_image": "ghcr.io/open-telemetry/demo:2.1.3-email", + "kubernetes.container_image_id": "ghcr.io/open-telemetry/demo@sha256:5a62bbc4c7f34292c37b7b04b5a85bb74fc7a50c99b0429bea684602472c211d", + "kubernetes.container_name": "email", + "kubernetes.namespace_labels.kubernetes.io/metadata.name": "play-otel", + "kubernetes.node_labels.beta.kubernetes.io/arch": "amd64", + "kubernetes.node_labels.beta.kubernetes.io/instance-type": "n2d-standard-8", + "kubernetes.node_labels.beta.kubernetes.io/os": "linux", + "kubernetes.node_labels.cloud.google.com/gke-boot-disk": "pd-balanced", + "kubernetes.node_labels.cloud.google.com/gke-container-runtime": "containerd", + "kubernetes.node_labels.cloud.google.com/gke-cpu-scaling-level": "8", + "kubernetes.node_labels.cloud.google.com/gke-logging-variant": "DEFAULT", + "kubernetes.node_labels.cloud.google.com/gke-max-pods-per-node": "110", + "kubernetes.node_labels.cloud.google.com/gke-memory-gb-scaling-level": "32", + "kubernetes.node_labels.cloud.google.com/gke-netd-ready": "true", + "kubernetes.node_labels.cloud.google.com/gke-nodepool": "n2d-std-8-20260330102605160300000001", + "kubernetes.node_labels.cloud.google.com/gke-os-distribution": "cos", + "kubernetes.node_labels.cloud.google.com/gke-provisioning": "standard", + "kubernetes.node_labels.cloud.google.com/gke-stack-type": "IPV4", + "kubernetes.node_labels.cloud.google.com/machine-family": "n2d", + "kubernetes.node_labels.disk-type.gke.io/hyperdisk-throughput": "true", + "kubernetes.node_labels.disk-type.gke.io/pd-balanced": "true", + "kubernetes.node_labels.disk-type.gke.io/pd-extreme": "true", + "kubernetes.node_labels.disk-type.gke.io/pd-ssd": "true", + "kubernetes.node_labels.disk-type.gke.io/pd-standard": "true", + "kubernetes.node_labels.failure-domain.beta.kubernetes.io/region": "us-east1", + "kubernetes.node_labels.failure-domain.beta.kubernetes.io/zone": "us-east1-b", + "kubernetes.node_labels.iam.gke.io/gke-metadata-server-enabled": "true", + "kubernetes.node_labels.kubernetes.io/arch": "amd64", + "kubernetes.node_labels.kubernetes.io/hostname": "gke-sandbox-n2d-std-8-202603301026051-852214e6-v88y", + "kubernetes.node_labels.kubernetes.io/os": "linux", + "kubernetes.node_labels.node.kubernetes.io/instance-type": "n2d-standard-8", + "kubernetes.node_labels.topology.gke.io/zone": "us-east1-b", + "kubernetes.node_labels.topology.kubernetes.io/region": "us-east1", + "kubernetes.node_labels.topology.kubernetes.io/zone": "us-east1-b", + "kubernetes.pod_ip": "10.71.1.150", + "kubernetes.pod_ips": "[\"10.71.1.150\"]", + "kubernetes.pod_labels.app.kubernetes.io/component": "email", + "kubernetes.pod_labels.app.kubernetes.io/name": "email", + "kubernetes.pod_labels.opentelemetry.io/name": "email", + "kubernetes.pod_labels.pod-template-hash": "686bdfd59b", + "kubernetes.pod_labels.topology.kubernetes.io/region": "us-east1", + "kubernetes.pod_labels.topology.kubernetes.io/zone": "us-east1-b", + "kubernetes.pod_name": "email-686bdfd59b-dd79g", + "kubernetes.pod_namespace": "play-otel", + "kubernetes.pod_node_name": "gke-sandbox-n2d-std-8-202603301026051-852214e6-v88y", + "kubernetes.pod_owner": "ReplicaSet/email-686bdfd59b", + "kubernetes.pod_uid": "745f66b9-2e21-4ee9-a053-a7cc849a0083", + "source_type": "kubernetes_logs", + "stream": "stdout" + }`, + `{ + "_msg": "Product Found", + "app.product.id": "L9ECAV7KIM", + "app.product.name": "Lens Cleaning Kit", + "collector": "otel-collector", + "host.name": "otel-collector-7f8479fcb7-s7kc2", + "k8s.deployment.name": "product-catalog", + "k8s.namespace.name": "play-otel", + "k8s.node.name": "gke-sandbox-n2d-std-8-202603301026051-852214e6-fx1b", + "k8s.pod.ip": "10.71.10.33", + "k8s.pod.name": "product-catalog-759d5f4b46-f6zfg", + "k8s.pod.start_time": "2026-06-05T10:07:49Z", + "k8s.pod.uid": "97463b02-e9d7-4a43-b164-e21c837e7b48", + "os.type": "linux", + "scope.name": "product-catalog", + "scope.version": "unknown", + "service.instance.id": "97463b02-e9d7-4a43-b164-e21c837e7b48", + "service.name": "product-catalog", + "service.namespace": "opentelemetry-demo", + "service.version": "2.1.3", + "severity_number": "9", + "severity_text": "INFO", + "span_id": "200124c35ba77ae1", + "telemetry.sdk.language": "go", + "telemetry.sdk.name": "opentelemetry", + "telemetry.sdk.version": "1.38.0", + "trace_id": "c56f09f96d570b55ae1890123ecbe4ce" + }`, + } + benchmarkTransforms(b, program, content) + }) +} + +func benchmarkTransforms(b *testing.B, program string, rows []string) { + b.Helper() + prog, err := ParseTransformsProgram(program) + if err != nil { + b.Fatal(err) + } + transformer := prog.NewTransformer(func(_ *LogRows) { + }) + + p := GetJSONParser() + defer PutJSONParser(p) + lr := GetLogRows(nil, nil, nil, nil, "missing _msg field in a bench") + defer PutLogRows(lr) + + rowsSize := 0 + for _, row := range rows { + rowsSize += len(row) + if err := p.ParseLogMessage([]byte(row), nil, ""); err != nil { + b.Fatal(err) + } + lr.MustAdd(TenantID{}, time.Now().UnixNano(), p.Fields, -1) + } + + b.ReportAllocs() + b.SetBytes(int64(rowsSize)) + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + transformer.Transform(lr) + } + }) +}