tester.go

// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
// or more contributor license agreements. Licensed under the Elastic License;
// you may not use this file except in compliance with the Elastic License.

package system

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"slices"
	"strings"
	"time"

	"github.com/Masterminds/semver/v3"
	"gopkg.in/yaml.v3"

	"github.com/elastic/elastic-package/internal/agentdeployer"
	"github.com/elastic/elastic-package/internal/common"
	"github.com/elastic/elastic-package/internal/configuration/locations"
	"github.com/elastic/elastic-package/internal/elasticsearch"
	"github.com/elastic/elastic-package/internal/elasticsearch/ingest"
	"github.com/elastic/elastic-package/internal/environment"
	"github.com/elastic/elastic-package/internal/fields"
	"github.com/elastic/elastic-package/internal/formatter"
	"github.com/elastic/elastic-package/internal/kibana"
	"github.com/elastic/elastic-package/internal/logger"
	"github.com/elastic/elastic-package/internal/multierror"
	"github.com/elastic/elastic-package/internal/packages"
	"github.com/elastic/elastic-package/internal/profile"
	"github.com/elastic/elastic-package/internal/resources"
	"github.com/elastic/elastic-package/internal/servicedeployer"
	"github.com/elastic/elastic-package/internal/stack"
	"github.com/elastic/elastic-package/internal/testrunner"
	"github.com/elastic/elastic-package/internal/wait"
)

const (
	checkFieldsBody = `{
		"fields": ["*"],
		"runtime_mappings": {
		  "my_ignored": {
			"type": "keyword",
			"script": {
			  "source": "for (def v : params['_fields']._ignored.values) { emit(v); }"
			}
		  }
		},
		"aggs": {
		  "all_ignored": {
			"filter": {
			  "exists": {
				"field": "_ignored"
			  }
			},
			"aggs": {
			  "ignored_fields": {
				"terms": {
				  "size": 100,
				  "field": "my_ignored"
				}
			  },
			  "ignored_docs": {
				"top_hits": {
				  "size": 5
				}
			  }
			}
		  }
		}
	  }`
	DevDeployDir = "_dev/deploy"

	// TestType defining system tests
	TestType testrunner.TestType = "system"

	// Maximum number of events to query.
	elasticsearchQuerySize = 500

	// ServiceLogsAgentDir is folder path where log files produced by the service
	// are stored on the Agent container's filesystem.
	ServiceLogsAgentDir = "/tmp/service_logs"

	waitForDataDefaultTimeout = 10 * time.Minute
)

type logsRegexp struct {
	includes *regexp.Regexp
	excludes []*regexp.Regexp
}

type logsByContainer struct {
	containerName string
	patterns      []logsRegexp
}

var (
	errorPatterns = []logsByContainer{
		{
			containerName: "elastic-agent",
			patterns: []logsRegexp{
				{
					includes: regexp.MustCompile("^Cannot index event publisher.Event"),
					excludes: []*regexp.Regexp{
						// this regex is excluded to ensure that logs coming from the `system` package installed by default are not taken into account
						regexp.MustCompile(`action \[indices:data\/write\/bulk\[s\]\] is unauthorized for API key id \[.*\] of user \[.*\] on indices \[.*\], this action is granted by the index privileges \[.*\]`),
					},
				},
				{
					includes: regexp.MustCompile("->(FAILED|DEGRADED)"),

					// this regex is excluded to avoid a regresion in 8.11 that can make a component to pass to a degraded state during some seconds after reassigning or removing a policy
					excludes: []*regexp.Regexp{
						regexp.MustCompile(`Component state changed .* \(HEALTHY->DEGRADED\): Degraded: pid .* missed .* check-in`),
					},
				},
			},
		},
	}
	enableIndependentAgents = environment.WithElasticPackagePrefix("TEST_ENABLE_INDEPENDENT_AGENT")
)

type tester struct {
	profile            *profile.Profile
	testFolder         testrunner.TestFolder
	packageRootPath    string
	generateTestResult bool
	esAPI              *elasticsearch.API
	esClient           *elasticsearch.Client
	kibanaClient       *kibana.Client

	runIndependentElasticAgent bool

	deferCleanup   time.Duration
	serviceVariant string
	configFileName string

	runSetup     bool
	runTearDown  bool
	runTestsOnly bool

	pipelines []ingest.Pipeline

	dataStreamPath     string
	stackVersion       kibana.VersionInfo
	locationManager    *locations.LocationManager
	resourcesManager   *resources.Manager
	pkgManifest        *packages.PackageManifest
	dataStreamManifest *packages.DataStreamManifest
	withCoverage       bool
	coverageType       string
	checkFailureStore  bool

	serviceStateFilePath string

	globalTestConfig testrunner.GlobalRunnerTestConfig

	// Execution order of following handlers is defined in runner.TearDown() method.
	removeAgentHandler        func(context.Context) error
	deleteTestPolicyHandler   func(context.Context) error
	cleanTestScenarioHandler  func(context.Context) error
	resetAgentPolicyHandler   func(context.Context) error
	resetAgentLogLevelHandler func(context.Context) error
	shutdownServiceHandler    func(context.Context) error
	shutdownAgentHandler      func(context.Context) error
}

type SystemTesterOptions struct {
	Profile            *profile.Profile
	TestFolder         testrunner.TestFolder
	PackageRootPath    string
	GenerateTestResult bool
	API                *elasticsearch.API
	KibanaClient       *kibana.Client

	// FIXME: Keeping Elasticsearch client to be able to do low-level requests for parameters not supported yet by the API.
	ESClient *elasticsearch.Client

	DeferCleanup      time.Duration
	ServiceVariant    string
	ConfigFileName    string
	GlobalTestConfig  testrunner.GlobalRunnerTestConfig
	WithCoverage      bool
	CoverageType      string
	CheckFailureStore bool

	RunSetup     bool
	RunTearDown  bool
	RunTestsOnly bool
}

func NewSystemTester(options SystemTesterOptions) (*tester, error) {
	r := tester{
		profile:                    options.Profile,
		testFolder:                 options.TestFolder,
		packageRootPath:            options.PackageRootPath,
		generateTestResult:         options.GenerateTestResult,
		esAPI:                      options.API,
		esClient:                   options.ESClient,
		kibanaClient:               options.KibanaClient,
		deferCleanup:               options.DeferCleanup,
		serviceVariant:             options.ServiceVariant,
		configFileName:             options.ConfigFileName,
		runSetup:                   options.RunSetup,
		runTestsOnly:               options.RunTestsOnly,
		runTearDown:                options.RunTearDown,
		globalTestConfig:           options.GlobalTestConfig,
		withCoverage:               options.WithCoverage,
		coverageType:               options.CoverageType,
		checkFailureStore:          options.CheckFailureStore,
		runIndependentElasticAgent: true,
	}
	r.resourcesManager = resources.NewManager()
	r.resourcesManager.RegisterProvider(resources.DefaultKibanaProviderName, &resources.KibanaProvider{Client: r.kibanaClient})

	r.serviceStateFilePath = filepath.Join(stateFolderPath(r.profile.ProfilePath), serviceStateFileName)

	var err error

	r.locationManager, err = locations.NewLocationManager()
	if err != nil {
		return nil, fmt.Errorf("reading service logs directory failed: %w", err)
	}

	r.dataStreamPath, _, err = packages.FindDataStreamRootForPath(r.testFolder.Path)
	if err != nil {
		return nil, fmt.Errorf("locating data stream root failed: %w", err)
	}

	if r.esAPI == nil {
		return nil, errors.New("missing Elasticsearch client")
	}
	if r.kibanaClient == nil {
		return nil, errors.New("missing Kibana client")
	}

	r.stackVersion, err = r.kibanaClient.Version()
	if err != nil {
		return nil, fmt.Errorf("cannot request Kibana version: %w", err)
	}

	r.pkgManifest, err = packages.ReadPackageManifestFromPackageRoot(r.packageRootPath)
	if err != nil {
		return nil, fmt.Errorf("reading package manifest failed: %w", err)
	}

	r.dataStreamManifest, err = packages.ReadDataStreamManifest(filepath.Join(r.dataStreamPath, packages.DataStreamManifestFile))
	if err != nil {
		return nil, fmt.Errorf("reading data stream manifest failed: %w", err)
	}

	// If the environment variable is present, it always has preference over the root
	// privileges value (if any) defined in the manifest file
	v, ok := os.LookupEnv(enableIndependentAgents)
	if ok {
		r.runIndependentElasticAgent = strings.ToLower(v) == "true"
	}

	return &r, nil
}

// Ensures that runner implements testrunner.Tester interface
var _ testrunner.Tester = new(tester)

// Type returns the type of test that can be run by this test runner.
func (r *tester) Type() testrunner.TestType {
	return TestType
}

// String returns the human-friendly name of the test runner.
func (r *tester) String() string {
	return "system"
}

// Parallel indicates if this tester can run in parallel or not.
func (r tester) Parallel() bool {
	// it is required independent Elastic Agents to run in parallel system tests
	return r.runIndependentElasticAgent && r.globalTestConfig.Parallel
}

