prometheus: Implement ParseMetrics

This turns multiple metrics lines into a MetricSet.

prometheus/metric.go

@@ -29,6 +29,10 @@ type Metric struct {
 	Milliseconds int64
 }
 
+func (m *Metric) String() string {
+	return fmt.Sprintf("<Metric: %s>", m.ToBytes())
+}
+
 // ToBytes renders the metric to wire format.
 func (m *Metric) ToBytes() []byte {
 	out := bytes.NewBuffer(m.Name)

prometheus/parse.go

@@ -17,10 +17,27 @@ package prometheus
 import (
 	"bytes"
 	"errors"
+	"fmt"
 
 	"github.com/jumptrading/influx-spout/convert"
 )
 
+// ParseMetrics parses multiple Promethesus metric lines, returning a MetricSet.
+func ParseMetrics(s []byte) (*MetricSet, error) {
+	set := NewMetricSet()
+	for i, line := range bytes.Split(s, []byte("\n")) {
+		if len(line) == 0 {
+			continue
+		}
+		m, err := ParseMetric(line)
+		if err != nil {
+			return nil, fmt.Errorf("line %d: %v", i+1, err)
+		}
+		set.Update(m)
+	}
+	return set, nil
+}
+
 // ParseMetric parses a single Promethesus metric line.
 //
 // Note: The implementation currently only supports integer values and

prometheus/parse_small_test.go

@@ -180,3 +180,106 @@ func TestParseLabelsAndTimestamp(t *testing.T) {
 		Milliseconds: 123456789,
 	}, m)
 }
+
+func TestParseMetrics(t *testing.T) {
+	input := []byte(`
+foo{host="nyc01",bar="definitely",thing="forgot"} 42 11111111
+foo{host="nyc02",bar="maybe"} 999 22222222
+bar 1234`[1:])
+
+	expected := prometheus.NewMetricSet()
+	expected.Update(&prometheus.Metric{
+		Name: []byte("foo"),
+		Labels: prometheus.LabelPairs{
+			{
+				Name:  []byte("host"),
+				Value: []byte("nyc01"),
+			},
+			{
+				Name:  []byte("bar"),
+				Value: []byte("definitely"),
+			},
+			{
+				Name:  []byte("thing"),
+				Value: []byte("forgot"),
+			},
+		},
+		Value:        42,
+		Milliseconds: 11111111,
+	})
+	expected.Update(&prometheus.Metric{
+		Name: []byte("foo"),
+		Labels: prometheus.LabelPairs{
+			{
+				Name:  []byte("host"),
+				Value: []byte("nyc02"),
+			},
+			{
+				Name:  []byte("bar"),
+				Value: []byte("maybe"),
+			},
+		},
+		Value:        999,
+		Milliseconds: 22222222,
+	})
+	expected.Update(&prometheus.Metric{
+		Name:  []byte("bar"),
+		Value: 1234,
+	})
+
+	actual, err := prometheus.ParseMetrics(input)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, expected.All(), actual.All())
+}
+
+func TestParseMetricsBlankLines(t *testing.T) {
+	input := []byte(`
+foo 111
+
+bar 222
+
+`)
+
+	expected := prometheus.NewMetricSet()
+	expected.Update(&prometheus.Metric{
+		Name:  []byte("foo"),
+		Value: 111,
+	})
+	expected.Update(&prometheus.Metric{
+		Name:  []byte("bar"),
+		Value: 222,
+	})
+
+	actual, err := prometheus.ParseMetrics(input)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, expected.All(), actual.All())
+}
+
+func TestParseMetricsLastWins(t *testing.T) {
+	input := []byte(`
+foo 1
+foo 2
+foo 3
+foo 4
+foo 5
+foo 6
+`[1:])
+
+	expected := prometheus.NewMetricSet()
+	expected.Update(&prometheus.Metric{
+		Name:  []byte("foo"),
+		Value: 6,
+	})
+
+	actual, err := prometheus.ParseMetrics(input)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, expected.All(), actual.All())
+}
+
+func TestParseMetricsEmpty(t *testing.T) {
+	expected := prometheus.NewMetricSet()
+
+	actual, err := prometheus.ParseMetrics([]byte{})
+	require.NoError(t, err)
+	assert.ElementsMatch(t, expected.All(), actual.All())
+}