Separate component type and name in Prometheus metrics

This simplifies querying of influx-spout metrics.

filter/filter.go

@@ -159,7 +159,8 @@ func (f *Filter) startStatistician(st *stats.Stats, rules *RuleSet) {
 	defer f.wg.Done()
 
 	generalLabels := map[string]string{
-		"filter": f.c.Name,
+		"component": "filter",
+		"name":      f.c.Name,
 	}
 
 	for {
@@ -177,8 +178,9 @@ func (f *Filter) startStatistician(st *stats.Stats, rules *RuleSet) {
 				ruleCounts[i],
 				now,
 				map[string]string{
-					"filter": f.c.Name,
-					"rule":   subject,
+					"component": "filter",
+					"name":      f.c.Name,
+					"rule":      subject,
 				},
 			))
 		}

filter/filter_medium_test.go

@@ -105,11 +105,11 @@ goodbye,host=gopher01
 
 	// Receive monitor metrics
 	spouttest.AssertMonitor(t, monitorCh, []string{
-		`passed{filter="particle"} 2`,
-		`processed{filter="particle"} 3`,
-		`rejected{filter="particle"} 1`,
-		`invalid_time{filter="particle"} 0`,
-		`triggered{filter="particle",rule="hello-subject"} 2`,
+		`passed{component="filter",name="particle"} 2`,
+		`processed{component="filter",name="particle"} 3`,
+		`rejected{component="filter",name="particle"} 1`,
+		`invalid_time{component="filter",name="particle"} 0`,
+		`triggered{component="filter",name="particle",rule="hello-subject"} 2`,
 	})
 }
 
@@ -162,10 +162,10 @@ func TestInvalidTimeStamps(t *testing.T) {
 
 	// Receive monitor metrics.
 	spouttest.AssertMonitor(t, monitorCh, []string{
-		`passed{filter="particle"} 2`,
-		`processed{filter="particle"} 4`,
-		`rejected{filter="particle"} 0`,
-		`invalid_time{filter="particle"} 2`,
-		`triggered{filter="particle",rule="hello-subject"} 2`,
+		`passed{component="filter",name="particle"} 2`,
+		`processed{component="filter",name="particle"} 4`,
+		`rejected{component="filter",name="particle"} 0`,
+		`invalid_time{component="filter",name="particle"} 2`,
+		`triggered{component="filter",name="particle",rule="hello-subject"} 2`,
 	})
 }

listener/listener.go

@@ -278,7 +278,8 @@ func (l *Listener) startStatistician() {
 	defer l.wg.Done()
 
 	labels := map[string]string{
-		"listener": l.c.Name,
+		"component": "listener",
+		"name":      l.c.Name,
 	}
 	for {
 		lines := stats.SnapshotToPrometheus(l.stats.Snapshot(), time.Now(), labels)

listener/listener_medium_test.go

@@ -296,9 +296,9 @@ func assertNoMore(t *testing.T, ch chan string) {
 
 func assertMonitor(t *testing.T, monitorCh chan string, received, sent int) {
 	expected := []string{
-		fmt.Sprintf(`received{listener="testlistener"} %d`, received),
-		fmt.Sprintf(`sent{listener="testlistener"} %d`, sent),
-		`read_errors{listener="testlistener"} 0`,
+		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`,
 	}
 	spouttest.AssertMonitor(t, monitorCh, expected)
 }

spouttest/e2e_test.go

@@ -129,17 +129,17 @@ func TestEndToEnd(t *testing.T) {
 
 	// Check metrics published by monitor component.
 	expectedMetrics := `
-failed_writes{influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",writer="writer"} 0
-invalid_time{filter="filter"} 0
-passed{filter="filter"} 10
-processed{filter="filter"} 20
-read_errors{listener="listener"} 0
-received{influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",writer="writer"} 2
-received{listener="listener"} 5
-rejected{filter="filter"} 10
-sent{listener="listener"} 1
-triggered{filter="filter",rule="system"} 10
-write_requests{influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",writer="writer"} 2
+failed_writes{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",name="writer"} 0
+invalid_time{component="filter",name="filter"} 0
+passed{component="filter",name="filter"} 10
+processed{component="filter",name="filter"} 20
+read_errors{component="listener",name="listener"} 0
+received{component="listener",name="listener"} 5
+received{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",name="writer"} 2
+rejected{component="filter",name="filter"} 10
+sent{component="listener",name="listener"} 1
+triggered{component="filter",name="filter",rule="system"} 10
+write_requests{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44501",name="writer"} 2
 `[1:]
 	var lines string
 	for try := 0; try < 20; try++ {

writer/writer.go

@@ -301,7 +301,8 @@ func (w *Writer) startStatistician() {
 
 func (w *Writer) metricsLabels() map[string]string {
 	return map[string]string{
-		"writer":           w.c.Name,
+		"component":        "writer",
+		"name":             w.c.Name,
 		"influxdb_address": w.c.InfluxDBAddress,
 		"influxdb_port":    strconv.Itoa(w.c.InfluxDBPort),
 		"influxdb_dbname":  w.c.DBName,

writer/writer_medium_test.go

@@ -96,10 +96,11 @@ func TestBasicWriter(t *testing.T) {
 
 	// Check the monitor output.
 	labels := "{" + strings.Join([]string{
+		`component="writer"`,
 		`influxdb_address="localhost"`,
 		`influxdb_dbname="metrics"`,
 		fmt.Sprintf(`influxdb_port="%d"`, influxPort),
-		`writer="foo"`,
+		`name="foo"`,
 	}, ",") + "}"
 	spouttest.AssertMonitor(t, monitorCh, []string{
 		`received` + labels + ` 5`,