writer: log once per flush window when LowLatencyBuffer overflows

spectator/writer/lowlatency_buffer.go

@@ -31,35 +31,47 @@ 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
-	overflowBytes int64    // Count the bytes that were dropped, 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
+	droppedSampleId string   // Meter id of one dropped line this window, for the overflow log line
+	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. It reports
-// whether space was reserved (true) or the data must be dropped (false); on
-// success b.chunkIndex identifies the chunk to append into (advancing it if the
-// line does not fit in the current chunk).
-func (b *bufferShard) reserveChunkForLine(lineLen int) bool {
-	totalWriteLength := lineLen
+// recordOverflow accounts for a dropped line's byte count and, on the first drop of the
+// window (overflows was 0), samples its meter id via meterIdKey so the once-per-window log
+// line can name a metric instead of just a count.
+func (b *bufferShard) recordOverflow(prefix string, totalWriteLength int) {
+	if b.overflows == 0 {
+		if id, ok := meterIdKey(prefix); ok {
+			b.droppedSampleId = id
+		} else {
+			b.droppedSampleId = prefix
+		}
+	}
+	b.overflows++
+	b.overflowBytes += int64(totalWriteLength)
+}
 
+// reserveChunkForLine reports whether space was reserved for a line of lineLen bytes (true)
+// or the buffer is full and the line must be dropped (false). On success b.chunkIndex
+// identifies the chunk to append into (advancing it if the line does not fit in the current
+// chunk). It does not record overflow stats itself: the caller has the line's prefix (needed
+// to sample a dropped meter id for the overflow log line) and calls recordOverflow when this
+// returns false.
+func (b *bufferShard) reserveChunkForLine(lineLen int) bool {
 	// All chunks are full for the shard, drop the data
 	if b.chunkIndex >= len(b.data) {
-		b.overflows++
-		b.overflowBytes += int64(totalWriteLength)
 		return false
 	}
 
 	// This should not happen, drop the data. The maximum length of a well-formed protocol line is 3.8KB.
 	if lineLen > chunkSize {
-		b.overflows++
-		b.overflowBytes += int64(totalWriteLength)
 		return false
 	}
 
+	totalWriteLength := lineLen
 	if len(b.data[b.chunkIndex]) > 0 {
 		// Chunk has data, so account for the separator character
 		totalWriteLength++
@@ -72,8 +84,6 @@ func (b *bufferShard) reserveChunkForLine(lineLen int) bool {
 
 	// Out of space in the shard, drop the data
 	if b.chunkIndex == len(b.data) {
-		b.overflows++
-		b.overflowBytes += int64(totalWriteLength)
 		return false
 	}
 
@@ -230,7 +240,7 @@ func (b *bufferShard) appendLine(state *atomic.Uint64, expected uint64, prefix,
 		return false
 	}
 	if !b.reserveChunkForLine(len(prefix) + len(value)) {
-		// overflows (drops) are counted in reserveChunkForLine, for metric reporting
+		b.recordOverflow(prefix, len(prefix)+len(value))
 		return true
 	}
 	if len(b.data[b.chunkIndex]) > 0 {
@@ -250,6 +260,7 @@ func (b *bufferShard) appendLineBytes(state *atomic.Uint64, expected uint64, pre
 		return false
 	}
 	if !b.reserveChunkForLine(len(prefix) + len(value)) {
+		b.recordOverflow(prefix, len(prefix)+len(value))
 		return true
 	}
 	if len(b.data[b.chunkIndex]) > 0 {
@@ -302,6 +313,23 @@ func orderSensitiveMeterIdKey(line string) (string, bool) {
 	return line[first+1 : last], true
 }
 
+// meterIdKey returns the id substring of a spectatord protocol line for any
+// meter type, order-sensitive or not. The input format is the same
+// "<type>[,<extra>]:<id>:<value>" line; the id is the substring between the
+// first and last ':'. Returns ok=false for malformed lines (fewer than two ':'
+// characters).
+func meterIdKey(line string) (string, bool) {
+	first := strings.IndexByte(line, ':')
+	if first < 0 {
+		return "", false
+	}
+	last := strings.LastIndexByte(line, ':')
+	if last <= first {
+		return "", false
+	}
+	return line[first+1 : last], true
+}
+
 // isOrderSensitiveMeterType identifies protocol types that carry absolute samples rather than
 // commutative observations. Reordering these samples can change what spectatord reports.
 func isOrderSensitiveMeterType(symbol string) bool {
@@ -367,9 +395,22 @@ func (llb *LowLatencyBuffer) swapAndFlush() {
 	}
 
 	// Flush each buffer shard, from the deactivated set
-	var bytesWritten int
+	var bytesWritten, totalOverflows int
+	var totalOverflowBytes int64
+	var droppeSampleId string
 	for _, buffer := range buffersToFlush {
-		bytesWritten += llb.flushBufferShard(buffer, bufferSet)
+		flushed := llb.flushBufferShard(buffer, bufferSet)
+		bytesWritten += flushed.bytesWritten
+		totalOverflows += flushed.overflows
+		totalOverflowBytes += flushed.overflowBytes
+		if droppeSampleId == "" {
+			droppeSampleId = flushed.droppedSampleId
+		}
+	}
+
+	// Log a single overflow for each window.
+	if totalOverflows > 0 {
+		llb.logger.Infof("LowLatencyBuffer overflow in bufferSet=%s: dropped %d lines (%d bytes) this window; sample dropped meter id: %s", bufferSet, totalOverflows, totalOverflowBytes, droppeSampleId)
 	}
 
 	pctUsage := float64(bytesWritten) / float64(llb.bufferSetSize)
@@ -383,14 +424,21 @@ func (llb *LowLatencyBuffer) swapAndFlush() {
 	llb.writer.WriteString(fmt.Sprintf("t:spectator-go.lowLatencyBuffer.flushTime,bufferSet=%s:%f", bufferSet, time.Since(start).Seconds()))
 }
 
+type flushBufferResult struct {
+	bytesWritten,
+	overflows int
+	overflowBytes   int64
+	droppedSampleId string
+}
+
 // flushBufferShard flushes a single bufferShard to the socket, iterating through all chunks
-func (llb *LowLatencyBuffer) flushBufferShard(buffer *bufferShard, bufferSet string) int {
+func (llb *LowLatencyBuffer) flushBufferShard(buffer *bufferShard, bufferSet string) flushBufferResult {
 	buffer.mu.Lock()
 	defer buffer.mu.Unlock()
 
 	// If there is no data to flush from the shard, then skip socket writes
 	if buffer.chunkIndex == 0 && len(buffer.data[0]) == 0 {
-		return 0
+		return flushBufferResult{}
 	}
 
 	// Write each chunk to the socket, and reset the chunk
@@ -404,14 +452,21 @@ 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))
+	overflows, overflowBytes, sampleId := buffer.overflows, buffer.overflowBytes, buffer.droppedSampleId
+	if overflows > 0 {
+		llb.writer.WriteString(fmt.Sprintf("c:spectator-go.lowLatencyBuffer.overflows,bufferSet=%s:%d", bufferSet, overflows))
+		llb.writer.WriteString(fmt.Sprintf("d:spectator-go.lowLatencyBuffer.overflowBytes,bufferSet=%s:%d", bufferSet, overflowBytes))
 		buffer.overflows = 0
 		buffer.overflowBytes = 0
+		buffer.droppedSampleId = ""
 	}
 	buffer.chunkIndex = 0
-	return bytesWritten
+	return flushBufferResult{
+		bytesWritten:    bytesWritten,
+		overflows:       overflows,
+		overflowBytes:   overflowBytes,
+		droppedSampleId: sampleId,
+	}
 }
 
 func (llb *LowLatencyBuffer) Close() {

spectator/writer/lowlatency_buffer_test.go

@@ -2,12 +2,14 @@ package writer
 
 import (
 	"fmt"
-	"github.com/Netflix/spectator-go/v2/spectator/logger"
 	"runtime"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
+
+	"github.com/Netflix/spectator-go/v2/spectator/logger"
 )
 
 func splitAndFilterMetricLines(memWriter *MemoryWriter) []string {
@@ -297,6 +299,42 @@ func TestLowLatencyBuffer_OrderSensitiveMeterIdKey(t *testing.T) {
 	}
 }
 
+func TestLowLatencyBuffer_MeterIdKey(t *testing.T) {
+	tests := []struct {
+		line string
+		key  string
+		ok   bool
+	}{
+		{line: "g:test.metric:1", key: "test.metric", ok: true},
+		{line: "g,60:test.metric:1", key: "test.metric", ok: true},
+		{line: "A:test.metric:1", key: "test.metric", ok: true},
+		// Unlike orderSensitiveMeterIdKey, commutative types yield an id
+		{line: "c:test.metric:1", key: "test.metric", ok: true},
+		{line: "d:test.metric:1", key: "test.metric", ok: true},
+		{line: "t:test.metric:1", key: "test.metric", ok: true},
+		{line: "gauge:test.metric:1", key: "test.metric", ok: true},
+		{line: "c:test.metric,app=foo:1", key: "test.metric,app=foo", ok: true},
+		// WriteLine passes the prefix alone, which ends with the value separator
+		{line: "c:test.metric:", key: "test.metric", ok: true},
+		{line: ":test.metric:1", key: "test.metric", ok: true},
+		{line: "message=1", ok: false},
+		{line: "g:test.metric", ok: false},
+		{line: "", ok: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.line, func(t *testing.T) {
+			key, ok := meterIdKey(test.line)
+			if ok != test.ok {
+				t.Fatalf("Expected ok=%t, got %t", test.ok, ok)
+			}
+			if key != test.key {
+				t.Fatalf("Expected key %q, got %q", test.key, key)
+			}
+		})
+	}
+}
+
 func TestLowLatencyBuffer_RoundRobinsSameCounterId(t *testing.T) {
 	shards := runtime.NumCPU()
 	if shards < 2 {
@@ -380,3 +418,137 @@ func TestLowLatencyBuffer_ChunkBoundaries_OverMaxSizeIsDropped(t *testing.T) {
 		t.Errorf("Expected %d lines, got %d", 0, len(lines))
 	}
 }
+
+// captureLogger records Infof messages, so the once-per-flush-window overflow log line
+// can be inspected.
+type captureLogger struct {
+	mu       sync.Mutex
+	messages []string
+}
+
+var _ logger.Logger = (*captureLogger)(nil)
+
+func (l *captureLogger) Debugf(_ string, _ ...interface{}) {}
+
+func (l *captureLogger) Infof(format string, v ...interface{}) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.messages = append(l.messages, fmt.Sprintf(format, v...))
+}
+
+func (l *captureLogger) Errorf(_ string, _ ...interface{}) {}
+
+// overflowMessages returns the overflow log lines, discarding the initialization message.
+func (l *captureLogger) overflowMessages() []string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	var messages []string
+	for _, message := range l.messages {
+		if strings.Contains(message, "LowLatencyBuffer overflow") {
+			messages = append(messages, message)
+		}
+	}
+	return messages
+}
+
+// newSingleChunkBuffer creates a buffer with one chunk per shard and a long flush interval
+// timer, to allow for manual flush trigger. One chunk per shard means a single chunkSize
+// line fills a shard, and every subsequent line written to it is dropped.
+func newSingleChunkBuffer(t *testing.T, memWriter *MemoryWriter, log logger.Logger) *LowLatencyBuffer {
+	t.Helper()
+	shards := runtime.NumCPU()
+	bufferSize := 2 * chunkSize * shards // two buffer sets, one 60 KB chunk/shard
+	buffer := NewLowLatencyBuffer(memWriter, log, bufferSize, 3*time.Minute)
+	t.Cleanup(buffer.Close)
+	return buffer
+}
+
+// appendToShard writes a line to a specific shard, bypassing shard selection, so tests can
+// drive overflows deterministically regardless of the number of CPUs.
+func appendToShard(t *testing.T, buffer *LowLatencyBuffer, shard *bufferShard, line string) {
+	t.Helper()
+	state := buffer.activeBufferState.Load()
+	if !shard.appendLine(&buffer.activeBufferState, state, line, "") {
+		t.Fatalf("Expected line %q to be accepted by the active buffer set", line)
+	}
+}
+
+// fillShard writes one chunkSize line to the shard, which leaves the shard with data to
+// flush and no room for any further line.
+func fillShard(t *testing.T, buffer *LowLatencyBuffer, shard *bufferShard) {
+	t.Helper()
+	const prefix = "g:overflow.fill:"
+	appendToShard(t, buffer, shard, prefix+strings.Repeat("1", chunkSize-len(prefix)))
+
+	shard.mu.Lock()
+	defer shard.mu.Unlock()
+	if shard.overflows != 0 {
+		t.Fatalf("Expected the fill line to be stored, got %d overflow(s)", shard.overflows)
+	}
+}
+
+func TestBufferShard_RecordOverflowSamplesFirstDroppedId(t *testing.T) {
+	shard := &bufferShard{}
+
+	shard.recordOverflow("g:first.dropped:1", 17)
+	shard.recordOverflow("g:second.dropped:1", 18)
+
+	if shard.overflows != 2 {
+		t.Errorf("Expected %d overflows, got %d", 2, shard.overflows)
+	}
+	if shard.overflowBytes != 35 {
+		t.Errorf("Expected %d overflow bytes, got %d", 35, shard.overflowBytes)
+	}
+	if shard.droppedSampleId != "first.dropped" {
+		t.Errorf("Expected the first dropped meter id to be sampled, got %q", shard.droppedSampleId)
+	}
+}
+
+func TestLowLatencyBuffer_OverflowLogsOncePerFlushWindow(t *testing.T) {
+	memWriter := &MemoryWriter{}
+	log := &captureLogger{}
+	buffer := newSingleChunkBuffer(t, memWriter, log)
+
+	shard := buffer.frontBuffers[0]
+	fillShard(t, buffer, shard)
+
+	dropped := []string{"g:overflow.dropped:1", "g:overflow.dropped:2", "c:other.dropped:3"}
+	var droppedBytes int
+	for _, line := range dropped {
+		appendToShard(t, buffer, shard, line)
+		droppedBytes += len(line)
+	}
+
+	buffer.swapAndFlush()
+
+	messages := log.overflowMessages()
+	if len(messages) != 1 {
+		t.Fatalf("Expected 1 overflow log line for the flush window, got %d: %v", len(messages), messages)
+	}
+	message := messages[0]
+	for _, want := range []string{
+		"bufferSet=front",
+		fmt.Sprintf("dropped %d lines (%d bytes)", len(dropped), droppedBytes),
+		"sample dropped meter id: overflow.dropped",
+	} {
+		if !strings.Contains(message, want) {
+			t.Errorf("Expected overflow log line to contain %q, got %q", want, message)
+		}
+	}
+
+	// The per-shard overflow metrics are unchanged by the log line
+	overflowMetrics := map[string]bool{
+		fmt.Sprintf("c:spectator-go.lowLatencyBuffer.overflows,bufferSet=front:%d", len(dropped)):     false,
+		fmt.Sprintf("d:spectator-go.lowLatencyBuffer.overflowBytes,bufferSet=front:%d", droppedBytes): false,
+	}
+	for _, line := range memWriter.Lines() {
+		if _, ok := overflowMetrics[line]; ok {
+			overflowMetrics[line] = true
+		}
+	}
+	for line, found := range overflowMetrics {
+		if !found {
+			t.Errorf("Expected metric line %q in MemoryWriter, got %v", line, memWriter.Lines())
+		}
+	}
+}