Add entry IDs to the memory queue

libbeat/publisher/pipeline/client.go

@@ -125,9 +125,9 @@ func (c *client) publish(e beat.Event) {
 
 	var published bool
 	if c.canDrop {
-		published = c.producer.TryPublish(pubEvent)
+		_, published = c.producer.TryPublish(pubEvent)
 	} else {
-		published = c.producer.Publish(pubEvent)
+		_, published = c.producer.Publish(pubEvent)
 	}
 
 	if published {

libbeat/publisher/pipeline/client_test.go

@@ -187,7 +187,6 @@ func TestClientWaitClose(t *testing.T) {
 		defer client.Close()
 
 		// Send an event which gets acknowledged immediately.
-		client.Publish(beat.Event{})
 		output := newMockClient(func(batch publisher.Batch) error {
 			batch.ACK()
 			return nil
@@ -196,6 +195,8 @@ func TestClientWaitClose(t *testing.T) {
 		pipeline.output.Set(outputs.Group{Clients: []outputs.Client{output}})
 		defer pipeline.output.Set(outputs.Group{})
 
+		client.Publish(beat.Event{})
+
 		closed := make(chan struct{})
 		go func() {
 			defer close(closed)

libbeat/publisher/pipeline/pipeline.go

@@ -266,9 +266,10 @@ func (p *Pipeline) ConnectWith(cfg beat.ClientConfig) (beat.Client, error) {
 	producerCfg := queue.ProducerConfig{}
 
 	if reportEvents || cfg.Events != nil {
-		producerCfg.OnDrop = func(event beat.Event) {
+		producerCfg.OnDrop = func(event interface{}) {
+			publisherEvent, _ := event.(publisher.Event)
 			if cfg.Events != nil {
-				cfg.Events.DroppedOnPublish(event)
+				cfg.Events.DroppedOnPublish(publisherEvent.Content)
 			}
 			if reportEvents {
 				p.waitCloseGroup.Add(-1)

libbeat/publisher/pipeline/pipeline_test.go

@@ -32,7 +32,7 @@ type testQueue struct {
 }
 
 type testProducer struct {
-	publish func(try bool, event interface{}) bool
+	publish func(try bool, event interface{}) (queue.EntryID, bool)
 	cancel  func() int
 }
 
@@ -68,18 +68,18 @@ func (q *testQueue) Get(sz int) (queue.Batch, error) {
 	return nil, nil
 }
 
-func (p *testProducer) Publish(event interface{}) bool {
+func (p *testProducer) Publish(event interface{}) (queue.EntryID, bool) {
 	if p.publish != nil {
 		return p.publish(false, event)
 	}
-	return false
+	return 0, false
 }
 
-func (p *testProducer) TryPublish(event interface{}) bool {
+func (p *testProducer) TryPublish(event interface{}) (queue.EntryID, bool) {
 	if p.publish != nil {
 		return p.publish(true, event)
 	}
-	return false
+	return 0, false
 }
 
 func (p *testProducer) Cancel() int {
@@ -114,7 +114,7 @@ func makeTestQueue() queue.Queue {
 			var producer *testProducer
 			p := blockingProducer(cfg)
 			producer = &testProducer{
-				publish: func(try bool, event interface{}) bool {
+				publish: func(try bool, event interface{}) (queue.EntryID, bool) {
 					if try {
 						return p.TryPublish(event)
 					}
@@ -146,10 +146,10 @@ func blockingProducer(_ queue.ProducerConfig) queue.Producer {
 	waiting := atomic.MakeInt(0)
 
 	return &testProducer{
-		publish: func(_ bool, _ interface{}) bool {
+		publish: func(_ bool, _ interface{}) (queue.EntryID, bool) {
 			waiting.Inc()
 			<-sig
-			return false
+			return 0, false
 		},
 
 		cancel: func() int {

libbeat/publisher/pipeline/ttl_batch.go

@@ -51,7 +51,7 @@ func newBatch(retryer retryer, original queue.Batch, ttl int) *ttlBatch {
 	count := original.Count()
 	events := make([]publisher.Event, 0, count)
 	for i := 0; i < count; i++ {
-		event, ok := original.Event(i).(publisher.Event)
+		event, ok := original.Entry(i).(publisher.Event)
 		if ok {
 			// In Beats this conversion will always succeed because only
 			// publisher.Event objects are inserted into the queue, but

libbeat/publisher/queue/diskqueue/benchmark_test.go

@@ -132,7 +132,7 @@ func publishEvents(p queue.Producer, num int, protobuf bool) {
 		} else {
 			e = makePublisherEvent()
 		}
-		ok := p.Publish(e)
+		_, ok := p.Publish(e)
 		if !ok {
 			panic("didn't publish")
 		}

libbeat/publisher/queue/diskqueue/consumer.go

@@ -86,10 +86,14 @@ func (batch *diskQueueBatch) Count() int {
 	return len(batch.frames)
 }
 
-func (batch *diskQueueBatch) Event(i int) interface{} {
+func (batch *diskQueueBatch) Entry(i int) interface{} {
 	return batch.frames[i].event
 }
 
+func (batch *diskQueueBatch) ID(i int) queue.EntryID {
+	return 0
+}
+
 func (batch *diskQueueBatch) Done() {
 	batch.queue.acks.addFrames(batch.frames)
 }

libbeat/publisher/queue/diskqueue/producer.go

@@ -49,12 +49,12 @@ type producerWriteRequest struct {
 // diskQueueProducer implementation of the queue.Producer interface
 //
 
-func (producer *diskQueueProducer) Publish(event interface{}) bool {
-	return producer.publish(event, true)
+func (producer *diskQueueProducer) Publish(event interface{}) (queue.EntryID, bool) {
+	return 0, producer.publish(event, true)
 }
 
-func (producer *diskQueueProducer) TryPublish(event interface{}) bool {
-	return producer.publish(event, false)
+func (producer *diskQueueProducer) TryPublish(event interface{}) (queue.EntryID, bool) {
+	return 0, producer.publish(event, false)
 }
 
 func (producer *diskQueueProducer) publish(

libbeat/publisher/queue/memqueue/batchbuf.go

@@ -17,11 +17,6 @@
 
 package memqueue
 
-type queueEntry struct {
-	event  interface{}
-	client clientState
-}
-
 type batchBuffer struct {
 	next    *batchBuffer
 	flushed bool
@@ -34,20 +29,20 @@ func newBatchBuffer(sz int) *batchBuffer {
 	return b
 }
 
-func (b *batchBuffer) add(event interface{}, st clientState) {
-	b.entries = append(b.entries, queueEntry{event, st})
+func (b *batchBuffer) add(entry queueEntry) {
+	b.entries = append(b.entries, entry)
 }
 
 func (b *batchBuffer) length() int {
 	return len(b.entries)
 }
 
-func (b *batchBuffer) cancel(st *produceState) int {
+func (b *batchBuffer) cancel(producer *ackProducer) int {
 	entries := b.entries[:0]
 
 	removedCount := 0
 	for _, entry := range b.entries {
-		if entry.client.state == st {
+		if entry.producer == producer {
 			removedCount++
 			continue
 		}

libbeat/publisher/queue/memqueue/broker.go

@@ -95,6 +95,14 @@ type Settings struct {
 	InputQueueSize int
 }
 
+type queueEntry struct {
+	event interface{}
+	id    queue.EntryID
+
+	producer   *ackProducer
+	producerID producerID // The order of this entry within its producer
+}
+
 type batch struct {
 	queue    *broker
 	entries  []queueEntry
@@ -153,7 +161,7 @@ func create(
 func NewQueue(
 	logger *logp.Logger,
 	settings Settings,
-) queue.Queue {
+) *broker {
 	var (
 		sz           = settings.Events
 		minEvents    = settings.FlushMinEvents
@@ -271,6 +279,7 @@ func (b *broker) Metrics() (queue.Metrics, error) {
 		EventCount:            opt.UintWith(uint64(resp.currentQueueSize)),
 		EventLimit:            opt.UintWith(uint64(b.bufSize)),
 		UnackedConsumedEvents: opt.UintWith(uint64(resp.occupiedRead)),
+		OldestEntryID:         resp.oldestEntryID,
 	}, nil
 }
 
@@ -380,10 +389,14 @@ func (b *batch) Count() int {
 	return len(b.entries)
 }
 
-func (b *batch) Event(i int) interface{} {
+func (b *batch) Entry(i int) interface{} {
 	return b.entries[i].event
 }
 
+func (b *batch) ID(i int) queue.EntryID {
+	return b.entries[i].id
+}
+
 func (b *batch) Done() {
 	b.doneChan <- batchDoneMsg{}
 }