Make the queue's public API accept interface{} instead of publisher.Event

CHANGELOG-developer.next.asciidoc

@@ -62,6 +62,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Wildcard fields no longer have a default ignore_above setting of 1024. {issue}30096[30096] {pull}30668[30668]
 - Remove `common.MapStr` and use `mapstr.M` from `github.com/elastic/elastic-agent-libs` instead. {pull}31420[31420]
 - Remove `queue.Consumer`. Queues can now be read via a `Get` call directly on the queue object. {pull}31502[31502]
+- The `queue.Batch` API now provides access to individual events instead of an array. {pull}31699[31699]
 
 ==== Bugfixes
 

libbeat/publisher/pipeline/pipeline_test.go

@@ -21,7 +21,6 @@ import (
 	"sync"
 
 	"github.com/elastic/beats/v7/libbeat/common/atomic"
-	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 )
 
@@ -33,7 +32,7 @@ type testQueue struct {
 }
 
 type testProducer struct {
-	publish func(try bool, event publisher.Event) bool
+	publish func(try bool, event interface{}) bool
 	cancel  func() int
 }
 
@@ -69,14 +68,14 @@ func (q *testQueue) Get(sz int) (queue.Batch, error) {
 	return nil, nil
 }
 
-func (p *testProducer) Publish(event publisher.Event) bool {
+func (p *testProducer) Publish(event interface{}) bool {
 	if p.publish != nil {
 		return p.publish(false, event)
 	}
 	return false
 }
 
-func (p *testProducer) TryPublish(event publisher.Event) bool {
+func (p *testProducer) TryPublish(event interface{}) bool {
 	if p.publish != nil {
 		return p.publish(true, event)
 	}
@@ -115,7 +114,7 @@ func makeTestQueue() queue.Queue {
 			var producer *testProducer
 			p := blockingProducer(cfg)
 			producer = &testProducer{
-				publish: func(try bool, event publisher.Event) bool {
+				publish: func(try bool, event interface{}) bool {
 					if try {
 						return p.TryPublish(event)
 					}
@@ -147,7 +146,7 @@ func blockingProducer(_ queue.ProducerConfig) queue.Producer {
 	waiting := atomic.MakeInt(0)
 
 	return &testProducer{
-		publish: func(_ bool, _ publisher.Event) bool {
+		publish: func(_ bool, _ interface{}) bool {
 			waiting.Inc()
 			<-sig
 			return false

libbeat/publisher/pipeline/ttl_batch.go

@@ -18,8 +18,6 @@
 package pipeline
 
 import (
-	"sync"
-
 	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 )
@@ -29,7 +27,9 @@ type retryer interface {
 }
 
 type ttlBatch struct {
-	original queue.Batch
+	// The callback to inform the queue (and possibly the producer)
+	// that this batch has been acknowledged.
+	ack func()
 
 	// The internal hook back to the eventConsumer, used to implement the
 	// publisher.Batch retry interface.
@@ -43,44 +43,42 @@ type ttlBatch struct {
 	events []publisher.Event
 }
 
-var batchPool = sync.Pool{
-	New: func() interface{} {
-		return &ttlBatch{}
-	},
-}
-
 func newBatch(retryer retryer, original queue.Batch, ttl int) *ttlBatch {
 	if original == nil {
 		panic("empty batch")
 	}
 
-	b := batchPool.Get().(*ttlBatch)
-	*b = ttlBatch{
-		original: original,
-		retryer:  retryer,
-		ttl:      ttl,
-		events:   original.Events(),
+	count := original.Count()
+	events := make([]publisher.Event, 0, count)
+	for i := 0; i < count; i++ {
+		event, ok := original.Event(i).(publisher.Event)
+		if ok {
+			// In Beats this conversion will always succeed because only
+			// publisher.Event objects are inserted into the queue, but
+			// there's no harm in making sure.
+			events = append(events, event)
+		}
 	}
-	return b
-}
 
-func releaseBatch(b *ttlBatch) {
-	*b = ttlBatch{} // clear batch
-	batchPool.Put(b)
+	b := &ttlBatch{
+		ack:     original.ACK,
+		retryer: retryer,
+		ttl:     ttl,
+		events:  events,
+	}
+	return b
 }
 
 func (b *ttlBatch) Events() []publisher.Event {
 	return b.events
 }
 
 func (b *ttlBatch) ACK() {
-	b.original.ACK()
-	releaseBatch(b)
+	b.ack()
 }
 
 func (b *ttlBatch) Drop() {
-	b.original.ACK()
-	releaseBatch(b)
+	b.ack()
 }
 
 func (b *ttlBatch) Retry() {

libbeat/publisher/queue/diskqueue/benchmark_test.go

@@ -101,9 +101,8 @@ func produceAndConsume(p queue.Producer, q *diskQueue, num_events int, batch_siz
 		if err != nil {
 			return err
 		}
-		events := batch.Events()
 		batch.ACK()
-		received = received + len(events)
+		received = received + batch.Count()
 		if received == num_events {
 			break
 		}

libbeat/publisher/queue/diskqueue/consumer.go

@@ -20,7 +20,6 @@ package diskqueue
 import (
 	"fmt"
 
-	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 )
 
@@ -83,12 +82,12 @@ eventLoop:
 // diskQueueBatch implementation of the queue.Batch interface
 //
 
-func (batch *diskQueueBatch) Events() []publisher.Event {
-	events := make([]publisher.Event, len(batch.frames))
-	for i, frame := range batch.frames {
-		events[i] = frame.event
-	}
-	return events
+func (batch *diskQueueBatch) Count() int {
+	return len(batch.frames)
+}
+
+func (batch *diskQueueBatch) Event(i int) interface{} {
+	return batch.frames[i].event
 }
 
 func (batch *diskQueueBatch) ACK() {

libbeat/publisher/queue/diskqueue/producer.go

@@ -18,7 +18,6 @@
 package diskqueue
 
 import (
-	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 )
 
@@ -50,21 +49,21 @@ type producerWriteRequest struct {
 // diskQueueProducer implementation of the queue.Producer interface
 //
 
-func (producer *diskQueueProducer) Publish(event publisher.Event) bool {
+func (producer *diskQueueProducer) Publish(event interface{}) bool {
 	return producer.publish(event, true)
 }
 
-func (producer *diskQueueProducer) TryPublish(event publisher.Event) bool {
+func (producer *diskQueueProducer) TryPublish(event interface{}) bool {
 	return producer.publish(event, false)
 }
 
 func (producer *diskQueueProducer) publish(
-	event publisher.Event, shouldBlock bool,
+	event interface{}, shouldBlock bool,
 ) bool {
 	if producer.cancelled {
 		return false
 	}
-	serialized, err := producer.encoder.encode(&event)
+	serialized, err := producer.encoder.encode(event)
 	if err != nil {
 		producer.queue.logger.Errorf(
 			"Couldn't serialize incoming event: %v", err)

libbeat/publisher/queue/diskqueue/serialize.go

@@ -22,6 +22,7 @@ package diskqueue
 
 import (
 	"bytes"
+	"fmt"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -82,7 +83,17 @@ func (e *eventEncoder) reset() {
 	e.folder = folder
 }
 
-func (e *eventEncoder) encode(event *publisher.Event) ([]byte, error) {
+func (e *eventEncoder) encode(evt interface{}) ([]byte, error) {
+	event, ok := evt.(publisher.Event)
+	if !ok {
+		// In order to support events of varying type, the disk queue needs
+		// to know how to encode them. When we decide to do this, we'll need
+		// to add an encoder to the settings passed in when creating a disk
+		// queue. See https://github.com/elastic/elastic-agent-shipper/issues/41.
+		// For now, just return an error.
+		return nil, fmt.Errorf("disk queue only supports publisher.Event")
+	}
+
 	e.buf.Reset()
 
 	err := e.folder.Fold(entry{

libbeat/publisher/queue/diskqueue/serialize_test.go

@@ -46,7 +46,7 @@ func TestSerialize(t *testing.T) {
 				},
 			},
 		}
-		serialized, err := encoder.encode(&event)
+		serialized, err := encoder.encode(event)
 		if err != nil {
 			t.Fatalf("[%v] Couldn't encode event: %v", test.name, err)
 		}

libbeat/publisher/queue/memqueue/ackloop.go

@@ -36,8 +36,6 @@ type ackLoop struct {
 
 func (l *ackLoop) run() {
 	var (
-		// log = l.broker.logger
-
 		// Buffer up event counter in ackCount. If ackCount > 0, acks will be set to
 		// the broker.acks channel for sending the ACKs while potentially receiving
 		// new batches from the broker event loop.
@@ -46,10 +44,11 @@ func (l *ackLoop) run() {
 		// loop, as the ack loop will not block on any channel
 		ackCount int
 		ackChan  chan int
-		sig      chan batchAckMsg
 	)
 
 	for {
+		nextBatchChan := l.ackChans.nextBatchChannel()
+
 		select {
 		case <-l.broker.done:
 			return
@@ -60,25 +59,12 @@ func (l *ackLoop) run() {
 		case chanList := <-l.broker.scheduledACKs:
 			l.ackChans.concat(&chanList)
 
-		case <-sig:
+		case <-nextBatchChan:
 			ackCount += l.handleBatchSig()
 			if ackCount > 0 {
 				ackChan = l.broker.ackChan
 			}
 		}
-
-		// log.Debug("ackloop INFO")
-		// log.Debug("ackloop:   total events scheduled = ", l.totalSched)
-		// log.Debug("ackloop:   total events ack = ", l.totalACK)
-		// log.Debug("ackloop:   total batches scheduled = ", l.batchesSched)
-		// log.Debug("ackloop:   total batches ack = ", l.batchesACKed)
-
-		sig = l.ackChans.channel()
-		// if l.sig == nil {
-		// 	log.Debug("ackloop: no ack scheduled")
-		// } else {
-		// 	log.Debug("ackloop: schedule ack: ", l.lst.head.seq)
-		// }
 	}
 }
 

libbeat/publisher/queue/memqueue/batchbuf.go

@@ -17,8 +17,6 @@
 
 package memqueue
 
-import "github.com/elastic/beats/v7/libbeat/publisher"
-
 type queueEntry struct {
 	event  interface{}
 	client clientState
@@ -36,7 +34,7 @@ func newBatchBuffer(sz int) *batchBuffer {
 	return b
 }
 
-func (b *batchBuffer) add(event *publisher.Event, st clientState) {
+func (b *batchBuffer) add(event interface{}, st clientState) {
 	b.entries = append(b.entries, queueEntry{event, st})
 }
 

libbeat/publisher/queue/memqueue/broker.go

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/feature"
-	"github.com/elastic/beats/v7/libbeat/publisher"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 	c "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -91,7 +90,7 @@ type Settings struct {
 
 type batch struct {
 	queue   *broker
-	events  []publisher.Event
+	entries []queueEntry
 	ackChan chan batchAckMsg
 }
 
@@ -242,15 +241,9 @@ func (b *broker) Get(count int) (queue.Batch, error) {
 
 	// if request has been sent, we have to wait for a response
 	resp := <-responseChan
-	events := make([]publisher.Event, 0, len(resp.entries))
-	for _, entry := range resp.entries {
-		if event, ok := entry.event.(*publisher.Event); ok {
-			events = append(events, *event)
-		}
-	}
 	return &batch{
 		queue:   b,
-		events:  events,
+		entries: resp.entries,
 		ackChan: resp.ackChan,
 	}, nil
 }
@@ -268,8 +261,7 @@ var ackChanPool = sync.Pool{
 }
 
 func newBatchACKState(start, count int, entries []queueEntry) *batchACKState {
-	//nolint: errcheck // Return value doesn't need to be checked before conversion.
-	ch := ackChanPool.Get().(*batchACKState)
+	ch, _ := ackChanPool.Get().(*batchACKState)
 	ch.next = nil
 	ch.start = start
 	ch.count = count
@@ -321,7 +313,7 @@ func (l *chanList) front() *batchACKState {
 	return l.head
 }
 
-func (l *chanList) channel() chan batchAckMsg {
+func (l *chanList) nextBatchChannel() chan batchAckMsg {
 	if l.head == nil {
 		return nil
 	}
@@ -362,8 +354,12 @@ func AdjustInputQueueSize(requested, mainQueueSize int) (actual int) {
 	return actual
 }
 
-func (b *batch) Events() []publisher.Event {
-	return b.events
+func (b *batch) Count() int {
+	return len(b.entries)
+}
+
+func (b *batch) Event(i int) interface{} {
+	return b.entries[i].event
 }
 
 func (b *batch) ACK() {