// Run runs the system tests defined under the given folder
func (r *tester) Run(ctx context.Context) ([]testrunner.TestResult, error) {
	if !r.runSetup && !r.runTearDown && !r.runTestsOnly {
		return r.run(ctx)
	}

	result := r.newResult("(init)")

	svcInfo, err := r.createServiceInfo()
	if err != nil {
		return result.WithError(err)
	}

	configFile := filepath.Join(r.testFolder.Path, r.configFileName)
	testConfig, err := newConfig(configFile, svcInfo, r.serviceVariant)
	if err != nil {
		return nil, fmt.Errorf("unable to load system test case file '%s': %w", configFile, err)
	}
	logger.Debugf("Using config: %q", testConfig.Name())

	resultName := ""
	switch {
	case r.runSetup:
		resultName = "setup"
	case r.runTearDown:
		resultName = "teardown"
	case r.runTestsOnly:
		resultName = "tests"
	}
	result = r.newResult(fmt.Sprintf("%s - %s", resultName, testConfig.Name()))

	scenario, err := r.prepareScenario(ctx, testConfig, svcInfo)
	if r.runSetup && err != nil {
		tdErr := r.tearDownTest(ctx)
		if tdErr != nil {
			logger.Errorf("failed to tear down runner: %s", tdErr.Error())
		}

		setupDirErr := r.removeServiceStateFile()
		if setupDirErr != nil {
			logger.Error(err.Error())
		}
		return result.WithError(err)
	}

	if r.runTestsOnly {
		if err != nil {
			return result.WithError(fmt.Errorf("failed to prepare scenario: %w", err))
		}
		results, err := r.validateTestScenario(ctx, result, scenario, testConfig)
		tdErr := r.tearDownTest(ctx)
		if tdErr != nil {
			logger.Errorf("failed to tear down runner: %s", tdErr.Error())
		}
		return results, err

	}

	if r.runTearDown {
		if err != nil {
			logger.Errorf("failed to prepare scenario: %s", err.Error())
			logger.Errorf("continue with the tear down process")
		}
		if err := r.tearDownTest(ctx); err != nil {
			return result.WithError(err)
		}

		err := r.removeServiceStateFile()
		if err != nil {
			return result.WithError(err)
		}
	}

	return result.WithSuccess()
}

type resourcesOptions struct {
	installedPackage bool
}

func (r *tester) createAgentOptions(policyName string) agentdeployer.FactoryOptions {
	return agentdeployer.FactoryOptions{
		Profile:            r.profile,
		PackageRootPath:    r.packageRootPath,
		DataStreamRootPath: r.dataStreamPath,
		DevDeployDir:       DevDeployDir,
		Type:               agentdeployer.TypeTest,
		StackVersion:       r.stackVersion.Version(),
		PackageName:        r.testFolder.Package,
		DataStream:         r.testFolder.DataStream,
		PolicyName:         policyName,
		RunTearDown:        r.runTearDown,
		RunTestsOnly:       r.runTestsOnly,
		RunSetup:           r.runSetup,
	}
}

func (r *tester) createServiceOptions(variantName string) servicedeployer.FactoryOptions {
	return servicedeployer.FactoryOptions{
		Profile:                r.profile,
		PackageRootPath:        r.packageRootPath,
		DataStreamRootPath:     r.dataStreamPath,
		DevDeployDir:           DevDeployDir,
		Variant:                variantName,
		Type:                   servicedeployer.TypeTest,
		StackVersion:           r.stackVersion.Version(),
		RunTearDown:            r.runTearDown,
		RunTestsOnly:           r.runTestsOnly,
		RunSetup:               r.runSetup,
		DeployIndependentAgent: r.runIndependentElasticAgent,
	}
}

func (r *tester) createAgentInfo(policy *kibana.Policy, config *testConfig, runID string, agentManifest packages.Agent) (agentdeployer.AgentInfo, error) {
	var info agentdeployer.AgentInfo

	info.Name = r.testFolder.Package
	info.Logs.Folder.Agent = ServiceLogsAgentDir
	info.Test.RunID = runID

	folderName := fmt.Sprintf("agent-%s", r.testFolder.Package)
	if r.testFolder.DataStream != "" {
		folderName = fmt.Sprintf("%s-%s", folderName, r.testFolder.DataStream)
	}
	folderName = fmt.Sprintf("%s-%s", folderName, runID)

	dirPath, err := agentdeployer.CreateServiceLogsDir(r.profile, folderName)
	if err != nil {
		return agentdeployer.AgentInfo{}, fmt.Errorf("failed to create service logs dir: %w", err)
	}
	info.Logs.Folder.Local = dirPath

	info.Policy.Name = policy.Name
	info.Policy.ID = policy.ID

	// Copy all agent settings from the test configuration file
	info.Agent.AgentSettings = config.Agent.AgentSettings

	// If user is defined in the configuration file, it has preference
	// and it should not be overwritten by the value in the manifest
	if info.Agent.User == "" && agentManifest.Privileges.Root {
		info.Agent.User = "root"
	}

	return info, nil
}

func (r *tester) createServiceInfo() (servicedeployer.ServiceInfo, error) {
	var svcInfo servicedeployer.ServiceInfo
	svcInfo.Name = r.testFolder.Package
	svcInfo.Logs.Folder.Local = r.locationManager.ServiceLogDir()
	svcInfo.Logs.Folder.Agent = ServiceLogsAgentDir
	svcInfo.Test.RunID = common.CreateTestRunID()

	if r.runTearDown || r.runTestsOnly {
		logger.Debug("Skip creating output directory")
	} else {
		outputDir, err := servicedeployer.CreateOutputDir(r.locationManager, svcInfo.Test.RunID)
		if err != nil {
			return servicedeployer.ServiceInfo{}, fmt.Errorf("could not create output dir for terraform deployer %w", err)
		}
		svcInfo.OutputDir = outputDir
	}

	svcInfo.Agent.Independent = false

	return svcInfo, nil
}

// TearDown method doesn't perform any global action as the "tear down" is executed per test case.
func (r *tester) TearDown(ctx context.Context) error {
	return nil
}

func (r *tester) tearDownTest(ctx context.Context) error {
	if r.deferCleanup > 0 {
		logger.Debugf("waiting for %s before tearing down...", r.deferCleanup)
		select {
		case <-time.After(r.deferCleanup):
		case <-ctx.Done():
		}
	}

	// Avoid cancellations during cleanup.
	cleanupCtx := context.WithoutCancel(ctx)

	// This handler should be run before shutting down Elastic Agents (agent deployer)
	// or services that could run agents like Custom Agents (service deployer)
	// or Kind deployer.
	if r.resetAgentPolicyHandler != nil {
		if err := r.resetAgentPolicyHandler(cleanupCtx); err != nil {
			return err
		}
		r.resetAgentPolicyHandler = nil
	}

	// Shutting down the service should be run one of the first actions
	// to ensure that resources created by terraform are deleted even if other
	// errors fail.
	if r.shutdownServiceHandler != nil {
		if err := r.shutdownServiceHandler(cleanupCtx); err != nil {
			return err
		}
		r.shutdownServiceHandler = nil
	}

	if r.cleanTestScenarioHandler != nil {
		if err := r.cleanTestScenarioHandler(cleanupCtx); err != nil {
			return err
		}
		r.cleanTestScenarioHandler = nil
	}

	if r.resetAgentLogLevelHandler != nil {
		if err := r.resetAgentLogLevelHandler(cleanupCtx); err != nil {
			return err
		}
		r.resetAgentLogLevelHandler = nil
	}

	if r.removeAgentHandler != nil {
		if err := r.removeAgentHandler(cleanupCtx); err != nil {
			return err
		}
		r.removeAgentHandler = nil
	}

	if r.deleteTestPolicyHandler != nil {
		if err := r.deleteTestPolicyHandler(cleanupCtx); err != nil {
			return err
		}
		r.deleteTestPolicyHandler = nil
	}

	if r.shutdownAgentHandler != nil {
		if err := r.shutdownAgentHandler(cleanupCtx); err != nil {
			return err
		}
		r.shutdownAgentHandler = nil
	}

	return nil
}

func (r *tester) newResult(name string) *testrunner.ResultComposer {
	return testrunner.NewResultComposer(testrunner.TestResult{
		TestType:   TestType,
		Name:       name,
		Package:    r.testFolder.Package,
		DataStream: r.testFolder.DataStream,
	})
}

func (r *tester) run(ctx context.Context) (results []testrunner.TestResult, err error) {
	result := r.newResult("(init)")

	startTesting := time.Now()

	results, err = r.runTestPerVariant(ctx, result, r.configFileName, r.serviceVariant)
	if err != nil {
		return results, err
	}

	// Every tester is in charge of just one test, so if there is no error,
	// then there should be just one result for tests. As an exception, there could
	// be two results if there is any issue checking Elastic Agent logs.
	if len(results) > 0 && results[0].Skipped != nil {
		logger.Debugf("Test skipped, avoid checking agent logs")
		return results, nil
	}

	tempDir, err := os.MkdirTemp("", "test-system-")
	if err != nil {
		return nil, fmt.Errorf("can't create temporal directory: %w", err)
	}
	defer os.RemoveAll(tempDir)

	stackConfig, err := stack.LoadConfig(r.profile)
	if err != nil {
		return nil, err
	}

	provider, err := stack.BuildProvider(stackConfig.Provider, r.profile)
	if err != nil {
		return nil, fmt.Errorf("failed to build stack provider: %w", err)
	}

	dumpOptions := stack.DumpOptions{
		Output:  tempDir,
		Profile: r.profile,
	}
	dump, err := provider.Dump(context.WithoutCancel(ctx), dumpOptions)
	if err != nil {
		return nil, fmt.Errorf("dump failed: %w", err)
	}

	logResults, err := r.checkAgentLogs(dump, startTesting, errorPatterns)
	if err != nil {
		return result.WithError(err)
	}
	results = append(results, logResults...)

	return results, nil
}

func (r *tester) runTestPerVariant(ctx context.Context, result *testrunner.ResultComposer, cfgFile, variantName string) ([]testrunner.TestResult, error) {
	svcInfo, err := r.createServiceInfo()
	if err != nil {
		return result.WithError(err)
	}

	configFile := filepath.Join(r.testFolder.Path, cfgFile)
	testConfig, err := newConfig(configFile, svcInfo, variantName)
	if err != nil {
		return nil, fmt.Errorf("unable to load system test case file '%s': %w", configFile, err)
	}
	logger.Debugf("Using config: %q", testConfig.Name())

	partial, err := r.runTest(ctx, testConfig, svcInfo)

	tdErr := r.tearDownTest(ctx)
	if err != nil {
		return partial, err
	}
	if tdErr != nil {
		return partial, fmt.Errorf("failed to tear down runner: %w", tdErr)
	}
	return partial, nil
}

