Repository: jumptrading/influx-spout (go)

Issue:
Add hostname/k8s pod name to influx-spout metrics
The Prometheus formatted metrics reported by influx-spout should include the hostname and/or Kubernetes pod name. 



Pull request:
Include hostname in Prometheus metrics
* Rename prometheus.LabelPair(s) to Label(s)  (better name)
* Added tests for Labels and Labels
* Added `With()` method to Labels 
* Pass labels around as `prometheus.Labels` instead of a map. This is more convenient and avoid unnecessary conversion.
* New `stats.NewLabels()` func for creating the base set of labels reported by all components.
* Include the hostname in the base set of labels.

Closes #89.


Patch:
diff --git a/filter/filter.go b/filter/filter.go
index c5e149f..3eabf8c 100644
--- a/filter/filter.go
+++ b/filter/filter.go
@@ -164,11 +164,7 @@ func (f *Filter) Stop() {
 func (f *Filter) startStatistician(st *stats.Stats, ruleSt *stats.AnonStats, rules *RuleSet) {
 	defer f.wg.Done()
 
-	generalLabels := map[string]string{
-		"component": "filter",
-		"name":      f.c.Name,
-	}
-
+	labels := stats.NewLabels("filter", f.c.Name)
 	for {
 		f.updateNATSDropped(st)
 
@@ -177,7 +173,7 @@ func (f *Filter) startStatistician(st *stats.Stats, ruleSt *stats.AnonStats, rul
 		ruleCounts := ruleSt.Snapshot()
 
 		// publish the general stats
-		lines := stats.SnapshotToPrometheus(snap, now, generalLabels)
+		lines := stats.SnapshotToPrometheus(snap, now, labels)
 		f.nc.Publish(f.c.NATSSubjectMonitor, lines)
 
 		// publish the per rule stats
@@ -186,11 +182,7 @@ func (f *Filter) startStatistician(st *stats.Stats, ruleSt *stats.AnonStats, rul
 				"triggered",
 				int(ruleCounts[i]),
 				now,
-				map[string]string{
-					"component": "filter",
-					"name":      f.c.Name,
-					"rule":      subject,
-				},
+				labels.With("rule", subject),
 			))
 		}
 
diff --git a/listener/listener.go b/listener/listener.go
index 12a5810..0799c67 100644
--- a/listener/listener.go
+++ b/listener/listener.go
@@ -413,11 +413,7 @@ func (l *Listener) handleNatsError(err error) {
 
 func (l *Listener) startStatistician() {
 	defer l.wg.Done()
-
-	labels := map[string]string{
-		"component": "listener",
-		"name":      l.c.Name,
-	}
+	labels := stats.NewLabels("listener", l.c.Name)
 	for {
 		lines := stats.SnapshotToPrometheus(l.stats.Snapshot(), time.Now(), labels)
 		l.nc.Publish(l.c.NATSSubjectMonitor, lines)
diff --git a/prometheus/labels.go b/prometheus/labels.go
new file mode 100644
index 0000000..a73c1b1
--- /dev/null
+++ b/prometheus/labels.go
@@ -0,0 +1,60 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package prometheus
+
+import (
+	"bytes"
+	"fmt"
+	"sort"
+)
+
+// Label contains a label name and value.
+type Label struct {
+	Name  []byte
+	Value []byte
+}
+
+// Labels contains the set of Label instances for a metric.
+type Labels []Label
+
+// ToBytes renders the label name and value to wire format.
+func (labels Labels) ToBytes() []byte {
+	labels.sort() // ensure consistent output order
+
+	out := new(bytes.Buffer)
+	out.WriteByte('{')
+	for i, label := range labels {
+		fmt.Fprintf(out, `%s="%s"`, label.Name, label.Value)
+		if i < len(labels)-1 {
+			out.WriteByte(',')
+		}
+	}
+	out.WriteByte('}')
+	return out.Bytes()
+}
+
+func (labels Labels) sort() {
+	sort.Slice(labels, func(i, j int) bool {
+		return bytes.Compare(labels[i].Name, labels[j].Name) < 0
+	})
+}
+
+// With returns a new Labels with the additional labels added.
+func (labels Labels) With(name, value string) Labels {
+	return append(labels, Label{
+		Name:  []byte(name),
+		Value: []byte(value),
+	})
+}
diff --git a/prometheus/metric.go b/prometheus/metric.go
index 2e0ceb2..ccc9ecc 100644
--- a/prometheus/metric.go
+++ b/prometheus/metric.go
@@ -17,14 +17,13 @@ package prometheus
 import (
 	"bytes"
 	"fmt"
-	"sort"
 )
 
 // Metric represents a single Prometheus metric line, including its
 // labels and timestamp.
 type Metric struct {
 	Name         []byte
-	Labels       LabelPairs
+	Labels       Labels
 	Value        int64
 	Milliseconds int64
 }
@@ -45,34 +44,3 @@ func (m *Metric) ToBytes() []byte {
 	}
 	return out.Bytes()
 }
-
-// LabelPairs contains the set of labels for a metric.
-type LabelPairs []LabelPair
-
-// ToBytes renders the label name and value to wire format.
-func (p LabelPairs) ToBytes() []byte {
-	p.sort() // ensure consistent output order
-
-	out := new(bytes.Buffer)
-	out.WriteByte('{')
-	for i, label := range p {
-		fmt.Fprintf(out, `%s="%s"`, label.Name, label.Value)
-		if i < len(p)-1 {
-			out.WriteByte(',')
-		}
-	}
-	out.WriteByte('}')
-	return out.Bytes()
-}
-
-func (p LabelPairs) sort() {
-	sort.Slice(p, func(i, j int) bool {
-		return bytes.Compare(p[i].Name, p[j].Name) < 0
-	})
-}
-
-// LabelPair contains a label name and value.
-type LabelPair struct {
-	Name  []byte
-	Value []byte
-}
diff --git a/prometheus/parse.go b/prometheus/parse.go
index 025e953..0559d68 100644
--- a/prometheus/parse.go
+++ b/prometheus/parse.go
@@ -90,15 +90,15 @@ func ParseMetric(s []byte) (*Metric, error) {
 	return out, nil
 }
 
-func parseLabels(s []byte) (LabelPairs, int, error) {
+func parseLabels(s []byte) (Labels, int, error) {
 	if s[0] == '}' {
 		return nil, 1, nil
 	}
 
 	i := 0
-	out := make(LabelPairs, 0, 1)
+	out := make(Labels, 0, 1)
 	for {
-		var label LabelPair
+		var label Label
 
 		j := bytes.Index(s[i:], []byte(`="`))
 		if j == -1 {
diff --git a/stats/prometheus.go b/stats/prometheus.go
index ce5299e..e8f1be8 100644
--- a/stats/prometheus.go
+++ b/stats/prometheus.go
@@ -15,34 +15,32 @@
 package stats
 
 import (
+	"os"
 	"time"
 
 	"github.com/jumptrading/influx-spout/prometheus"
 )
 
+// NewLabels creates a new prometheus.Labels with the component and
+// name set as given.
+func NewLabels(component, name string) prometheus.Labels {
+	return prometheus.Labels{
+		{Name: []byte("component"), Value: []byte(component)},
+		{Name: []byte("name"), Value: []byte(name)},
+		{Name: []byte("host"), Value: []byte(hostname)},
+	}
+}
+
 // SnapshotToPrometheus takes Snapshot produced by a Stats instance
 // and formats it into Prometheus metrics lines using the timestamp
 // and labels provided.
-func SnapshotToPrometheus(
-	snap Snapshot,
-	now time.Time,
-	labels map[string]string,
-) []byte {
+func SnapshotToPrometheus(snap Snapshot, now time.Time, labels prometheus.Labels) []byte {
 	millis := timeToMillis(now)
-
-	labelPairs := make(prometheus.LabelPairs, 0, len(labels))
-	for name, value := range labels {
-		labelPairs = append(labelPairs, prometheus.LabelPair{
-			Name:  []byte(name),
-			Value: []byte(value),
-		})
-	}
-
 	set := prometheus.NewMetricSet()
 	for _, counter := range snap {
 		set.Update(&prometheus.Metric{
 			Name:         []byte(counter.Name),
-			Labels:       labelPairs,
+			Labels:       labels,
 			Value:        int64(counter.Value),
 			Milliseconds: millis,
 		})
@@ -51,10 +49,10 @@ func SnapshotToPrometheus(
 }
 
 // CounterToPrometheus generates a single Prometheus line for a counter.
-func CounterToPrometheus(name string, value int, now time.Time, labels map[string]string) []byte {
+func CounterToPrometheus(name string, value int, now time.Time, labels prometheus.Labels) []byte {
 	metric := &prometheus.Metric{
 		Name:         []byte(name),
-		Labels:       toLabelPairs(labels),
+		Labels:       labels,
 		Value:        int64(value),
 		Milliseconds: timeToMillis(now),
 	}
@@ -65,13 +63,18 @@ func timeToMillis(t time.Time) int64 {
 	return t.UnixNano() / int64(time.Millisecond)
 }
 
-func toLabelPairs(labels map[string]string) prometheus.LabelPairs {
-	labelPairs := make(prometheus.LabelPairs, 0, len(labels))
-	for name, value := range labels {
-		labelPairs = append(labelPairs, prometheus.LabelPair{
-			Name:  []byte(name),
-			Value: []byte(value),
-		})
+var hostname string
+
+// SetHostname allows the hostname reported for metrics to be
+// overridden. This is mainly intended for tests.
+func SetHostname(h string) {
+	hostname = h
+}
+
+func init() {
+	h, err := os.Hostname()
+	if err != nil {
+		panic(err)
 	}
-	return labelPairs
+	SetHostname(h)
 }
diff --git a/writer/writer.go b/writer/writer.go
index f80f754..cb601a7 100644
--- a/writer/writer.go
+++ b/writer/writer.go
@@ -241,14 +241,12 @@ func (w *Writer) sendBatch(batch *batch.Batch, client *http.Client) error {
 func (w *Writer) startStatistician(subs []*nats.Subscription) {
 	defer w.wg.Done()
 
+	labels := stats.NewLabels("writer", w.c.Name).
+		With("influxdb_address", w.c.InfluxDBAddress).
+		With("influxdb_port", strconv.Itoa(w.c.InfluxDBPort)).
+		With("influxdb_dbname", w.c.DBName)
+
 	for {
-		labels := map[string]string{
-			"component":        "writer",
-			"name":             w.c.Name,
-			"influxdb_address": w.c.InfluxDBAddress,
-			"influxdb_port":    strconv.Itoa(w.c.InfluxDBPort),
-			"influxdb_dbname":  w.c.DBName,
-		}
 		now := time.Now()
 
 		// Publish general stats.
@@ -262,8 +260,11 @@ func (w *Writer) startStatistician(subs []*nats.Subscription) {
 				log.Printf("NATS: failed to get dropped count: %v", err)
 				continue
 			}
-			labels["subject"] = sub.Subject
-			line := stats.CounterToPrometheus(statNATSDropped, dropped, now, labels)
+			line := stats.CounterToPrometheus(
+				statNATSDropped,
+				dropped,
+				now,
+				labels.With("subject", sub.Subject))
 			w.nc.Publish(w.c.NATSSubjectMonitor, line)
 		}
 


Test patch:
diff --git a/cmd/influx-spout/e2e_large_test.go b/cmd/influx-spout/e2e_large_test.go
index f46fae5..bc9a64e 100644
--- a/cmd/influx-spout/e2e_large_test.go
+++ b/cmd/influx-spout/e2e_large_test.go
@@ -34,6 +34,7 @@ import (
 
 	"github.com/jumptrading/influx-spout/config"
 	"github.com/jumptrading/influx-spout/spouttest"
+	"github.com/jumptrading/influx-spout/stats"
 )
 
 const (
@@ -58,6 +59,8 @@ const (
 )
 
 func TestEndToEnd(t *testing.T) {
+	stats.SetHostname("h")
+
 	// Start gnatsd.
 	gnatsd := spouttest.RunGnatsd(natsPort)
 	defer gnatsd.Shutdown()
@@ -140,22 +143,22 @@ 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="44601",name="writer"} 0
-invalid_time{component="filter",name="filter"} 0
-max_pending{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} \d+
-nats_dropped{component="filter",name="filter"} 0
-nats_dropped{component="writer",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer",subject="system"} 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="44601",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="44601",name="writer"} 2
+failed_nats_publish{component="filter",host="h",name="filter"} 0
+failed_nats_publish{component="listener",host="h",name="listener"} 0
+failed_writes{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} 0
+invalid_time{component="filter",host="h",name="filter"} 0
+max_pending{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} \d+
+nats_dropped{component="filter",host="h",name="filter"} 0
+nats_dropped{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer",subject="system"} 0
+passed{component="filter",host="h",name="filter"} 10
+processed{component="filter",host="h",name="filter"} 20
+read_errors{component="listener",host="h",name="listener"} 0
+received{component="listener",host="h",name="listener"} 5
+received{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} 2
+rejected{component="filter",host="h",name="filter"} 10
+sent{component="listener",host="h",name="listener"} 1
+triggered{component="filter",host="h",name="filter",rule="system"} 10
+write_requests{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} 2
 $`[1:])
 	var lines string
 	for try := 0; try < 20; try++ {
diff --git a/filter/filter_medium_test.go b/filter/filter_medium_test.go
index ffa8b9b..3498997 100644
--- a/filter/filter_medium_test.go
+++ b/filter/filter_medium_test.go
@@ -28,6 +28,7 @@ import (
 
 	"github.com/jumptrading/influx-spout/config"
 	"github.com/jumptrading/influx-spout/spouttest"
+	"github.com/jumptrading/influx-spout/stats"
 )
 
 const natsPort = 44100
@@ -53,6 +54,8 @@ func testConfig() *config.Config {
 }
 
 func TestFilterWorker(t *testing.T) {
+	stats.SetHostname("h")
+
 	gnatsd := spouttest.RunGnatsd(natsPort)
 	defer gnatsd.Shutdown()
 
@@ -108,17 +111,19 @@ goodbye,host=gopher01
 
 	// Receive monitor metrics
 	spouttest.AssertMonitor(t, monitorCh, []string{
-		`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`,
-		`failed_nats_publish{component="filter",name="particle"} 0`,
-		`nats_dropped{component="filter",name="particle"} 0`,
-		`triggered{component="filter",name="particle",rule="hello-subject"} 2`,
+		`passed{component="filter",host="h",name="particle"} 2`,
+		`processed{component="filter",host="h",name="particle"} 3`,
+		`rejected{component="filter",host="h",name="particle"} 1`,
+		`invalid_time{component="filter",host="h",name="particle"} 0`,
+		`failed_nats_publish{component="filter",host="h",name="particle"} 0`,
+		`nats_dropped{component="filter",host="h",name="particle"} 0`,
+		`triggered{component="filter",host="h",name="particle",rule="hello-subject"} 2`,
 	})
 }
 
 func TestInvalidTimeStamps(t *testing.T) {
+	stats.SetHostname("h")
+
 	gnatsd := spouttest.RunGnatsd(natsPort)
 	defer gnatsd.Shutdown()
 
@@ -166,13 +171,13 @@ func TestInvalidTimeStamps(t *testing.T) {
 
 	// Receive monitor metrics.
 	spouttest.AssertMonitor(t, monitorCh, []string{
-		`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`,
-		`failed_nats_publish{component="filter",name="particle"} 0`,
-		`nats_dropped{component="filter",name="particle"} 0`,
-		`triggered{component="filter",name="particle",rule="hello-subject"} 2`,
+		`passed{component="filter",host="h",name="particle"} 2`,
+		`processed{component="filter",host="h",name="particle"} 4`,
+		`rejected{component="filter",host="h",name="particle"} 0`,
+		`invalid_time{component="filter",host="h",name="particle"} 2`,
+		`failed_nats_publish{component="filter",host="h",name="particle"} 0`,
+		`nats_dropped{component="filter",host="h",name="particle"} 0`,
+		`triggered{component="filter",host="h",name="particle",rule="hello-subject"} 2`,
 	})
 }
 
diff --git a/listener/listener_medium_test.go b/listener/listener_medium_test.go
index 7f58bde..51ef692 100644
--- a/listener/listener_medium_test.go
+++ b/listener/listener_medium_test.go
@@ -32,6 +32,7 @@ import (
 
 	"github.com/jumptrading/influx-spout/config"
 	"github.com/jumptrading/influx-spout/spouttest"
+	"github.com/jumptrading/influx-spout/stats"
 )
 
 const (
@@ -79,6 +80,8 @@ func testConfig() *config.Config {
 }
 
 func TestBatching(t *testing.T) {
+	stats.SetHostname("h")
+
 	s := spouttest.RunGnatsd(natsPort)
 	defer s.Shutdown()
 
@@ -110,6 +113,8 @@ func TestBatching(t *testing.T) {
 }
 
 func TestWhatComesAroundGoesAround(t *testing.T) {
+	stats.SetHostname("h")
+
 	s := spouttest.RunGnatsd(natsPort)
 	defer s.Shutdown()
 
@@ -212,6 +217,8 @@ func TestBatchAge(t *testing.T) {
 }
 
 func TestHTTPListener(t *testing.T) {
+	stats.SetHostname("h")
+
 	s := spouttest.RunGnatsd(natsPort)
 	defer s.Shutdown()
 
@@ -241,6 +248,8 @@ func TestHTTPListener(t *testing.T) {
 }
 
 func TestHTTPListenerBigPOST(t *testing.T) {
+	stats.SetHostname("h")
+
 	s := spouttest.RunGnatsd(natsPort)
 	defer s.Shutdown()
 
@@ -552,10 +561,10 @@ func assertNoMore(t *testing.T, ch chan string) {
 
 func assertMonitor(t *testing.T, monitorCh chan string, received, sent int) {
 	expected := []string{
-		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`,
+		fmt.Sprintf(`received{component="listener",host="h",name="testlistener"} %d`, received),
+		fmt.Sprintf(`sent{component="listener",host="h",name="testlistener"} %d`, sent),
+		`read_errors{component="listener",host="h",name="testlistener"} 0`,
+		`failed_nats_publish{component="listener",host="h",name="testlistener"} 0`,
 	}
 	spouttest.AssertMonitor(t, monitorCh, expected)
 }
diff --git a/monitor/monitor_medium_test.go b/monitor/monitor_medium_test.go
index d2d8b61..551ecc7 100644
--- a/monitor/monitor_medium_test.go
+++ b/monitor/monitor_medium_test.go
@@ -70,7 +70,7 @@ func TestMonitor(t *testing.T) {
 	// monitor's metric endpoint.
 	m0 := &prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc01"),
@@ -93,7 +93,7 @@ func TestMonitor(t *testing.T) {
 	nextUpdate := prometheus.NewMetricSet()
 	nextUpdate.Update(&prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc01"),
@@ -108,7 +108,7 @@ func TestMonitor(t *testing.T) {
 	})
 	nextUpdate.Update(&prometheus.Metric{
 		Name: []byte("bar"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc02"),
diff --git a/prometheus/labels_small_test.go b/prometheus/labels_small_test.go
new file mode 100644
index 0000000..01be384
--- /dev/null
+++ b/prometheus/labels_small_test.go
@@ -0,0 +1,76 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build small
+
+package prometheus_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/jumptrading/influx-spout/prometheus"
+)
+
+func TestToBytesEmpty(t *testing.T) {
+	var labels prometheus.Labels
+	assert.Equal(t, labels.ToBytes(), []byte("{}"))
+}
+
+func TestToBytesSingle(t *testing.T) {
+	labels := prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+	}
+	assert.Equal(t, labels.ToBytes(), []byte(`{foo="aaa"}`))
+}
+
+func TestToBytesMultiple(t *testing.T) {
+	labels := prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+		{Name: []byte("bar"), Value: []byte("bbbb")},
+	}
+	// Note reordering.
+	assert.Equal(t, labels.ToBytes(), []byte(`{bar="bbbb",foo="aaa"}`))
+}
+
+func TestWith(t *testing.T) {
+	labels0 := prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+	}
+	labels1 := labels0.With("bar", "bbb")
+	labels2 := labels0.With("bar", "none")
+	labels3 := labels2.With("che", "sne")
+
+	assert.Equal(t, labels1, prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+		{Name: []byte("bar"), Value: []byte("bbb")},
+	})
+
+	assert.Equal(t, labels2, prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+		{Name: []byte("bar"), Value: []byte("none")},
+	})
+
+	assert.Equal(t, labels3, prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+		{Name: []byte("bar"), Value: []byte("none")},
+		{Name: []byte("che"), Value: []byte("sne")},
+	})
+
+	// Original should be unchanged.
+	assert.Equal(t, labels0, prometheus.Labels{
+		{Name: []byte("foo"), Value: []byte("aaa")},
+	})
+}
diff --git a/prometheus/metric_small_test.go b/prometheus/metric_small_test.go
index dd93209..3c97113 100644
--- a/prometheus/metric_small_test.go
+++ b/prometheus/metric_small_test.go
@@ -44,7 +44,7 @@ func TestToBytesTimestamp(t *testing.T) {
 func TestToBytesLabels(t *testing.T) {
 	m := &prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc01"),
diff --git a/prometheus/metricset_small_test.go b/prometheus/metricset_small_test.go
index 873eee6..94b80ad 100644
--- a/prometheus/metricset_small_test.go
+++ b/prometheus/metricset_small_test.go
@@ -67,7 +67,7 @@ func TestUpdateWithLabels(t *testing.T) {
 
 	m0 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -78,7 +78,7 @@ func TestUpdateWithLabels(t *testing.T) {
 
 	m1 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("1")},
 		},
@@ -89,7 +89,7 @@ func TestUpdateWithLabels(t *testing.T) {
 
 	m2 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("auk01")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -100,7 +100,7 @@ func TestUpdateWithLabels(t *testing.T) {
 
 	m3 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("auk02")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -130,7 +130,7 @@ func TestUpdateLabelOrdering(t *testing.T) {
 
 	m0 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -143,7 +143,7 @@ func TestUpdateLabelOrdering(t *testing.T) {
 	// switched.
 	m1 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("core"), Value: []byte("0")},
 			{Name: []byte("host"), Value: []byte("nyc02")},
 		},
@@ -160,7 +160,7 @@ func TestUpdateFromSet(t *testing.T) {
 	set0 := prometheus.NewMetricSet()
 	m0 := &prometheus.Metric{
 		Name: []byte("uptime"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc01")},
 		},
 		Value:        222,
@@ -169,7 +169,7 @@ func TestUpdateFromSet(t *testing.T) {
 	set0.Update(m0)
 	set0.Update(&prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -181,7 +181,7 @@ func TestUpdateFromSet(t *testing.T) {
 	set1 := prometheus.NewMetricSet()
 	m1 := &prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -191,7 +191,7 @@ func TestUpdateFromSet(t *testing.T) {
 	set1.Update(m1)
 	m2 := &prometheus.Metric{
 		Name: []byte("uptime"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 		},
 		Value:        1234,
@@ -209,7 +209,7 @@ func TestToBytes(t *testing.T) {
 
 	set.Update(&prometheus.Metric{
 		Name: []byte("uptime"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc01")},
 		},
 		Value:        1234,
@@ -217,7 +217,7 @@ func TestToBytes(t *testing.T) {
 	})
 	set.Update(&prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("0")},
 		},
@@ -226,7 +226,7 @@ func TestToBytes(t *testing.T) {
 	})
 	set.Update(&prometheus.Metric{
 		Name: []byte("temp"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 			{Name: []byte("core"), Value: []byte("1")},
 		},
@@ -235,7 +235,7 @@ func TestToBytes(t *testing.T) {
 	})
 	set.Update(&prometheus.Metric{
 		Name: []byte("uptime"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{Name: []byte("host"), Value: []byte("nyc02")},
 		},
 		Value:        4444,
diff --git a/prometheus/parse_small_test.go b/prometheus/parse_small_test.go
index 96084cd..d3c82f0 100644
--- a/prometheus/parse_small_test.go
+++ b/prometheus/parse_small_test.go
@@ -69,7 +69,7 @@ func TestParseLabel(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, &prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{{
+		Labels: prometheus.Labels{{
 			Name:  []byte("method"),
 			Value: []byte("post"),
 		}},
@@ -91,7 +91,7 @@ func TestParseMultipleLabels(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, &prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("method"),
 				Value: []byte("post"),
@@ -162,7 +162,7 @@ func TestParseLabelsAndTimestamp(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, &prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc01"),
@@ -190,7 +190,7 @@ bar 1234`[1:])
 	expected := prometheus.NewMetricSet()
 	expected.Update(&prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc01"),
@@ -209,7 +209,7 @@ bar 1234`[1:])
 	})
 	expected.Update(&prometheus.Metric{
 		Name: []byte("foo"),
-		Labels: prometheus.LabelPairs{
+		Labels: prometheus.Labels{
 			{
 				Name:  []byte("host"),
 				Value: []byte("nyc02"),
diff --git a/stats/prometheus_small_test.go b/stats/prometheus_small_test.go
index dc9b1db..bfcbdb1 100644
--- a/stats/prometheus_small_test.go
+++ b/stats/prometheus_small_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/jumptrading/influx-spout/prometheus"
 	"github.com/jumptrading/influx-spout/stats"
 	"github.com/stretchr/testify/assert"
 )
@@ -43,16 +44,12 @@ qaz 0 1234567890000
 
 func TestToPrometheusWithLabels(t *testing.T) {
 	snap := stats.Snapshot{
-		{"foo", 42},
-		{"bar", 99},
-		{"qaz", 0},
+		{Name: "foo", Value: 42},
+		{Name: "bar", Value: 99},
+		{Name: "qaz", Value: 0},
 	}
 	ts := time.Date(2009, 2, 13, 23, 31, 30, 0, time.UTC)
-	labels := map[string]string{
-		"host":  "nyc01",
-		"level": "high",
-	}
-
+	labels := prometheus.Labels{}.With("host", "nyc01").With("level", "high")
 	actual := stats.SnapshotToPrometheus(snap, ts, labels)
 	expected := []byte(`
 bar{host="nyc01",level="high"} 99 1234567890000
@@ -64,7 +61,7 @@ qaz{host="nyc01",level="high"} 0 1234567890000
 
 func TestCounterToPrometheus(t *testing.T) {
 	ts := time.Date(2009, 2, 13, 23, 31, 30, 0, time.UTC)
-	labels := map[string]string{"host": "nyc01"}
+	labels := prometheus.Labels{}.With("host", "nyc01")
 
 	actual := stats.CounterToPrometheus("foo", 99, ts, labels)
 	expected := []byte(`foo{host="nyc01"} 99 1234567890000`)
diff --git a/writer/writer_medium_test.go b/writer/writer_medium_test.go
index 0057450..76594f9 100644
--- a/writer/writer_medium_test.go
+++ b/writer/writer_medium_test.go
@@ -32,6 +32,7 @@ import (
 
 	"github.com/jumptrading/influx-spout/config"
 	"github.com/jumptrading/influx-spout/spouttest"
+	"github.com/jumptrading/influx-spout/stats"
 )
 
 const natsPort = 44200
@@ -61,6 +62,8 @@ func testConfig() *config.Config {
 }
 
 func TestBasicWriter(t *testing.T) {
+	stats.SetHostname("h")
+
 	nc, closeNATS := runGnatsd(t)
 	defer closeNATS()
 
@@ -100,6 +103,7 @@ func TestBasicWriter(t *testing.T) {
 	// Check the monitor output.
 	labels := "{" + strings.Join([]string{
 		`component="writer"`,
+		`host="h"`,
 		`influxdb_address="localhost"`,
 		`influxdb_dbname="metrics"`,
 		fmt.Sprintf(`influxdb_port="%d"`, influxPort),