Durable Emitter - Metrics and Alerts

pkg/durableemitter/durable_emitter.go

@@ -10,6 +10,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"google.golang.org/protobuf/proto"
 
@@ -182,6 +183,11 @@ type DurableEmitter struct {
 	pendingCount atomic.Int64
 	pendingMax   atomic.Int64
 
+	// fallbackInFlightCount is incremented when a single-event fallback
+	// goroutine starts and decremented on its defer. metricsLoop samples
+	// this into the fallback.in_flight gauge.
+	fallbackInFlightCount atomic.Int64
+
 	// stopCh signals background loops to exit.
 	stopCh services.StopChan
 	wg     sync.WaitGroup
@@ -416,6 +422,10 @@ func (d *DurableEmitter) Emit(ctx context.Context, body []byte, attrKVs ...any)
 		t0Publish := time.Now()
 		if qErr := d.batchEmitter.QueueMessage(eventPb, d.deliveryCallback(id, eventPb, t0Publish)); qErr != nil {
 			d.eng.Warnw("DurableEmitter: batch emitter buffer full, relying on retransmit", "id", id)
+			if d.metrics != nil {
+				d.metrics.batchEnqueueBufferFull.Add(ctx, 1,
+					metric.WithAttributes(attribute.String("phase", "immediate")))
+			}
 		}
 		return nil
 	})
@@ -480,8 +490,10 @@ func (d *DurableEmitter) tryFallback(id int64, eventPb *chipingress.CloudEventPb
 		return
 	}
 	d.fallbackWg.Add(1)
+	d.fallbackInFlightCount.Add(1)
 	go func() {
 		defer d.fallbackWg.Done()
+		defer d.fallbackInFlightCount.Add(-1)
 		d.singleEventFallback(id, eventPb)
 	}()
 }
@@ -679,12 +691,12 @@ func (d *DurableEmitter) retransmitPending() {
 		return
 	}
 
-	d.retransmit(pending)
+	d.retransmit(ctx, pending)
 }
 
 // retransmit re-enqueues pending DB rows through the batch emitter. Each row
 // gets its own delivery callback that marks it delivered on success.
-func (d *DurableEmitter) retransmit(pending []DurableEvent) {
+func (d *DurableEmitter) retransmit(ctx context.Context, pending []DurableEvent) {
 	var enqueued, skipped int
 
 	for _, pe := range pending {
@@ -703,6 +715,10 @@ func (d *DurableEmitter) retransmit(pending []DurableEvent) {
 		id := pe.ID
 		if err := d.batchEmitter.QueueMessage(eventPb, d.deliveryCallback(id, eventPb, time.Now())); err != nil {
 			skipped++
+			if d.metrics != nil {
+				d.metrics.batchEnqueueBufferFull.Add(ctx, 1,
+					metric.WithAttributes(attribute.String("phase", "retransmit")))
+			}
 		} else {
 			enqueued++
 		}
@@ -802,6 +818,14 @@ func (d *DurableEmitter) metricsLoop() {
 		case <-ticker.C:
 			d.metrics.queueDepth.Record(ctx, d.pendingCount.Load())
 			d.metrics.queueDepthMax.Record(ctx, d.pendingMax.Load())
+			d.metrics.fallbackInFlight.Record(ctx, d.fallbackInFlightCount.Load())
+			if d.insertCh != nil {
+				if c := cap(d.insertCh); c > 0 {
+					d.metrics.insertCoalescerFill.Record(ctx, float64(len(d.insertCh))/float64(c))
+				}
+			} else {
+				d.metrics.insertCoalescerFill.Record(ctx, 0)
+			}
 			if obs, ok := d.store.(DurableQueueObserver); ok {
 				d.metrics.pollQueueGauges(ctx, obs, d.cfg.EventTTL, d.queueStatsNearExpiryLead(), mc.MaxQueuePayloadBytes)
 			}

pkg/durableemitter/durable_emitter_metrics.go

@@ -52,6 +52,16 @@ type durableEmitterMetrics struct {
 	procHeapSys        metric.Int64Gauge
 	procCPUUser        metric.Float64Gauge
 	procCPUSys         metric.Float64Gauge
+	// batchEnqueueBufferFull counts events that could not be handed to the
+	// batch emitter because its internal queue was full and must be picked up
+	// by the retransmit loop instead. Labels: phase={immediate,retransmit}.
+	batchEnqueueBufferFull metric.Int64Counter
+	// insertCoalescerFill reports the write-coalescer channel fill ratio
+	// (len/cap). Only meaningful when InsertBatchSize > 0; otherwise 0.
+	insertCoalescerFill metric.Float64Gauge
+	// fallbackInFlight reports the number of single-event fallback Publish
+	// goroutines currently in flight.
+	fallbackInFlight metric.Int64Gauge
 }
 
 // durationBuckets provides histogram boundaries (in seconds) tuned for
@@ -251,6 +261,27 @@ func newDurableEmitterMetrics(meter metric.Meter) (*durableEmitterMetrics, error
 	); err != nil {
 		return nil, err
 	}
+	if m.batchEnqueueBufferFull, err = meter.Int64Counter(
+		"durable_emitter.batch_enqueue.buffer_full",
+		metric.WithUnit("{event}"),
+		metric.WithDescription("Events that could not be handed to the batch emitter (buffer full); event remains in DB for retransmit. Labels: phase={immediate,retransmit}."),
+	); err != nil {
+		return nil, err
+	}
+	if m.insertCoalescerFill, err = meter.Float64Gauge(
+		"durable_emitter.insert_coalescer.queue_fill_ratio",
+		metric.WithUnit("1"),
+		metric.WithDescription("Write-coalescer channel fill ratio (len/cap); 0 when write coalescing is disabled"),
+	); err != nil {
+		return nil, err
+	}
+	if m.fallbackInFlight, err = meter.Int64Gauge(
+		"durable_emitter.fallback.in_flight",
+		metric.WithUnit("{goroutine}"),
+		metric.WithDescription("Single-event fallback Publish goroutines currently in flight"),
+	); err != nil {
+		return nil, err
+	}
 	return m, nil
 }