func isSyntheticSourceModeEnabled(ctx context.Context, api *elasticsearch.API, dataStreamName string) (bool, error) {
	// We append a suffix so we don't use an existing resource, what may cause conflicts in old versions of
	// Elasticsearch, such as https://github.com/elastic/elasticsearch/issues/84256.
	resp, err := api.Indices.SimulateIndexTemplate(dataStreamName+"simulated",
		api.Indices.SimulateIndexTemplate.WithContext(ctx),
	)
	if err != nil {
		return false, fmt.Errorf("could not simulate index template for %s: %w", dataStreamName, err)
	}
	defer resp.Body.Close()

	if resp.IsError() {
		return false, fmt.Errorf("could not simulate index template for %s: %s", dataStreamName, resp.String())
	}

	var results struct {
		Template struct {
			Mappings struct {
				Source struct {
					Mode string `json:"mode"`
				} `json:"_source"`
			} `json:"mappings"`
			Settings struct {
				Index struct {
					Mode string `json:"mode"`
				} `json:"index"`
			} `json:"settings"`
		} `json:"template"`
	}

	if err := json.NewDecoder(resp.Body).Decode(&results); err != nil {
		return false, fmt.Errorf("could not decode index template simulation response: %w", err)
	}

	if results.Template.Mappings.Source.Mode == "synthetic" {
		return true, nil
	}

	// It seems that some index modes enable synthetic source mode even when it is not explicitly mentioned
	// in the mappings. So assume that when these index modes are used, the synthetic mode is also used.
	var syntheticsIndexModes = []string{
		"logs", // Replaced in 8.15.0 with "logsdb", see https://github.com/elastic/elasticsearch/pull/111054
		"logsdb",
		"time_series",
	}
	if slices.Contains(syntheticsIndexModes, results.Template.Settings.Index.Mode) {
		return true, nil
	}

	return false, nil
}

type hits struct {
	Source        []common.MapStr `json:"_source"`
	Fields        []common.MapStr `json:"fields"`
	IgnoredFields []string
	DegradedDocs  []common.MapStr
}

func (h hits) getDocs(syntheticsEnabled bool) []common.MapStr {
	if syntheticsEnabled {
		return h.Fields
	}
	return h.Source
}

func (h hits) size() int {
	return len(h.Source)
}

