Merge pull request #73 from mjs/listener-failed-nats-publish

listener: Add `failed_nats_publish` metric

listener/listener.go

@@ -34,9 +34,10 @@ import (
 
 const (
 	// Listener stats counters
-	statReceived   = "received"
-	statSent       = "sent"
-	statReadErrors = "read_errors"
+	statReceived          = "received"
+	statSent              = "sent"
+	statReadErrors        = "read_errors"
+	statFailedNATSPublish = "failed_nats_publish"
 
 	// The maximum possible UDP read size.
 	udpMaxDatagramSize = 65536
@@ -129,8 +130,13 @@ func newListener(c *config.Config) (*Listener, error) {
 		c:     c,
 		ready: make(chan struct{}),
 		stop:  make(chan struct{}),
-		stats: stats.New(statReceived, statSent, statReadErrors),
-		buf:   make([]byte, c.ListenerBatchBytes),
+		stats: stats.New(
+			statReceived,
+			statSent,
+			statReadErrors,
+			statFailedNATSPublish,
+		),
+		buf: make([]byte, c.ListenerBatchBytes),
 
 		// If more than batchSizeThreshold bytes has been written to
 		// the current batch buffer, the batch will be sent. We allow
@@ -262,6 +268,7 @@ func (l *Listener) processRead(sz int) {
 	if statReceived%l.c.BatchMessages == 0 || l.batchSize > l.batchSizeThreshold {
 		l.stats.Inc(statSent)
 		if err := l.nc.Publish(l.c.NATSSubject[0], l.buf[:l.batchSize]); err != nil {
+			l.stats.Inc(statFailedNATSPublish)
 			l.handleNatsError(err)
 		}
 		l.batchSize = 0

listener/listener_medium_test.go

@@ -299,6 +299,7 @@ func assertMonitor(t *testing.T, monitorCh chan string, received, sent int) {
 		fmt.Sprintf(`received{component="listener",name="testlistener"} %d`, received),
 		fmt.Sprintf(`sent{component="listener",name="testlistener"} %d`, sent),
 		`read_errors{component="listener",name="testlistener"} 0`,
+		`failed_nats_publish{component="listener",name="testlistener"} 0`,
 	}
 	spouttest.AssertMonitor(t, monitorCh, expected)
 }

spouttest/e2e_test.go

@@ -131,6 +131,7 @@ func TestEndToEnd(t *testing.T) {
 	// Check metrics published by monitor component.
 	expectedMetrics := regexp.MustCompile(`
 failed_nats_publish{component="filter",name="filter"} 0
+failed_nats_publish{component="listener",name="listener"} 0
 failed_writes{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",name="writer"} 0
 invalid_time{component="filter",name="filter"} 0
 max_pending{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",name="writer"} \d+