Validate the proper value of event.dataset

docs/howto/update_major_package_spec.md

@@ -59,3 +59,32 @@ some of the expected values.
 
 For example if a document contains `event.category: web`, the value of
 `event.type` must be `access`, `error` or `info` according to ECS 8.4.
+
+### field "event.dataset" should have value ..., it has ...
+
+The fields `event.dataset` and `data_stream.dataset` should contain the name of
+the package and the name of the data stream that generates it, separated by a
+dot. For example for documents of the "access" data stream of the Apache module,
+it should be `apache.access`.
+
+If these fields are not being correctly populated, look for the source of the
+value.
+
+If it is a constant keyword, review the configured value.
+```
+- name: event.dataset
+  type: constant_keyword
+  external: ecs
+  value: "apache.access"
+```
+
+If the value comes with an unexpected value from the collector, you can override
+it in the pipeline:
+```
+- set:
+    field: event.dataset
+    value: "apache.access"
+```
+
+Changing the value of `event.dataset` can be considered a breaking change, take
+this into account in your package when adding the changelog entry.

internal/fields/validate.go

@@ -38,6 +38,9 @@ type Validator struct {
 	// SpecVersion contains the version of the spec used by the package.
 	specVersion semver.Version
 
+	// expectedDataset contains the value expected for dataset fields.
+	expectedDataset string
+
 	defaultNumericConversion bool
 	numericKeywordFields     map[string]struct{}
 
@@ -98,6 +101,14 @@ func WithEnabledAllowedIPCheck() ValidatorOption {
 	}
 }
 
+// WithExpectedDataset configures the validator to check if the dataset fields have the expected values.
+func WithExpectedDataset(dataset string) ValidatorOption {
+	return func(v *Validator) error {
+		v.expectedDataset = dataset
+		return nil
+	}
+}
+
 // CreateValidatorForDirectory function creates a validator for the directory.
 func CreateValidatorForDirectory(fieldsParentDir string, opts ...ValidatorOption) (v *Validator, err error) {
 	v = new(Validator)
@@ -202,13 +213,38 @@ func (v *Validator) ValidateDocumentBody(body json.RawMessage) multierror.Error
 
 // ValidateDocumentMap validates the provided document as common.MapStr.
 func (v *Validator) ValidateDocumentMap(body common.MapStr) multierror.Error {
-	errs := v.validateMapElement("", body, body)
+	errs := v.validateDocumentValues(body)
+	errs = append(errs, v.validateMapElement("", body, body)...)
 	if len(errs) == 0 {
 		return nil
 	}
 	return errs
 }
 
+var datasetFieldNames = []string{
+	"event.dataset",
+	"data_stream.dataset",
+}
+
+func (v *Validator) validateDocumentValues(body common.MapStr) multierror.Error {
+	var errs multierror.Error
+	if !v.specVersion.LessThan(semver2_0_0) && v.expectedDataset != "" {
+		for _, datasetField := range datasetFieldNames {
+			value, err := body.GetValue(datasetField)
+			if err == common.ErrKeyNotFound {
+				continue
+			}
+			str, ok := value.(string)
+			if !ok || str != v.expectedDataset {
+				err := errors.Errorf("field %q should have value %q, it has \"%v\"",
+					datasetField, v.expectedDataset, value)
+				errs = append(errs, err)
+			}
+		}
+	}
+	return errs
+}
+
 func (v *Validator) validateMapElement(root string, elem common.MapStr, doc common.MapStr) multierror.Error {
 	var errs multierror.Error
 	for name, val := range elem {

internal/fields/validate_test.go

@@ -165,6 +165,61 @@ func TestValidate_ExpectedEventType(t *testing.T) {
 	}
 }
 
+func TestValidate_ExpectedDataset(t *testing.T) {
+	validator, err := CreateValidatorForDirectory("testdata",
+		WithSpecVersion("2.0.0"),
+		WithExpectedDataset("apache.status"),
+	)
+	require.NoError(t, err)
+	require.NotNil(t, validator)
+
+	cases := []struct {
+		title string
+		doc   common.MapStr
+		valid bool
+	}{
+		{
+			title: "valid dataset",
+			doc: common.MapStr{
+				"event.dataset": "apache.status",
+			},
+			valid: true,
+		},
+		{
+			title: "empty dataset",
+			doc: common.MapStr{
+				"event.dataset": "",
+			},
+			valid: false,
+		},
+		{
+			title: "absent dataset",
+			doc:   common.MapStr{},
+			valid: true,
+		},
+		{
+			title: "wrong dataset",
+			doc: common.MapStr{
+				"event.dataset": "httpd.status",
+			},
+			valid: false,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			errs := validator.ValidateDocumentMap(c.doc)
+			if c.valid {
+				assert.Empty(t, errs)
+			} else {
+				if assert.Len(t, errs, 1) {
+					assert.Contains(t, errs[0].Error(), `field "event.dataset" should have value`)
+				}
+			}
+		})
+	}
+}
+
 func Test_parseElementValue(t *testing.T) {
 	for _, test := range []struct {
 		key        string

internal/testrunner/runners/pipeline/runner.go

@@ -139,12 +139,14 @@ func (r *runner) run() ([]testrunner.TestResult, error) {
 		}
 
 		tr.TimeElapsed = time.Since(startTime)
+		expectedDataset := pkgManifest.Name + "." + r.options.TestFolder.DataStream
 		fieldsValidator, err := fields.CreateValidatorForDirectory(dataStreamPath,
 			fields.WithSpecVersion(pkgManifest.SpecVersion),
 			fields.WithNumericKeywordFields(tc.config.NumericKeywordFields),
 			// explicitly enabled for pipeline tests only
 			// since system tests can have dynamic public IPs
 			fields.WithEnabledAllowedIPCheck(),
+			fields.WithExpectedDataset(expectedDataset),
 		)
 		if err != nil {
 			return nil, errors.Wrapf(err, "creating fields validator for data stream failed (path: %s, test case file: %s)", dataStreamPath, testCaseFile)

internal/testrunner/runners/static/runner.go

@@ -95,9 +95,12 @@ func (r runner) verifySampleEvent(pkgManifest *packages.PackageManifest) []testr
 		return results
 	}
 
+	expectedDataset := pkgManifest.Name + "." + r.options.TestFolder.DataStream
 	fieldsValidator, err := fields.CreateValidatorForDirectory(dataStreamPath,
 		fields.WithSpecVersion(pkgManifest.SpecVersion),
-		fields.WithDefaultNumericConversion())
+		fields.WithDefaultNumericConversion(),
+		fields.WithExpectedDataset(expectedDataset),
+	)
 	if err != nil {
 		results, _ := resultComposer.WithError(errors.Wrap(err, "creating fields validator for data stream failed"))
 		return results

internal/testrunner/runners/system/runner.go

@@ -447,9 +447,12 @@ func (r *runner) runTest(config *testConfig, ctxt servicedeployer.ServiceContext
 	}
 
 	// Validate fields in docs
+	expectedDataset := pkgManifest.Name + "." + r.options.TestFolder.DataStream
 	fieldsValidator, err := fields.CreateValidatorForDirectory(serviceOptions.DataStreamRootPath,
 		fields.WithSpecVersion(pkgManifest.SpecVersion),
-		fields.WithNumericKeywordFields(config.NumericKeywordFields))
+		fields.WithNumericKeywordFields(config.NumericKeywordFields),
+		fields.WithExpectedDataset(expectedDataset),
+	)
 	if err != nil {
 		return result.WithError(errors.Wrapf(err, "creating fields validator for data stream failed (path: %s)", serviceOptions.DataStreamRootPath))
 	}