func (r *tester) getDocs(ctx context.Context, dataStream string) (*hits, error) {
	resp, err := r.esAPI.Search(
		r.esAPI.Search.WithContext(ctx),
		r.esAPI.Search.WithIndex(dataStream),
		r.esAPI.Search.WithSort("@timestamp:asc"),
		r.esAPI.Search.WithSize(elasticsearchQuerySize),
		r.esAPI.Search.WithSource("true"),
		r.esAPI.Search.WithBody(strings.NewReader(checkFieldsBody)),
		r.esAPI.Search.WithIgnoreUnavailable(true),
	)
	if err != nil {
		return nil, fmt.Errorf("could not search data stream: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusServiceUnavailable && strings.Contains(resp.String(), "no_shard_available_action_exception") {
		// Index is being created, but no shards are available yet.
		// See https://github.com/elastic/elasticsearch/issues/65846
		return &hits{}, nil
	}
	if resp.IsError() {
		return nil, fmt.Errorf("failed to search docs for data stream %s: %s", dataStream, resp.String())
	}

	var results struct {
		Hits struct {
			Total struct {
				Value int
			}
			Hits []struct {
				Source common.MapStr `json:"_source"`
				Fields common.MapStr `json:"fields"`
			}
		}
		Aggregations struct {
			AllIgnored struct {
				DocCount      int `json:"doc_count"`
				IgnoredFields struct {
					Buckets []struct {
						Key string `json:"key"`
					} `json:"buckets"`
				} `json:"ignored_fields"`
				IgnoredDocs struct {
					Hits struct {
						Hits []common.MapStr `json:"hits"`
					} `json:"hits"`
				} `json:"ignored_docs"`
			} `json:"all_ignored"`
		} `json:"aggregations"`
		Error *struct {
			Type   string
			Reason string
		}
		Status int
	}

	if err := json.NewDecoder(resp.Body).Decode(&results); err != nil {
		return nil, fmt.Errorf("could not decode search results response: %w", err)
	}

	numHits := results.Hits.Total.Value
	if results.Error != nil {
		logger.Debugf("found %d hits in %s data stream: %s: %s Status=%d",
			numHits, dataStream, results.Error.Type, results.Error.Reason, results.Status)
	} else {
		logger.Debugf("found %d hits in %s data stream", numHits, dataStream)
	}

	var hits hits
	for _, hit := range results.Hits.Hits {
		hits.Source = append(hits.Source, hit.Source)
		hits.Fields = append(hits.Fields, hit.Fields)
	}
	for _, bucket := range results.Aggregations.AllIgnored.IgnoredFields.Buckets {
		hits.IgnoredFields = append(hits.IgnoredFields, bucket.Key)
	}
	hits.DegradedDocs = results.Aggregations.AllIgnored.IgnoredDocs.Hits.Hits

	return &hits, nil
}

func (r *tester) getFailureStoreDocs(ctx context.Context, dataStream string) ([]failureStoreDocument, error) {
	query := map[string]any{
		"query": map[string]any{
			"bool": map[string]any{
				// Ignoring failures with error.type version_conflict_engine_exception because there are packages which
				// explicitly set the _id with the fingerprint processor to avoid duplicates.
				"must_not": map[string]any{
					"term": map[string]any{
						"error.type": "version_conflict_engine_exception",
					},
				},
			},
		},
	}
	body, err := json.Marshal(query)
	if err != nil {
		return nil, fmt.Errorf("failed to encode search query: %w", err)
	}
	// FIXME: Using the low-level transport till the API SDK supports the failure store.
	request, err := http.NewRequest(http.MethodPost, fmt.Sprintf("/%s/_search?failure_store=only", dataStream), bytes.NewReader(body))
	if err != nil {
		return nil, fmt.Errorf("failed to create search request: %w", err)
	}
	request.Header.Set("Content-Type", "application/json")

	resp, err := r.esClient.Transport.Perform(request)
	if err != nil {
		return nil, fmt.Errorf("failed to perform search request: %w", err)
	}
	defer resp.Body.Close()

	switch {
	case resp.StatusCode == http.StatusNotFound:
		// Can happen if the data stream hasn't been created yet.
		return nil, nil
	case resp.StatusCode == http.StatusServiceUnavailable:
		// Index is being created, but no shards are available yet.
		// See https://github.com/elastic/elasticsearch/issues/65846
		return nil, nil
	case resp.StatusCode >= 400:
		return nil, fmt.Errorf("search request returned status code %d", resp.StatusCode)
	}

	var results struct {
		Hits struct {
			Hits []struct {
				Source failureStoreDocument `json:"_source"`
				Fields common.MapStr        `json:"fields"`
			} `json:"hits"`
		} `json:"hits"`
	}
	err = json.NewDecoder(resp.Body).Decode(&results)
	if err != nil {
		return nil, fmt.Errorf("failed to decode search response: %w", err)
	}

	var docs []failureStoreDocument
	for _, hit := range results.Hits.Hits {
		docs = append(docs, hit.Source)
	}

	return docs, nil
}

type scenarioTest struct {
	dataStream         string
	policyTemplateName string
	kibanaDataStream   kibana.PackageDataStream
	syntheticEnabled   bool
	docs               []common.MapStr
	failureStore       []failureStoreDocument
	ignoredFields      []string
	degradedDocs       []common.MapStr
	agent              agentdeployer.DeployedAgent
	startTestTime      time.Time
}

type failureStoreDocument struct {
	Error struct {
		Type          string   `json:"type"`
		Message       string   `json:"message"`
		StackTrace    string   `json:"stack_trace"`
		PipelineTrace []string `json:"pipeline_trace"`
		Pipeline      string   `json:"pipeline"`
		ProcessorType string   `json:"processor_type"`
	} `json:"error"`
}

func (r *tester) deleteDataStream(ctx context.Context, dataStream string) error {
	resp, err := r.esAPI.Indices.DeleteDataStream([]string{dataStream},
		r.esAPI.Indices.DeleteDataStream.WithContext(ctx),
	)
	if err != nil {
		return fmt.Errorf("delete request failed for data stream %s: %w", dataStream, err)
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		// Data stream doesn't exist, there was nothing to do.
		return nil
	}
	if resp.IsError() {
		return fmt.Errorf("delete request failed for data stream %s: %s", dataStream, resp.String())
	}
	return nil
}

func (r *tester) prepareScenario(ctx context.Context, config *testConfig, svcInfo servicedeployer.ServiceInfo) (*scenarioTest, error) {
	serviceOptions := r.createServiceOptions(config.ServiceVariantName)

	var err error
	var serviceStateData ServiceState
	if r.runSetup {
		err = r.createServiceStateDir()
		if err != nil {
			return nil, fmt.Errorf("failed to create setup services dir: %w", err)
		}
	}
	scenario := scenarioTest{}

	if r.runTearDown || r.runTestsOnly {
		serviceStateData, err = readServiceStateData(r.serviceStateFilePath)
		if err != nil {
			return nil, fmt.Errorf("failed to read service setup data: %w", err)
		}
	}

	serviceOptions.DeployIndependentAgent = r.runIndependentElasticAgent

	policyTemplateName := config.PolicyTemplate
	if policyTemplateName == "" {
		policyTemplateName, err = findPolicyTemplateForInput(*r.pkgManifest, *r.dataStreamManifest, config.Input)
		if err != nil {
			return nil, fmt.Errorf("failed to determine the associated policy_template: %w", err)
		}
	}
	scenario.policyTemplateName = policyTemplateName

	policyTemplate, err := selectPolicyTemplateByName(r.pkgManifest.PolicyTemplates, scenario.policyTemplateName)
	if err != nil {
		return nil, fmt.Errorf("failed to find the selected policy_template: %w", err)
	}

	// Configure package (single data stream) via Fleet APIs.
	testTime := time.Now().Format("20060102T15:04:05Z")
	var policyToTest, policyCurrent, policyToEnroll *kibana.Policy
	if r.runTearDown || r.runTestsOnly {
		policyCurrent = &serviceStateData.CurrentPolicy
		policyToEnroll = &serviceStateData.EnrollPolicy
		logger.Debugf("Got current policy from file: %q - %q", policyCurrent.Name, policyCurrent.ID)
	} else {
		// Created a specific Agent Policy to enrolling purposes
		// There are some issues when the stack is running for some time,
		// agents cannot enroll with the default policy
		// This enroll policy must be created even if independent Elastic Agents are not used. Agents created
		// in Kubernetes or Custom Agents require this enroll policy too (service deployer).
		logger.Debug("creating enroll policy...")
		policyEnroll := kibana.Policy{
			Name:        fmt.Sprintf("ep-test-system-enroll-%s-%s-%s-%s-%s", r.testFolder.Package, r.testFolder.DataStream, r.serviceVariant, r.configFileName, testTime),
			Description: fmt.Sprintf("test policy created by elastic-package to enroll agent for data stream %s/%s", r.testFolder.Package, r.testFolder.DataStream),
			Namespace:   common.CreateTestRunID(),
		}

		policyToEnroll, err = r.kibanaClient.CreatePolicy(ctx, policyEnroll)
		if err != nil {
			return nil, fmt.Errorf("could not create test policy: %w", err)
		}
	}

	r.deleteTestPolicyHandler = func(ctx context.Context) error {
		// ensure that policyToEnroll policy gets deleted if the execution receives a signal
		// before creating the test policy
		// This handler is going to be redefined after creating the test policy
		if r.runTestsOnly {
			return nil
		}
		if err := r.kibanaClient.DeletePolicy(ctx, policyToEnroll.ID); err != nil {
			return fmt.Errorf("error cleaning up test policy: %w", err)
		}
		return nil
	}

	if r.runTearDown {
		// required to assign the policy stored in the service state file
		// so data stream related to this Agent Policy can be obtained (and deleted)
		// in the cleanTestScenarioHandler handler
		policyToTest = policyCurrent
	} else {
		// Create a specific Agent Policy just for testing this test.
		// This allows us to ensure that the Agent Policy used for testing is
		// assigned to the agent with all the required changes (e.g. Package DataStream)
		logger.Debug("creating test policy...")
		policy := kibana.Policy{
			Name:        fmt.Sprintf("ep-test-system-%s-%s-%s-%s-%s", r.testFolder.Package, r.testFolder.DataStream, r.serviceVariant, r.configFileName, testTime),
			Description: fmt.Sprintf("test policy created by elastic-package test system for data stream %s/%s", r.testFolder.Package, r.testFolder.DataStream),
			Namespace:   common.CreateTestRunID(),
		}
		// Assign the data_output_id to the agent policy to configure the output to logstash. The value is inferred from stack/_static/kibana.yml.tmpl
		if r.profile.Config("stack.logstash_enabled", "false") == "true" {
			policy.DataOutputID = "fleet-logstash-output"
		}
		policyToTest, err = r.kibanaClient.CreatePolicy(ctx, policy)
		if err != nil {
			return nil, fmt.Errorf("could not create test policy: %w", err)
		}
	}

	r.deleteTestPolicyHandler = func(ctx context.Context) error {
		logger.Debug("deleting test policies...")
		if err := r.kibanaClient.DeletePolicy(ctx, policyToTest.ID); err != nil {
			return fmt.Errorf("error cleaning up test policy: %w", err)
		}
		if r.runTestsOnly {
			return nil
		}
		if err := r.kibanaClient.DeletePolicy(ctx, policyToEnroll.ID); err != nil {
			return fmt.Errorf("error cleaning up test policy: %w", err)
		}
		return nil
	}

	// policyToEnroll is used in both independent agents and agents created by servicedeployer (custom or kubernetes agents)
	policy := policyToEnroll
	if r.runTearDown || r.runTestsOnly {
		// required in order to be able select the right agent in `checkEnrolledAgents` when
		// using independent agents or custom/kubernetes agents since policy data is set into `agentInfo` variable`
		policy = policyCurrent
	}

	agentDeployed, agentInfo, err := r.setupAgent(ctx, config, serviceStateData, policy, r.pkgManifest.Agent)
	if err != nil {
		return nil, err
	}

	scenario.agent = agentDeployed

	service, svcInfo, err := r.setupService(ctx, config, serviceOptions, svcInfo, agentInfo, agentDeployed, policy, serviceStateData)
	if errors.Is(err, os.ErrNotExist) {
		logger.Debugf("No service deployer defined for this test")
	} else if err != nil {
		return nil, err
	}

	// Reload test config with ctx variable substitution.
	config, err = newConfig(config.Path, svcInfo, serviceOptions.Variant)
	if err != nil {
		return nil, fmt.Errorf("unable to reload system test case configuration: %w", err)
	}

	// store the time just before adding the Test Policy, this time will be used to check
	// the agent logs from that time onwards to avoid possible previous errors present in logs
	scenario.startTestTime = time.Now()

	logger.Debug("adding package data stream to test policy...")
	ds := createPackageDatastream(*policyToTest, *r.pkgManifest, policyTemplate, *r.dataStreamManifest, *config, policyToTest.Namespace)
	if r.runTearDown {
		logger.Debug("Skip adding data stream config to policy")
	} else {
		if err := r.kibanaClient.AddPackageDataStreamToPolicy(ctx, ds); err != nil {
			return nil, fmt.Errorf("could not add data stream config to policy: %w", err)
		}
	}
	scenario.kibanaDataStream = ds

	// Delete old data
	logger.Debug("deleting old data in data stream...")

	// Input packages can set `data_stream.dataset` by convention to customize the dataset.
	dataStreamDataset := ds.Inputs[0].Streams[0].DataStream.Dataset
	if r.pkgManifest.Type == "input" {
		v, _ := config.Vars.GetValue("data_stream.dataset")
		if dataset, ok := v.(string); ok && dataset != "" {
			dataStreamDataset = dataset
		}
	}
	scenario.dataStream = fmt.Sprintf(
		"%s-%s-%s",
		ds.Inputs[0].Streams[0].DataStream.Type,
		dataStreamDataset,
		ds.Namespace,
	)

	r.cleanTestScenarioHandler = func(ctx context.Context) error {
		logger.Debugf("Deleting data stream for testing %s", scenario.dataStream)
		err := r.deleteDataStream(ctx, scenario.dataStream)
		if err != nil {
			return fmt.Errorf("failed to delete data stream %s: %w", scenario.dataStream, err)
		}
		return nil
	}

	// FIXME: running per stages does not work when multiple agents are created
	var origPolicy kibana.Policy
	// While there could be created Elastic Agents within `setupService()` (custom agents and k8s agents),
	// this "checkEnrolledAgents" call to must be located after creating the service.
	agents, err := checkEnrolledAgents(ctx, r.kibanaClient, agentInfo, svcInfo, r.runIndependentElasticAgent)
	if err != nil {
		return nil, fmt.Errorf("can't check enrolled agents: %w", err)
	}
	agent := agents[0]
	logger.Debugf("Selected enrolled agent %q", agent.ID)

	r.removeAgentHandler = func(ctx context.Context) error {
		if r.runTestsOnly {
			return nil
		}
		// When not using independent agents, service deployers like kubernetes or custom agents create new Elastic Agent
		if !r.runIndependentElasticAgent && !svcInfo.Agent.Independent {
			return nil
		}
		logger.Debug("removing agent...")
		err := r.kibanaClient.RemoveAgent(ctx, agent)
		if err != nil {
			return fmt.Errorf("failed to remove agent %q: %w", agent.ID, err)
		}
		return nil
	}

	if r.runTearDown {
		origPolicy = serviceStateData.OrigPolicy
		logger.Debugf("Got orig policy from file: %q - %q", origPolicy.Name, origPolicy.ID)
	} else {
		// Store previous agent policy assigned to the agent
		origPolicy = kibana.Policy{
			ID:       agent.PolicyID,
			Revision: agent.PolicyRevision,
		}
	}

	r.resetAgentPolicyHandler = func(ctx context.Context) error {
		if r.runSetup {
			// it should be kept the same policy just when system tests are
			// triggered with the flags for running spolicyToAssignDatastreamTestsetup stage (--setup)
			return nil
		}
		logger.Debug("reassigning original policy back to agent...")
		if err := r.kibanaClient.AssignPolicyToAgent(ctx, agent, origPolicy); err != nil {
			return fmt.Errorf("error reassigning original policy to agent: %w", err)
		}
		return nil
	}

	origAgent := agent
	origLogLevel := ""
	if r.runTearDown {
		logger.Debug("Skip assiging log level debug to agent")
		origLogLevel = serviceStateData.Agent.LocalMetadata.Elastic.Agent.LogLevel
	} else {
		logger.Debug("Set Debug log level to agent")
		origLogLevel = agent.LocalMetadata.Elastic.Agent.LogLevel
		err = r.kibanaClient.SetAgentLogLevel(ctx, agent.ID, "debug")
		if err != nil {
			return nil, fmt.Errorf("error setting log level debug for agent %s: %w", agent.ID, err)
		}
	}
	r.resetAgentLogLevelHandler = func(ctx context.Context) error {
		if r.runTestsOnly {
			return nil
		}
		logger.Debugf("reassigning original log level %q back to agent...", origLogLevel)

		if err := r.kibanaClient.SetAgentLogLevel(ctx, agent.ID, origLogLevel); err != nil {
			return fmt.Errorf("error reassigning original log level to agent: %w", err)
		}
		return nil
	}

	if r.runTearDown {
		logger.Debug("Skip assigning package data stream to agent")
	} else {
		policyWithDataStream, err := r.kibanaClient.GetPolicy(ctx, policyToTest.ID)
		if err != nil {
			return nil, fmt.Errorf("could not read the policy with data stream: %w", err)
		}

		logger.Debug("assigning package data stream to agent...")
		if err := r.kibanaClient.AssignPolicyToAgent(ctx, agent, *policyWithDataStream); err != nil {
			return nil, fmt.Errorf("could not assign policy to agent: %w", err)
		}
	}

	// Signal to the service that the agent is ready (policy is assigned).
	if service != nil && config.ServiceNotifySignal != "" {
		if err = service.Signal(ctx, config.ServiceNotifySignal); err != nil {
			return nil, fmt.Errorf("failed to notify test service: %w", err)
		}
	}

	if r.runTearDown {
		return &scenario, nil
	}

	// Use custom timeout if the service can't collect data immediately.
	waitForDataTimeout := waitForDataDefaultTimeout
	if config.WaitForDataTimeout > 0 {
		waitForDataTimeout = config.WaitForDataTimeout
	}

	// (TODO in future) Optionally exercise service to generate load.
	logger.Debugf("checking for expected data in data stream (%s)...", waitForDataTimeout)
	var hits *hits
	oldHits := 0
	passed, waitErr := wait.UntilTrue(ctx, func(ctx context.Context) (bool, error) {
		var err error
		hits, err = r.getDocs(ctx, scenario.dataStream)
		if err != nil {
			return false, err
		}

		if r.checkFailureStore {
			failureStore, err := r.getFailureStoreDocs(ctx, scenario.dataStream)
			if err != nil {
				return false, fmt.Errorf("failed to check failure store: %w", err)
			}
			if n := len(failureStore); n > 0 {
				// Interrupt loop earlier if there are failures in the document store.
				logger.Debugf("Found %d hits in the failure store for %s", len(failureStore), scenario.dataStream)
				return true, nil
			}
		}

		if config.Assert.HitCount > 0 {
			if hits.size() < config.Assert.HitCount {
				return false, nil
			}

			ret := hits.size() == oldHits
			if !ret {
				oldHits = hits.size()
				time.Sleep(4 * time.Second)
			}

			return ret, nil
		}

		return hits.size() > 0, nil
	}, 1*time.Second, waitForDataTimeout)

	if service != nil && config.Service != "" && !config.IgnoreServiceError {
		exited, code, err := service.ExitCode(ctx, config.Service)
		if err != nil && !errors.Is(err, servicedeployer.ErrNotSupported) {
			return nil, err
		}
		if exited && code > 0 {
			return nil, testrunner.ErrTestCaseFailed{Reason: fmt.Sprintf("the test service %s unexpectedly exited with code %d", config.Service, code)}
		}
	}

	if waitErr != nil {
		return nil, waitErr
	}

	if !passed {
		return nil, testrunner.ErrTestCaseFailed{Reason: fmt.Sprintf("could not find hits in %s data stream", scenario.dataStream)}
	}

	logger.Debugf("Check whether or not synthetic source mode is enabled (data stream %s)...", scenario.dataStream)
	scenario.syntheticEnabled, err = isSyntheticSourceModeEnabled(ctx, r.esAPI, scenario.dataStream)
	if err != nil {
		return nil, fmt.Errorf("failed to check if synthetic source mode is enabled for data stream %s: %w", scenario.dataStream, err)
	}
	logger.Debugf("Data stream %s has synthetic source mode enabled: %t", scenario.dataStream, scenario.syntheticEnabled)

	scenario.docs = hits.getDocs(scenario.syntheticEnabled)
	scenario.ignoredFields = hits.IgnoredFields
	scenario.degradedDocs = hits.DegradedDocs
	if r.checkFailureStore {
		logger.Debugf("Checking failure store for data stream %s", scenario.dataStream)
		scenario.failureStore, err = r.getFailureStoreDocs(ctx, scenario.dataStream)
		if err != nil {
			return nil, fmt.Errorf("failed to get documents from the failure store for data stream %s: %w", scenario.dataStream, err)
		}
		logger.Debugf("Found %d docs in failure store for data stream %s", len(scenario.failureStore), scenario.dataStream)
	}

	if r.runSetup {
		opts := scenarioStateOpts{
			origPolicy:    &origPolicy,
			enrollPolicy:  policyToEnroll,
			currentPolicy: policyToTest,
			config:        config,
			agent:         origAgent,
			agentInfo:     agentInfo,
			svcInfo:       svcInfo,
		}
		err = writeScenarioState(opts, r.serviceStateFilePath)
		if err != nil {
			return nil, err
		}
	}

	return &scenario, nil
}

func (r *tester) setupService(ctx context.Context, config *testConfig, serviceOptions servicedeployer.FactoryOptions, svcInfo servicedeployer.ServiceInfo, agentInfo agentdeployer.AgentInfo, agentDeployed agentdeployer.DeployedAgent, policy *kibana.Policy, state ServiceState) (servicedeployer.DeployedService, servicedeployer.ServiceInfo, error) {
	logger.Debug("setting up service...")
	if r.runTearDown || r.runTestsOnly {
		svcInfo.Test.RunID = state.ServiceRunID
		svcInfo.OutputDir = state.ServiceOutputDir
	}

	// By default using agent running in the Elastic stack
	svcInfo.AgentNetworkName = stack.Network(r.profile)
	if agentDeployed != nil {
		svcInfo.AgentNetworkName = agentInfo.NetworkName
	}

	// Set the right folder for logs execpt for custom agents that are still deployed using "servicedeployer"
	if r.runIndependentElasticAgent && agentDeployed != nil {
		svcInfo.Logs.Folder.Local = agentInfo.Logs.Folder.Local
	}

	// In case of custom or kubernetes agents (servicedeployer) it is needed also the Agent Policy created
	// for each test execution
	serviceOptions.PolicyName = policy.Name

	if config.Service != "" {
		svcInfo.Name = config.Service
	}

	serviceDeployer, err := servicedeployer.Factory(serviceOptions)
	if err != nil {
		return nil, svcInfo, fmt.Errorf("could not create service runner: %w", err)
	}

	service, err := serviceDeployer.SetUp(ctx, svcInfo)
	if err != nil {
		return nil, svcInfo, fmt.Errorf("could not setup service: %w", err)
	}

	r.shutdownServiceHandler = func(ctx context.Context) error {
		if r.runTestsOnly {
			return nil
		}
		if service == nil {
			return nil
		}
		logger.Debug("tearing down service...")
		if err := service.TearDown(ctx); err != nil {
			return fmt.Errorf("error tearing down service: %w", err)
		}

		return nil
	}

	return service, service.Info(), nil
}

func (r *tester) setupAgent(ctx context.Context, config *testConfig, state ServiceState, policy *kibana.Policy, agentManifest packages.Agent) (agentdeployer.DeployedAgent, agentdeployer.AgentInfo, error) {
	if !r.runIndependentElasticAgent {
		return nil, agentdeployer.AgentInfo{}, nil
	}
	agentRunID := common.CreateTestRunID()
	if r.runTearDown || r.runTestsOnly {
		agentRunID = state.AgentRunID
	}
	logger.Debug("setting up independent Elastic Agent...")
	agentInfo, err := r.createAgentInfo(policy, config, agentRunID, agentManifest)
	if err != nil {
		return nil, agentdeployer.AgentInfo{}, err
	}

	agentOptions := r.createAgentOptions(agentInfo.Policy.Name)
	agentDeployer, err := agentdeployer.Factory(agentOptions)
	if err != nil {
		return nil, agentInfo, fmt.Errorf("could not create agent runner: %w", err)
	}
	if agentDeployer == nil {
		logger.Debug("Not found agent deployer. Agent will be created along with the service.")
		return nil, agentInfo, nil
	}

	agentDeployed, err := agentDeployer.SetUp(ctx, agentInfo)
	if err != nil {
		return nil, agentInfo, fmt.Errorf("could not setup agent: %w", err)
	}
	r.shutdownAgentHandler = func(ctx context.Context) error {
		if r.runTestsOnly {
			return nil
		}
		if agentDeployer == nil {
			return nil
		}
		logger.Debug("tearing down agent...")
		if err := agentDeployed.TearDown(ctx); err != nil {
			return fmt.Errorf("error tearing down agent: %w", err)
		}

		return nil
	}
	return agentDeployed, agentDeployed.Info(), nil
}

func (r *tester) removeServiceStateFile() error {
	err := os.Remove(r.serviceStateFilePath)
	if err != nil {
		return fmt.Errorf("failed to remove file %q: %w", r.serviceStateFilePath, err)
	}
	return nil
}

func (r *tester) createServiceStateDir() error {
	dirPath := filepath.Dir(r.serviceStateFilePath)
	err := os.MkdirAll(dirPath, 0755)
	if err != nil {
		return fmt.Errorf("mkdir failed (path: %s): %w", dirPath, err)
	}
	return nil
}

func (r *tester) validateTestScenario(ctx context.Context, result *testrunner.ResultComposer, scenario *scenarioTest, config *testConfig) ([]testrunner.TestResult, error) {
	if err := validateFailureStore(scenario.failureStore); err != nil {
		return result.WithError(err)
	}

	// Validate fields in docs
	// when reroute processors are used, expectedDatasets should be set depends on the processor config
	var expectedDatasets []string
	for _, pipeline := range r.pipelines {
		var esIngestPipeline map[string]any
		err := yaml.Unmarshal(pipeline.Content, &esIngestPipeline)
		if err != nil {
			return nil, fmt.Errorf("unmarshalling ingest pipeline content failed: %w", err)
		}
		processors, _ := esIngestPipeline["processors"].([]any)
		for _, p := range processors {
			processor, ok := p.(map[string]any)
			if !ok {
				return nil, fmt.Errorf("unexpected processor %+v", p)
			}
			if reroute, ok := processor["reroute"]; ok {
				if rerouteP, ok := reroute.(ingest.RerouteProcessor); ok {
					expectedDatasets = append(expectedDatasets, rerouteP.Dataset...)
				}
			}
		}
	}

	if expectedDatasets == nil {
		var expectedDataset string
		if ds := r.testFolder.DataStream; ds != "" {
			expectedDataset = getDataStreamDataset(*r.pkgManifest, *r.dataStreamManifest)
		} else {
			expectedDataset = r.pkgManifest.Name + "." + scenario.policyTemplateName
		}
		expectedDatasets = []string{expectedDataset}
	}
	if r.pkgManifest.Type == "input" {
		v, _ := config.Vars.GetValue("data_stream.dataset")
		if dataset, ok := v.(string); ok && dataset != "" {
			expectedDatasets = append(expectedDatasets, dataset)
		}
	}

	fieldsValidator, err := fields.CreateValidatorForDirectory(r.dataStreamPath,
		fields.WithSpecVersion(r.pkgManifest.SpecVersion),
		fields.WithNumericKeywordFields(config.NumericKeywordFields),
		fields.WithExpectedDatasets(expectedDatasets),
		fields.WithEnabledImportAllECSSChema(true),
		fields.WithDisableNormalization(scenario.syntheticEnabled),
	)
	if err != nil {
		return result.WithErrorf("creating fields validator for data stream failed (path: %s): %w", r.dataStreamPath, err)
	}
	if err := validateFields(scenario.docs, fieldsValidator, scenario.dataStream); err != nil {
		return result.WithError(err)
	}

	err = validateIgnoredFields(r.stackVersion.Number, scenario, config)
	if err != nil {
		return result.WithError(err)
	}

	docs := scenario.docs
	if scenario.syntheticEnabled {
		docs, err = fieldsValidator.SanitizeSyntheticSourceDocs(scenario.docs)
		if err != nil {
			results, _ := result.WithErrorf("failed to sanitize synthetic source docs: %w", err)
			return results, nil
		}
	}

	specVersion, err := semver.NewVersion(r.pkgManifest.SpecVersion)
	if err != nil {
		return result.WithErrorf("failed to parse format version %q: %w", r.pkgManifest.SpecVersion, err)
	}

	// Write sample events file from first doc, if requested
	if err := r.generateTestResultFile(docs, *specVersion); err != nil {
		return result.WithError(err)
	}

	// Check Hit Count within docs, if 0 then it has not been specified
	if assertionPass, message := assertHitCount(config.Assert.HitCount, docs); !assertionPass {
		result.FailureMsg = message
	}

	// Check transforms if present
	if err := r.checkTransforms(ctx, config, r.pkgManifest, scenario.kibanaDataStream, scenario.dataStream); err != nil {
		results, _ := result.WithError(err)
		return results, nil
	}

	if scenario.agent != nil {
		logResults, err := r.checkNewAgentLogs(ctx, scenario.agent, scenario.startTestTime, errorPatterns, config.Name())
		if err != nil {
			return result.WithError(err)
		}
		if len(logResults) > 0 {
			return logResults, nil
		}
	}

	if r.withCoverage {
		coverage, err := r.generateCoverageReport(result.CoveragePackageName())
		if err != nil {
			return result.WithErrorf("coverage report generation failed: %w", err)
		}
		result = result.WithCoverage(coverage)
	}

	return result.WithSuccess()
}

func (r *tester) runTest(ctx context.Context, config *testConfig, svcInfo servicedeployer.ServiceInfo) ([]testrunner.TestResult, error) {
	result := r.newResult(config.Name())

	if skip := testrunner.AnySkipConfig(config.Skip, r.globalTestConfig.Skip); skip != nil {
		logger.Warnf("skipping %s test for %s/%s: %s (details: %s)",
			TestType, r.testFolder.Package, r.testFolder.DataStream,
			skip.Reason, skip.Link)
		return result.WithSkip(skip)
	}

	logger.Debugf("running test with configuration '%s'", config.Name())

	scenario, err := r.prepareScenario(ctx, config, svcInfo)
	if err != nil {
		return result.WithError(err)
	}

	return r.validateTestScenario(ctx, result, scenario, config)
}

func checkEnrolledAgents(ctx context.Context, client *kibana.Client, agentInfo agentdeployer.AgentInfo, svcInfo servicedeployer.ServiceInfo, runIndependentElasticAgent bool) ([]kibana.Agent, error) {
	var agents []kibana.Agent

	enrolled, err := wait.UntilTrue(ctx, func(ctx context.Context) (bool, error) {
		allAgents, err := client.ListAgents(ctx)
		if err != nil {
			return false, fmt.Errorf("could not list agents: %w", err)
		}

		if runIndependentElasticAgent {
			agents = filterIndependentAgents(allAgents, agentInfo)
		} else {
			agents = filterAgents(allAgents, svcInfo)
		}
		logger.Debugf("found %d enrolled agent(s)", len(agents))
		if len(agents) == 0 {
			return false, nil // selected agents are unavailable yet
		}
		return true, nil
	}, 1*time.Second, 5*time.Minute)
	if err != nil {
		return nil, fmt.Errorf("agent enrollment failed: %w", err)
	}
	if !enrolled {
		return nil, errors.New("no agent enrolled in time")
	}
	return agents, nil
}

func createPackageDatastream(
	kibanaPolicy kibana.Policy,
	pkg packages.PackageManifest,
	policyTemplate packages.PolicyTemplate,
	ds packages.DataStreamManifest,
	config testConfig,
	suffix string,
) kibana.PackageDataStream {
	if pkg.Type == "input" {
		return createInputPackageDatastream(kibanaPolicy, pkg, policyTemplate, config, suffix)
	}
	return createIntegrationPackageDatastream(kibanaPolicy, pkg, policyTemplate, ds, config, suffix)
}

func createIntegrationPackageDatastream(
	kibanaPolicy kibana.Policy,
	pkg packages.PackageManifest,
	policyTemplate packages.PolicyTemplate,
	ds packages.DataStreamManifest,
	config testConfig,
	suffix string,
) kibana.PackageDataStream {
	r := kibana.PackageDataStream{
		Name:      fmt.Sprintf("%s-%s-%s", pkg.Name, ds.Name, suffix),
		Namespace: kibanaPolicy.Namespace,
		PolicyID:  kibanaPolicy.ID,
		Enabled:   true,
		Inputs: []kibana.Input{
			{
				PolicyTemplate: policyTemplate.Name,
				Enabled:        true,
			},
		},
	}
	r.Package.Name = pkg.Name
	r.Package.Title = pkg.Title
	r.Package.Version = pkg.Version

	stream := ds.Streams[getDataStreamIndex(config.Input, ds)]
	streamInput := stream.Input
	r.Inputs[0].Type = streamInput

	streams := []kibana.Stream{
		{
			ID:      fmt.Sprintf("%s-%s.%s", streamInput, pkg.Name, ds.Name),
			Enabled: true,
			DataStream: kibana.DataStream{
				Type:    ds.Type,
				Dataset: getDataStreamDataset(pkg, ds),
			},
		},
	}

	// Add dataStream-level vars
	streams[0].Vars = setKibanaVariables(stream.Vars, config.DataStream.Vars)
	r.Inputs[0].Streams = streams

	// Add input-level vars
	input := policyTemplate.FindInputByType(streamInput)
	if input != nil {
		r.Inputs[0].Vars = setKibanaVariables(input.Vars, config.Vars)
	}

	// Add package-level vars
	r.Vars = setKibanaVariables(pkg.Vars, config.Vars)

	return r
}

func createInputPackageDatastream(
	kibanaPolicy kibana.Policy,
	pkg packages.PackageManifest,
	policyTemplate packages.PolicyTemplate,
	config testConfig,
	suffix string,
) kibana.PackageDataStream {
	r := kibana.PackageDataStream{
		Name:      fmt.Sprintf("%s-%s-%s", pkg.Name, policyTemplate.Name, suffix),
		Namespace: kibanaPolicy.Namespace,
		PolicyID:  kibanaPolicy.ID,
		Enabled:   true,
	}
	r.Package.Name = pkg.Name
	r.Package.Title = pkg.Title
	r.Package.Version = pkg.Version
	r.Inputs = []kibana.Input{
		{
			PolicyTemplate: policyTemplate.Name,
			Enabled:        true,
			Vars:           kibana.Vars{},
		},
	}

	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),
			Enabled: true,
			DataStream: kibana.DataStream{
				Type:    policyTemplate.Type,
				Dataset: dataset,
			},
		},
	}

	// Add policyTemplate-level vars.
	vars := setKibanaVariables(policyTemplate.Vars, config.Vars)
	if _, found := vars["data_stream.dataset"]; !found {
		dataStreamDataset := dataset
		v, _ := config.Vars.GetValue("data_stream.dataset")
		if dataset, ok := v.(string); ok && dataset != "" {
			dataStreamDataset = dataset
		}

		var value packages.VarValue
		value.Unpack(dataStreamDataset)
		vars["data_stream.dataset"] = kibana.Var{
			Value: value,
			Type:  "text",
		}
	}

	streams[0].Vars = vars
	r.Inputs[0].Streams = streams
	return r
}

