Add basic metrics to disk queue

libbeat/publisher/queue/diskqueue/core_loop.go

@@ -84,10 +84,23 @@ func (dq *diskQueue) run() {
 			// If there were blocked producers waiting for more queue space,
 			// we might be able to unblock them now.
 			dq.maybeUnblockProducers()
+
+		case metricsReq := <-dq.metricsRequestChan:
+			dq.handleMetricsRequest(metricsReq)
 		}
 	}
 }
 
+// handleMetricsRequest responds to an event on the metricsRequestChan chan
+func (dq *diskQueue) handleMetricsRequest(request metricsRequest) {
+	resp := metricsRequestResponse{
+		sizeOnDisk:    dq.segments.sizeOnDisk(),
+		OccupiedRead:  dq.segments.unACKedReadBytes(),
+		oldestEntryID: dq.segments.oldestIDOnDisk(),
+	}
+	request.response <- resp
+}
+
 func (dq *diskQueue) handleProducerWriteRequest(request producerWriteRequest) {
 	// Pathological case checking: make sure the incoming frame isn't bigger
 	// than an entire segment all by itself (as long as it isn't, it is

libbeat/publisher/queue/diskqueue/queue.go

@@ -20,13 +20,15 @@ package diskqueue
 import (
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"sync"
 
 	"github.com/elastic/beats/v7/libbeat/feature"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/opt"
 )
 
 // diskQueue is the internal type representing a disk-based implementation
@@ -71,6 +73,9 @@ type diskQueue struct {
 	// The API channel used by diskQueueProducer to write events.
 	producerWriteRequestChan chan producerWriteRequest
 
+	// API channel used by the public Metrics() API to request queue metrics
+	metricsRequestChan chan metricsRequest
+
 	// pendingFrames is a list of all incoming data frames that have been
 	// accepted by the queue and are waiting to be sent to the writer loop.
 	// Segment ids in this list always appear in sorted order, even between
@@ -86,6 +91,18 @@ type diskQueue struct {
 	done chan struct{}
 }
 
+// channel request for metrics from an external client
+type metricsRequest struct {
+	response chan metricsRequestResponse
+}
+
+// metrics response from the disk queue
+type metricsRequestResponse struct {
+	sizeOnDisk    uint64
+	oldestEntryID segmentID
+	OccupiedRead  uint64
+}
+
 func init() {
 	queue.RegisterQueueType(
 		"disk",
@@ -220,6 +237,7 @@ func NewQueue(logger *logp.Logger, settings Settings) (*diskQueue, error) {
 		deleterLoop: newDeleterLoop(settings),
 
 		producerWriteRequestChan: make(chan producerWriteRequest),
+		metricsRequestChan:       make(chan metricsRequest),
 
 		done: make(chan struct{}),
 	}
@@ -281,6 +299,23 @@ func (dq *diskQueue) Producer(cfg queue.ProducerConfig) queue.Producer {
 	}
 }
 
+// Metrics returns current disk metrics
 func (dq *diskQueue) Metrics() (queue.Metrics, error) {
-	return queue.Metrics{}, queue.ErrMetricsNotImplemented
+	respChan := make(chan metricsRequestResponse, 1)
+	req := metricsRequest{response: respChan}
+
+	select {
+	case <-dq.done:
+		return queue.Metrics{}, io.EOF
+	case dq.metricsRequestChan <- req:
+	}
+	resp := <-respChan
+
+	maxSize := dq.settings.MaxBufferSize
+	return queue.Metrics{
+		ByteLimit:            opt.UintWith(maxSize),
+		ByteCount:            opt.UintWith(resp.sizeOnDisk),
+		UnackedConsumedBytes: opt.UintWith(resp.OccupiedRead),
+		OldestEntryID:        queue.EntryID(resp.oldestEntryID),
+	}, nil
 }

libbeat/publisher/queue/diskqueue/queue_test.go

@@ -28,6 +28,9 @@ import (
 	"github.com/elastic/beats/v7/libbeat/publisher/queue"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue/queuetest"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+
+	"github.com/stretchr/testify/require"
 )
 
 var seed int64
@@ -75,6 +78,65 @@ func TestProduceConsumer(t *testing.T) {
 	t.Run("direct", testWith(makeTestQueue()))
 }
 
+func TestMetrics(t *testing.T) {
+	dir, err := ioutil.TempDir("", "diskqueue_metrics")
+	defer func() {
+		_ = os.RemoveAll(dir)
+	}()
+	require.NoError(t, err)
+	settings := DefaultSettings()
+	settings.Path = dir
+	// lower max segment size so we can get multiple segments
+	settings.MaxSegmentSize = 100
+
+	testQueue, err := NewQueue(logp.L(), settings)
+	require.NoError(t, err)
+	defer testQueue.Close()
+
+	eventsToTest := 100
+
+	// Send events to queue
+	producer := testQueue.Producer(queue.ProducerConfig{})
+	sendEventsToQueue(eventsToTest, producer)
+
+	// fetch metrics before we read any events
+	time.Sleep(time.Millisecond * 500)
+	testMetrics, err := testQueue.Metrics()
+	require.NoError(t, err)
+
+	require.Equal(t, testMetrics.ByteLimit.ValueOr(0), uint64((1 << 30)))
+	require.NotZero(t, testMetrics.ByteCount.ValueOr(0))
+	t.Logf("got %d bytes written", testMetrics.ByteCount.ValueOr(0))
+
+	// now read & ACK the events
+	batch, err := testQueue.Get(eventsToTest)
+	require.NoError(t, err, "error in Get")
+	time.Sleep(time.Millisecond * 100)
+	require.NotZero(t, testMetrics.UnackedConsumedBytes.ValueOr(0))
+	// ACK
+	batch.Done()
+
+	// Occupied read should now be zero
+	testMetricsACKed, err := testQueue.Metrics()
+	require.NoError(t, err)
+	require.Zero(t, testMetricsACKed.UnackedConsumedBytes.ValueOr(0))
+
+	// insert again
+	sendEventsToQueue(eventsToTest, producer)
+	// This time, the oldest segment ID should be  > 0
+	time.Sleep(time.Millisecond * 100)
+	testMetricsSecond, err := testQueue.Metrics()
+	require.NoError(t, err)
+	require.NotZero(t, testMetricsSecond.OldestEntryID)
+
+}
+
+func sendEventsToQueue(count int, prod queue.Producer) {
+	for i := 0; i < count; i++ {
+		prod.Publish(queuetest.MakeEvent(mapstr.M{"count": i}))
+	}
+}
+
 func makeTestQueue() queuetest.QueueFactory {
 	return func(t *testing.T) queue.Queue {
 		dir, err := ioutil.TempDir("", "diskqueue_test")

libbeat/publisher/queue/diskqueue/segments.go

@@ -470,6 +470,35 @@ func (segments *diskQueueSegments) sizeOnDisk() uint64 {
 	return total
 }
 
+// Iterates through all segment types to find the oldest ID in the queue
+func (segments *diskQueueSegments) oldestIDOnDisk() segmentID {
+	// not all segment types are pre-sorted, hence us appending some, taking [0] from others
+	oldSegments := []*queueSegment{}
+
+	if len(segments.reading) > 0 {
+		oldSegments = append(oldSegments, segments.reading[0])
+	}
+	if len(segments.acking) > 0 {
+		oldSegments = append(oldSegments, segments.acking[0])
+	}
+
+	oldSegments = append(oldSegments, segments.writing...)
+	oldSegments = append(oldSegments, segments.acked...)
+
+	sort.Slice(oldSegments, func(i, j int) bool { return oldSegments[i].id < oldSegments[j].id })
+
+	return oldSegments[0].id
+}
+
+// unACKedReadBytes returns the total count of bytes that have been read, but not ack'ed by the consumer
+func (segments *diskQueueSegments) unACKedReadBytes() uint64 {
+	var acc uint64
+	for _, seg := range segments.acking {
+		acc += seg.byteCount
+	}
+	return acc
+}
+
 // segmentReader handles reading of segments.  getReader sets up the
 // reader and handles setting up the Reader to deal with the different
 // schema version.  With Schema version 2 there is the option for

libbeat/publisher/queue/queue.go

@@ -47,6 +47,8 @@ type Metrics struct {
 
 	//UnackedConsumedEvents is the count of events that an output consumer has read, but not yet ack'ed
 	UnackedConsumedEvents opt.Uint
+	//UnackedConsumedEvents is the count of bytes that an output consumer has read, but not yet ack'ed
+	UnackedConsumedBytes opt.Uint
 
 	//OldestActiveTimestamp is the timestamp of the oldest item in the queue.
 	OldestActiveTimestamp common.Time