Render Mustache expressions in routing rules datasets values (#1393)

* Render Mustache expressions in datasets value

This is a quick PoC to test using the github.com/hoisie/mustache
library to render Mustache expressions in datasets values.

Datasets values containing mustache expressions like:

    {{kubernetes.labels.elastic_co/dataset}}

are rendered using document values.

Currently, the principal limit is the validator is using processed
documents (docs returned by the ingest pipeline), but for this
purpose of using the original documents instead (docs read from the
test case file).

* Add tests for dataset expressions rendering

* Switch to the github.com/cbroglie/mustache library

I just noticed the original github.com/hoisie/mustache seems
unmaintained (last commit happened on Aug 5, 2016).

So I switched to github.com/cbroglie/mustache, a fork of the same
repo that looks active.

They added essentials improvements, like returning an error on the
`Render()` API.

* Fix routing_rules.yml format

I didn't run `elastic-package check` in the kubernetes test package.

Without a proper format, the test detects a dirty git working
directory at the end of the test run.

* Remove the kubernetes test package

We already have one at:

    test/packages/with-kind/kubernetes

So this one is redundant.

Dropping it!

* Add a test rule with expression

To test that routing rules work with datasets and namespaces, we add
another rule that uses the `{{cloud.account.id}}` for both if the
`cloud.region` is `eu-west-1`.

* Update expression tests

Changing the expression test to make it more realistic. The main
challenge here is to find a source event field value I can use for a
dataset or namespace: it can't contain `-`.

* Update internal/fields/validate.go

Co-authored-by: Jaime Soriano Pastor <[email protected]>

* Make static checker happy

* Make more realistic tests adding a `labels` field

Even if Firehose currently does not provide an `aws.labels` field,
using it in the routing rules tests is probably less confusing than
using the field `{{cloud.account.id}}` as target dataset or namespace.

---------

Co-authored-by: Jaime Soriano Pastor <[email protected]>

Author: zmoog

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/ProtonMail/gopenpgp/v2 v2.7.2
 	github.com/aymerick/raymond v2.0.2+incompatible
 	github.com/boumenot/gocover-cobertura v1.2.0
+	github.com/cbroglie/mustache v1.4.0
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/dustin/go-humanize v1.0.1
 	github.com/elastic/elastic-integration-corpus-generator-tool v0.5.0

go.sum

@@ -51,6 +51,8 @@ github.com/aymerick/raymond v2.0.2+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76e
 github.com/boumenot/gocover-cobertura v1.2.0 h1:g+VROIASoEHBrEilIyaCmgo7HGm+AV5yKEPLk0qIY+s=
 github.com/boumenot/gocover-cobertura v1.2.0/go.mod h1:fz7ly8dslE42VRR5ZWLt2OHGDHjkTiA2oNvKgJEjLT0=
 github.com/bwesterb/go-ristretto v1.2.3/go.mod h1:fUIoIZaG73pV5biE2Blr2xEzDoMj7NFEuV9ekS419A0=
+github.com/cbroglie/mustache v1.4.0 h1:Azg0dVhxTml5me+7PsZ7WPrQq1Gkf3WApcHMjMprYoU=
+github.com/cbroglie/mustache v1.4.0/go.mod h1:SS1FTIghy0sjse4DUVGV1k/40B1qE1XkD9DtDsHo9iM=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=

internal/fields/validate.go

@@ -18,6 +18,7 @@ import (
 	"strings"
 
 	"github.com/Masterminds/semver/v3"
+	"github.com/cbroglie/mustache"
 	"gopkg.in/yaml.v3"
 
 	"github.com/elastic/elastic-package/internal/common"
@@ -316,12 +317,33 @@ func (v *Validator) validateDocumentValues(body common.MapStr) multierror.Error
 	if !v.specVersion.LessThan(semver2_0_0) && v.expectedDatasets != nil {
 		for _, datasetField := range datasetFieldNames {
 			value, err := body.GetValue(datasetField)
-			if err == common.ErrKeyNotFound {
+			if errors.Is(err, common.ErrKeyNotFound) {
 				continue
 			}
 
+			// Why do we render the expected datasets here?
+			// Because the expected datasets can contain
+			// mustache templates, and not just static
+			// strings.
+			//
+			// For example, the expected datasets for the
+			// Kubernetes container logs dataset can be:
+			//
+			//   - "{{kubernetes.labels.elastic_co/dataset}}"
+			//
+			var renderedExpectedDatasets []string
+			for _, dataset := range v.expectedDatasets {
+				renderedDataset, err := mustache.Render(dataset, body)
+				if err != nil {
+					err := fmt.Errorf("can't render expected dataset %q: %w", dataset, err)
+					errs = append(errs, err)
+					return errs
+				}
+				renderedExpectedDatasets = append(renderedExpectedDatasets, renderedDataset)
+			}
+
 			str, ok := valueToString(value, v.disabledNormalization)
-			exists := stringInArray(str, v.expectedDatasets)
+			exists := stringInArray(str, renderedExpectedDatasets)
 			if !ok || !exists {
 				err := fmt.Errorf("field %q should have value in %q, it has \"%v\"",
 					datasetField, v.expectedDatasets, value)

test/packages/parallel/awsfirehose/data_stream/log/_dev/test/pipeline/test-vpcflow-log.json

@@ -0,0 +1,34 @@
+{
+    "events": [
+        {
+            "cloud": {
+                "region": "us-east-2",
+                "account": {
+                    "id": "428152502467"
+                },
+                "provider": "aws"
+            },
+            "aws": {
+                "firehose": {
+                    "arn": "arn:aws:firehose:us-east-2:428152502467:deliverystream/test-vpcflow-logs",
+                    "request_id": "1cfbed13-d631-4b8b-b20a-b7c5bf8fcd00"
+                },
+                "kinesis": {
+                    "name": "test-vpcflow-logs",
+                    "type": "deliverystream"
+                },
+                "labels": {
+                    "elastic_co/dataset": "mydataset",
+                    "elastic_co/namespace": "mynamespace"
+                }
+            },
+            "data_stream": {
+                "namespace": "default",
+                "type": "logs",
+                "dataset": "generic"
+            },
+            "@timestamp": "2023-08-23T16:47:26Z",
+            "message": "{\"message\":\"2 428152502467 eni-0b584e1c714317ac6 176.111.174.91 10.0.0.102 41536 1135 6 1 40 1692809104 1692809162 REJECT OK\"}\n"
+        }
+    ]
+}

test/packages/parallel/awsfirehose/data_stream/log/_dev/test/pipeline/test-vpcflow-log.json-expected.json

@@ -0,0 +1,37 @@
+{
+    "expected": [
+        {
+            "@timestamp": "2023-08-23T16:47:26Z",
+            "aws": {
+                "firehose": {
+                    "arn": "arn:aws:firehose:us-east-2:428152502467:deliverystream/test-vpcflow-logs",
+                    "request_id": "1cfbed13-d631-4b8b-b20a-b7c5bf8fcd00"
+                },
+                "kinesis": {
+                    "name": "test-vpcflow-logs",
+                    "type": "deliverystream"
+                },
+                "labels": {
+                    "elastic_co/dataset": "mydataset",
+                    "elastic_co/namespace": "mynamespace"
+                }
+            },
+            "cloud": {
+                "account": {
+                    "id": "428152502467"
+                },
+                "provider": "aws",
+                "region": "us-east-2"
+            },
+            "data_stream": {
+                "dataset": "mydataset",
+                "namespace": "mynamespace",
+                "type": "logs"
+            },
+            "ecs": {
+                "version": "8.0.0"
+            },
+            "message": "{\"message\":\"2 428152502467 eni-0b584e1c714317ac6 176.111.174.91 10.0.0.102 41536 1135 6 1 40 1692809104 1692809162 REJECT OK\"}\n"
+        }
+    ]
+}

test/packages/parallel/awsfirehose/data_stream/log/fields/fields.yml

@@ -38,3 +38,12 @@
           type: keyword
           description: |-
             Kinesis type.
+    - name: labels
+      type: object
+      fields:
+        - name: elastic_co/dataset
+          type: keyword
+          description: Fictional field to test datasets routing rules.
+        - name: elastic_co/namespace
+          type: keyword
+          description: Fictional field to test namespaces routing rules.

test/packages/parallel/awsfirehose/data_stream/log/routing_rules.yml

@@ -1,6 +1,11 @@
 - source_dataset: awsfirehose.log
   rules:
+    - target_dataset:
+        - "{{aws.labels.elastic_co/dataset}}"
+      namespace:
+        - "{{aws.labels.elastic_co/namespace}}"
+      if: "ctx.aws?.labels != null"
     - target_dataset: aws.cloudtrail
-      if: ctx['aws.cloudwatch.log_stream'].contains('CloudTrail')
+      if: ctx['aws.cloudwatch.log_stream'].contains('CloudTrail') == true
       namespace:
         - default