Support system tests otel input (#2885)

This PR enables to run system tests also in input packages
using the OTEL collector input (otelcol).
In order to achieve that the validation process for fields and
mappings is currently disabled because we need to determine
the specific requirements for validation when the OTEL collector
input is configured.


* Update index template name

* Remove outdated log message

* Extract function to create kibana policies

* Revert "Extract function to create kibana policies"

This reverts commit 77de02ed6bc3c765f8e88954fa3bee5e780ad37a.

* Update package datastream - integration policy

* Add otel as suffic for dataset

* Keep package datastream / integration policy with the previous values

* Enable system test for httpcheck package

* httpcheck - Added field definitions

* httpcheck - replace string by keyword type

* Update comments

* Skip some OTEL fields - to be validated

* Update index template and data stream names for system tests

* Skip validation per fields in OTEL collector input

* Remove skip fields for OTEL

Packages using OTEL collector input will perform the validation based on
the mappings found before and after ingesting documents.

* Skip dynamic templates without match condition

* Replace stringInArray

* Validate dataset fields also in OTEL

* Remove duplicated assignment

* Remove unnecessary else block

* Add warning message

* Update test package

* Add OTEL option when validating fields in transform

* Remove all field definitions

* Restore base-fields field definitions

* Remove skip test

Author: mrodm

internal/fields/validate.go

@@ -154,6 +154,8 @@ type Validator struct {
 
 	disabledNormalization bool
 
+	enabledOTELValidation bool
+
 	injectFieldsOptions InjectFieldsOptions
 }
 
@@ -246,6 +248,14 @@ func WithInjectFieldsOptions(options InjectFieldsOptions) ValidatorOption {
 	}
 }
 
+// WithOTELValidation configures the validator to enable or disable OpenTelemetry specific validation.
+func WithOTELValidation(otelValidation bool) ValidatorOption {
+	return func(v *Validator) error {
+		v.enabledOTELValidation = otelValidation
+		return nil
+	}
+}
+
 type packageRootFinder interface {
 	FindPackageRoot() (string, bool, error)
 }
@@ -559,7 +569,13 @@ 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.validateDocumentValues(body)
-	errs = append(errs, v.validateMapElement("", body, body)...)
+
+	// If package uses OpenTelemetry Collector, skip field validation and just
+	// validate document values (datasets).
+	if !v.enabledOTELValidation {
+		errs = append(errs, v.validateMapElement("", body, body)...)
+	}
+
 	if len(errs) == 0 {
 		return nil
 	}
@@ -602,7 +618,7 @@ func (v *Validator) validateDocumentValues(body common.MapStr) multierror.Error
 			}
 
 			str, ok := valueToString(value, v.disabledNormalization)
-			exists := stringInArray(str, renderedExpectedDatasets)
+			exists := slices.Contains(renderedExpectedDatasets, str)
 			if !ok || !exists {
 				err := fmt.Errorf("field %q should have value in %q, it has \"%v\"",
 					datasetField, v.expectedDatasets, value)
@@ -613,18 +629,6 @@ func (v *Validator) validateDocumentValues(body common.MapStr) multierror.Error
 	return errs
 }
 
