Render Mustache expressions in routing rules datasets values

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

@@ -55,6 +55,8 @@ github.com/boumenot/gocover-cobertura v1.2.0 h1:g+VROIASoEHBrEilIyaCmgo7HGm+AV5y
 github.com/boumenot/gocover-cobertura v1.2.0/go.mod h1:fz7ly8dslE42VRR5ZWLt2OHGDHjkTiA2oNvKgJEjLT0=
 github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
 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 v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=

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