[ES output] Log event fields in events log file

The Elasticsearch output, when faced with ingestion errors was logging
the raw publisher.Event that had already been encoded, hence no event
fields were present in the logs.

This commit fixes it by adding a String method to the encodedEvent
type and using the encodedEvent in the logs instead of the
publisher.Event.

CHANGELOG.next.asciidoc

@@ -148,6 +148,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix bug in Okta entity analytics rate limit logic. {issue}40106[40106] {pull}40267[40267]
 - Fix crashes in the journald input. {pull}40061[40061]
 - Fix order of configuration for EntraID entity analytics provider. {pull}40487[40487]
+- The Elasticsearch output now correctly logs the event fields to the event log file {issue}40509[40509] {pull}40512[40512]
 
 *Heartbeat*
 

filebeat/tests/integration/event_log_file_test.go

@@ -122,7 +122,7 @@ func TestEventsLoggerESOutput(t *testing.T) {
 	}
 
 	strData := string(data)
-	eventMsg := "not a number"
+	eventMsg := `\"int\":\"not a number\"`
 	if !strings.Contains(strData, eventMsg) {
 		t.Errorf("expecting to find '%s' on '%s'", eventMsg, eventsLogFile)
 		t.Errorf("Contents:\n%s", strData)

libbeat/outputs/elasticsearch/client.go

@@ -479,14 +479,15 @@ func (client *Client) applyItemStatus(
 			// Fatal error while sending an already-failed event to the dead letter
 			// index, drop.
 			client.log.Errorf("Can't deliver to dead letter index event (status=%v). Look at the event log to view the event and cause.", itemStatus)
-			client.log.Errorw(fmt.Sprintf("Can't deliver to dead letter index event %#v (status=%v): %s", event, itemStatus, itemMessage), logp.TypeKey, logp.EventType)
+			client.log.Errorw(fmt.Sprintf("Can't deliver to dead letter index event '%s' (status=%v): %s", encodedEvent, itemStatus, itemMessage), logp.TypeKey, logp.EventType)
 			stats.nonIndexable++
 			return false
 		}
 		if client.deadLetterIndex == "" {
 			// Fatal error and no dead letter index, drop.
 			client.log.Warnf("Cannot index event (status=%v): dropping event! Look at the event log to view the event and cause.", itemStatus)
-			client.log.Warnw(fmt.Sprintf("Cannot index event %#v (status=%v): %s, dropping event!", event, itemStatus, itemMessage), logp.TypeKey, logp.EventType)
+			// This line
+			client.log.Warnw(fmt.Sprintf("Cannot index event '%s' (status=%v): %s, dropping event!", encodedEvent, itemStatus, itemMessage), logp.TypeKey, logp.EventType)
 			stats.nonIndexable++
 			return false
 		}
@@ -495,7 +496,7 @@ func (client *Client) applyItemStatus(
 		// ingestion succeeds it is counted in the "deadLetter" counter
 		// rather than the "acked" counter.
 		client.log.Warnf("Cannot index event (status=%v), trying dead letter index. Look at the event log to view the event and cause.", itemStatus)
-		client.log.Warnw(fmt.Sprintf("Cannot index event %#v (status=%v): %s, trying dead letter index", event, itemStatus, itemMessage), logp.TypeKey, logp.EventType)
+		client.log.Warnw(fmt.Sprintf("Cannot index event '%s' (status=%v): %s, trying dead letter index", encodedEvent, itemStatus, itemMessage), logp.TypeKey, logp.EventType)
 		encodedEvent.setDeadLetter(client.deadLetterIndex, itemStatus, string(itemMessage))
 	}
 

libbeat/outputs/elasticsearch/event_encoder.go

@@ -151,3 +151,10 @@ func (e *encodedEvent) setDeadLetter(
 	}
 	e.encoding = []byte(deadLetterReencoding.String())
 }
+
+// String converts e.encoding to string and returns it.
+// The goal of this method is to provide an easy way to log
+// the event encoded.
+func (e *encodedEvent) String() string {
+	return string(e.encoding)
+}