Preserve original meta fields during early encoding of events

libbeat/outputs/elasticsearch/event_encoder.go

@@ -55,6 +55,10 @@ type encodedEvent struct {
 	// there's an ingestion error.
 	timestamp time.Time
 
+	// The meta fields from the original event (which aren't included in the
+	// encoding but may still need to be logged if there is an error).
+	meta mapstr.M
+
 	id       string
 	opType   events.OpType
 	pipeline string
@@ -130,6 +134,7 @@ func (pe *eventEncoder) encodeRawEvent(e *beat.Event) *encodedEvent {
 	copy(bytes, bufBytes)
 	return &encodedEvent{
 		id:        id,
+		meta:      e.Meta,
 		timestamp: e.Timestamp,
 		opType:    opType,
 		pipeline:  pipeline,
@@ -152,9 +157,14 @@ func (e *encodedEvent) setDeadLetter(
 	e.encoding = []byte(deadLetterReencoding.String())
 }
 
-// String converts e.encoding to string and returns it.
+// String converts e.encoding (and meta fields if present)
+// 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)
+	metaString := "none"
+	if e.meta != nil {
+		metaString = e.meta.String()
+	}
+	return string(e.encoding) + ", Meta: " + metaString
 }

libbeat/outputs/elasticsearch/event_encoder_test.go

@@ -42,6 +42,12 @@
 
 	encoder := newEventEncoder(true, indexSelector, nil)
 
+	metaFields := mapstr.M{
+		events.FieldMetaOpType:   "create",
+		events.FieldMetaPipeline: "TEST_PIPELINE",
+		events.FieldMetaID:       "test_id",
+	}
+
 	timestamp := time.Date(1980, time.January, 1, 0, 0, 0, 0, time.UTC)
 	pubEvent := publisher.Event{
 		Content: beat.Event{
@@ -53,11 +59,7 @@
 					"nested_field": "nested_value",
 				},
 			},
-			Meta: mapstr.M{
-				events.FieldMetaOpType:   "create",
-				events.FieldMetaPipeline: "TEST_PIPELINE",
-				events.FieldMetaID:       "test_id",
-			},
+			Meta: metaFields,
 		},
 	}
 
@@ -81,6 +83,7 @@
 	assert.Equal(t, timestamp, encBeatEvent.timestamp, "encodedEvent.timestamp should match the original event")
 	assert.Equal(t, events.OpTypeCreate, encBeatEvent.opType, "encoded opType should match the original metadata")
 	assert.False(t, encBeatEvent.deadLetter, "encoded event shouldn't have deadLetter flag set")
+	assert.Equal(t, encBeatEvent.meta, metaFields, "encoded event struct should include original event's meta fields")
 
 	// Check encoded fields
 	var eventContent struct {
@@ -97,6 +100,9 @@
 	assert.Equal(t, "test_value", eventContent.TestField, "Encoded field should match original")
 	assert.Equal(t, 5, eventContent.NumberField, "Encoded field should match original")
 	assert.Equal(t, "nested_value", eventContent.Nested.NestedField, "Encoded field should match original")
+
+	// Check string representation includes meta fields
+	assert.Contains(t, encBeatEvent.String(), `"pipeline":"TEST_PIPELINE"`, "String representation of encoded event should include the original event's meta fields")
 }
 
 // encodeBatch encodes a publisher.Batch so it can be provided to
@@ -124,14 +130,14 @@
 		// Skip encoding if there's already encoded data present
 		if events[i].EncodedEvent == nil {
 			encoded, _ := encoder.EncodeEntry(events[i])
 			event := encoded.(publisher.Event)
 			events[i] = event
 		}
 	}
 	return events
 }
 
 func encodeEvent(client *Client, event publisher.Event) publisher.Event {
 	encoder := newEventEncoder(
 		client.conn.EscapeHTML,
 		client.indexSelector,