[m3msg] Add receive and handle latency to consumers

src/msg/consumer/consumer.go

@@ -29,6 +29,7 @@ import (
 	"github.com/m3db/m3/src/msg/protocol/proto"
 	"github.com/m3db/m3/src/x/clock"
 	xio "github.com/m3db/m3/src/x/io"
+	xtime "github.com/m3db/m3/src/x/time"
 
 	"github.com/uber-go/tally"
 )
@@ -75,6 +76,10 @@ type metrics struct {
 	ackSent            tally.Counter
 	ackEncodeError     tally.Counter
 	ackWriteError      tally.Counter
+	// the duration between the producer sending the message and the consumer reading the message.
+	receiveLatency tally.Histogram
+	// the duration between the consumer reading the message and sending an ack to the producer.
+	handleLatency tally.Histogram
 }
 
 func newConsumerMetrics(scope tally.Scope) metrics {
@@ -84,6 +89,12 @@ func newConsumerMetrics(scope tally.Scope) metrics {
 		ackSent:            scope.Counter("ack-sent"),
 		ackEncodeError:     scope.Counter("ack-encode-error"),
 		ackWriteError:      scope.Counter("ack-write-error"),
+		receiveLatency: scope.Histogram("receive-latency",
+			// 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1.2s, 2.4s, 4.8s, 9.6s
+			tally.MustMakeExponentialDurationBuckets(time.Millisecond*10, 2, 11)),
+		handleLatency: scope.Histogram("handle-latency",
+			// 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1.2s, 2.4s, 4.8s, 9.6s
+			tally.MustMakeExponentialDurationBuckets(time.Millisecond*10, 2, 11)),
 	}
 }
 
@@ -156,6 +167,9 @@ func (c *consumer) Message() (Message, error) {
 		c.m.messageDecodeError.Inc(1)
 		return nil, err
 	}
+	if m.Metadata.SentAtNanos > 0 {
+		c.m.receiveLatency.RecordDuration(xtime.Since(xtime.UnixNano(m.Metadata.SentAtNanos)))
+	}
 	c.m.messageReceived.Inc(1)
 	return m, nil
 }

src/msg/consumer/handlers.go

@@ -23,6 +23,7 @@ package consumer
 import (
 	"io"
 	"net"
+	"time"
 
 	"github.com/m3db/m3/src/x/server"
 
@@ -60,7 +61,9 @@ func (h *messageHandler) Handle(conn net.Conn) {
 		if msgErr != nil {
 			break
 		}
+		start := time.Now()
 		c.process(msg)
+		h.m.handleLatency.RecordDuration(time.Since(start))
 	}
 	if msgErr != nil && msgErr != io.EOF {
 		h.opts.InstrumentOptions().Logger().With(zap.Error(msgErr)).Error("could not read message from consumer")

src/msg/generated/proto/msgpb/msg.proto

@@ -7,6 +7,7 @@ import "github.com/gogo/protobuf/gogoproto/gogo.proto";
 message Metadata {
     uint64 shard = 1;
     uint64 id = 2;
+    uint64 sentAtNanos = 3;
 }
 
 message Message {

src/msg/producer/writer/message.go

@@ -122,6 +122,11 @@ func (m *message) Metadata() metadata {
 	return m.meta
 }
 
+// SetSentAt sets the sentAtNanos on the metadata.
+func (m *message) SetSentAt(nanos int64) {
+	m.meta.sentAtNanos = uint64(nanos)
+}
+
 // Marshaler returns the marshaler and a bool to indicate whether the marshaler is valid.
 func (m *message) Marshaler() (proto.Marshaler, bool) {
 	return &m.pb, !m.RefCountedMessage.IsDroppedOrConsumed()

src/msg/producer/writer/message_writer.go

@@ -318,6 +318,7 @@ func (w *messageWriterImpl) write(
 	m *message,
 ) error {
 	m.IncReads()
+	m.SetSentAt(w.nowFn().UnixNano())
 	msg, isValid := m.Marshaler()
 	if !isValid {
 		m.DecReads()

src/msg/producer/writer/metadata.go

@@ -24,18 +24,21 @@ import "github.com/m3db/m3/src/msg/generated/proto/msgpb"
 
 // metadata is the metadata for a message.
 type metadata struct {
-	shard uint64
-	id    uint64
+	shard       uint64
+	id          uint64
+	sentAtNanos uint64
 }
 
 func (m metadata) ToProto(pb *msgpb.Metadata) {
 	pb.Shard = m.shard
 	pb.Id = m.id
+	pb.SentAtNanos = m.sentAtNanos
 }
 
 func (m *metadata) FromProto(pb msgpb.Metadata) {
 	m.shard = pb.Shard
 	m.id = pb.Id
+	m.sentAtNanos = pb.SentAtNanos
 }
 
 func newMetadataFromProto(pb msgpb.Metadata) metadata {

src/x/time/time.go

@@ -35,6 +35,12 @@ func Now() UnixNano {
 	return ToUnixNano(time.Now())
 }
 
+// Since returns the time elapsed since t.
+// It is shorthand for time.Now().Sub(t).
+func Since(t UnixNano) time.Duration {
+	return Now().Sub(t)
+}
+
 // ToNormalizedTime returns the normalized units of time given a time unit.
 func ToNormalizedTime(t time.Time, u time.Duration) int64 {
 	return t.UnixNano() / u.Nanoseconds()