[Blocked] Add test for running Log input as Filestream in Beatsreceivers

NOTICE-fips.txt

@@ -787,11 +787,11 @@ Contents of probable licence file $GOMODCACHE/github.com/dolmen-go/contextio@v0.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/beats/v7
-Version: v7.0.0-alpha2.0.20251202130319-deec5c55e9b0
+Version: v7.0.0-alpha2.0.20251203173126-96dc1b39c163
 Licence type (autodetected): Elastic
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/[email protected].20251202130319-deec5c55e9b0/LICENSE.txt:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/[email protected].20251203173126-96dc1b39c163/LICENSE.txt:
 
 Source code in this repository is variously licensed under the Apache License
 Version 2.0, an Apache compatible license, or the Elastic License. Outside of

NOTICE.txt

@@ -787,11 +787,11 @@ Contents of probable licence file $GOMODCACHE/github.com/dolmen-go/contextio@v0.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/beats/v7
-Version: v7.0.0-alpha2.0.20251202130319-deec5c55e9b0
+Version: v7.0.0-alpha2.0.20251203173126-96dc1b39c163
 Licence type (autodetected): Elastic
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/[email protected].20251202130319-deec5c55e9b0/LICENSE.txt:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/beats/[email protected].20251203173126-96dc1b39c163/LICENSE.txt:
 
 Source code in this repository is variously licensed under the Apache License
 Version 2.0, an Apache compatible license, or the Elastic License. Outside of

dev-tools/notice/overrides.json

@@ -1,4 +1,5 @@
 {"name": "github.com/elastic/beats/v7", "licenceType": "Elastic"}
+{"name": "github.com/belimawr/beats/v7", "licenceType": "Elastic"}
 {"name": "github.com/elastic/elastic-agent-client/v7", "licenceType": "Elastic"}
 {"name": "github.com/gorhill/cronexpr", "licenceType": "Apache-2.0", "licenceFile":"APLv2"}
 {"name": "github.com/hashicorp/cronexpr", "licenceType": "Apache-2.0", "licenceFile":"APLv2"}

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/docker/docker v28.5.1+incompatible
 	github.com/docker/go-units v0.5.0
 	github.com/dolmen-go/contextio v0.0.0-20200217195037-68fc5150bcd5
-	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251202130319-deec5c55e9b0
+	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251203173126-96dc1b39c163
 	github.com/elastic/cloud-on-k8s/v2 v2.0.0-20250327073047-b624240832ae
 	github.com/elastic/elastic-agent-autodiscover v0.10.0
 	github.com/elastic/elastic-agent-client/v7 v7.17.2
