[Filebeat/Journald] Fix flakiness from TestDoubleStarCanBeUsed (#46913) (#47175)

---------


(cherry picked from commit f13a0bd)

Co-authored-by: Tiago Queiroz <[email protected]>
Co-authored-by: Anderson Queiroz <[email protected]>

Author: 3 people

.buildkite/filebeat/filebeat-pipeline.yml

@@ -86,6 +86,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
         plugins:
           - test-collector#v1.10.2:
               files: "filebeat/build/TEST-*.xml"
@@ -112,6 +113,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
         plugins:
           - test-collector#v1.10.2:
               files: "filebeat/build/TEST-*.xml"
@@ -138,6 +140,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
         plugins:
           - test-collector#v1.10.2:
               files: "filebeat/build/TEST-*.xml"
@@ -162,6 +165,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
           - "filebeat/build/integration-tests/*"
           - "filebeat/build/integration-tests/Test*/*"
           - "filebeat/build/integration-tests/Test*/data/**/*"
@@ -193,6 +197,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
           - "filebeat/build/integration-tests/*"
           - "filebeat/build/integration-tests/Test*/*"
           - "filebeat/build/integration-tests/Test*/data/**/*"
@@ -220,6 +225,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
           - "filebeat/build/integration-tests/*"
           - "filebeat/build/integration-tests/Test*/*"
           - "filebeat/build/integration-tests/Test*/data/**/*"
@@ -338,6 +344,7 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/input-test/**/*"
         plugins:
           - test-collector#v1.10.2:
               files: "filebeat/build/TEST-*.xml"

filebeat/input/journald/environment_test.go

@@ -20,15 +20,21 @@
 package journald
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/gofrs/uuid/v5"
 	"github.com/stretchr/testify/require"
+	"go.elastic.co/ecszap"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -52,6 +58,9 @@ type inputTestingEnvironment struct {
 	pluginInitOnce sync.Once
 	plugin         v2.Plugin
 
+	inputLogger *logp.Logger
+	logBuffer   *bytes.Buffer
+
 	wg  sync.WaitGroup
 	grp unison.TaskGroup
 }
@@ -92,6 +101,39 @@ func (e *inputTestingEnvironment) mustCreateInput(config map[string]interface{})
 
 func (e *inputTestingEnvironment) startInput(ctx context.Context, inp v2.Input) {
 	e.wg.Add(1)
+	t := e.t
+
+	e.inputLogger, e.logBuffer = newInMemoryJSON()
+	e.t.Cleanup(func() {
+		if t.Failed() {
+			folder := filepath.Join("..", "..", "build", "input-test")
+			if err := os.MkdirAll(folder, 0o750); err != nil {
+				t.Logf("cannot create folder for error logs: %s", err)
+				return
+			}
+
+			cleanTestName := strings.Replace(t.Name(), "\\", "_", -1)
+
+			f, err := os.CreateTemp(folder, cleanTestName+"-*")
+			if err != nil {
+				t.Logf("cannot create file for error logs: %s", err)
+				return
+			}
+			defer f.Close()
+			fullLogPath, err := filepath.Abs(f.Name())
+			if err != nil {
+				t.Logf("cannot get full path from log file: %s", err)
+			}
+
+			if _, err := f.Write(e.logBuffer.Bytes()); err != nil {
+				t.Logf("cannot write to file: %s", err)
+				return
+			}
+
+			t.Logf("Test Failed, logs from input at %q", fullLogPath)
+		}
+	})
+
 	go func(wg *sync.WaitGroup, grp *unison.TaskGroup) {
 		defer wg.Done()
 		defer func() {
@@ -108,7 +150,7 @@ func (e *inputTestingEnvironment) startInput(ctx context.Context, inp v2.Input)
 			Cancelation:     ctx,
 			StatusReporter:  e.statusReporter,
 			MetricsRegistry: monitoring.NewRegistry(),
-			Logger:          logp.L(),
+			Logger:          e.inputLogger,
 		}
 		if err := inp.Run(inputCtx, e.pipeline); err != nil {
 			e.t.Errorf("input 'Run' method returned an error: %s", err)
@@ -139,6 +181,23 @@ func (e *inputTestingEnvironment) waitUntilEventCount(count int) {
 	}, 5*time.Second, 10*time.Millisecond, &msg)
 }
 
+// waitUntilEventCount waits until total count events arrive to the client.
+func (e *inputTestingEnvironment) waitUntilEventsPublished(published int) {
+	e.t.Helper()
+	msg := strings.Builder{}
+	require.Eventually(e.t, func() bool {
+		sum := len(e.pipeline.GetAllEvents())
+		if sum >= published {
+			return true
+		}
+
+		msg.Reset()
+		fmt.Fprintf(&msg, "too few events; expected: %d, actual: %d", published, sum)
+
+		return false
+	}, 5*time.Second, 10*time.Millisecond, &msg)
+}
+
 func (e *inputTestingEnvironment) RequireStatuses(expected []statusUpdate) {
 	t := e.t
 	t.Helper()
@@ -308,3 +367,24 @@ func (m *mockStatusReporter) GetUpdates() []statusUpdate {
 	defer m.mutex.RUnlock()
 	return append([]statusUpdate{}, m.updates...)
 }
+
+func newInMemoryJSON() (*logp.Logger, *bytes.Buffer) {
+	buff := bytes.Buffer{}
+	encoderConfig := ecszap.ECSCompatibleEncoderConfig(logp.JSONEncoderConfig())
+	encoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
+	encoder := zapcore.NewJSONEncoder(encoderConfig)
+
+	core := zapcore.NewCore(
+		encoder,
+		zapcore.Lock(zapcore.AddSync(&buff)),
+		zap.NewAtomicLevelAt(zap.DebugLevel))
+	ecszap.ECSCompatibleEncoderConfig(logp.ConsoleEncoderConfig())
+
+	logger, _ := logp.NewDevelopmentLogger(
+		"journald",
+		zap.WrapCore(func(in zapcore.Core) zapcore.Core {
+			return core
+		}))
+
+	return logger, &buff
+}

filebeat/input/journald/input_test.go

@@ -509,7 +509,18 @@ func TestDoubleStarCanBeUsed(t *testing.T) {
 	}
 
 	env.startInput(ctx, inp)
-	env.waitUntilEventCount(len(srcFiles) * 10)
+	// Wait for at least 11 events, this means more than one journal file
+	// has been read and ingested.
+	//
+	// When many small journal files are ingested, the journalctl process
+	// may exit before the input has fully read its stdout, which makes us
+	// discard the last few lines/entries.
+	//
+	// We still correctly track the cursor of events published to the output,
+	// however the cursor returned by journalctl on this set of handcrafted
+	// journal files leads to us skipping some events.
+	// See https://github.com/elastic/beats/issues/46904.
+	env.waitUntilEventsPublished(11)
 }
 
 func decompress(t *testing.T, namegz string) string {