func setKibanaVariables(definitions []packages.Variable, values common.MapStr) kibana.Vars {
	vars := kibana.Vars{}
	for _, definition := range definitions {
		val := definition.Default

		value, err := values.GetValue(definition.Name)
		if err == nil {
			val = packages.VarValue{}
			val.Unpack(value)
		}

		vars[definition.Name] = kibana.Var{
			Type:  definition.Type,
			Value: val,
		}
	}
	return vars
}

// getDataStreamIndex returns the index of the data stream whose input name
// matches. Otherwise it returns the 0.
func getDataStreamIndex(inputName string, ds packages.DataStreamManifest) int {
	for i, s := range ds.Streams {
		if s.Input == inputName {
			return i
		}
	}
	return 0
}

func getDataStreamDataset(pkg packages.PackageManifest, ds packages.DataStreamManifest) string {
	if len(ds.Dataset) > 0 {
		return ds.Dataset
	}
	return fmt.Sprintf("%s.%s", pkg.Name, ds.Name)
}

// findPolicyTemplateForInput returns the name of the policy_template that
// applies to the input under test. An error is returned if no policy template
// matches or if multiple policy templates match and the response is ambiguous.
func findPolicyTemplateForInput(pkg packages.PackageManifest, ds packages.DataStreamManifest, inputName string) (string, error) {
	if pkg.Type == "input" {
		return findPolicyTemplateForInputPackage(pkg, inputName)
	}
	return findPolicyTemplateForDataStream(pkg, ds, inputName)
}

