Expand Buffer Metrics

Expand metrics emitted by low latency buffer

Author: fvallenilla

spectator/writer/lowlatency_buffer.go

@@ -2,11 +2,12 @@ package writer
 
 import (
 	"fmt"
-	"github.com/Netflix/spectator-go/v2/spectator/logger"
 	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/Netflix/spectator-go/v2/spectator/logger"
 )
 
 // chunkSize is set to 60KB, to ensure each message fits in the socket buffer (64KB), with some room
@@ -26,28 +27,32 @@ const separator = "\n"
 // is less than the impact of the front and back buffer design, but it is still important for
 // throughput reasons.
 type bufferShard struct {
-	data       [][]byte // Array of chunkSize chunks of spectatord protocol lines, stored as bytes
-	chunkIndex int      // Index of the chunk available for writes
-	overflows  int      // Count the buffer overflows, which correspond to data drops, for reporting metrics
-	mu         sync.Mutex
+	data          [][]byte // Array of chunkSize chunks of spectatord protocol lines, stored as bytes
+	chunkIndex    int      // Index of the chunk available for writes
+	overflows     int      // Count the buffer overflows, which correspond to data drops, for reporting metrics
+	overflowBytes int64    // Count the bytes that were dropped, for reporting metrics
+	mu            sync.Mutex
 }
 
 // getChunkIndexForLine returns the chunkIndex that should be used for storing the line, or -1, if there is
 // an overflow and the line cannot be stored in the bufferShard.
 func (b *bufferShard) getChunkIndexForLine(line []byte) int {
+	totalWriteLength := len(line)
+
 	// All chunks are full for the shard, drop the data
 	if b.chunkIndex >= len(b.data) {
 		b.overflows++
+		b.overflowBytes += int64(totalWriteLength)
 		return -1
 	}
 
 	// This should not happen, drop the data. The maximum length of a well-formed protocol line is 3.8KB.
 	if len(line) > chunkSize {
 		b.overflows++
+		b.overflowBytes += int64(totalWriteLength)
 		return -1
 	}
 
-	totalWriteLength := len(line)
 	if len(b.data[b.chunkIndex]) > 0 {
 		// Chunk has data, so account for the separator character
 		totalWriteLength++
@@ -61,6 +66,7 @@ func (b *bufferShard) getChunkIndexForLine(line []byte) int {
 	// Out of space in the shard, drop the data
 	if b.chunkIndex == len(b.data) {
 		b.overflows++
+		b.overflowBytes += int64(totalWriteLength)
 		return -1
 	}
 
@@ -190,6 +196,8 @@ func (llb *LowLatencyBuffer) flushLoop() {
 
 // swapAndFlush swaps the front and back buffers and flushes the deactivated buffers
 func (llb *LowLatencyBuffer) swapAndFlush() {
+	start := time.Now()
+
 	// Swap the buffer sets, so one can be drained, while the other accepts application writes
 	old := llb.useFrontBuffers.Load()
 	llb.useFrontBuffers.CompareAndSwap(old, !old)
@@ -219,6 +227,8 @@ func (llb *LowLatencyBuffer) swapAndFlush() {
 	if pctUsage > 0 {
 		llb.writer.WriteString(fmt.Sprintf("g,1:spectator-go.lowLatencyBuffer.pctUsage,bufferSet=%s:%f", bufferSet, pctUsage))
 	}
+
+	llb.writer.WriteString(fmt.Sprintf("t:spectator-go.lowLatencyBuffer.flushTime,bufferSet=%s:%f", bufferSet, time.Since(start).Seconds()))
 }
 
 // flushBufferShard flushes a single bufferShard to the socket, iterating through all chunks
@@ -244,7 +254,9 @@ func (llb *LowLatencyBuffer) flushBufferShard(buffer *bufferShard, bufferSet str
 	// record status metrics and reset shard statistics
 	if buffer.overflows > 0 {
 		llb.writer.WriteString(fmt.Sprintf("c:spectator-go.lowLatencyBuffer.overflows,bufferSet=%s:%d", bufferSet, buffer.overflows))
+		llb.writer.WriteString(fmt.Sprintf("d:spectator-go.lowLatencyBuffer.overflowBytes,bufferSet=%s:%d", bufferSet, buffer.overflowBytes))
 		buffer.overflows = 0
+		buffer.overflowBytes = 0
 	}
 	buffer.chunkIndex = 0
 	return bytesWritten