[9.1](backport #46308) Accomodate logger changes from tlscommon package (#46427)

mergify[bot] · khushijain21 · web-flow · commit 9636d83d9286 · 2025-09-09T11:19:06.000+05:30
* Accomodate logger changes from `tlscommon` package (#46308) This PR accommodates breaking changes made in elastic-agent-libs . It effectively uses local logger in place of global ones (cherry picked from commit 0788d61) # Conflicts: # x-pack/filebeat/input/awss3/decoding_parquet.go # x-pack/filebeat/input/awss3/s3_objects.go # x-pack/filebeat/input/azureblobstorage/decoding.go # x-pack/filebeat/input/azureblobstorage/job.go # x-pack/filebeat/input/gcs/job.go * fix conflicts * fix ci --------- Co-authored-by: Khushi Jain <khushi.jain@elastic.co>
diff --git a/NOTICE.txt b/NOTICE.txt
@@ -12350,11 +12350,11 @@ SOFTWARE
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-libs
-Version: v0.23.0
+Version: v0.24.0
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-libs@v0.23.0/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-libs@v0.24.0/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004
diff --git a/go.mod b/go.mod
@@ -174,7 +174,7 @@ require (
 	github.com/elastic/bayeux v1.0.5
 	github.com/elastic/ebpfevents v0.7.0
 	github.com/elastic/elastic-agent-autodiscover v0.10.0
-	github.com/elastic/elastic-agent-libs v0.23.0
+	github.com/elastic/elastic-agent-libs v0.24.0
 	github.com/elastic/elastic-agent-system-metrics v0.13.2
 	github.com/elastic/go-elasticsearch/v8 v8.18.1
 	github.com/elastic/go-freelru v0.16.0
diff --git a/go.sum b/go.sum
@@ -364,8 +364,8 @@ github.com/elastic/elastic-agent-autodiscover v0.10.0 h1:WJ4zl9uSfk1kHmn2B/0byQB
 github.com/elastic/elastic-agent-autodiscover v0.10.0/go.mod h1:Nf3zh9FcJ9nTTswTwDTUAqXmvQllOrNliM6xmORSxwE=
 github.com/elastic/elastic-agent-client/v7 v7.15.0 h1:nDB7v8TBoNuD6IIzC3z7Q0y+7bMgXoT2DsHfolO2CHE=
 github.com/elastic/elastic-agent-client/v7 v7.15.0/go.mod h1:6h+f9QdIr3GO2ODC0Y8+aEXRwzbA5W4eV4dd/67z7nI=
-github.com/elastic/elastic-agent-libs v0.23.0 h1:xpMKkrw59QUYWUx/q2TMpJU+6vwB3Mw1VPhRPcGhBMo=
-github.com/elastic/elastic-agent-libs v0.23.0/go.mod h1:xSeIP3NtOIT4N2pPS4EyURmS1Q8mK0lWZ8Wd1Du6q3w=
+github.com/elastic/elastic-agent-libs v0.24.0 h1:8skd0le8Y/zAGVaj4m44ljiJMAp0xuJz07ngDTL/1kM=
+github.com/elastic/elastic-agent-libs v0.24.0/go.mod h1:xSeIP3NtOIT4N2pPS4EyURmS1Q8mK0lWZ8Wd1Du6q3w=
 github.com/elastic/elastic-agent-system-metrics v0.13.2 h1:R4ogKHESuWhWTtopnw/aHnBxxSZbxd7KHV4GefdwT2M=
 github.com/elastic/elastic-agent-system-metrics v0.13.2/go.mod h1:ezM1kzDUT+vWXFh5oK8QXB/AEB0UoLWqWA8rkRicFFo=
 github.com/elastic/elastic-transport-go/v8 v8.7.0 h1:OgTneVuXP2uip4BA658Xi6Hfw+PeIOod2rY3GVMGoVE=
diff --git a/heartbeat/monitors/active/dialchain/tls.go b/heartbeat/monitors/active/dialchain/tls.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/elastic/beats/v7/heartbeat/monitors/active/dialchain/tlsmeta"
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/transport"
 	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
 )
@@ -40,7 +41,7 @@ func TLSLayer(cfg *tlscommon.TLSConfig, to time.Duration) Layer {
 		// This gets us the timestamp for when the TLS layer will start the handshake.
 		next = startTimerAfterDial(&timer, next)
 
-		dialer := transport.TLSDialer(next, cfg, to)
+		dialer := transport.TLSDialer(next, cfg, to, logp.NewLogger(""))
 		return afterDial(dialer, func(conn net.Conn) (net.Conn, error) {
 			tlsConn, ok := conn.(*cryptoTLS.Conn)
 			if !ok {
diff --git a/libbeat/autodiscover/providers/kubernetes/config.go b/libbeat/autodiscover/providers/kubernetes/config.go
@@ -113,9 +113,13 @@ func (c *Config) Validate() error {
 	if c.Scope != "node" && c.Scope != "cluster" {
 		return fmt.Errorf("invalid `scope` configured. supported values are `node` and `cluster`")
 	}
-	if c.Unique && c.Scope != "cluster" {
-		logp.L().Warnf("can only set `unique` when scope is `cluster`")
-	}
 
 	return nil
 }
+
+// checkUnsupportedParams checks if unsupported/deprecated/discouraged paramaters are set and logs a warning
+func (c Config) checkUnsupportedParams(logger *logp.Logger) {
+	if c.Unique && c.Scope != "cluster" {
+		logger.Warn("can only set `unique` when scope is `cluster`")
+	}
+}
diff --git a/libbeat/autodiscover/providers/kubernetes/kubernetes.go b/libbeat/autodiscover/providers/kubernetes/kubernetes.go
@@ -110,6 +110,9 @@ func AutodiscoverBuilder(
 		return nil, errWrap(err)
 	}
 
+	// log warning about any unsupported params
+	config.checkUnsupportedParams(logger)
+
 	client, err := kubernetes.GetKubernetesClient(config.KubeConfig, config.KubeClientOptions)
 	if err != nil {
 		return nil, errWrap(err)
diff --git a/libbeat/autodiscover/providers/kubernetes/node.go b/libbeat/autodiscover/providers/kubernetes/node.go
@@ -63,6 +63,9 @@ func NewNodeEventer(
 		return nil, err
 	}
 
+	// log warning about any unsupported params
+	config.checkUnsupportedParams(logger)
+
 	// Ensure that node is set correctly whenever the scope is set to "node". Make sure that node is empty
 	// when cluster scope is enforced.
 	if config.Scope == "node" {
diff --git a/libbeat/autodiscover/providers/kubernetes/pod.go b/libbeat/autodiscover/providers/kubernetes/pod.go
@@ -75,6 +75,9 @@ func NewPodEventer(
 		return nil, err
 	}
 
+	// log warning about any unsupported params
+	config.checkUnsupportedParams(logger)
+
 	// Ensure that node is set correctly whenever the scope is set to "node". Make sure that node is empty
 	// when cluster scope is enforced.
 	if config.Scope == "node" {
diff --git a/libbeat/autodiscover/providers/kubernetes/service.go b/libbeat/autodiscover/providers/kubernetes/service.go
@@ -63,6 +63,9 @@ func NewServiceEventer(
 		return nil, err
 	}
 
+	// log warning about any unsupported params
+	config.checkUnsupportedParams(logger)
+
 	watcher, err := kubernetes.NewNamedWatcher("service", client, &kubernetes.Service{}, kubernetes.WatchOptions{
 		SyncTimeout:  config.SyncPeriod,
 		Namespace:    config.Namespace,
diff --git a/libbeat/common/transport/transptest/testing.go b/libbeat/common/transport/transptest/testing.go
@@ -171,7 +171,7 @@ func connectTCP(timeout time.Duration) TransportFactory {
 			Proxy:   proxy,
 			Timeout: timeout,
 		}
-		return transport.NewClient(cfg, "tcp", addr, 0)
+		return transport.NewClient(cfg, "tcp", addr, 0, logp.NewNopLogger())
 	}
 }
 
@@ -189,7 +189,7 @@ func connectTLS(timeout time.Duration, certName string) TransportFactory {
 			TLS:     tlsConfig,
 			Timeout: timeout,
 		}
-		return transport.NewClient(cfg, "tcp", addr, 0)
+		return transport.NewClient(cfg, "tcp", addr, 0, logp.NewNopLogger())
 	}
 }
 
diff --git a/libbeat/esleg/eslegclient/connection.go b/libbeat/esleg/eslegclient/connection.go
@@ -354,7 +354,7 @@ func (conn *Connection) Test(d testing.Driver) {
 				}
 
 				netDialer := transport.NetDialer(conn.Transport.Timeout)
-				tlsDialer := transport.TestTLSDialer(d, netDialer, tls, conn.Transport.Timeout)
+				tlsDialer := transport.TestTLSDialer(d, netDialer, tls, conn.Transport.Timeout, conn.log)
 				_, err = tlsDialer.Dial("tcp", address)
 				d.Fatal("dial up", err)
 			})
diff --git a/libbeat/outputs/logstash/logstash.go b/libbeat/outputs/logstash/logstash.go
@@ -67,7 +67,7 @@ func makeLogstash(
 	for i, host := range hosts {
 		var client outputs.NetworkClient
 
-		conn, err := transport.NewClient(transp, "tcp", host, defaultPort)
+		conn, err := transport.NewClient(transp, "tcp", host, defaultPort, beat.Logger)
 		if err != nil {
 			return outputs.Fail(err)
 		}
diff --git a/libbeat/outputs/redis/redis.go b/libbeat/outputs/redis/redis.go
@@ -144,7 +144,7 @@ func makeRedis(
 			}
 		}
 
-		conn, err := transport.NewClient(transp, "tcp", hostUrl.Host, defaultPort)
+		conn, err := transport.NewClient(transp, "tcp", hostUrl.Host, defaultPort, beat.Logger)
 		if err != nil {
 			return outputs.Fail(err)
 		}
diff --git a/x-pack/filebeat/input/awss3/decoding.go b/x-pack/filebeat/input/awss3/decoding.go
@@ -7,6 +7,8 @@ package awss3
 import (
 	"fmt"
 	"io"
+
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 // decoder is an interface for decoding data from an io.Reader.
@@ -34,12 +36,12 @@ type valueDecoder interface {
 // newDecoder creates a new decoder based on the codec type.
 // It returns a decoder type and an error if the codec type is not supported.
 // If the reader config codec option is not set, it returns a nil decoder and nil error.
-func newDecoder(config decoderConfig, r io.Reader) (decoder, error) {
+func newDecoder(config decoderConfig, r io.Reader, logger *logp.Logger) (decoder, error) {
 	switch {
 	case config.Codec == nil:
 		return nil, nil
 	case config.Codec.Parquet != nil:
-		return newParquetDecoder(config, r)
+		return newParquetDecoder(config, r, logger)
 	case config.Codec.CSV != nil:
 		return newCSVDecoder(config, r)
 	default:
diff --git a/x-pack/filebeat/input/awss3/decoding_parquet.go b/x-pack/filebeat/input/awss3/decoding_parquet.go
@@ -9,6 +9,7 @@ import (
 	"io"
 
 	"github.com/elastic/beats/v7/x-pack/libbeat/reader/parquet"
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 // parquetDecoder is a decoder for parquet data.
@@ -18,11 +19,11 @@ type parquetDecoder struct {
 
 // newParquetDecoder creates a new parquet decoder. It uses the libbeat parquet reader under the hood.
 // It returns an error if the parquet reader cannot be created.
-func newParquetDecoder(config decoderConfig, r io.Reader) (decoder, error) {
+func newParquetDecoder(config decoderConfig, r io.Reader, logger *logp.Logger) (decoder, error) {
 	reader, err := parquet.NewBufferedReader(r, &parquet.Config{
 		ProcessParallel: config.Codec.Parquet.ProcessParallel,
 		BatchSize:       config.Codec.Parquet.BatchSize,
-	})
+	}, logger)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create parquet decoder: %w", err)
 	}
diff --git a/x-pack/filebeat/input/awss3/s3_objects.go b/x-pack/filebeat/input/awss3/s3_objects.go
@@ -164,7 +164,7 @@ func (p *s3ObjectProcessor) ProcessS3Object(log *logp.Logger, eventCallback func
 	}
 
 	// try to create a dec from the using the codec config
-	dec, err := newDecoder(p.readerConfig.Decoding, streamReader)
+	dec, err := newDecoder(p.readerConfig.Decoding, streamReader, log)
 	if err != nil {
 		return err
 	}
diff --git a/x-pack/filebeat/input/cel/config.go b/x-pack/filebeat/input/cel/config.go
@@ -115,7 +115,7 @@ func (c config) Validate() error {
 		patterns = map[string]*regexp.Regexp{".": nil}
 	}
 	wantDump := c.FailureDump.enabled() && c.FailureDump.Filename != ""
-	_, _, _, err = newProgram(context.Background(), c.Program, root, nil, &http.Client{}, lib.HTTPOptions{}, patterns, c.XSDs, logp.L().Named("input.cel"), nil, wantDump, false)
+	_, _, _, err = newProgram(context.Background(), c.Program, root, nil, &http.Client{}, lib.HTTPOptions{}, patterns, c.XSDs, logp.NewNopLogger(), nil, wantDump, false)
 	if err != nil {
 		return fmt.Errorf("failed to check program: %w", err)
 	}
diff --git a/x-pack/filebeat/input/cel/input_test.go b/x-pack/filebeat/input/cel/input_test.go
@@ -2145,7 +2145,6 @@ func TestInput(t *testing.T) {
 		t.Fatalf("failed to remove failure_dumps directory: %v", err)
 	}
 
-	logp.TestingSetup()
 	for _, test := range inputTests {
 		t.Run(test.name, func(t *testing.T) {
 			if reason, skip := skipOnWindows[test.name]; runtime.GOOS == "windows" && skip {
diff --git a/x-pack/filebeat/input/streaming/config.go b/x-pack/filebeat/input/streaming/config.go
@@ -168,7 +168,7 @@ func (c config) Validate() error {
 		patterns = map[string]*regexp.Regexp{".": nil}
 	}
 	if c.Program != "" {
-		_, _, err = newProgram(context.Background(), c.Program, root, patterns, logp.L().Named("input.websocket"))
+		_, _, err = newProgram(context.Background(), c.Program, root, patterns, logp.NewNopLogger())
 		if err != nil {
 			return fmt.Errorf("failed to check program: %w", err)
 		}
diff --git a/x-pack/libbeat/reader/parquet/parquet.go b/x-pack/libbeat/reader/parquet/parquet.go
@@ -30,8 +30,8 @@ type BufferedReader struct {
 // It will return an error if the parquet data stream cannot be read.
 // Note: As io.ReadAll is used, the entire data stream would be read into memory, so very large data streams
 // may cause memory bottleneck issues.
-func NewBufferedReader(r io.Reader, cfg *Config) (*BufferedReader, error) {
-	log := logp.L().Named("reader.parquet")
+func NewBufferedReader(r io.Reader, cfg *Config, logger *logp.Logger) (*BufferedReader, error) {
+	log := logger.Named("reader.parquet")
 
 	if cfg.BatchSize == 0 {
 		cfg.BatchSize = 1
@@ -52,7 +52,7 @@ func NewBufferedReader(r io.Reader, cfg *Config) (*BufferedReader, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create parquet reader: %w", err)
 	}
-	log.Debugw("created parquet reader")
+	log.Debug("created parquet reader")
 
 	// constructs a reader for converting to Arrow objects from an existing parquet file reader object
 	reader, err := pqarrow.NewFileReader(pf, pqarrow.ArrowReadProperties{
@@ -62,14 +62,14 @@ func NewBufferedReader(r io.Reader, cfg *Config) (*BufferedReader, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create pqarrow parquet reader: %w", err)
 	}
-	log.Debugw("created pqarrow parquet reader")
+	log.Debug("created pqarrow parquet reader")
 
 	// constructs a record reader that is capable of reding entire sets of arrow records
 	rr, err := reader.GetRecordReader(context.Background(), nil, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create parquet record reader: %w", err)
 	}
-	log.Debugw("initialization process completed")
+	log.Debug("initialization process completed")
 
 	return &BufferedReader{
 		cfg:          cfg,
diff --git a/x-pack/libbeat/reader/parquet/parquet_benchmark_test.go b/x-pack/libbeat/reader/parquet/parquet_benchmark_test.go
@@ -9,6 +9,8 @@ import (
 	"path/filepath"
 	"sync"
 	"testing"
+
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 // parquetFile is a struct that contains the name of the parquet
@@ -241,7 +243,7 @@ func BenchmarkReadParquet(b *testing.B) {
 
 // readParquetFile reads entire parquet file
 func readParquetFile(b *testing.B, cfg *Config, file *os.File) {
-	sReader, err := NewBufferedReader(file, cfg)
+	sReader, err := NewBufferedReader(file, cfg, logp.NewNopLogger())
 	if err != nil {
 		b.Fatalf("failed to init stream reader: %v", err)
 	}
@@ -260,7 +262,7 @@ func readParquetFile(b *testing.B, cfg *Config, file *os.File) {
 
 // readParquetSingleRow reads only the first row of parquet files
 func readParquetSingleRow(b *testing.B, cfg *Config, file *os.File) {
-	sReader, err := NewBufferedReader(file, cfg)
+	sReader, err := NewBufferedReader(file, cfg, logp.NewNopLogger())
 	if err != nil {
 		b.Fatalf("failed to init stream reader: %v", err)
 	}
@@ -279,7 +281,7 @@ func readParquetSingleRow(b *testing.B, cfg *Config, file *os.File) {
 
 // constructBufferedReader constructs a stream reader for reading parquet files
 func constructBufferedReader(b *testing.B, cfg *Config, file *os.File) {
-	sReader, err := NewBufferedReader(file, cfg)
+	sReader, err := NewBufferedReader(file, cfg, logp.NewNopLogger())
 	if err != nil {
 		b.Fatalf("failed to init stream reader: %v", err)
 	}
diff --git a/x-pack/libbeat/reader/parquet/parquet_test.go b/x-pack/libbeat/reader/parquet/parquet_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/logp/logptest"
 )
 
 // all test files are read from/stored within the "testdata" directory
@@ -85,7 +86,7 @@ func TestParquetWithRandomData(t *testing.T) {
 
 // readAndValidateParquetFile reads the parquet file and validates the data
 func readAndValidateParquetFile(t *testing.T, cfg *Config, file *os.File, data map[string]bool) int {
-	sReader, err := NewBufferedReader(file, cfg)
+	sReader, err := NewBufferedReader(file, cfg, logptest.NewTestingLogger(t, ""))
 	if err != nil {
 		t.Fatalf("failed to init stream reader: %v", err)
 	}
@@ -235,7 +236,7 @@ func readJSONFromFile(t *testing.T, filepath string) (map[int]string, int) {
 
 // readAndCompareParquetFile reads the parquet file and compares the data with the input data
 func readAndCompareParquetFile(t *testing.T, cfg *Config, file *os.File, data map[int]string, rows int, maxRowsToCompare int) {
-	sReader, err := NewBufferedReader(file, cfg)
+	sReader, err := NewBufferedReader(file, cfg, logptest.NewTestingLogger(t, ""))
 	if err != nil {
 		t.Fatalf("failed to init stream reader: %v", err)
 	}

Original file line number	Diff line number	Diff line change
`@@ -110,6 +110,9 @@ func AutodiscoverBuilder(`
`110`	`110`	`return nil, errWrap(err)`
`111`	`111`	`}`
`112`	`112`
	`113`	`+ // log warning about any unsupported params`
	`114`	`+ config.checkUnsupportedParams(logger)`
	`115`	`+`
`113`	`116`	`client, err := kubernetes.GetKubernetesClient(config.KubeConfig, config.KubeClientOptions)`
`114`	`117`	`if err != nil {`
`115`	`118`	`return nil, errWrap(err)`
Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,9 @@ func NewNodeEventer(`
`63`	`63`	`return nil, err`
`64`	`64`	`}`
`65`	`65`
	`66`	`+ // log warning about any unsupported params`
	`67`	`+ config.checkUnsupportedParams(logger)`
	`68`	`+`
`66`	`69`	`// Ensure that node is set correctly whenever the scope is set to "node". Make sure that node is empty`
`67`	`70`	`// when cluster scope is enforced.`
`68`	`71`	`if config.Scope == "node" {`
Original file line number	Diff line number	Diff line change
`@@ -75,6 +75,9 @@ func NewPodEventer(`
`75`	`75`	`return nil, err`
`76`	`76`	`}`
`77`	`77`
	`78`	`+ // log warning about any unsupported params`
	`79`	`+ config.checkUnsupportedParams(logger)`
	`80`	`+`
`78`	`81`	`// Ensure that node is set correctly whenever the scope is set to "node". Make sure that node is empty`
`79`	`82`	`// when cluster scope is enforced.`
`80`	`83`	`if config.Scope == "node" {`
Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,9 @@ func NewServiceEventer(`
`63`	`63`	`return nil, err`
`64`	`64`	`}`
`65`	`65`
	`66`	`+ // log warning about any unsupported params`
	`67`	`+ config.checkUnsupportedParams(logger)`
	`68`	`+`
`66`	`69`	`watcher, err := kubernetes.NewNamedWatcher("service", client, &kubernetes.Service{}, kubernetes.WatchOptions{`
`67`	`70`	`SyncTimeout: config.SyncPeriod,`
`68`	`71`	`Namespace: config.Namespace,`
Original file line number	Diff line number	Diff line change
`@@ -171,7 +171,7 @@ func connectTCP(timeout time.Duration) TransportFactory {`
`171`	`171`	`Proxy: proxy,`
`172`	`172`	`Timeout: timeout,`
`173`	`173`	`}`
`174`		`- return transport.NewClient(cfg, "tcp", addr, 0)`
	`174`	`+ return transport.NewClient(cfg, "tcp", addr, 0, logp.NewNopLogger())`
`175`	`175`	`}`
`176`	`176`	`}`
`177`	`177`
`@@ -189,7 +189,7 @@ func connectTLS(timeout time.Duration, certName string) TransportFactory {`
`189`	`189`	`TLS: tlsConfig,`
`190`	`190`	`Timeout: timeout,`
`191`	`191`	`}`
`192`		`- return transport.NewClient(cfg, "tcp", addr, 0)`
	`192`	`+ return transport.NewClient(cfg, "tcp", addr, 0, logp.NewNopLogger())`
`193`	`193`	`}`
`194`	`194`	`}`
`195`	`195`