func findPolicyTemplateForDataStream(pkg packages.PackageManifest, ds packages.DataStreamManifest, inputName string) (string, error) {
	if inputName == "" {
		if len(ds.Streams) == 0 {
			return "", errors.New("no streams declared in data stream manifest")
		}
		inputName = ds.Streams[getDataStreamIndex(inputName, ds)].Input
	}

	var matchedPolicyTemplates []string
	for _, policyTemplate := range pkg.PolicyTemplates {
		// Does this policy_template include this input type?
		if policyTemplate.FindInputByType(inputName) == nil {
			continue
		}

		// Does the policy_template apply to this data stream (when data streams are specified)?
		if len(policyTemplate.DataStreams) > 0 && !slices.Contains(policyTemplate.DataStreams, ds.Name) {
			continue
		}

		matchedPolicyTemplates = append(matchedPolicyTemplates, policyTemplate.Name)
	}

	switch len(matchedPolicyTemplates) {
	case 1:
		return matchedPolicyTemplates[0], nil
	case 0:
		return "", fmt.Errorf("no policy template was found for data stream %q "+
			"with input type %q: verify that you have included the data stream "+
			"and input in the package's policy_template list", ds.Name, inputName)
	default:
		return "", fmt.Errorf("ambiguous result: multiple policy templates ([%s]) "+
			"were found that apply to data stream %q with input type %q: please "+
			"specify the 'policy_template' in the system test config",
			strings.Join(matchedPolicyTemplates, ", "), ds.Name, inputName)
	}
}

