Support system tests otel input

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