Testing: fix namespace error (#3036)

* Fixing namespace by removing special char and setting as lowercase * Update enroll_test.go Cleaner way found from testFQDN * Fixing import * Add retries for ES queries * Fixing Namespace globally * Pass namespace to ES queries --------- Co-authored-by: Shaunak Kashyap <[email protected]>
elastic · Jul 8, 2023 · 98913e1 · 98913e1
1 parent 05e7d68
commit 98913e1
Show file tree

Hide file tree

Showing 4 changed files with 100 additions and 41 deletions.
diff --git a/pkg/testing/define/define.go b/pkg/testing/define/define.go
@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
 	"runtime"
 	"strings"
 	"sync"
@@ -29,6 +30,7 @@ import (
 var osInfo *types.OSInfo
 var osInfoErr error
 var osInfoOnce sync.Once
+var noSpecialCharsRegexp = regexp.MustCompile("[^a-zA-Z0-9]+")
 
 // Require defines what this test requires for it to be run by the test runner.
 //
@@ -191,7 +193,13 @@ func getNamespace(t *testing.T, local bool) (string, error) {
 	name := fmt.Sprintf("%s-%s", prefix, t.Name())
 	hasher := sha256.New()
 	hasher.Write([]byte(name))
-	return base64.URLEncoding.EncodeToString(hasher.Sum(nil)), nil
+
+	// Fleet API requires the namespace to be lowercased and not contain
+	// special characters.
+	namespace := strings.ToLower(base64.URLEncoding.EncodeToString(hasher.Sum(nil)))
+	namespace = noSpecialCharsRegexp.ReplaceAllString(namespace, "")
+
+	return namespace, nil
 }
 
 // getESClient creates the elasticsearch client from the information passed from the test runner.

diff --git a/pkg/testing/tools/elasticsearch.go b/pkg/testing/tools/elasticsearch.go
@@ -104,16 +104,29 @@ func GetIndicesWithContext(ctx context.Context, client elastictransport.Interfac
 }
 
 // FindMatchingLogLines returns any logs with message fields that match the given line
-func FindMatchingLogLines(client elastictransport.Interface, line string) (Documents, error) {
-	return FindMatchingLogLinesWithContext(context.Background(), client, line)
+func FindMatchingLogLines(client elastictransport.Interface, namespace, line string) (Documents, error) {
+	return FindMatchingLogLinesWithContext(context.Background(), client, namespace, line)
 }
 
 // FindMatchingLogLinesWithContext returns any logs with message fields that match the given line
-func FindMatchingLogLinesWithContext(ctx context.Context, client elastictransport.Interface, line string) (Documents, error) {
+func FindMatchingLogLinesWithContext(ctx context.Context, client elastictransport.Interface, namespace, line string) (Documents, error) {
 	queryRaw := map[string]interface{}{
 		"query": map[string]interface{}{
-			"match_phrase": map[string]interface{}{
-				"message": line,
+			"bool": map[string]interface{}{
+				"must": []map[string]interface{}{
+					{
+						"match_phrase": map[string]interface{}{
+							"message": line,
+						},
+					},
+					{
+						"term": map[string]interface{}{
+							"data_stream.namespace": map[string]interface{}{
+								"value": namespace,
+							},
+						},
+					},
+				},
 			},
 		},
 	}
@@ -142,17 +155,30 @@ func FindMatchingLogLinesWithContext(ctx context.Context, client elastictranspor
 
 // CheckForErrorsInLogs checks to see if any error-level lines exist
 // excludeStrings can be used to remove any particular error strings from logs
-func CheckForErrorsInLogs(client elastictransport.Interface, excludeStrings []string) (Documents, error) {
-	return CheckForErrorsInLogsWithContext(context.Background(), client, excludeStrings)
+func CheckForErrorsInLogs(client elastictransport.Interface, namespace string, excludeStrings []string) (Documents, error) {
+	return CheckForErrorsInLogsWithContext(context.Background(), client, namespace, excludeStrings)
 }
 
 // CheckForErrorsInLogsWithContext checks to see if any error-level lines exist
 // excludeStrings can be used to remove any particular error strings from logs
-func CheckForErrorsInLogsWithContext(ctx context.Context, client elastictransport.Interface, excludeStrings []string) (Documents, error) {
+func CheckForErrorsInLogsWithContext(ctx context.Context, client elastictransport.Interface, namespace string, excludeStrings []string) (Documents, error) {
 	queryRaw := map[string]interface{}{
 		"query": map[string]interface{}{
-			"match": map[string]interface{}{
-				"log.level": "error",
+			"bool": map[string]interface{}{
+				"must": []map[string]interface{}{
+					{
+						"match": map[string]interface{}{
+							"log.level": "error",
+						},
+					},
+					{
+						"term": map[string]interface{}{
+							"data_stream.namespace": map[string]interface{}{
+								"value": namespace,
+							},
+						},
+					},
+				},
 			},
 		},
 	}

diff --git a/testing/integration/enroll_test.go b/testing/integration/enroll_test.go
@@ -29,13 +29,12 @@ func TestEnrollAndLog(t *testing.T) {
 	})
 	t.Logf("got namespace: %s", info.Namespace)
 
-	t.Logf("In SetupSuite")
 	agentFixture, err := define.NewFixture(t, define.Version())
 	require.NoError(t, err)
 
-	t.Logf("In TestEnroll")
 	kibClient := info.KibanaClient
-	// Enroll agent in Fleet with a test policy
+
+	t.Log("Enrolling agent in Fleet with a test policy")
 	createPolicyReq := kibana.AgentPolicy{
 		Name:        fmt.Sprintf("test-policy-enroll-%d", time.Now().Unix()),
 		Namespace:   info.Namespace,
@@ -51,14 +50,14 @@ func TestEnrollAndLog(t *testing.T) {
 			},
 		},
 	}
-	// Stage 1: Install
+
 	// As part of the cleanup process, we'll uninstall the agent
 	policy, err := tools.InstallAgentWithPolicy(t, agentFixture, kibClient, createPolicyReq)
 	require.NoError(t, err)
 	t.Logf("created policy: %s", policy.ID)
 
 	t.Cleanup(func() {
-		// After: unenroll
+		t.Logf("Cleanup: unenrolling agent")
 		err = tools.UnEnrollAgent(info.KibanaClient)
 		require.NoError(t, err)
 	})
@@ -74,38 +73,70 @@ func TestEnrollAndLog(t *testing.T) {
 		t.Logf("%s: %d/%d deleted: %d\n", run.Index, run.DocsCount, run.StoreSizeBytes, run.DocsDeleted)
 	}
 
-	// Stage 3: Make sure metricbeat logs are populated
-	docs, err := tools.GetLogsForDatastream(info.ESClient, "elastic_agent.metricbeat")
-	require.NoError(t, err)
+	t.Log("Making sure metricbeat logs are populated")
+	docs := findESDocs(t, func() (tools.Documents, error) {
+		return tools.GetLogsForDatastream(info.ESClient, "elastic_agent.metricbeat")
+	})
 	require.NotZero(t, len(docs.Hits.Hits))
 	t.Logf("metricbeat: Got %d documents", len(docs.Hits.Hits))
 
-	// Stage 4: Make sure filebeat logs are populated
-	docs, err = tools.GetLogsForDatastream(info.ESClient, "elastic_agent.filebeat")
-	require.NoError(t, err)
+	t.Log("Making sure filebeat logs are populated")
+	docs = findESDocs(t, func() (tools.Documents, error) {
+		return tools.GetLogsForDatastream(info.ESClient, "elastic_agent.filebeat")
+	})
 	require.NotZero(t, len(docs.Hits.Hits))
 	t.Logf("Filebeat: Got %d documents", len(docs.Hits.Hits))
 
-	// Stage 5: make sure we have no errors
-	docs, err = tools.CheckForErrorsInLogs(info.ESClient, []string{})
-	require.NoError(t, err)
+	t.Log("Making sure there are no error logs")
+	docs = findESDocs(t, func() (tools.Documents, error) {
+		return tools.CheckForErrorsInLogs(info.ESClient, info.Namespace, []string{})
+	})
 	t.Logf("errors: Got %d documents", len(docs.Hits.Hits))
 	for _, doc := range docs.Hits.Hits {
 		t.Logf("%#v", doc.Source)
 	}
 	require.Empty(t, docs.Hits.Hits)
 
-	// Stage 6: Make sure we have message confirming central management is running
-	docs, err = tools.FindMatchingLogLines(info.ESClient, "Parsed configuration and determined agent is managed by Fleet")
-	require.NoError(t, err)
+	t.Log("Making sure we have message confirming central management is running")
+	docs = findESDocs(t, func() (tools.Documents, error) {
+		return tools.FindMatchingLogLines(info.ESClient, info.Namespace, "Parsed configuration and determined agent is managed by Fleet")
+	})
 	require.NotZero(t, len(docs.Hits.Hits))
 
+	t.Log("Check for metricbeat starting message")
 	// Stage 7: check for starting messages
-	docs, err = tools.FindMatchingLogLines(info.ESClient, "metricbeat start running")
-	require.NoError(t, err)
+	docs = findESDocs(t, func() (tools.Documents, error) {
+		return tools.FindMatchingLogLines(info.ESClient, info.Namespace, "metricbeat start running")
+	})
 	require.NotZero(t, len(docs.Hits.Hits))
 
-	docs, err = tools.FindMatchingLogLines(info.ESClient, "filebeat start running")
-	require.NoError(t, err)
+	t.Log("Check for filebeat starting message")
+	docs = findESDocs(t, func() (tools.Documents, error) {
+		return tools.FindMatchingLogLines(info.ESClient, info.Namespace, "filebeat start running")
+	})
 	require.NotZero(t, len(docs.Hits.Hits))
 }
+
+func findESDocs(t *testing.T, findFn func() (tools.Documents, error)) tools.Documents {
+	var docs tools.Documents
+
+	require.Eventually(
+		t,
+		func() bool {
+			var err error
+			docs, err = findFn()
+			return err == nil
+		},
+		3*time.Minute,
+		15*time.Second,
+	)
+
+	// TODO: remove after debugging
+	t.Log("--- debugging: results from ES --- START ---")
+	for _, doc := range docs.Hits.Hits {
+		t.Logf("%#v", doc.Source)
+	}
+	t.Log("--- debugging: results from ES --- END ---")
+
+	return docs
+}
diff --git a/testing/integration/fqdn_test.go b/testing/integration/fqdn_test.go
@@ -14,7 +14,6 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
-	"regexp"
 	"strings"
 	"testing"
 	"time"
@@ -76,15 +75,10 @@ func TestFQDN(t *testing.T) {
 	err = setHostFQDN(ctx, origEtcHosts, externalIP, fqdn, t.Log)
 	require.NoError(t, err)
 
-	// Fleet API requires the namespace to be lowercased and not contain
-	// special characters.
-	policyNamespace := strings.ToLower(info.Namespace)
-	policyNamespace = regexp.MustCompile("[^a-zA-Z0-9]+").ReplaceAllString(policyNamespace, "")
-
 	t.Log("Enroll agent in Fleet with a test policy")
 	createPolicyReq := kibana.AgentPolicy{
 		Name:        "test-policy-fqdn-" + strings.ReplaceAll(fqdn, ".", "-"),
-		Namespace:   policyNamespace,
+		Namespace:   info.Namespace,
 		Description: fmt.Sprintf("Test policy for FQDN E2E test (%s)", fqdn),
 		MonitoringEnabled: []kibana.MonitoringEnabledOption{
 			kibana.MonitoringEnabledLogs,
@@ -116,7 +110,7 @@ func TestFQDN(t *testing.T) {
 	}
 	updatePolicyReq := kibana.AgentPolicyUpdateRequest{
 		Name:          policy.Name,
-		Namespace:     policyNamespace,
+		Namespace:     info.Namespace,
 		AgentFeatures: policy.AgentFeatures,
 	}
 	_, err = kibClient.UpdatePolicy(policy.ID, updatePolicyReq)
@@ -147,7 +141,7 @@ func TestFQDN(t *testing.T) {
 	}
 	updatePolicyReq = kibana.AgentPolicyUpdateRequest{
 		Name:          policy.Name,
-		Namespace:     policyNamespace,
+		Namespace:     info.Namespace,
 		AgentFeatures: policy.AgentFeatures,
 	}
 	_, err = kibClient.UpdatePolicy(policy.ID, updatePolicyReq)