Merge pull request #43117 from crassirostris/fix-es-cluster-logging-tests

test/e2e/cluster_logging_es_utils.go

@@ -166,15 +166,15 @@ func (logsProvider *esLogsProvider) ReadEntries(pod *loggingPod) []*logEntry {
 		return nil
 	}
 
+	query := fmt.Sprintf("kubernetes.pod_name:%s AND kubernetes.namespace_name:%s", pod.Name, f.Namespace.Name)
+	framework.Logf("Sending a search request to Elasticsearch with the following query: %s", query)
+
 	// Ask Elasticsearch to return all the log lines that were tagged with the
 	// pod name. Ask for ten times as many log lines because duplication is possible.
 	body, err := proxyRequest.Namespace(api.NamespaceSystem).
 		Name("elasticsearch-logging").
 		Suffix("_search").
-		// TODO: Change filter to only match records from current test run
-		// after fluent-plugin-kubernetes_metadata_filter is enabled
-		// and optimize current query
-		Param("q", fmt.Sprintf("tag:*%s*", pod.Name)).
+		Param("q", query).
 		// Ask for more in case we included some unrelated records in our query
 		Param("size", strconv.Itoa(pod.ExpectedLinesNumber*10)).
 		DoRaw()

test/e2e/cluster_logging_utils.go

@@ -18,8 +18,8 @@ package e2e
 
 import (
 	"fmt"
+	"regexp"
 	"strconv"
-	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -42,6 +42,11 @@ const (
 	loggingContainerMemoryRequest = 10 * 1024 * 1024
 )
 
+var (
+	// Regexp, matching the contents of log entries, parsed or not
+	logEntryMessageRegex = regexp.MustCompile("(?:I\\d+ \\d+:\\d+:\\d+.\\d+       \\d+ logs_generator.go:67] )?(\\d+) .*")
+)
+
 // Type to track the progress of logs generating pod
 type loggingPod struct {
 	// Name of the pod
@@ -77,8 +82,12 @@ type loggingTestConfig struct {
 }
 
 func (entry *logEntry) getLogEntryNumber() (int, bool) {
-	chunks := strings.Split(entry.Payload, " ")
-	lineNumber, err := strconv.Atoi(strings.TrimSpace(chunks[0]))
+	submatch := logEntryMessageRegex.FindStringSubmatch(entry.Payload)
+	if submatch == nil || len(submatch) < 2 {
+		return 0, false
+	}
+
+	lineNumber, err := strconv.Atoi(submatch[1])
 	return lineNumber, err == nil
 }
 
@@ -212,6 +221,8 @@ func pullMissingLogsCount(logsProvider logsProvider, pod *loggingPod) int {
 func getMissingLinesCount(logsProvider logsProvider, pod *loggingPod) (int, error) {
 	entries := logsProvider.ReadEntries(pod)
 
+	framework.Logf("Got %d entries from provider", len(entries))
+
 	for _, entry := range entries {
 		lineNumber, ok := entry.getLogEntryNumber()
 		if !ok {