Add queue processing metrics

src/msg/producer/writer/message_writer.go

@@ -62,6 +62,14 @@ type messageWriter interface {
 	// RemoveConsumerWriter removes the consumer writer for the given address.
 	RemoveConsumerWriter(addr string)
 
+	// Metrics returns the metrics
+	Metrics() messageWriterMetrics
+
+	// SetMetrics sets the metrics
+	//
+	// This allows changing the labels of the metrics when the downstream consumer instance changes.
+	SetMetrics(m messageWriterMetrics)
+
 	// ReplicatedShardID returns the replicated shard id.
 	ReplicatedShardID() uint64
 
@@ -88,6 +96,8 @@ type messageWriter interface {
 }
 
 type messageWriterMetrics struct {
+	scope tally.Scope
+	opts instrument.TimerOptions
 	writeSuccess             tally.Counter
 	oneConsumerWriteError    tally.Counter
 	allConsumersWriteError   tally.Counter
@@ -103,40 +113,83 @@ type messageWriterMetrics struct {
 	messageWriteDelay        tally.Timer
 	scanBatchLatency         tally.Timer
 	scanTotalLatency         tally.Timer
+	enqueuedMessages         tally.Counter
+	dequeuedMessages         tally.Counter
+	processedWrite           tally.Counter
+	processedClosed          tally.Counter
+	processedNotReady        tally.Counter
+	processedTTL             tally.Counter
+	processedAck             tally.Counter
+	processedDrop            tally.Counter
+}
+
+func (m messageWriterMetrics) withConsumer(consumer string) messageWriterMetrics {
+	return newMessageWriterMetricsWithConsumer(m.scope, m.opts, consumer)
 }
 
 func newMessageWriterMetrics(
 	scope tally.Scope,
 	opts instrument.TimerOptions,
 ) messageWriterMetrics {
+	return newMessageWriterMetricsWithConsumer(scope, opts, "")
+}
+
+func newMessageWriterMetricsWithConsumer(
+	scope tally.Scope,
+	opts instrument.TimerOptions,
+	consumer string,
+) messageWriterMetrics {
+	consumerScope := scope.Tagged(map[string]string{"consumer" : consumer})
 	return messageWriterMetrics{
-		writeSuccess:          scope.Counter("write-success"),
+		scope: scope,
+		opts: opts,
+		writeSuccess:          consumerScope.Counter("write-success"),
 		oneConsumerWriteError: scope.Counter("write-error-one-consumer"),
-		allConsumersWriteError: scope.
+		allConsumersWriteError: consumerScope.
 			Tagged(map[string]string{"error-type": "all-consumers"}).
 			Counter("write-error"),
-		noWritersError: scope.
+		noWritersError: consumerScope.
 			Tagged(map[string]string{"error-type": "no-writers"}).
 			Counter("write-error"),
-		writeAfterCutoff: scope.
+		writeAfterCutoff: consumerScope.
 			Tagged(map[string]string{"reason": "after-cutoff"}).
 			Counter("invalid-write"),
-		writeBeforeCutover: scope.
+		writeBeforeCutover: consumerScope.
 			Tagged(map[string]string{"reason": "before-cutover"}).
 			Counter("invalid-write"),
-		messageAcked:  scope.Counter("message-acked"),
-		messageClosed: scope.Counter("message-closed"),
-		messageDroppedBufferFull: scope.Tagged(
+		messageAcked:  consumerScope.Counter("message-acked"),
+		messageClosed: consumerScope.Counter("message-closed"),
+		messageDroppedBufferFull: consumerScope.Tagged(
 			map[string]string{"reason": "buffer-full"},
 		).Counter("message-dropped"),
-		messageDroppedTTLExpire: scope.Tagged(
+		messageDroppedTTLExpire: consumerScope.Tagged(
 			map[string]string{"reason": "ttl-expire"},
 		).Counter("message-dropped"),
-		messageRetry:          scope.Counter("message-retry"),
-		messageConsumeLatency: instrument.NewTimer(scope, "message-consume-latency", opts),
-		messageWriteDelay:     instrument.NewTimer(scope, "message-write-delay", opts),
-		scanBatchLatency:      instrument.NewTimer(scope, "scan-batch-latency", opts),
-		scanTotalLatency:      instrument.NewTimer(scope, "scan-total-latency", opts),
+		messageRetry:          consumerScope.Counter("message-retry"),
+		messageConsumeLatency: instrument.NewTimer(consumerScope, "message-consume-latency", opts),
+		messageWriteDelay:     instrument.NewTimer(consumerScope, "message-write-delay", opts),
+		scanBatchLatency:      instrument.NewTimer(consumerScope, "scan-batch-latency", opts),
+		scanTotalLatency:      instrument.NewTimer(consumerScope, "scan-total-latency", opts),
+		enqueuedMessages:      consumerScope.Counter("message-enqueue"),
+		dequeuedMessages:      consumerScope.Counter("message-dequeue"),
+		processedWrite: consumerScope.
+			Tagged(map[string]string{"result": "write"}).
+			Counter("message-processed"),
+		processedClosed: consumerScope.
+			Tagged(map[string]string{"result": "closed"}).
+			Counter("message-processed"),
+		processedNotReady: consumerScope.
+			Tagged(map[string]string{"result": "retry"}).
+			Counter("message-processed"),
+		processedTTL: consumerScope.
+			Tagged(map[string]string{"result": "ttl"}).
+			Counter("message-processed"),
+		processedAck: consumerScope.
+			Tagged(map[string]string{"result": "ack"}).
+			Counter("message-processed"),
+		processedDrop: consumerScope.
+			Tagged(map[string]string{"result": "drop"}).
+			Counter("message-processed"),
 	}
 }
 
@@ -221,6 +274,7 @@ func (w *messageWriterImpl) Write(rm *producer.RefCountedMessage) {
 	msg.Set(meta, rm, nowNanos)
 	w.acks.add(meta, msg)
 	// Make sure all the new writes are ordered in queue.
+	w.m.enqueuedMessages.Inc(1)
 	if w.lastNewWrite != nil {
 		w.lastNewWrite = w.queue.InsertAfter(msg, w.lastNewWrite)
 	} else {
@@ -429,6 +483,7 @@ func (w *messageWriterImpl) scanBatchWithLock(
 		next = e.Next()
 		m := e.Value.(*message)
 		if w.isClosed {
+			w.m.processedClosed.Inc(1)
 			// Simply ack the messages here to mark them as consumed for this
 			// message writer, this is useful when user removes a consumer service
 			// during runtime that may be unhealthy to consume the messages.
@@ -441,6 +496,7 @@ func (w *messageWriterImpl) scanBatchWithLock(
 			continue
 		}
 		if m.RetryAtNanos() >= nowNanos {
+			w.m.processedNotReady.Inc(1)
 			if !fullScan {
 				// If this is not a full scan, bail after the first element that
 				// is not a new write.
@@ -451,6 +507,7 @@ func (w *messageWriterImpl) scanBatchWithLock(
 		// If the message exceeded its allowed ttl of the consumer service,
 		// remove it from the buffer.
 		if w.messageTTLNanos > 0 && m.InitNanos()+w.messageTTLNanos <= nowNanos {
+			w.m.processedTTL.Inc(1)
 			// There is a chance the message was acked right before the ack is
 			// called, in which case just remove it from the queue.
 			if acked, _ := w.acks.ack(m.Metadata()); acked {
@@ -460,10 +517,12 @@ func (w *messageWriterImpl) scanBatchWithLock(
 			continue
 		}
 		if m.IsAcked() {
+			w.m.processedAck.Inc(1)
 			w.removeFromQueueWithLock(e, m)
 			continue
 		}
 		if m.IsDroppedOrConsumed() {
+			w.m.processedDrop.Inc(1)
 			// There is a chance the message could be acked between m.Acked()
 			// and m.IsDroppedOrConsumed() check, in which case we should not
 			// mark it as dropped, just continue and next tick will remove it
@@ -482,6 +541,7 @@ func (w *messageWriterImpl) scanBatchWithLock(
 		if writeTimes > 1 {
 			w.m.messageRetry.Inc(1)
 		}
+		w.m.processedWrite.Inc(1)
 		w.msgsToWrite = append(w.msgsToWrite, m)
 	}
 	return next, w.msgsToWrite
@@ -599,6 +659,18 @@ func (w *messageWriterImpl) RemoveConsumerWriter(addr string) {
 	w.Unlock()
 }
 
+func (w *messageWriterImpl) Metrics() messageWriterMetrics {
+	w.RLock()
+	defer w.RUnlock()
+	return w.m
+}
+
+func (w *messageWriterImpl) SetMetrics(m messageWriterMetrics) {
+	w.Lock()
+	w.m = m
+	w.Unlock()
+}
+
 func (w *messageWriterImpl) QueueSize() int {
 	return w.acks.size()
 }
@@ -612,6 +684,7 @@ func (w *messageWriterImpl) newMessage() *message {
 
 func (w *messageWriterImpl) removeFromQueueWithLock(e *list.Element, m *message) {
 	w.queue.Remove(e)
+	w.m.dequeuedMessages.Inc(1)
 	w.close(m)
 }
 

src/msg/producer/writer/message_writer_test.go

@@ -799,7 +799,9 @@ func TestMessageWriterQueueFullScanOnWriteErrors(t *testing.T) {
 	defer ctrl.Finish()
 
 	opts := testOptions().SetMessageQueueScanBatchSize(1)
-	w := newMessageWriter(200, nil, opts, testMessageWriterMetrics()).(*messageWriterImpl)
+	scope := tally.NewTestScope("", nil)
+	metrics := testMessageWriterMetricsWithScope(scope).withConsumer("c1")
+	w := newMessageWriter(200, nil, opts, metrics).(*messageWriterImpl)
 	w.AddConsumerWriter(newConsumerWriter("bad", nil, opts, testConsumerWriterMetrics()))
 
 	mm1 := producer.NewMockMessage(ctrl)
@@ -820,6 +822,11 @@ func TestMessageWriterQueueFullScanOnWriteErrors(t *testing.T) {
 	rm1.Drop()
 	w.scanMessageQueue()
 	require.Equal(t, 1, w.queue.Len())
+
+	snapshot := scope.Snapshot()
+	counters := snapshot.Counters()
+	require.Equal(t, int64(1), counters["message-processed+consumer=c1,result=write"].Value())
+	require.Equal(t, int64(1), counters["message-processed+consumer=c1,result=drop"].Value())
 }
 
 func isEmptyWithLock(h *acks) bool {
@@ -838,6 +845,10 @@ func testMessageWriterMetrics() messageWriterMetrics {
 	return newMessageWriterMetrics(tally.NoopScope, instrument.TimerOptions{})
 }
 
+func testMessageWriterMetricsWithScope(scope tally.TestScope) messageWriterMetrics {
+	return newMessageWriterMetrics(scope, instrument.TimerOptions{})
+}
+
 func validateMessages(t *testing.T, msgs []*producer.RefCountedMessage, w *messageWriterImpl) {
 	w.RLock()
 	idx := 0

src/msg/producer/writer/shard_writer.go

@@ -203,6 +203,9 @@ func (w *replicatedShardWriter) UpdateInstances(
 		if instance, cw, ok := anyKeyValueInMap(toBeAdded); ok {
 			mw.AddConsumerWriter(cw)
 			mw.RemoveConsumerWriter(id)
+			// a replicated writer only has a single downstream consumer instance at a time so we can update the
+			// metrics with a useful consumer label.
+			mw.SetMetrics(mw.Metrics().withConsumer(instance.ID()))
 			w.updateCutoverCutoffNanos(mw, instance)
 			newMessageWriters[instance.Endpoint()] = mw
 			delete(toBeAdded, instance)
@@ -218,6 +221,7 @@ func (w *replicatedShardWriter) UpdateInstances(
 		w.replicaID++
 		mw := newMessageWriter(replicatedShardID, w.mPool, w.opts, w.m)
 		mw.AddConsumerWriter(cw)
+		mw.SetMetrics(mw.Metrics().withConsumer(instance.ID()))
 		w.updateCutoverCutoffNanos(mw, instance)
 		mw.Init()
 		w.ackRouter.Register(replicatedShardID, mw)