func findPolicyTemplateForInputPackage(pkg packages.PackageManifest, inputName string) (string, error) {
	if inputName == "" {
		if len(pkg.PolicyTemplates) == 0 {
			return "", errors.New("no policy templates specified for input package")
		}
		inputName = pkg.PolicyTemplates[0].Input
	}

	var matched []string
	for _, policyTemplate := range pkg.PolicyTemplates {
		if policyTemplate.Input != inputName {
			continue
		}

		matched = append(matched, policyTemplate.Name)
	}

	switch len(matched) {
	case 1:
		return matched[0], nil
	case 0:
		return "", fmt.Errorf("no policy template was found"+
			"with input type %q: verify that you have included the data stream "+
			"and input in the package's policy_template list", inputName)
	default:
		return "", fmt.Errorf("ambiguous result: multiple policy templates ([%s]) "+
			"with input type %q: please "+
			"specify the 'policy_template' in the system test config",
			strings.Join(matched, ", "), inputName)
	}
}

func selectPolicyTemplateByName(policies []packages.PolicyTemplate, name string) (packages.PolicyTemplate, error) {
	for _, policy := range policies {
		if policy.Name == name {
			return policy, nil
		}
	}
	return packages.PolicyTemplate{}, fmt.Errorf("policy template %q not found", name)
}

func (r *tester) checkTransforms(ctx context.Context, config *testConfig, pkgManifest *packages.PackageManifest, ds kibana.PackageDataStream, dataStream string) error {
	transforms, err := packages.ReadTransformsFromPackageRoot(r.packageRootPath)
	if err != nil {
		return fmt.Errorf("loading transforms for package failed (root: %s): %w", r.packageRootPath, err)
	}
	for _, transform := range transforms {
		hasSource, err := transform.HasSource(dataStream)
		if err != nil {
			return fmt.Errorf("failed to check if transform %q has %s as source: %w", transform.Name, dataStream, err)
		}
		if !hasSource {
			logger.Debugf("transform %q does not match %q as source (sources: %s)", transform.Name, dataStream, transform.Definition.Source.Index)
			continue
		}

		logger.Debugf("checking transform %q", transform.Name)

		// IDs format is: "<type>-<package>.<transform>-<namespace>-<version>"
		// For instance: "logs-ti_anomali.latest_ioc-default-0.1.0"
		transformPattern := fmt.Sprintf("%s-%s.%s-*-%s",
			ds.Inputs[0].Streams[0].DataStream.Type,
			pkgManifest.Name,
			transform.Name,
			transform.Definition.Meta.FleetTransformVersion,
		)
		transformId, err := r.getTransformId(ctx, transformPattern)
		if err != nil {
			return fmt.Errorf("failed to determine transform ID: %w", err)
		}

		// Using the preview instead of checking the actual index because
		// transforms with retention policies may be deleting the documents based
		// on old fixtures as soon as they are indexed.
		transformDocs, err := r.previewTransform(ctx, transformId)
		if err != nil {
			return fmt.Errorf("failed to preview transform %q: %w", transformId, err)
		}
		if len(transformDocs) == 0 {
			return fmt.Errorf("no documents found in preview for transform %q", transformId)
		}

		transformRootPath := filepath.Dir(transform.Path)
		fieldsValidator, err := fields.CreateValidatorForDirectory(transformRootPath,
			fields.WithSpecVersion(pkgManifest.SpecVersion),
			fields.WithNumericKeywordFields(config.NumericKeywordFields),
			fields.WithEnabledImportAllECSSChema(true),
		)
		if err != nil {
			return fmt.Errorf("creating fields validator for data stream failed (path: %s): %w", transformRootPath, err)
		}
		if err := validateFields(transformDocs, fieldsValidator, dataStream); err != nil {
			return err
		}
	}

	return nil
}

func (r *tester) getTransformId(ctx context.Context, transformPattern string) (string, error) {
	resp, err := r.esAPI.TransformGetTransform(
		r.esAPI.TransformGetTransform.WithContext(ctx),
		r.esAPI.TransformGetTransform.WithTransformID(transformPattern),
	)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.IsError() {
		return "", fmt.Errorf("failed to get transforms: %s", resp.String())
	}

	var transforms struct {
		Transforms []struct {
			ID string `json:"id"`
		} `json:"transforms"`
	}

	err = json.NewDecoder(resp.Body).Decode(&transforms)
	switch {
	case err != nil:
		return "", fmt.Errorf("failed to decode response: %w", err)
	case len(transforms.Transforms) == 0:
		return "", fmt.Errorf("no transform found with pattern %q", transformPattern)
	case len(transforms.Transforms) > 1:
		return "", fmt.Errorf("multiple transforms (%d) found with pattern %q", len(transforms.Transforms), transformPattern)
	}
	id := transforms.Transforms[0].ID
	if id == "" {
		return "", fmt.Errorf("empty ID found with pattern %q", transformPattern)
	}
	return id, nil
}

func (r *tester) previewTransform(ctx context.Context, transformId string) ([]common.MapStr, error) {
	resp, err := r.esAPI.TransformPreviewTransform(
		r.esAPI.TransformPreviewTransform.WithContext(ctx),
		r.esAPI.TransformPreviewTransform.WithTransformID(transformId),
	)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if resp.IsError() {
		return nil, fmt.Errorf("failed to preview transform %q: %s", transformId, resp.String())
	}

	var preview struct {
		Documents []common.MapStr `json:"preview"`
	}
	err = json.NewDecoder(resp.Body).Decode(&preview)
	if err != nil {
		return nil, fmt.Errorf("failed to decode response: %w", err)
	}

	return preview.Documents, nil
}

func filterAgents(allAgents []kibana.Agent, svcInfo servicedeployer.ServiceInfo) []kibana.Agent {
	if svcInfo.Agent.Host.NamePrefix != "" {
		logger.Debugf("filter agents using criteria: NamePrefix=%s", svcInfo.Agent.Host.NamePrefix)
	}

	var filtered []kibana.Agent
	for _, agent := range allAgents {
		if agent.PolicyRevision == 0 {
			continue // For some reason Kibana doesn't always return a valid policy revision (eventually it will be present and valid)
		}

		if svcInfo.Agent.Host.NamePrefix != "" && !strings.HasPrefix(agent.LocalMetadata.Host.Name, svcInfo.Agent.Host.NamePrefix) {
			continue
		}
		filtered = append(filtered, agent)
	}
	return filtered
}

func filterIndependentAgents(allAgents []kibana.Agent, agentInfo agentdeployer.AgentInfo) []kibana.Agent {
	// filtered list of agents must contain all agents started by the stack
	// they could be assigned the default policy (elastic-agent-managed-ep) or the test policy (ep-test-system-*)
	var filtered []kibana.Agent
	for _, agent := range allAgents {
		if agent.PolicyRevision == 0 {
			continue // For some reason Kibana doesn't always return a valid policy revision (eventually it will be present and valid)
		}

		if agent.Status != "online" {
			continue
		}

		if agent.PolicyID != agentInfo.Policy.ID {
			continue
		}

		filtered = append(filtered, agent)
	}
	return filtered
}

