Add early encoding support to the Logstash output

libbeat/outputs/logstash/async.go

@@ -77,12 +77,10 @@ func newAsyncClient(
 		log.Warn(`The async Logstash client does not support the "ttl" option`)
 	}
 
-	enc := makeLogstashEventEncoder(log, beat, config.EscapeHTML, config.Index)
-
 	queueSize := config.Pipelining - 1
 	timeout := config.Timeout
 	compressLvl := config.CompressionLevel
-	clientFactory := makeClientFactory(queueSize, timeout, enc, compressLvl)
+	clientFactory := makeClientFactory(queueSize, timeout, compressLvl)
 
 	var err error
 	c.client, err = clientFactory(c.Client)
@@ -104,12 +102,11 @@ func newAsyncClient(
 func makeClientFactory(
 	queueSize int,
 	timeout time.Duration,
-	enc func(interface{}) ([]byte, error),
 	compressLvl int,
 ) func(net.Conn) (*v2.AsyncClient, error) {
 	return func(conn net.Conn) (*v2.AsyncClient, error) {
 		return v2.NewAsyncClientWithConn(conn, queueSize,
-			v2.JSONEncoder(enc),
+			v2.JSONEncoder(logstashEventUnwrapper),
 			v2.Timeout(timeout),
 			v2.CompressionLevel(compressLvl),
 		)
@@ -216,7 +213,7 @@ func (c *asyncClient) sendEvents(ref *msgRef, events []publisher.Event) error {
 	}
 	window := make([]interface{}, len(events))
 	for i := range events {
-		window[i] = &events[i].Content
+		window[i] = events[i].EncodedEvent
 	}
 	ref.count.Inc()
 	return client.Send(ref.callback, window)

libbeat/outputs/logstash/enc.go

@@ -18,24 +18,79 @@
 package logstash
 
 import (
-	"strings"
+	"fmt"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/outputs/codec/json"
+	"github.com/elastic/beats/v7/libbeat/publisher"
+	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 	"github.com/elastic/elastic-agent-libs/logp"
 )
 
-func makeLogstashEventEncoder(log *logp.Logger, info beat.Info, escapeHTML bool, index string) func(interface{}) ([]byte, error) {
+type eventEncoder struct {
+	log   *logp.Logger
+	enc   *json.Encoder
+	index string
+}
+
+type encodedEvent struct {
+	encoding []byte
+	err      error
+}
+
+func newEventEncoderFactory(
+	log *logp.Logger,
+	info beat.Info,
+	escapeHTML bool,
+	index string,
+) queue.EncoderFactory {
+	return func() queue.Encoder {
+		return newEventEncoder(log, info, escapeHTML, index)
+	}
+}
+
+func newEventEncoder(
+	log *logp.Logger,
+	info beat.Info,
+	escapeHTML bool,
+	index string,
+) queue.Encoder {
 	enc := json.New(info.Version, json.Config{
 		Pretty:     false,
 		EscapeHTML: escapeHTML,
 	})
-	index = strings.ToLower(index)
-	return func(event interface{}) (d []byte, err error) {
-		d, err = enc.Encode(index, event.(*beat.Event))
-		if err != nil {
-			log.Debugf("Failed to encode event: %v", event)
-		}
-		return
+	return &eventEncoder{
+		log:   log,
+		enc:   enc,
+		index: index,
+	}
+}
+
+func (e *eventEncoder) EncodeEntry(entry queue.Entry) (queue.Entry, int) {
+	pubEvent, ok := entry.(publisher.Event)
+	if !ok {
+		// Currently all queue entries are publisher.Events but let's be cautious.
+		return entry, 0
+	}
+	encoding, err := e.enc.Encode(e.index, &pubEvent.Content)
+	if err != nil {
+		e.log.Debugf("Failed to encode event: %v", pubEvent.Content)
+	}
+
+	copiedEncoding := make([]byte, len(encoding))
+	copy(copiedEncoding, encoding)
+	pubEvent.EncodedEvent = &encodedEvent{
+		encoding: copiedEncoding,
+		err:      err,
+	}
+	pubEvent.Content = beat.Event{}
+	return pubEvent, len(encoding)
+}
+
+func logstashEventUnwrapper(event interface{}) ([]byte, error) {
+	encoded, ok := event.(*encodedEvent)
+	if !ok {
+		return nil, fmt.Errorf("event is wrong type (expected *encodedEvent)")
 	}
+	return encoded.encoding, encoded.err
 }

libbeat/outputs/logstash/logstash.go

@@ -21,6 +21,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/outputs"
 	conf "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/transport"
 	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
 )
@@ -63,6 +64,12 @@ func makeLogstash(
 		Stats:   observer,
 	}
 
+	encoderFactory := newEventEncoderFactory(
+		logp.NewLogger("logstash"),
+		beat,
+		lsConfig.EscapeHTML,
+		lsConfig.Index)
+
 	clients := make([]outputs.NetworkClient, len(hosts))
 	for i, host := range hosts {
 		var client outputs.NetworkClient
@@ -85,5 +92,5 @@ func makeLogstash(
 		clients[i] = client
 	}
 
-	return outputs.SuccessNet(lsConfig.Queue, lsConfig.LoadBalance, lsConfig.BulkMaxSize, lsConfig.MaxRetries, nil, clients)
+	return outputs.SuccessNet(lsConfig.Queue, lsConfig.LoadBalance, lsConfig.BulkMaxSize, lsConfig.MaxRetries, encoderFactory, clients)
 }

libbeat/outputs/logstash/sync.go

@@ -61,9 +61,8 @@ func newSyncClient(
 	}
 
 	var err error
-	enc := makeLogstashEventEncoder(log, beat, config.EscapeHTML, config.Index)
 	c.client, err = v2.NewSyncClientWithConn(conn,
-		v2.JSONEncoder(enc),
+		v2.JSONEncoder(logstashEventUnwrapper),
 		v2.Timeout(config.Timeout),
 		v2.CompressionLevel(config.CompressionLevel),
 	)
@@ -196,7 +195,7 @@ func (c *syncClient) publishWindowed(events []publisher.Event) (int, error) {
 func (c *syncClient) sendEvents(events []publisher.Event) (int, error) {
 	window := make([]interface{}, len(events))
 	for i := range events {
-		window[i] = &events[i].Content
+		window[i] = events[i].EncodedEvent
 	}
 	return c.client.Send(window)
 }