@@ -243,6 +243,7 @@ require (
 	github.com/bitfield/gotestdox v0.2.2 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/bmatcuk/doublestar/v4 v4.9.1 // indirect
+	github.com/brianvoe/gofakeit v3.18.0+incompatible // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/chai2010/gettext-go v1.0.2 // indirect
 	github.com/cilium/ebpf v0.19.0 // indirect

go.sum

@@ -503,8 +503,8 @@ github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumpti
 github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumption v1.1.0-elastic/go.mod h1:0vCBR1wgGwZeGmloJ+eCWIZF2S47grTXRzj2mftg2Nk=
 github.com/elastic/bayeux v1.0.5 h1:UceFq01ipmT3S8DzFK+uVAkbCdiPR0Bqei8qIGmUeY0=
 github.com/elastic/bayeux v1.0.5/go.mod h1:CSI4iP7qeo5MMlkznGvYKftp8M7qqP/3nzmVZoXHY68=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251202130319-deec5c55e9b0 h1:NLqI2no2x3cSjcKH76C9ItRw0GiQVC7/uoJ/DLurEcE=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251202130319-deec5c55e9b0/go.mod h1:jFJMJMOG18CdEbvM3UKQnn3Ft66j2vgMfStZwLAcK+A=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251203173126-96dc1b39c163 h1:U/zcMEPIsEhzXJ8PjEFyn5bgy4GJKIFuTQ/MVr9Zry4=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251203173126-96dc1b39c163/go.mod h1:jFJMJMOG18CdEbvM3UKQnn3Ft66j2vgMfStZwLAcK+A=
 github.com/elastic/cloud-on-k8s/v2 v2.0.0-20250327073047-b624240832ae h1:OiShmbWAyGU0MS0ADJWr1/QgeLIZliMk9xsrFicR3/s=
 github.com/elastic/cloud-on-k8s/v2 v2.0.0-20250327073047-b624240832ae/go.mod h1:D2IckZVXARugvikE4fv1glvaJMohKSZRzrPsxCjo9O0=
 github.com/elastic/elastic-agent-autodiscover v0.10.0 h1:WJ4zl9uSfk1kHmn2B/0byQBLIL607Zt4LNfOgV7+XN0=

internal/edot/go.mod

@@ -5,7 +5,7 @@ go 1.24.11
 replace github.com/elastic/elastic-agent => ../../
 
 require (
-	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251202130319-deec5c55e9b0
+	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251203173126-96dc1b39c163
 	github.com/elastic/elastic-agent v0.0.0-00010101000000-000000000000
 	github.com/elastic/elastic-agent-libs v0.26.2
 	github.com/elastic/opentelemetry-collector-components/connector/elasticapmconnector v0.20.0

internal/edot/go.sum

@@ -432,8 +432,8 @@ github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumpti
 github.com/elastic/azure-sdk-for-go/sdk/resourcemanager/consumption/armconsumption v1.1.0-elastic/go.mod h1:0vCBR1wgGwZeGmloJ+eCWIZF2S47grTXRzj2mftg2Nk=
 github.com/elastic/bayeux v1.0.5 h1:UceFq01ipmT3S8DzFK+uVAkbCdiPR0Bqei8qIGmUeY0=
 github.com/elastic/bayeux v1.0.5/go.mod h1:CSI4iP7qeo5MMlkznGvYKftp8M7qqP/3nzmVZoXHY68=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251202130319-deec5c55e9b0 h1:NLqI2no2x3cSjcKH76C9ItRw0GiQVC7/uoJ/DLurEcE=
-github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251202130319-deec5c55e9b0/go.mod h1:jFJMJMOG18CdEbvM3UKQnn3Ft66j2vgMfStZwLAcK+A=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251203173126-96dc1b39c163 h1:U/zcMEPIsEhzXJ8PjEFyn5bgy4GJKIFuTQ/MVr9Zry4=
+github.com/elastic/beats/v7 v7.0.0-alpha2.0.20251203173126-96dc1b39c163/go.mod h1:jFJMJMOG18CdEbvM3UKQnn3Ft66j2vgMfStZwLAcK+A=
 github.com/elastic/elastic-agent-autodiscover v0.10.0 h1:WJ4zl9uSfk1kHmn2B/0byQBLIL607Zt4LNfOgV7+XN0=
 github.com/elastic/elastic-agent-autodiscover v0.10.0/go.mod h1:Nf3zh9FcJ9nTTswTwDTUAqXmvQllOrNliM6xmORSxwE=
 github.com/elastic/elastic-agent-client/v7 v7.17.2 h1:Cl2TeABqWZgW40t5fchGWT/sRk4MDDLWA0d8iHHOxLA=

magefile.go

@@ -2301,7 +2301,6 @@ func (Integration) Local(ctx context.Context, testName string) error {
 	params.Packages = []string{
 		"github.com/elastic/elastic-agent/testing/integration/...",
 	}
-
 	var goTestFlags []string
 	rawTestFlags := os.Getenv("GOTEST_FLAGS")
 	if rawTestFlags != "" {

pkg/testing/linux/linux.go

@@ -81,7 +81,7 @@ func linuxCopy(ctx context.Context, sshClient ssh.SSHClient, logger common.Logge
 			err, stdout, stderr)
 	}
 
-	stdOut, errOut, err := sshClient.Exec(ctx, "unzip", []string{destRepoName, "-d", "agent"}, nil)
+	stdOut, errOut, err := sshClient.Exec(ctx, "unzip", []string{"-o", destRepoName, "-d", "agent"}, nil)
 	if err != nil {
 		return fmt.Errorf("failed to unzip %s to agent directory: %w (stdout: %s, stderr: %s)", destRepoName, err, stdOut, errOut)
 	}

testing/integration/common.go

@@ -7,6 +7,7 @@
 package integration
 
 import (
+	"errors"
 	"fmt"
 	"net/url"
 	"os"
@@ -21,6 +22,9 @@ import (
 
 func GetESHost() (string, error) {
 	fixedESHost := os.Getenv("ELASTICSEARCH_HOST")
+	if len(fixedESHost) == 0 {
+		return "", errors.New("ELASTICSEARCH_HOST cannot be empty")
+	}
 	parsedES, err := url.Parse(fixedESHost)
 	if err != nil {
 		return "", err

testing/integration/ess/beat_receivers_test.go

@@ -188,9 +188,7 @@ func TestClassicAndReceiverAgentMonitoring(t *testing.T) {
 	require.NoError(t, err, "error reading policy response")
 	defer resp.Body.Close()
 
-	apiKeyResponse, err := createESApiKey(info.ESClient)
-	require.NoError(t, err, "failed to get api key")
-	require.True(t, len(apiKeyResponse.Encoded) > 1, "api key is invalid %q", apiKeyResponse)
+	apiKeyResponse := createESApiKey(t, info.ESClient)
 	apiKey, err := getDecodedApiKey(apiKeyResponse)
 	require.NoError(t, err, "error decoding api key")
 
@@ -464,9 +462,7 @@ outputs:
 
 	esEndpoint, err := integration.GetESHost()
 	require.NoError(t, err, "error getting elasticsearch endpoint")
-	esApiKey, err := createESApiKey(info.ESClient)
-	require.NoError(t, err, "error creating API key")
-	require.True(t, len(esApiKey.Encoded) > 1, "api key is invalid %q", esApiKey)
+	esApiKey := createESApiKey(t, info.ESClient)
 
 	beatsApiKey, err := base64.StdEncoding.DecodeString(esApiKey.Encoded)
 	require.NoError(t, err, "error decoding api key")
@@ -1203,9 +1199,7 @@ func TestSensitiveLogsESExporter(t *testing.T) {
 	}
 	esEndpoint, err := integration.GetESHost()
 	require.NoError(t, err, "error getting elasticsearch endpoint")
-	esApiKey, err := createESApiKey(info.ESClient)
-	require.NoError(t, err, "error creating API key")
-	require.True(t, len(esApiKey.Encoded) > 1, "api key is invalid %q", esApiKey)
+	esApiKey := createESApiKey(t, info.ESClient)
 	decodedApiKey, err := getDecodedApiKey(esApiKey)
 	require.NoError(t, err)
 
@@ -1386,9 +1380,7 @@ func TestSensitiveIncludeSourceOnError(t *testing.T) {
 	}
 	esEndpoint, err := integration.GetESHost()
 	require.NoError(t, err, "error getting elasticsearch endpoint")
-	esApiKey, err := createESApiKey(info.ESClient)
-	require.NoError(t, err, "error creating API key")
-	require.True(t, len(esApiKey.Encoded) > 1, "api key is invalid %q", esApiKey)
+	esApiKey := createESApiKey(t, info.ESClient)
 	decodedApiKey, err := getDecodedApiKey(esApiKey)
 	require.NoError(t, err)
 

testing/integration/ess/logfile.go

@@ -0,0 +1,160 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package ess
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// LogFile wraps a *os.File and makes it more suitable for tests.
+// Key features:
+//   - On failures, the file is kept and its path printed
+//   - Methods to search and wait for substrings in lines are provided,
+//     they keep track of the offset, ensuring ordering when
+//     when searching.
+type LogFile struct {
+	*os.File
+	offset               int64
+	KeepLogFileOnSuccess bool
+}
+
+// NewLogFile returns a new LogFile, path must be the components of a path,
+// they will be joined using the OS path separator.
+// If path is not provided, os.TempDir is used as the base path for the file.
+func NewLogFile(t testing.TB, path ...string) *LogFile {
+	dir := filepath.Join(path...)
+	if dir == "" {
+		dir = os.TempDir()
+	}
+
+	if err := os.MkdirAll(dir, 0o750); err != nil {
+		t.Fatalf("cannot create folder for logs: %s", err)
+	}
+
+	f, err := os.CreateTemp(dir, "elastic-agent-*.ndjson")
+	if err != nil {
+		t.Fatalf("cannot create log file: %s", err)
+	}
+
+	lf := &LogFile{
+		File: f,
+	}
+
+	t.Cleanup(func() {
+		if err := f.Sync(); err != nil {
+			t.Logf("cannot sync log file: %s", err)
+		}
+
+		if err := f.Close(); err != nil {
+			t.Logf("cannot close log file: %s", err)
+		}
+
+		// If the test failed, print the log file location,
+		// otherwise remove it.
+		if t.Failed() || lf.KeepLogFileOnSuccess {
+			t.Logf("Full logs written to %s", f.Name())
+			return
+		}
+
+		if err := os.Remove(f.Name()); err != nil {
+			t.Logf("could not remove temporary log file: %s", err)
+		}
+	})
+
+	return lf
+}
+
+// WaitLogsContains waits for the specified string s to be present in the logs within
+// the given timeout duration and fails the test if s is not found.
+// It keeps track of the log file offset, reading only new lines. Each
+// subsequent call to WaitLogsContains will only check logs not yet evaluated.
+// msgAndArgs should be a format string and arguments that will be printed
+// if the logs are not found, providing additional context for debugging.
+func (l *LogFile) WaitLogsContains(t testing.TB, s string, timeout time.Duration, msgAndArgs ...any) {
+	t.Helper()
+	require.EventuallyWithT(
+		t,
+		func(c *assert.CollectT) {
+			found, err := l.FindInLogs(s)
+			if err != nil {
+				c.Errorf("cannot check the log file: %s", err)
+				return
+			}
+
+			if !found {
+				c.Errorf("did not find '%s' in the logs", s)
+			}
+		},
+		timeout,
+		100*time.Millisecond,
+		msgAndArgs...)
+}
+
+// LogContains searches for str in the log file keeping track of the
+// offset. If there is any issue reading the log file, then t.Fatalf is called,
+// if str is not present in the logs, t.Errorf is called.
+func (l *LogFile) LogContains(t testing.TB, str string) {
+	t.Helper()
+	found, err := l.FindInLogs(str)
+	if err != nil {
+		t.Fatalf("cannot read log file: %s", err)
+	}
+
+	if !found {
+		t.Errorf("'%s' not found in logs", str)
+	}
+}
+
+// FindInLogs searches for str in the log file keeping track of the offset.
+// It returns true if str is found in the logs. If there are any errors,
+// it returns false and the error
+func (l *LogFile) FindInLogs(str string) (bool, error) {
+	// Open the file again so we can seek and not interfere with
+	// the logger writing to it.
+	f, err := os.Open(l.Name())
+	if err != nil {
+		return false, fmt.Errorf("cannot open log file for reading: %w", err)
+	}
+
+	if _, err := f.Seek(l.offset, io.SeekStart); err != nil {
+		return false, fmt.Errorf("cannot seek log file: %w", err)
+	}
+
+	r := bufio.NewReader(f)
+	for {
+		data, err := r.ReadBytes('\n')
+		line := string(data)
+		l.offset += int64(len(data))
+
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				return false, fmt.Errorf("error reading log file '%s': %w", l.Name(), err)
+			}
+			break
+		}
+
+		if strings.Contains(line, str) {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
+// ResetOffset resets the log file offset
+func (l *LogFile) ResetOffset() {
+	l.offset = 0
+}