writer: Include InfluxDB details in metrics tags

writer/writer.go

@@ -23,16 +23,12 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
+	"strconv"
 	"sync"
 	"time"
 
-	// for profiling a nasty memleak
 	"net/http"
-	_ "net/http/pprof"
-
-	// This would be nice, but it's too unstable for now
-	// revisit eventually
-	//"github.com/valyala/fasthttp"
+	_ "net/http/pprof" // for profiling a nasty memleak
 
 	"github.com/nats-io/go-nats"
 
@@ -299,12 +295,22 @@ func (w *Writer) startStatistician() {
 	// sending it to the monitoring backend.
 	statsLine := lineformatter.New(
 		"spout_stat_writer",
-		[]string{"writer"}, // tag keys
+		[]string{ // tag keys
+			"writer",
+			"influxdb_address",
+			"influxdb_port",
+			"influxdb_dbname",
+		},
 		"received",
 		"write_requests",
 		"failed_writes",
 	)
-	tagVals := []string{w.c.Name}
+	tagVals := []string{
+		w.c.Name,
+		w.c.InfluxDBAddress,
+		strconv.Itoa(w.c.InfluxDBPort),
+		w.c.DBName,
+	}
 	for {
 		stats := w.stats.Clone()
 		w.nc.Publish(w.c.NATSSubjectMonitor, statsLine.Format(

writer/writer_medium_test.go

@@ -21,6 +21,8 @@ import (
 	"io/ioutil"
 	"net/http"
 	"os"
+	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -42,16 +44,18 @@ var natsAddress = fmt.Sprintf("nats://127.0.0.1:%d", natsPort)
 
 func testConfig() *config.Config {
 	return &config.Config{
+		Mode:               "writer",
+		Name:               "foo",
 		NATSAddress:        natsAddress,
 		NATSSubject:        []string{"writer-test"},
 		NATSSubjectMonitor: "writer-test-monitor",
 		InfluxDBAddress:    "localhost",
 		InfluxDBPort:       influxPort,
+		DBName:             "metrics",
 		BatchMessages:      1,
 		BatchMaxMB:         10,
 		BatchMaxSecs:       300,
 		Port:               influxPort,
-		Mode:               "writer",
 		Workers:            96,
 		NATSPendingMaxMB:   32,
 	}
@@ -96,15 +100,22 @@ func TestBasicWriter(t *testing.T) {
 	w := startWriter(t, conf)
 	defer w.Stop()
 
-	// publish 5 messages to the bus
+	// Subscribe to stats output.
+	statsCh := make(chan string, 10)
+	_, err := nc.Subscribe(conf.NATSSubjectMonitor, func(msg *nats.Msg) {
+		statsCh <- string(msg.Data)
+	})
+	require.NoError(t, err)
+
+	// Publish 5 messages to the bus.
 	subject := conf.NATSSubject[0]
 	publish(t, subject, "To be, or not to be: that is the question:")
 	publish(t, subject, "Whether ’tis nobler in the mind to suffer")
 	publish(t, subject, "The slings and arrows of outrageous fortune,")
 	publish(t, subject, "Or to take arms against a sea of troubles,")
 	publish(t, subject, "And by opposing end them. To die: to sleep;")
 
-	// wait for confirmation that they were written
+	// Wait for confirmation that they were written.
 	timeout := time.After(spouttest.LongWait)
 	for i := 0; i < 5; i++ {
 		select {
@@ -113,6 +124,16 @@ func TestBasicWriter(t *testing.T) {
 			t.Fatal("timed out waiting for messages")
 		}
 	}
+
+	// Check the stats output.
+	spouttest.AssertRecvMulti(t, statsCh, "stats",
+		strings.Join([]string{
+			"spout_stat_writer",
+			"writer=foo",
+			"influxdb_address=localhost",
+			"influxdb_port=" + strconv.Itoa(influxPort),
+			"influxdb_dbname=metrics",
+		}, ",")+" received=5,write_requests=5,failed_writes=0\n")
 }
 
 func TestBatchMBLimit(t *testing.T) {