[9.0](backport #47585) libbeat/processors/translate_ldap_attribute: fix double locking and improve logging

changelog/fragments/1762903519-s6615-ldap_logging.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Fix double locking in translate_ldap_attribute processor and improve logging.
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

libbeat/processors/translate_ldap_attribute/ldap.go

@@ -25,13 +25,18 @@ import (
 	"sync"
 
 	"github.com/go-ldap/ldap/v3"
+
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 // ldapClient manages a single reusable LDAP connection
 type ldapClient struct {
-	conn *ldap.Conn
-	mu   sync.Mutex
 	*ldapConfig
+
+	mu   sync.Mutex
+	conn *ldap.Conn
+
+	log *logp.Logger
 }
 
 type ldapConfig struct {
@@ -46,21 +51,22 @@ type ldapConfig struct {
 }
 
 // newLDAPClient initializes a new ldapClient with a single connection
-func newLDAPClient(config *ldapConfig) (*ldapClient, error) {
-	client := &ldapClient{ldapConfig: config}
+func newLDAPClient(config *ldapConfig, log *logp.Logger) (*ldapClient, error) {
+	client := &ldapClient{ldapConfig: config, log: log}
 
 	// Establish initial connection
-	if err := client.connect(); err != nil {
+	conn, err := client.dial()
+	if err != nil {
 		return nil, err
 	}
+	client.conn = conn
 
 	return client, nil
 }
 
-// connect establishes a new connection to the LDAP server
-func (client *ldapClient) connect() error {
-	client.mu.Lock()
-	defer client.mu.Unlock()
+// dial establishes a new connection to the LDAP server
+func (client *ldapClient) dial() (*ldap.Conn, error) {
+	client.log.Debugw("ldap client connecting")
 
 	// Connect with or without TLS based on configuration
 	var opts []ldap.DialOpt
@@ -69,46 +75,49 @@ func (client *ldapClient) connect() error {
 	}
 	conn, err := ldap.DialURL(client.address, opts...)
 	if err != nil {
-		return fmt.Errorf("failed to dial LDAP server: %w", err)
+		return nil, fmt.Errorf("failed to dial LDAP server: %w", err)
 	}
 
 	if client.password != "" {
+		client.log.Debugw("ldap client bind")
 		err = conn.Bind(client.username, client.password)
 	} else {
+		client.log.Debugw("ldap client unauthenticated bind")
 		err = conn.UnauthenticatedBind(client.username)
 	}
 
 	if err != nil {
 		conn.Close()
-		return fmt.Errorf("failed to bind to LDAP server: %w", err)
+		return nil, fmt.Errorf("failed to bind to LDAP server: %w", err)
 	}
 
-	client.conn = conn
-	return nil
+	return conn, nil
 }
 
 // reconnect checks the connection's health and reconnects if necessary
-func (client *ldapClient) reconnect() error {
+func (client *ldapClient) connection() (*ldap.Conn, error) {
 	client.mu.Lock()
 	defer client.mu.Unlock()
 
 	// Check if the connection is still alive
-	if client.conn.IsClosing() {
-		return client.connect()
+	if client.conn == nil || client.conn.IsClosing() {
+		conn, err := client.dial()
+		if err != nil {
+			return nil, err
+		}
+		client.conn = conn
 	}
-	return nil
+	return client.conn, nil
 }
 
 // findObjectBy searches for an object and returns its mapped values.
 func (client *ldapClient) findObjectBy(searchBy string) ([]string, error) {
 	// Ensure the connection is alive or reconnect if necessary
-	if err := client.reconnect(); err != nil {
+	conn, err := client.connection()
+	if err != nil {
 		return nil, fmt.Errorf("failed to reconnect: %w", err)
 	}
 
-	client.mu.Lock()
-	defer client.mu.Unlock()
-
 	// Format the filter and perform the search
 	filter := fmt.Sprintf("(%s=%s)", client.searchAttr, searchBy)
 	searchRequest := ldap.NewSearchRequest(
@@ -118,7 +127,7 @@ func (client *ldapClient) findObjectBy(searchBy string) ([]string, error) {
 	)
 
 	// Execute search
-	result, err := client.conn.Search(searchRequest)
+	result, err := conn.Search(searchRequest)
 	if err != nil {
 		return nil, fmt.Errorf("search failed: %w", err)
 	}
@@ -137,5 +146,6 @@ func (client *ldapClient) close() {
 	defer client.mu.Unlock()
 	if client.conn != nil {
 		client.conn.Close()
+		client.conn = nil
 	}
 }

libbeat/processors/translate_ldap_attribute/translate_ldap_attribute.go

@@ -73,15 +73,14 @@ func newFromConfig(c config) (*processor, error) {
 		}
 		ldapConfig.tlsConfig = tlsConfig.ToConfig()
 	}
-	client, err := newLDAPClient(ldapConfig)
+	p := &processor{config: c}
+	p.log = logp.NewLogger(logName).With(logp.Stringer("processor", p))
+	client, err := newLDAPClient(ldapConfig, p.log)
 	if err != nil {
 		return nil, err
 	}
-	return &processor{
-		config: c,
-		client: client,
-		log:    logp.NewLogger(logName),
-	}, nil
+	p.client = client
+	return p, nil
 }
 
 func (p *processor) String() string {
@@ -90,7 +89,15 @@ func (p *processor) String() string {
 }
 
 func (p *processor) Run(event *beat.Event) (*beat.Event, error) {
+	p.log.Debugw("run ldap translation")
 	err := p.translateLDAPAttr(event)
+	if err != nil {
+		// Always log errors at debug level, even when we are
+		// ignoring failures.
+		p.log.Debugw("ldap translation error", "error", err)
+	} else {
+		p.log.Debugw("ldap translation complete")
+	}
 	if err == nil || p.IgnoreFailure || (p.IgnoreMissing && errors.Is(err, mapstr.ErrKeyNotFound)) {
 		return event, nil
 	}
@@ -108,7 +115,9 @@ func (p *processor) translateLDAPAttr(event *beat.Event) error {
 		return errInvalidType
 	}
 
+	p.log.Debugw("ldap search", "guid", guidString)
 	cn, err := p.client.findObjectBy(guidString)
+	p.log.Debugw("ldap result", "common_name", cn)
 	if err != nil {
 		return err
 	}