func writeSampleEvent(path string, doc common.MapStr, specVersion semver.Version) error {
	jsonFormatter := formatter.JSONFormatterBuilder(specVersion)
	body, err := jsonFormatter.Encode(doc)
	if err != nil {
		return fmt.Errorf("marshalling sample event failed: %w", err)
	}

	err = os.WriteFile(filepath.Join(path, "sample_event.json"), body, 0644)
	if err != nil {
		return fmt.Errorf("writing sample event failed: %w", err)
	}

	return nil
}

func validateFailureStore(failureStore []failureStoreDocument) error {
	var multiErr multierror.Error
	for _, doc := range failureStore {
		// TODO: Move this to the trace log level when available.
		logger.Debug("Error found in failure store: ", doc.Error.StackTrace)
		multiErr = append(multiErr,
			fmt.Errorf("%s: %s (processor: %s, pipelines: %s)",
				doc.Error.Type,
				doc.Error.Message,
				doc.Error.ProcessorType,
				strings.Join(doc.Error.PipelineTrace, ",")))
	}

	if len(multiErr) > 0 {
		multiErr = multiErr.Unique()
		return testrunner.ErrTestCaseFailed{
			Reason:  "one or more documents found in the failure store",
			Details: multiErr.Error(),
		}
	}

	return nil
}

func validateFields(docs []common.MapStr, fieldsValidator *fields.Validator, dataStream string) error {
	var multiErr multierror.Error
	for _, doc := range docs {
		if message, err := doc.GetValue("error.message"); err != common.ErrKeyNotFound {
			multiErr = append(multiErr, fmt.Errorf("found error.message in event: %v", message))
			continue
		}

		errs := fieldsValidator.ValidateDocumentMap(doc)
		if errs != nil {
			multiErr = append(multiErr, errs...)
			continue
		}
	}

	if len(multiErr) > 0 {
		multiErr = multiErr.Unique()
		return testrunner.ErrTestCaseFailed{
			Reason:  fmt.Sprintf("one or more errors found in documents stored in %s data stream", dataStream),
			Details: multiErr.Error(),
		}
	}

	return nil
}

func validateIgnoredFields(stackVersionString string, scenario *scenarioTest, config *testConfig) error {
	skipIgnoredFields := append([]string(nil), config.SkipIgnoredFields...)
	stackVersion, err := semver.NewVersion(stackVersionString)
	if err != nil {
		return fmt.Errorf("failed to parse stack version: %w", err)
	}
	if stackVersion.LessThan(semver.MustParse("8.14.0")) {
		// Pre 8.14 Elasticsearch commonly has event.original not mapped correctly, exclude from check: https://github.com/elastic/elasticsearch/pull/106714
		skipIgnoredFields = append(skipIgnoredFields, "event.original")
	}

	ignoredFields := make([]string, 0, len(scenario.ignoredFields))

	for _, field := range scenario.ignoredFields {
		if !slices.Contains(skipIgnoredFields, field) {
			ignoredFields = append(ignoredFields, field)
		}
	}

	if len(ignoredFields) > 0 {
		issues := make([]struct {
			ID            any `json:"_id"`
			Timestamp     any `json:"@timestamp,omitempty"`
			IgnoredFields any `json:"ignored_field_values"`
		}, len(scenario.degradedDocs))
		for i, d := range scenario.degradedDocs {
			issues[i].ID = d["_id"]
			if source, ok := d["_source"].(map[string]any); ok {
				if ts, ok := source["@timestamp"]; ok {
					issues[i].Timestamp = ts
				}
			}
			issues[i].IgnoredFields = d["ignored_field_values"]
		}
		degradedDocsJSON, err := json.MarshalIndent(issues, "", "  ")
		if err != nil {
			return fmt.Errorf("failed to marshal degraded docs to JSON: %w", err)
		}

		return testrunner.ErrTestCaseFailed{
			Reason:  "found ignored fields in data stream",
			Details: fmt.Sprintf("found ignored fields in data stream %s: %v. Affected documents: %s", scenario.dataStream, ignoredFields, degradedDocsJSON),
		}
	}

	return nil
}

func assertHitCount(expected int, docs []common.MapStr) (pass bool, message string) {
	if expected != 0 {
		observed := len(docs)
		logger.Debugf("assert hit count expected %d, observed %d", expected, observed)
		if observed != expected {
			return false, fmt.Sprintf("observed hit count %d did not match expected hit count %d", observed, expected)
		}
	}
	return true, ""
}

func (r *tester) generateTestResultFile(docs []common.MapStr, specVersion semver.Version) error {
	if !r.generateTestResult {
		return nil
	}

	rootPath := r.packageRootPath
	if ds := r.testFolder.DataStream; ds != "" {
		rootPath = filepath.Join(rootPath, "data_stream", ds)
	}

	if err := writeSampleEvent(rootPath, docs[0], specVersion); err != nil {
		return fmt.Errorf("failed to write sample event file: %w", err)
	}

	return nil
}

func (r *tester) checkNewAgentLogs(ctx context.Context, agent agentdeployer.DeployedAgent, startTesting time.Time, errorPatterns []logsByContainer, configName string) (results []testrunner.TestResult, err error) {
	if agent == nil {
		return nil, nil
	}

	f, err := os.CreateTemp("", "elastic-agent.logs")
	if err != nil {
		return nil, fmt.Errorf("failed to create temp file for logs: %w", err)
	}
	defer os.Remove(f.Name())

	for _, patternsContainer := range errorPatterns {
		if patternsContainer.containerName != "elastic-agent" {
			continue
		}

		startTime := time.Now()

		outputBytes, err := agent.Logs(ctx, startTesting)
		if err != nil {
			return nil, fmt.Errorf("check log messages failed: %s", err)
		}
		_, err = f.Write(outputBytes)
		if err != nil {
			return nil, fmt.Errorf("write log messages failed: %s", err)
		}

		err = r.anyErrorMessages(f.Name(), startTesting, patternsContainer.patterns)
		if e, ok := err.(testrunner.ErrTestCaseFailed); ok {
			tr := testrunner.TestResult{
				TestType:   TestType,
				Name:       fmt.Sprintf("(%s logs - %s)", patternsContainer.containerName, configName),
				Package:    r.testFolder.Package,
				DataStream: r.testFolder.DataStream,
			}
			tr.FailureMsg = e.Error()
			tr.FailureDetails = e.Details
			tr.TimeElapsed = time.Since(startTime)
			results = append(results, tr)
			// Just check elastic-agent
			break
		}

		if err != nil {
			return nil, fmt.Errorf("check log messages failed: %s", err)
		}
		// Just check elastic-agent
		break
	}
	return results, nil
}

func (r *tester) checkAgentLogs(dump []stack.DumpResult, startTesting time.Time, errorPatterns []logsByContainer) (results []testrunner.TestResult, err error) {
	for _, patternsContainer := range errorPatterns {
		startTime := time.Now()

		serviceDumpIndex := slices.IndexFunc(dump, func(d stack.DumpResult) bool {
			return d.ServiceName == patternsContainer.containerName
		})
		if serviceDumpIndex < 0 {
			return nil, fmt.Errorf("could not find logs dump for service %s", patternsContainer.containerName)
		}
		serviceLogsFile := dump[serviceDumpIndex].LogsFile

		err = r.anyErrorMessages(serviceLogsFile, startTesting, patternsContainer.patterns)
		if e, ok := err.(testrunner.ErrTestCaseFailed); ok {
			tr := testrunner.TestResult{
				TestType:   TestType,
				Name:       fmt.Sprintf("(%s logs)", patternsContainer.containerName),
				Package:    r.testFolder.Package,
				DataStream: r.testFolder.DataStream,
			}
			tr.FailureMsg = e.Error()
			tr.FailureDetails = e.Details
			tr.TimeElapsed = time.Since(startTime)
			results = append(results, tr)
			continue
		}

		if err != nil {
			return nil, fmt.Errorf("check log messages failed: %s", err)
		}
	}
	return results, nil
}

func (r *tester) anyErrorMessages(logsFilePath string, startTime time.Time, errorPatterns []logsRegexp) error {
	var multiErr multierror.Error
	processLog := func(log stack.LogLine) error {
		for _, pattern := range errorPatterns {
			if !pattern.includes.MatchString(log.Message) {
				continue
			}
			isExcluded := false
			for _, excludes := range pattern.excludes {
				if excludes.MatchString(log.Message) {
					isExcluded = true
					break
				}
			}
			if isExcluded {
				continue
			}

			multiErr = append(multiErr, fmt.Errorf("found error %q", log.Message))
		}
		return nil
	}
	err := stack.ParseLogs(stack.ParseLogsOptions{
		LogsFilePath: logsFilePath,
		StartTime:    startTime,
	}, processLog)
	if err != nil {
		return err
	}

	if len(multiErr) > 0 {
		return testrunner.ErrTestCaseFailed{
			Reason:  fmt.Sprintf("one or more errors found while examining %s", filepath.Base(logsFilePath)),
			Details: multiErr.Error(),
		}
	}
	return nil
}

func (r *tester) generateCoverageReport(pkgName string) (testrunner.CoverageReport, error) {
	dsPattern := "*"
	if r.dataStreamManifest != nil && r.dataStreamManifest.Name != "" {
		dsPattern = r.dataStreamManifest.Name
	}

	// This list of patterns includes patterns for all types of packages. It should not be a problem if some path doesn't exist.
	patterns := []string{
		filepath.Join(r.packageRootPath, "manifest.yml"),
		filepath.Join(r.packageRootPath, "fields", "*.yml"),
		filepath.Join(r.packageRootPath, "data_stream", dsPattern, "manifest.yml"),
		filepath.Join(r.packageRootPath, "data_stream", dsPattern, "fields", "*.yml"),
	}

	return testrunner.GenerateBaseFileCoverageReportGlob(pkgName, patterns, r.coverageType, true)
}