-func stringInArray(target string, arr []string) bool {
-	// Check if target is part of the array
-	found := false
-	for _, item := range arr {
-		if item == target {
-			found = true
-			break
-		}
-	}
-	return found
-}
-
 func valueToString(value any, disabledNormalization bool) (string, bool) {
 	if disabledNormalization {
 		// when synthetics mode is enabled, each field present in the document is an array

internal/testrunner/runners/system/tester.go

@@ -145,6 +145,7 @@ var (
 type fieldValidationMethod int
 
 const (
+	// Required to allow setting `fields` as an option via environment variable
 	fieldsMethod fieldValidationMethod = iota
 	mappingsMethod
 )
@@ -1616,19 +1617,25 @@ func (r *tester) waitForDocs(ctx context.Context, config *testConfig, dataStream
 }
 
 func (r *tester) validateTestScenario(ctx context.Context, result *testrunner.ResultComposer, scenario *scenarioTest, config *testConfig) ([]testrunner.TestResult, error) {
+	logger.Info("Validating test case...")
 	expectedDatasets, err := r.expectedDatasets(scenario, config)
 	if err != nil {
 		return nil, err
 	}
 
-	// Validate fields in docs
+	if r.isTestUsingOTELCollectorInput(scenario.policyTemplateInput) {
+		logger.Warn("Validation for packages using OpenTelemetry Collector input is experimental")
+	}
+
 	fieldsValidator, err := fields.CreateValidatorForDirectory(r.dataStreamPath,
 		fields.WithSpecVersion(r.pkgManifest.SpecVersion),
 		fields.WithNumericKeywordFields(config.NumericKeywordFields),
 		fields.WithStringNumberFields(config.StringNumberFields),
 		fields.WithExpectedDatasets(expectedDatasets),
 		fields.WithEnabledImportAllECSSChema(true),
 		fields.WithDisableNormalization(scenario.syntheticEnabled),
+		// When using the OTEL collector input, just a subset of validations are performed (e.g. check expected datasets)
+		fields.WithOTELValidation(r.isTestUsingOTELCollectorInput(scenario.policyTemplateInput)),
 	)
 	if err != nil {
 		return result.WithErrorf("creating fields validator for data stream failed (path: %s): %w", r.dataStreamPath, err)
@@ -1641,7 +1648,7 @@ func (r *tester) validateTestScenario(ctx context.Context, result *testrunner.Re
 		})
 	}
 
-	if r.fieldValidationMethod == mappingsMethod {
+	if !r.isTestUsingOTELCollectorInput(scenario.policyTemplateInput) && r.fieldValidationMethod == mappingsMethod {
 		logger.Debug("Performing validation based on mappings")
 		exceptionFields := listExceptionFields(scenario.docs, fieldsValidator)
 
@@ -1698,7 +1705,7 @@ func (r *tester) validateTestScenario(ctx context.Context, result *testrunner.Re
 	}
 
 	// Check transforms if present
-	if err := r.checkTransforms(ctx, config, r.pkgManifest, scenario.dataStream, scenario.syntheticEnabled); err != nil {
+	if err := r.checkTransforms(ctx, config, r.pkgManifest, scenario.dataStream, scenario.policyTemplateInput, scenario.syntheticEnabled); err != nil {
 		results, _ := result.WithError(err)
 		return results, nil
 	}
@@ -1817,6 +1824,19 @@ func (r *tester) runTest(ctx context.Context, config *testConfig, stackConfig st
 	return r.validateTestScenario(ctx, result, scenario, config)
 }
 
+func (r *tester) isTestUsingOTELCollectorInput(policyTemplateInput string) bool {
+	// Just supported for input packages currently
+	if r.pkgManifest.Type != "input" {
+		return false
+	}
+
+	if policyTemplateInput != otelCollectorInputName {
+		return false
+	}
+
+	return true
+}
+
 func dumpScenarioDocs(docs any) error {
 	timestamp := time.Now().Format("20060102150405")
 	path := filepath.Join(os.TempDir(), fmt.Sprintf("elastic-package-test-docs-dump-%s.json", timestamp))
@@ -1983,16 +2003,14 @@ func createInputPackageDatastream(
 			PolicyTemplate: policyTemplate.Name,
 			Enabled:        true,
 			Vars:           kibana.Vars{},
+			Type:           policyTemplate.Input,
 		},
 	}
 
-	streamInput := policyTemplate.Input
-	r.Inputs[0].Type = streamInput
-
 	dataset := fmt.Sprintf("%s.%s", pkg.Name, policyTemplate.Name)
 	streams := []kibana.Stream{
 		{
-			ID:      fmt.Sprintf("%s-%s.%s", streamInput, pkg.Name, policyTemplate.Name),
+			ID:      fmt.Sprintf("%s-%s.%s", policyTemplate.Input, pkg.Name, policyTemplate.Name),
 			Enabled: true,
 			DataStream: kibana.DataStream{
 				Type:    policyTemplate.Type,
@@ -2145,7 +2163,7 @@ func selectPolicyTemplateByName(policies []packages.PolicyTemplate, name string)
 	return packages.PolicyTemplate{}, fmt.Errorf("policy template %q not found", name)
 }
 
-func (r *tester) checkTransforms(ctx context.Context, config *testConfig, pkgManifest *packages.PackageManifest, dataStream string, syntheticEnabled bool) error {
+func (r *tester) checkTransforms(ctx context.Context, config *testConfig, pkgManifest *packages.PackageManifest, dataStream, policyTemplateInput string, syntheticEnabled bool) error {
 	if config.SkipTransformValidation {
 		return nil
 	}
@@ -2199,6 +2217,8 @@ func (r *tester) checkTransforms(ctx context.Context, config *testConfig, pkgMan
 			fields.WithNumericKeywordFields(config.NumericKeywordFields),
 			fields.WithEnabledImportAllECSSChema(true),
 			fields.WithDisableNormalization(syntheticEnabled),
+			// When using the OTEL collector input, just a subset of validations are performed (e.g. check expected datasets)
+			fields.WithOTELValidation(r.isTestUsingOTELCollectorInput(policyTemplateInput)),
 		)
 		if err != nil {
 			return fmt.Errorf("creating fields validator for data stream failed (path: %s): %w", transformRootPath, err)

test/packages/parallel/httpcheck/_dev/test/system/test-local-config.yml

@@ -1,8 +1,7 @@
-skip:
-  reason: Not supported system tests with input type otelcol.
-  link: https://github.com/elastic/elastic-package/issues/2835
 service: web
 vars:
   period: 1s
   endpoints:
     - http://{{Hostname}}:{{Port}}
+assert:
+  min_count: 150