[Ingest-Manager] Implementation for the Deploy scenario of the stand-alone mode

e2e/_suites/ingest-manager/ingest-manager_test.go

@@ -21,12 +21,22 @@ var stackVersion = "8.0.0-SNAPSHOT"
 // affecting the runtime dependencies (or profile)
 var profileEnv map[string]string
 
+// queryMaxAttempts is the number of attempts to query elasticsearch before aborting
+// It can be overriden by OP_QUERY_MAX_ATTEMPTS env var
+var queryMaxAttempts = 5
+
+// queryRetryTimeout is the number of seconds between elasticsearch retry queries.
+// It can be overriden by OP_RETRY_TIMEOUT env var
+var queryRetryTimeout = 3
+
 // All URLs running on localhost as Kibana is expected to be exposed there
 const kibanaBaseURL = "http://localhost:5601"
 
 func init() {
 	config.Init()
 
+	queryMaxAttempts = e2e.GetIntegerFromEnv("OP_QUERY_MAX_ATTEMPTS", queryMaxAttempts)
+	queryRetryTimeout = e2e.GetIntegerFromEnv("OP_RETRY_TIMEOUT", queryRetryTimeout)
 	stackVersion = e2e.GetEnv("OP_STACK_VERSION", stackVersion)
 }
 

e2e/_suites/ingest-manager/stand-alone.go

@@ -1,7 +1,11 @@
 package main
 
 import (
+	"context"
+	"time"
+
 	"github.com/cucumber/godog"
+	"github.com/elastic/e2e-testing/cli/docker"
 	"github.com/elastic/e2e-testing/cli/services"
 	"github.com/elastic/e2e-testing/e2e"
 	log "github.com/sirupsen/logrus"
@@ -66,7 +70,113 @@ func (sats *StandAloneTestSuite) aStandaloneAgentIsDeployed() error {
 }
 
 func (sats *StandAloneTestSuite) thereIsNewDataInTheIndexFromAgent() error {
-	return godog.ErrPending
+	timezone := "America/New_York"
+	now := time.Now()
+	startDate := now.Add(-15 * time.Minute)
+
+	serviceName := "ingest-manager_elastic-agent_1"
+	hostname, err := getContainerName(serviceName)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"error":   err,
+			"service": serviceName,
+		}).Error("Could not retrieve container name from the Docker client")
+		return err
+	}
+
+	esQuery := map[string]interface{}{
+		"version": true,
+		"size":    500,
+		"docvalue_fields": []map[string]interface{}{
+			{
+				"field":  "@timestamp",
+				"format": "date_time",
+			},
+			{
+				"field":  "system.process.cpu.start_time",
+				"format": "date_time",
+			},
+			{
+				"field":  "system.service.state_since",
+				"format": "date_time",
+			},
+		},
+		"_source": map[string]interface{}{
+			"excludes": []map[string]interface{}{},
+		},
+		"query": map[string]interface{}{
+			"bool": map[string]interface{}{
+				"must": []map[string]interface{}{},
+				"filter": []map[string]interface{}{
+					{
+						"bool": map[string]interface{}{
+							"filter": []map[string]interface{}{
+								{
+									"bool": map[string]interface{}{
+										"should": []map[string]interface{}{
+											{
+												"match_phrase": map[string]interface{}{
+													"host.name": hostname,
+												},
+											},
+										},
+										"minimum_should_match": 1,
+									},
+								},
+								{
+									"bool": map[string]interface{}{
+										"should": []map[string]interface{}{
+											{
+												"range": map[string]interface{}{
+													"@timestamp": map[string]interface{}{
+														"gte":       now,
+														"time_zone": timezone,
+													},
+												},
+											},
+										},
+										"minimum_should_match": 1,
+									},
+								},
+							},
+						},
+					},
+					{
+						"range": map[string]interface{}{
+							"@timestamp": map[string]interface{}{
+								"gte":    startDate,
+								"format": "strict_date_optional_time",
+							},
+						},
+					},
+				},
+				"should":   []map[string]interface{}{},
+				"must_not": []map[string]interface{}{},
+			},
+		},
+	}
+
+	// wait an amount of time for documents
+	e2e.Sleep("5")
+
+	indexName := "logs-agent-default"
+
+	result, err := e2e.RetrySearch(indexName, esQuery, queryMaxAttempts, queryRetryTimeout)
+	if err != nil {
+		return err
+	}
+
+	log.Debugf("Search result: %v", result)
+
+	err = e2e.AssertHitsArePresent(result)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"index": indexName,
+		}).Error(err.Error())
+		return err
+	}
+
+	return nil
 }
 
 func (sats *StandAloneTestSuite) theDockerContainerIsStopped(arg1 string) error {
@@ -76,3 +186,21 @@ func (sats *StandAloneTestSuite) theDockerContainerIsStopped(arg1 string) error
 func (sats *StandAloneTestSuite) thereIsNoNewDataInTheIndexAfterAgentShutsDown() error {
 	return godog.ErrPending
 }
+
+func getContainerName(serviceName string) (string, error) {
+	log.WithFields(log.Fields{
+		"service": serviceName,
+	}).Debug("Retrieving container name from the Docker client")
+
+	containerName, err := docker.ExecCommandIntoContainer(context.Background(), serviceName, "root", []string{"hostname"})
+	if err != nil {
+		return "", err
+	}
+
+	log.WithFields(log.Fields{
+		"containerName": containerName,
+		"service":       serviceName,
+	}).Debug("Container name retrieved from the Docker client")
+
+	return containerName, nil
+}

e2e/_suites/metricbeat/metricbeat_test.go

@@ -408,7 +408,17 @@ func (mts *MetricbeatTestSuite) thereAreEventsInTheIndex() error {
 		return err
 	}
 
-	return e2e.AssertHitsArePresent(result, mts.Query)
+	err = e2e.AssertHitsArePresent(result)
+	if err != nil {
+		log.WithFields(log.Fields{
+			"eventModule":    mts.Query.EventModule,
+			"index":          mts.Query.IndexName,
+			"serviceVersion": mts.Query.ServiceVersion,
+		}).Error(err.Error())
+		return err
+	}
+
+	return nil
 }
 
 func (mts *MetricbeatTestSuite) thereAreNoErrorsInTheIndex() error {

e2e/assertions.go

@@ -7,12 +7,10 @@ import (
 )
 
 // AssertHitsArePresent returns an error if no hits are present
-func AssertHitsArePresent(hits map[string]interface{}, q ElasticsearchQuery) error {
+func AssertHitsArePresent(hits map[string]interface{}) error {
 	hitsCount := len(hits["hits"].(map[string]interface{})["hits"].([]interface{}))
 	if hitsCount == 0 {
-		return fmt.Errorf(
-			"There aren't documents for %s-%s on Metricbeat index %s",
-			q.EventModule, q.ServiceVersion, q.IndexName)
+		return fmt.Errorf("There aren't documents in the index")
 	}
 
 	return nil

e2e/elasticsearch.go

@@ -83,6 +83,8 @@ func getElasticsearchClientFromHostPort(host string, port int) (*es.Client, erro
 
 	cfg := es.Config{
 		Addresses: []string{fmt.Sprintf("http://%s:%d", host, port)},
+		Username:  "elastic",
+		Password:  "changeme",
 	}
 	esClient, err := es.NewClient(cfg)
 	if err != nil {