Merge pull request #11785 from timstclair/tests

Use prometheus extraction library for metric parsing in e2e tests

Godeps/Godeps.json

@@ -137,7 +137,7 @@
 			"Rev": "97e243d21a8e232e9d8af38ba2366dfcfceebeba"
 		},
 		{
-			"ImportPath": "github.com/cpuguy83/go-md2man",
+			"ImportPath": "github.com/cpuguy83/go-md2man/mangen",
 			"Comment": "v1.0.2-5-g2831f11",
 			"Rev": "2831f11f66ff4008f10e2cd7ed9a85e3d3fc2bed"
 		},
@@ -435,6 +435,11 @@
 			"Comment": "v1.0-28-g8adf9e1730c5",
 			"Rev": "8adf9e1730c55cdc590de7d49766cb2acc88d8f2"
 		},
+		{
+			"ImportPath": "github.com/prometheus/client_golang/extraction",
+			"Comment": "0.4.0-1-g692492e",
+			"Rev": "692492e54b553a81013254cc1fba4b6dd76fad30"
+		},
 		{
 			"ImportPath": "github.com/prometheus/client_golang/model",
 			"Comment": "0.4.0-1-g692492e",

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/discriminator.go

@@ -0,0 +1,74 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"errors"
+	"fmt"
+	"mime"
+	"net/http"
+)
+
+// ProcessorForRequestHeader interprets a HTTP request header to determine
+// what Processor should be used for the given input.  If no acceptable
+// Processor can be found, an error is returned.
+func ProcessorForRequestHeader(header http.Header) (Processor, error) {
+	if header == nil {
+		return nil, errors.New("received illegal and nil header")
+	}
+
+	mediatype, params, err := mime.ParseMediaType(header.Get("Content-Type"))
+	if err != nil {
+		return nil, fmt.Errorf("invalid Content-Type header %q: %s", header.Get("Content-Type"), err)
+	}
+	switch mediatype {
+	case "application/vnd.google.protobuf":
+		if params["proto"] != "io.prometheus.client.MetricFamily" {
+			return nil, fmt.Errorf("unrecognized protocol message %s", params["proto"])
+		}
+		if params["encoding"] != "delimited" {
+			return nil, fmt.Errorf("unsupported encoding %s", params["encoding"])
+		}
+		return MetricFamilyProcessor, nil
+	case "text/plain":
+		switch params["version"] {
+		case "0.0.4":
+			return Processor004, nil
+		case "":
+			// Fallback: most recent version.
+			return Processor004, nil
+		default:
+			return nil, fmt.Errorf("unrecognized API version %s", params["version"])
+		}
+	case "application/json":
+		var prometheusAPIVersion string
+
+		if params["schema"] == "prometheus/telemetry" && params["version"] != "" {
+			prometheusAPIVersion = params["version"]
+		} else {
+			prometheusAPIVersion = header.Get("X-Prometheus-API-Version")
+		}
+
+		switch prometheusAPIVersion {
+		case "0.0.2":
+			return Processor002, nil
+		case "0.0.1":
+			return Processor001, nil
+		default:
+			return nil, fmt.Errorf("unrecognized API version %s", prometheusAPIVersion)
+		}
+	default:
+		return nil, fmt.Errorf("unsupported media type %q, expected %q", mediatype, "application/json")
+	}
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/discriminator_test.go

@@ -0,0 +1,126 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"errors"
+	"net/http"
+	"testing"
+)
+
+func testDiscriminatorHTTPHeader(t testing.TB) {
+	var scenarios = []struct {
+		input  map[string]string
+		output Processor
+		err    error
+	}{
+		{
+			output: nil,
+			err:    errors.New("received illegal and nil header"),
+		},
+		{
+			input:  map[string]string{"Content-Type": "application/json", "X-Prometheus-API-Version": "0.0.0"},
+			output: nil,
+			err:    errors.New("unrecognized API version 0.0.0"),
+		},
+		{
+			input:  map[string]string{"Content-Type": "application/json", "X-Prometheus-API-Version": "0.0.1"},
+			output: Processor001,
+			err:    nil,
+		},
+		{
+			input:  map[string]string{"Content-Type": `application/json; schema="prometheus/telemetry"; version=0.0.0`},
+			output: nil,
+			err:    errors.New("unrecognized API version 0.0.0"),
+		},
+		{
+			input:  map[string]string{"Content-Type": `application/json; schema="prometheus/telemetry"; version=0.0.1`},
+			output: Processor001,
+			err:    nil,
+		},
+		{
+			input:  map[string]string{"Content-Type": `application/json; schema="prometheus/telemetry"; version=0.0.2`},
+			output: Processor002,
+			err:    nil,
+		},
+		{
+			input:  map[string]string{"Content-Type": `application/vnd.google.protobuf; proto="io.prometheus.client.MetricFamily"; encoding="delimited"`},
+			output: MetricFamilyProcessor,
+			err:    nil,
+		},
+		{
+			input:  map[string]string{"Content-Type": `application/vnd.google.protobuf; proto="illegal"; encoding="delimited"`},
+			output: nil,
+			err:    errors.New("unrecognized protocol message illegal"),
+		},
+		{
+			input:  map[string]string{"Content-Type": `application/vnd.google.protobuf; proto="io.prometheus.client.MetricFamily"; encoding="illegal"`},
+			output: nil,
+			err:    errors.New("unsupported encoding illegal"),
+		},
+		{
+			input:  map[string]string{"Content-Type": `text/plain; version=0.0.4`},
+			output: Processor004,
+			err:    nil,
+		},
+		{
+			input:  map[string]string{"Content-Type": `text/plain`},
+			output: Processor004,
+			err:    nil,
+		},
+		{
+			input:  map[string]string{"Content-Type": `text/plain; version=0.0.3`},
+			output: nil,
+			err:    errors.New("unrecognized API version 0.0.3"),
+		},
+	}
+
+	for i, scenario := range scenarios {
+		var header http.Header
+
+		if len(scenario.input) > 0 {
+			header = http.Header{}
+		}
+
+		for key, value := range scenario.input {
+			header.Add(key, value)
+		}
+
+		actual, err := ProcessorForRequestHeader(header)
+
+		if scenario.err != err {
+			if scenario.err != nil && err != nil {
+				if scenario.err.Error() != err.Error() {
+					t.Errorf("%d. expected %s, got %s", i, scenario.err, err)
+				}
+			} else if scenario.err != nil || err != nil {
+				t.Errorf("%d. expected %s, got %s", i, scenario.err, err)
+			}
+		}
+
+		if scenario.output != actual {
+			t.Errorf("%d. expected %s, got %s", i, scenario.output, actual)
+		}
+	}
+}
+
+func TestDiscriminatorHTTPHeader(t *testing.T) {
+	testDiscriminatorHTTPHeader(t)
+}
+
+func BenchmarkDiscriminatorHTTPHeader(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		testDiscriminatorHTTPHeader(b)
+	}
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/extraction.go

@@ -0,0 +1,15 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction decodes Prometheus clients' data streams for consumers.
+package extraction

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/fixtures/test0_0_1-0_0_2.json

@@ -0,0 +1,79 @@
+[
+  {
+    "baseLabels": {
+      "__name__": "rpc_calls_total",
+      "job": "batch_job"
+    },
+    "docstring": "RPC calls.",
+    "metric": {
+      "type": "counter",
+      "value": [
+        {
+          "labels": {
+            "service": "zed"
+          },
+          "value": 25
+        },
+        {
+          "labels": {
+            "service": "bar"
+          },
+          "value": 25
+        },
+        {
+          "labels": {
+            "service": "foo"
+          },
+          "value": 25
+        }
+      ]
+    }
+  },
+  {
+    "baseLabels": {
+      "__name__": "rpc_latency_microseconds"
+    },
+      "docstring": "RPC latency.",
+      "metric": {
+        "type": "histogram",
+        "value": [
+          {
+            "labels": {
+              "service": "foo"
+            },
+            "value": {
+              "0.010000": 15.890724674774395,
+              "0.050000": 15.890724674774395,
+              "0.500000": 84.63044031436561,
+              "0.900000": 160.21100853053224,
+              "0.990000": 172.49828748957728
+            }
+          },
+          {
+            "labels": {
+              "service": "zed"
+            },
+            "value": {
+              "0.010000": 0.0459814091918713,
+              "0.050000": 0.0459814091918713,
+              "0.500000": 0.6120456642749681,
+              "0.900000": 1.355915069887731,
+              "0.990000": 1.772733213161236
+            }
+          },
+          {
+            "labels": {
+              "service": "bar"
+            },
+            "value": {
+              "0.010000": 78.48563317257356,
+              "0.050000": 78.48563317257356,
+              "0.500000": 97.31798360385088,
+              "0.900000": 109.89202084295582,
+              "0.990000": 109.99626121011262
+            }
+          }
+        ]
+      }
+  }
+]

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/metricfamilyprocessor.go

@@ -0,0 +1,318 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"fmt"
+	"io"
+	"math"
+
+	dto "github.com/prometheus/client_model/go"
+
+	"github.com/matttproud/golang_protobuf_extensions/pbutil"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+type metricFamilyProcessor struct{}
+
+// MetricFamilyProcessor decodes varint encoded record length-delimited streams
+// of io.prometheus.client.MetricFamily.
+//
+// See http://godoc.org/github.com/matttproud/golang_protobuf_extensions/ext for
+// more details.
+var MetricFamilyProcessor = &metricFamilyProcessor{}
+
+func (m *metricFamilyProcessor) ProcessSingle(i io.Reader, out Ingester, o *ProcessOptions) error {
+	family := &dto.MetricFamily{}
+
+	for {
+		family.Reset()
+
+		if _, err := pbutil.ReadDelimited(i, family); err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return err
+		}
+		if err := extractMetricFamily(out, o, family); err != nil {
+			return err
+		}
+	}
+}
+
+func extractMetricFamily(out Ingester, o *ProcessOptions, family *dto.MetricFamily) error {
+	switch family.GetType() {
+	case dto.MetricType_COUNTER:
+		if err := extractCounter(out, o, family); err != nil {
+			return err
+		}
+	case dto.MetricType_GAUGE:
+		if err := extractGauge(out, o, family); err != nil {
+			return err
+		}
+	case dto.MetricType_SUMMARY:
+		if err := extractSummary(out, o, family); err != nil {
+			return err
+		}
+	case dto.MetricType_UNTYPED:
+		if err := extractUntyped(out, o, family); err != nil {
+			return err
+		}
+	case dto.MetricType_HISTOGRAM:
+		if err := extractHistogram(out, o, family); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func extractCounter(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
+	samples := make(model.Samples, 0, len(f.Metric))
+
+	for _, m := range f.Metric {
+		if m.Counter == nil {
+			continue
+		}
+
+		sample := &model.Sample{
+			Metric: model.Metric{},
+			Value:  model.SampleValue(m.Counter.GetValue()),
+		}
+		samples = append(samples, sample)
+
+		if m.TimestampMs != nil {
+			sample.Timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
+		} else {
+			sample.Timestamp = o.Timestamp
+		}
+
+		metric := sample.Metric
+		for _, p := range m.Label {
+			metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+		}
+		metric[model.MetricNameLabel] = model.LabelValue(f.GetName())
+	}
+
+	return out.Ingest(samples)
+}
+
+func extractGauge(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
+	samples := make(model.Samples, 0, len(f.Metric))
+
+	for _, m := range f.Metric {
+		if m.Gauge == nil {
+			continue
+		}
+
+		sample := &model.Sample{
+			Metric: model.Metric{},
+			Value:  model.SampleValue(m.Gauge.GetValue()),
+		}
+		samples = append(samples, sample)
+
+		if m.TimestampMs != nil {
+			sample.Timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
+		} else {
+			sample.Timestamp = o.Timestamp
+		}
+
+		metric := sample.Metric
+		for _, p := range m.Label {
+			metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+		}
+		metric[model.MetricNameLabel] = model.LabelValue(f.GetName())
+	}
+
+	return out.Ingest(samples)
+}
+
+func extractSummary(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
+	samples := make(model.Samples, 0, len(f.Metric))
+
+	for _, m := range f.Metric {
+		if m.Summary == nil {
+			continue
+		}
+
+		timestamp := o.Timestamp
+		if m.TimestampMs != nil {
+			timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
+		}
+
+		for _, q := range m.Summary.Quantile {
+			sample := &model.Sample{
+				Metric:    model.Metric{},
+				Value:     model.SampleValue(q.GetValue()),
+				Timestamp: timestamp,
+			}
+			samples = append(samples, sample)
+
+			metric := sample.Metric
+			for _, p := range m.Label {
+				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+			}
+			// BUG(matt): Update other names to "quantile".
+			metric[model.LabelName(model.QuantileLabel)] = model.LabelValue(fmt.Sprint(q.GetQuantile()))
+			metric[model.MetricNameLabel] = model.LabelValue(f.GetName())
+		}
+
+		if m.Summary.SampleSum != nil {
+			sum := &model.Sample{
+				Metric:    model.Metric{},
+				Value:     model.SampleValue(m.Summary.GetSampleSum()),
+				Timestamp: timestamp,
+			}
+			samples = append(samples, sum)
+
+			metric := sum.Metric
+			for _, p := range m.Label {
+				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+			}
+			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
+		}
+
+		if m.Summary.SampleCount != nil {
+			count := &model.Sample{
+				Metric:    model.Metric{},
+				Value:     model.SampleValue(m.Summary.GetSampleCount()),
+				Timestamp: timestamp,
+			}
+			samples = append(samples, count)
+
+			metric := count.Metric
+			for _, p := range m.Label {
+				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+			}
+			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
+		}
+	}
+
+	return out.Ingest(samples)
+}
+
+func extractUntyped(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
+	samples := make(model.Samples, 0, len(f.Metric))
+
+	for _, m := range f.Metric {
+		if m.Untyped == nil {
+			continue
+		}
+
+		sample := &model.Sample{
+			Metric: model.Metric{},
+			Value:  model.SampleValue(m.Untyped.GetValue()),
+		}
+		samples = append(samples, sample)
+
+		if m.TimestampMs != nil {
+			sample.Timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
+		} else {
+			sample.Timestamp = o.Timestamp
+		}
+
+		metric := sample.Metric
+		for _, p := range m.Label {
+			metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+		}
+		metric[model.MetricNameLabel] = model.LabelValue(f.GetName())
+	}
+
+	return out.Ingest(samples)
+}
+
+func extractHistogram(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
+	samples := make(model.Samples, 0, len(f.Metric))
+
+	for _, m := range f.Metric {
+		if m.Histogram == nil {
+			continue
+		}
+
+		timestamp := o.Timestamp
+		if m.TimestampMs != nil {
+			timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
+		}
+
+		infSeen := false
+
+		for _, q := range m.Histogram.Bucket {
+			sample := &model.Sample{
+				Metric:    model.Metric{},
+				Value:     model.SampleValue(q.GetCumulativeCount()),
+				Timestamp: timestamp,
+			}
+			samples = append(samples, sample)
+
+			metric := sample.Metric
+			for _, p := range m.Label {
+				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+			}
+			metric[model.LabelName(model.BucketLabel)] = model.LabelValue(fmt.Sprint(q.GetUpperBound()))
+			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
+
+			if math.IsInf(q.GetUpperBound(), +1) {
+				infSeen = true
+			}
+		}
+
+		if m.Histogram.SampleSum != nil {
+			sum := &model.Sample{
+				Metric:    model.Metric{},
+				Value:     model.SampleValue(m.Histogram.GetSampleSum()),
+				Timestamp: timestamp,
+			}
+			samples = append(samples, sum)
+
+			metric := sum.Metric
+			for _, p := range m.Label {
+				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+			}
+			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
+		}
+
+		if m.Histogram.SampleCount != nil {
+			count := &model.Sample{
+				Metric:    model.Metric{},
+				Value:     model.SampleValue(m.Histogram.GetSampleCount()),
+				Timestamp: timestamp,
+			}
+			samples = append(samples, count)
+
+			metric := count.Metric
+			for _, p := range m.Label {
+				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+			}
+			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
+
+			if !infSeen {
+				infBucket := &model.Sample{
+					Metric:    model.Metric{},
+					Value:     count.Value,
+					Timestamp: timestamp,
+				}
+				samples = append(samples, infBucket)
+
+				metric := infBucket.Metric
+				for _, p := range m.Label {
+					metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
+				}
+				metric[model.LabelName(model.BucketLabel)] = model.LabelValue("+Inf")
+				metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
+			}
+		}
+	}
+
+	return out.Ingest(samples)
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/metricfamilyprocessor_test.go

@@ -0,0 +1,153 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+var testTime = model.Now()
+
+type metricFamilyProcessorScenario struct {
+	in               string
+	expected, actual []model.Samples
+}
+
+func (s *metricFamilyProcessorScenario) Ingest(samples model.Samples) error {
+	s.actual = append(s.actual, samples)
+	return nil
+}
+
+func (s *metricFamilyProcessorScenario) test(t *testing.T, set int) {
+	i := strings.NewReader(s.in)
+
+	o := &ProcessOptions{
+		Timestamp: testTime,
+	}
+
+	err := MetricFamilyProcessor.ProcessSingle(i, s, o)
+	if err != nil {
+		t.Fatalf("%d. got error: %s", set, err)
+	}
+
+	if len(s.expected) != len(s.actual) {
+		t.Fatalf("%d. expected length %d, got %d", set, len(s.expected), len(s.actual))
+	}
+
+	for i, expected := range s.expected {
+		sort.Sort(s.actual[i])
+		sort.Sort(expected)
+
+		if !expected.Equal(s.actual[i]) {
+			t.Errorf("%d.%d. expected %s, got %s", set, i, expected, s.actual[i])
+		}
+	}
+}
+
+func TestMetricFamilyProcessor(t *testing.T) {
+	scenarios := []metricFamilyProcessorScenario{
+		{
+			in: "",
+		},
+		{
+			in: "\x8f\x01\n\rrequest_count\x12\x12Number of requests\x18\x00\"0\n#\n\x0fsome_label_name\x12\x10some_label_value\x1a\t\t\x00\x00\x00\x00\x00\x00E\xc0\"6\n)\n\x12another_label_name\x12\x13another_label_value\x1a\t\t\x00\x00\x00\x00\x00\x00U@",
+			expected: []model.Samples{
+				model.Samples{
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_count", "some_label_name": "some_label_value"},
+						Value:     -42,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_count", "another_label_name": "another_label_value"},
+						Value:     84,
+						Timestamp: testTime,
+					},
+				},
+			},
+		},
+		{
+			in: "\xb9\x01\n\rrequest_count\x12\x12Number of requests\x18\x02\"O\n#\n\x0fsome_label_name\x12\x10some_label_value\"(\x1a\x12\t\xaeG\xe1z\x14\xae\xef?\x11\x00\x00\x00\x00\x00\x00E\xc0\x1a\x12\t+\x87\x16\xd9\xce\xf7\xef?\x11\x00\x00\x00\x00\x00\x00U\xc0\"A\n)\n\x12another_label_name\x12\x13another_label_value\"\x14\x1a\x12\t\x00\x00\x00\x00\x00\x00\xe0?\x11\x00\x00\x00\x00\x00\x00$@",
+			expected: []model.Samples{
+				model.Samples{
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_count", "some_label_name": "some_label_value", "quantile": "0.99"},
+						Value:     -42,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_count", "some_label_name": "some_label_value", "quantile": "0.999"},
+						Value:     -84,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_count", "another_label_name": "another_label_value", "quantile": "0.5"},
+						Value:     10,
+						Timestamp: testTime,
+					},
+				},
+			},
+		},
+		{
+			in: "\x8d\x01\n\x1drequest_duration_microseconds\x12\x15The response latency.\x18\x04\"S:Q\b\x85\x15\x11\xcd\xcc\xccL\x8f\xcb:A\x1a\v\b{\x11\x00\x00\x00\x00\x00\x00Y@\x1a\f\b\x9c\x03\x11\x00\x00\x00\x00\x00\x00^@\x1a\f\b\xd0\x04\x11\x00\x00\x00\x00\x00\x00b@\x1a\f\b\xf4\v\x11\x9a\x99\x99\x99\x99\x99e@\x1a\f\b\x85\x15\x11\x00\x00\x00\x00\x00\x00\xf0\u007f",
+			expected: []model.Samples{
+				model.Samples{
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_bucket", "le": "100"},
+						Value:     123,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_bucket", "le": "120"},
+						Value:     412,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_bucket", "le": "144"},
+						Value:     592,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_bucket", "le": "172.8"},
+						Value:     1524,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_bucket", "le": "+Inf"},
+						Value:     2693,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_sum"},
+						Value:     1756047.3,
+						Timestamp: testTime,
+					},
+					&model.Sample{
+						Metric:    model.Metric{model.MetricNameLabel: "request_duration_microseconds_count"},
+						Value:     2693,
+						Timestamp: testTime,
+					},
+				},
+			},
+		},
+	}
+
+	for i, scenario := range scenarios {
+		scenario.test(t, i)
+	}
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/processor.go

@@ -0,0 +1,84 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"io"
+	"time"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+// ProcessOptions dictates how the interpreted stream should be rendered for
+// consumption.
+type ProcessOptions struct {
+	// Timestamp is added to each value from the stream that has no explicit
+	// timestamp set.
+	Timestamp model.Timestamp
+}
+
+// Ingester consumes result streams in whatever way is desired by the user.
+type Ingester interface {
+	Ingest(model.Samples) error
+}
+
+// Processor is responsible for decoding the actual message responses from
+// stream into a format that can be consumed with the end result written
+// to the results channel.
+type Processor interface {
+	// ProcessSingle treats the input as a single self-contained message body and
+	// transforms it accordingly.  It has no support for streaming.
+	ProcessSingle(in io.Reader, out Ingester, o *ProcessOptions) error
+}
+
+// Helper function to convert map[string]string into LabelSet.
+//
+// NOTE: This should be deleted when support for go 1.0.3 is removed; 1.1 is
+//       smart enough to unmarshal JSON objects into LabelSet directly.
+func labelSet(labels map[string]string) model.LabelSet {
+	labelset := make(model.LabelSet, len(labels))
+
+	for k, v := range labels {
+		labelset[model.LabelName(k)] = model.LabelValue(v)
+	}
+
+	return labelset
+}
+
+// A basic interface only useful in testing contexts for dispensing the time
+// in a controlled manner.
+type instantProvider interface {
+	// The current instant.
+	Now() time.Time
+}
+
+// Clock is a simple means for fluently wrapping around standard Go timekeeping
+// mechanisms to enhance testability without compromising code readability.
+//
+// It is sufficient for use on bare initialization.  A provider should be
+// set only for test contexts.  When not provided, it emits the current
+// system time.
+type clock struct {
+	// The underlying means through which time is provided, if supplied.
+	Provider instantProvider
+}
+
+// Emit the current instant.
+func (t *clock) Now() time.Time {
+	if t.Provider == nil {
+		return time.Now()
+	}
+
+	return t.Provider.Now()
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/processor0_0_1.go

@@ -0,0 +1,127 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+const (
+	baseLabels001 = "baseLabels"
+	counter001    = "counter"
+	docstring001  = "docstring"
+	gauge001      = "gauge"
+	histogram001  = "histogram"
+	labels001     = "labels"
+	metric001     = "metric"
+	type001       = "type"
+	value001      = "value"
+	percentile001 = "percentile"
+)
+
+// Processor001 is responsible for decoding payloads from protocol version
+// 0.0.1.
+var Processor001 = &processor001{}
+
+// processor001 is responsible for handling API version 0.0.1.
+type processor001 struct{}
+
+// entity001 represents a the JSON structure that 0.0.1 uses.
+type entity001 []struct {
+	BaseLabels map[string]string `json:"baseLabels"`
+	Docstring  string            `json:"docstring"`
+	Metric     struct {
+		MetricType string `json:"type"`
+		Value      []struct {
+			Labels map[string]string `json:"labels"`
+			Value  interface{}       `json:"value"`
+		} `json:"value"`
+	} `json:"metric"`
+}
+
+func (p *processor001) ProcessSingle(in io.Reader, out Ingester, o *ProcessOptions) error {
+	// TODO(matt): Replace with plain-jane JSON unmarshalling.
+	buffer, err := ioutil.ReadAll(in)
+	if err != nil {
+		return err
+	}
+
+	entities := entity001{}
+	if err = json.Unmarshal(buffer, &entities); err != nil {
+		return err
+	}
+
+	// TODO(matt): This outer loop is a great basis for parallelization.
+	pendingSamples := model.Samples{}
+	for _, entity := range entities {
+		for _, value := range entity.Metric.Value {
+			labels := labelSet(entity.BaseLabels).Merge(labelSet(value.Labels))
+
+			switch entity.Metric.MetricType {
+			case gauge001, counter001:
+				sampleValue, ok := value.Value.(float64)
+				if !ok {
+					return fmt.Errorf("could not convert value from %s %s to float64", entity, value)
+				}
+
+				pendingSamples = append(pendingSamples, &model.Sample{
+					Metric:    model.Metric(labels),
+					Timestamp: o.Timestamp,
+					Value:     model.SampleValue(sampleValue),
+				})
+
+				break
+
+			case histogram001:
+				sampleValue, ok := value.Value.(map[string]interface{})
+				if !ok {
+					return fmt.Errorf("could not convert value from %q to a map[string]interface{}", value.Value)
+				}
+
+				for percentile, percentileValue := range sampleValue {
+					individualValue, ok := percentileValue.(float64)
+					if !ok {
+						return fmt.Errorf("could not convert value from %q to a float64", percentileValue)
+					}
+
+					childMetric := make(map[model.LabelName]model.LabelValue, len(labels)+1)
+
+					for k, v := range labels {
+						childMetric[k] = v
+					}
+
+					childMetric[model.LabelName(percentile001)] = model.LabelValue(percentile)
+
+					pendingSamples = append(pendingSamples, &model.Sample{
+						Metric:    model.Metric(childMetric),
+						Timestamp: o.Timestamp,
+						Value:     model.SampleValue(individualValue),
+					})
+				}
+
+				break
+			}
+		}
+	}
+	if len(pendingSamples) > 0 {
+		return out.Ingest(pendingSamples)
+	}
+
+	return nil
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/processor0_0_1_test.go

@@ -0,0 +1,185 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"errors"
+	"os"
+	"path"
+	"sort"
+	"testing"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+var test001Time = model.Now()
+
+type testProcessor001ProcessScenario struct {
+	in               string
+	expected, actual []model.Samples
+	err              error
+}
+
+func (s *testProcessor001ProcessScenario) Ingest(samples model.Samples) error {
+	s.actual = append(s.actual, samples)
+	return nil
+}
+
+func (s *testProcessor001ProcessScenario) test(t testing.TB, set int) {
+	reader, err := os.Open(path.Join("fixtures", s.in))
+	if err != nil {
+		t.Fatalf("%d. couldn't open scenario input file %s: %s", set, s.in, err)
+	}
+
+	options := &ProcessOptions{
+		Timestamp: test001Time,
+	}
+	err = Processor001.ProcessSingle(reader, s, options)
+	if s.err != err && (s.err == nil || err == nil || err.Error() != s.err.Error()) {
+		t.Fatalf("%d. expected err of %s, got %s", set, s.err, err)
+	}
+
+	if len(s.actual) != len(s.expected) {
+		t.Fatalf("%d. expected output length of %d, got %d", set, len(s.expected), len(s.actual))
+	}
+
+	for i, expected := range s.expected {
+		sort.Sort(s.actual[i])
+		sort.Sort(expected)
+
+		if !expected.Equal(s.actual[i]) {
+			t.Errorf("%d.%d. expected %s, got %s", set, i, expected, s.actual[i])
+		}
+	}
+}
+
+func testProcessor001Process(t testing.TB) {
+	var scenarios = []testProcessor001ProcessScenario{
+		{
+			in:  "empty.json",
+			err: errors.New("unexpected end of JSON input"),
+		},
+		{
+			in: "test0_0_1-0_0_2.json",
+			expected: []model.Samples{
+				model.Samples{
+					&model.Sample{
+						Metric:    model.Metric{"service": "zed", model.MetricNameLabel: "rpc_calls_total", "job": "batch_job"},
+						Value:     25,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"service": "bar", model.MetricNameLabel: "rpc_calls_total", "job": "batch_job"},
+						Value:     25,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"service": "foo", model.MetricNameLabel: "rpc_calls_total", "job": "batch_job"},
+						Value:     25,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.010000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     0.0459814091918713,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.010000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     78.48563317257356,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.010000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     15.890724674774395,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.050000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     0.0459814091918713,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.050000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     78.48563317257356,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.050000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     15.890724674774395,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.500000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     0.6120456642749681,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.500000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     97.31798360385088,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.500000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     84.63044031436561,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.900000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     1.355915069887731,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.900000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     109.89202084295582,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.900000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     160.21100853053224,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.990000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     1.772733213161236,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.990000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     109.99626121011262,
+						Timestamp: test001Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.990000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     172.49828748957728,
+						Timestamp: test001Time,
+					},
+				},
+			},
+		},
+	}
+
+	for i, scenario := range scenarios {
+		scenario.test(t, i)
+	}
+}
+
+func TestProcessor001Process(t *testing.T) {
+	testProcessor001Process(t)
+}
+
+func BenchmarkProcessor001Process(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		testProcessor001Process(b)
+	}
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/processor0_0_2.go

@@ -0,0 +1,106 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+// Processor002 is responsible for decoding payloads from protocol version
+// 0.0.2.
+var Processor002 = &processor002{}
+
+type histogram002 struct {
+	Labels map[string]string            `json:"labels"`
+	Values map[string]model.SampleValue `json:"value"`
+}
+
+type counter002 struct {
+	Labels map[string]string `json:"labels"`
+	Value  model.SampleValue `json:"value"`
+}
+
+type processor002 struct{}
+
+func (p *processor002) ProcessSingle(in io.Reader, out Ingester, o *ProcessOptions) error {
+	// Processor for telemetry schema version 0.0.2.
+	// container for telemetry data
+	var entities []struct {
+		BaseLabels map[string]string `json:"baseLabels"`
+		Docstring  string            `json:"docstring"`
+		Metric     struct {
+			Type   string          `json:"type"`
+			Values json.RawMessage `json:"value"`
+		} `json:"metric"`
+	}
+
+	if err := json.NewDecoder(in).Decode(&entities); err != nil {
+		return err
+	}
+
+	pendingSamples := model.Samples{}
+	for _, entity := range entities {
+		switch entity.Metric.Type {
+		case "counter", "gauge":
+			var values []counter002
+
+			if err := json.Unmarshal(entity.Metric.Values, &values); err != nil {
+				return fmt.Errorf("could not extract %s value: %s", entity.Metric.Type, err)
+			}
+
+			for _, counter := range values {
+				labels := labelSet(entity.BaseLabels).Merge(labelSet(counter.Labels))
+
+				pendingSamples = append(pendingSamples, &model.Sample{
+					Metric:    model.Metric(labels),
+					Timestamp: o.Timestamp,
+					Value:     counter.Value,
+				})
+			}
+
+		case "histogram":
+			var values []histogram002
+
+			if err := json.Unmarshal(entity.Metric.Values, &values); err != nil {
+				return fmt.Errorf("could not extract %s value: %s", entity.Metric.Type, err)
+			}
+
+			for _, histogram := range values {
+				for percentile, value := range histogram.Values {
+					labels := labelSet(entity.BaseLabels).Merge(labelSet(histogram.Labels))
+					labels[model.LabelName("percentile")] = model.LabelValue(percentile)
+
+					pendingSamples = append(pendingSamples, &model.Sample{
+						Metric:    model.Metric(labels),
+						Timestamp: o.Timestamp,
+						Value:     value,
+					})
+				}
+			}
+
+		default:
+			return fmt.Errorf("unknown metric type %q", entity.Metric.Type)
+		}
+	}
+
+	if len(pendingSamples) > 0 {
+		return out.Ingest(pendingSamples)
+	}
+
+	return nil
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/processor0_0_2_test.go

@@ -0,0 +1,225 @@
+// Copyright 2013 The Prometheus Authors
+// 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 extraction
+
+import (
+	"bytes"
+	"errors"
+	"io/ioutil"
+	"os"
+	"path"
+	"runtime"
+	"sort"
+	"testing"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+var test002Time = model.Now()
+
+type testProcessor002ProcessScenario struct {
+	in               string
+	expected, actual []model.Samples
+	err              error
+}
+
+func (s *testProcessor002ProcessScenario) Ingest(samples model.Samples) error {
+	s.actual = append(s.actual, samples)
+	return nil
+}
+
+func (s *testProcessor002ProcessScenario) test(t testing.TB, set int) {
+	reader, err := os.Open(path.Join("fixtures", s.in))
+	if err != nil {
+		t.Fatalf("%d. couldn't open scenario input file %s: %s", set, s.in, err)
+	}
+
+	options := &ProcessOptions{
+		Timestamp: test002Time,
+	}
+	err = Processor002.ProcessSingle(reader, s, options)
+	if s.err != err && (s.err == nil || err == nil || err.Error() != s.err.Error()) {
+		t.Fatalf("%d. expected err of %s, got %s", set, s.err, err)
+	}
+
+	if len(s.actual) != len(s.expected) {
+		t.Fatalf("%d. expected output length of %d, got %d", set, len(s.expected), len(s.actual))
+	}
+
+	for i, expected := range s.expected {
+		sort.Sort(s.actual[i])
+		sort.Sort(expected)
+
+		if !expected.Equal(s.actual[i]) {
+			t.Fatalf("%d.%d. expected %s, got %s", set, i, expected, s.actual[i])
+		}
+	}
+}
+
+func testProcessor002Process(t testing.TB) {
+	var scenarios = []testProcessor002ProcessScenario{
+		{
+			in:  "empty.json",
+			err: errors.New("EOF"),
+		},
+		{
+			in: "test0_0_1-0_0_2.json",
+			expected: []model.Samples{
+				model.Samples{
+					&model.Sample{
+						Metric:    model.Metric{"service": "zed", model.MetricNameLabel: "rpc_calls_total", "job": "batch_job"},
+						Value:     25,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"service": "bar", model.MetricNameLabel: "rpc_calls_total", "job": "batch_job"},
+						Value:     25,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"service": "foo", model.MetricNameLabel: "rpc_calls_total", "job": "batch_job"},
+						Value:     25,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.010000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     0.0459814091918713,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.010000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     78.48563317257356,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.010000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     15.890724674774395,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+
+						Metric:    model.Metric{"percentile": "0.050000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     0.0459814091918713,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.050000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     78.48563317257356,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.050000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     15.890724674774395,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.500000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     0.6120456642749681,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.500000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     97.31798360385088,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.500000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     84.63044031436561,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.900000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     1.355915069887731,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.900000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     109.89202084295582,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.900000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     160.21100853053224,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.990000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "zed"},
+						Value:     1.772733213161236,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.990000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "bar"},
+						Value:     109.99626121011262,
+						Timestamp: test002Time,
+					},
+					&model.Sample{
+						Metric:    model.Metric{"percentile": "0.990000", model.MetricNameLabel: "rpc_latency_microseconds", "service": "foo"},
+						Value:     172.49828748957728,
+						Timestamp: test002Time,
+					},
+				},
+			},
+		},
+	}
+
+	for i, scenario := range scenarios {
+		scenario.test(t, i)
+	}
+}
+
+func TestProcessor002Process(t *testing.T) {
+	testProcessor002Process(t)
+}
+
+func BenchmarkProcessor002Process(b *testing.B) {
+	b.StopTimer()
+
+	pre := runtime.MemStats{}
+	runtime.ReadMemStats(&pre)
+
+	b.StartTimer()
+
+	for i := 0; i < b.N; i++ {
+		testProcessor002Process(b)
+	}
+
+	post := runtime.MemStats{}
+	runtime.ReadMemStats(&post)
+
+	allocated := post.TotalAlloc - pre.TotalAlloc
+
+	b.Logf("Allocated %d at %f per cycle with %d cycles.", allocated, float64(allocated)/float64(b.N), b.N)
+}
+
+func BenchmarkProcessor002ParseOnly(b *testing.B) {
+	b.StopTimer()
+	data, err := ioutil.ReadFile("fixtures/test0_0_1-0_0_2-large.json")
+	if err != nil {
+		b.Fatal(err)
+	}
+	ing := fakeIngester{}
+	b.StartTimer()
+
+	for i := 0; i < b.N; i++ {
+		if err := Processor002.ProcessSingle(bytes.NewReader(data), ing, &ProcessOptions{}); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+type fakeIngester struct{}
+
+func (i fakeIngester) Ingest(model.Samples) error {
+	return nil
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/textprocessor.go

@@ -0,0 +1,40 @@
+// Copyright 2014 The Prometheus Authors
+// 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 extraction
+
+import (
+	"io"
+
+	"github.com/prometheus/client_golang/text"
+)
+
+type processor004 struct{}
+
+// Processor004 s responsible for decoding payloads from the text based variety
+// of protocol version 0.0.4.
+var Processor004 = &processor004{}
+
+func (t *processor004) ProcessSingle(i io.Reader, out Ingester, o *ProcessOptions) error {
+	var parser text.Parser
+	metricFamilies, err := parser.TextToMetricFamilies(i)
+	if err != nil {
+		return err
+	}
+	for _, metricFamily := range metricFamilies {
+		if err := extractMetricFamily(out, o, metricFamily); err != nil {
+			return err
+		}
+	}
+	return nil
+}

Godeps/_workspace/src/github.com/prometheus/client_golang/extraction/textprocessor_test.go

@@ -0,0 +1,100 @@
+// Copyright 2014 The Prometheus Authors
+// 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 extraction
+
+import (
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/model"
+)
+
+var (
+	ts = model.Now()
+	in = `
+# Only a quite simple scenario with two metric families.
+# More complicated tests of the parser itself can be found in the text package.
+# TYPE mf2 counter
+mf2 3
+mf1{label="value1"} -3.14 123456
+mf1{label="value2"} 42
+mf2 4
+`
+	out = map[model.LabelValue]model.Samples{
+		"mf1": model.Samples{
+			&model.Sample{
+				Metric:    model.Metric{model.MetricNameLabel: "mf1", "label": "value1"},
+				Value:     -3.14,
+				Timestamp: 123456,
+			},
+			&model.Sample{
+				Metric:    model.Metric{model.MetricNameLabel: "mf1", "label": "value2"},
+				Value:     42,
+				Timestamp: ts,
+			},
+		},
+		"mf2": model.Samples{
+			&model.Sample{
+				Metric:    model.Metric{model.MetricNameLabel: "mf2"},
+				Value:     3,
+				Timestamp: ts,
+			},
+			&model.Sample{
+				Metric:    model.Metric{model.MetricNameLabel: "mf2"},
+				Value:     4,
+				Timestamp: ts,
+			},
+		},
+	}
+)
+
+type testIngester struct {
+	results []model.Samples
+}
+
+func (i *testIngester) Ingest(s model.Samples) error {
+	i.results = append(i.results, s)
+	return nil
+}
+
+func TestTextProcessor(t *testing.T) {
+	var ingester testIngester
+	i := strings.NewReader(in)
+	o := &ProcessOptions{
+		Timestamp: ts,
+	}
+
+	err := Processor004.ProcessSingle(i, &ingester, o)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if expected, got := len(out), len(ingester.results); expected != got {
+		t.Fatalf("Expected length %d, got %d", expected, got)
+	}
+	for _, r := range ingester.results {
+		expected, ok := out[r[0].Metric[model.MetricNameLabel]]
+		if !ok {
+			t.Fatalf(
+				"Unexpected metric name %q",
+				r[0].Metric[model.MetricNameLabel],
+			)
+		}
+		sort.Sort(expected)
+		sort.Sort(r)
+		if !expected.Equal(r) {
+			t.Errorf("expected %s, got %s", expected, r)
+		}
+	}
+}

test/e2e/kubelet_stats.go

@@ -33,7 +33,11 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/metrics"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	cadvisor "github.com/google/cadvisor/info/v1"
+
+	"github.com/prometheus/client_golang/extraction"
+	"github.com/prometheus/client_golang/model"
 )
 
 // KubeletMetric stores metrics scraped from the kubelet server's /metric endpoint.
@@ -56,96 +60,58 @@ func (a KubeletMetricByLatency) Len() int           { return len(a) }
 func (a KubeletMetricByLatency) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
 func (a KubeletMetricByLatency) Less(i, j int) bool { return a[i].Latency > a[j].Latency }
 
+// KubeletMetricIngester implements extraction.Ingester
+type kubeletMetricIngester []KubeletMetric
+
+func (k *kubeletMetricIngester) Ingest(samples model.Samples) error {
+	acceptedMethods := util.NewStringSet(
+		metrics.PodWorkerLatencyKey,
+		metrics.PodWorkerStartLatencyKey,
+		metrics.SyncPodsLatencyKey,
+		metrics.PodStartLatencyKey,
+		metrics.PodStatusLatencyKey,
+		metrics.ContainerManagerOperationsKey,
+		metrics.DockerOperationsKey,
+		metrics.DockerErrorsKey,
+	)
+
+	for _, sample := range samples {
+		const prefix = metrics.KubeletSubsystem + "_"
+		metricName := string(sample.Metric[model.MetricNameLabel])
+		if !strings.HasPrefix(metricName, prefix) {
+			// Not a kubelet metric.
+			continue
+		}
+
+		method := strings.TrimPrefix(metricName, prefix)
+		if !acceptedMethods.Has(method) {
+			continue
+		}
+
+		if method == metrics.DockerErrorsKey {
+			Logf("ERROR %v", sample)
+		}
+
+		latency := sample.Value
+		operation := string(sample.Metric["operation_type"])
+		var quantile float64
+		if val, ok := sample.Metric[model.QuantileLabel]; ok {
+			var err error
+			if quantile, err = strconv.ParseFloat(string(val), 64); err != nil {
+				continue
+			}
+		}
+
+		*k = append(*k, KubeletMetric{operation, method, quantile, time.Duration(int64(latency)) * time.Microsecond})
+	}
+	return nil
+}
+
 // ReadKubeletMetrics reads metrics from the kubelet server running on the given node
 func ParseKubeletMetrics(metricsBlob string) ([]KubeletMetric, error) {
-	metric := make([]KubeletMetric, 0)
-	for _, line := range strings.Split(metricsBlob, "\n") {
-
-		// A kubelet stats line starts with the KubeletSubsystem marker, followed by a stat name, followed by fields
-		// that vary by stat described on a case by case basis below.
-		// TODO: String parsing is such a hack, but getting our rest client/proxy to cooperate with prometheus
-		// client is weird, we should eventually invest some time in doing this the right way.
-		if !strings.HasPrefix(line, fmt.Sprintf("%v_", metrics.KubeletSubsystem)) {
-			continue
-		}
-		keyVal := strings.Split(line, " ")
-		if len(keyVal) != 2 {
-			return nil, fmt.Errorf("Error parsing metric %q", line)
-		}
-		keyElems := strings.Split(line, "\"")
-
-		latency, err := strconv.ParseFloat(keyVal[1], 64)
-		if err != nil {
-			continue
-		}
-
-		methodLine := strings.Split(keyElems[0], "{")
-		methodList := strings.Split(methodLine[0], "_")
-		if len(methodLine) != 2 || len(methodList) == 1 {
-			continue
-		}
-		method := strings.Join(methodList[1:], "_")
-
-		var operation, rawQuantile string
-		var quantile float64
-
-		switch method {
-		case metrics.PodWorkerLatencyKey:
-			// eg: kubelet_pod_worker_latency_microseconds{operation_type="create",pod_name="foopause3_default",quantile="0.99"} 1344
-			if len(keyElems) != 7 {
-				continue
-			}
-			operation = keyElems[1]
-			rawQuantile = keyElems[5]
-			break
-
-		case metrics.PodWorkerStartLatencyKey:
-			// eg: kubelet_pod_worker_start_latency_microseconds{quantile="0.99"} 12
-			fallthrough
-
-		case metrics.SyncPodsLatencyKey:
-			// eg:  kubelet_sync_pods_latency_microseconds{quantile="0.5"} 9949
-			fallthrough
-
-		case metrics.PodStartLatencyKey:
-			// eg: kubelet_pod_start_latency_microseconds{quantile="0.5"} 123
-			fallthrough
-
-		case metrics.PodStatusLatencyKey:
-			// eg: kubelet_generate_pod_status_latency_microseconds{quantile="0.5"} 12715
-			if len(keyElems) != 3 {
-				continue
-			}
-			operation = ""
-			rawQuantile = keyElems[1]
-			break
-
-		case metrics.ContainerManagerOperationsKey:
-			// eg: kubelet_container_manager_latency_microseconds{operation_type="SyncPod",quantile="0.5"} 6705
-			fallthrough
-
-		case metrics.DockerOperationsKey:
-			// eg: kubelet_docker_operations_latency_microseconds{operation_type="info",quantile="0.5"} 31590
-			if len(keyElems) != 5 {
-				continue
-			}
-			operation = keyElems[1]
-			rawQuantile = keyElems[3]
-			break
-
-		case metrics.DockerErrorsKey:
-			Logf("ERROR %v", line)
-
-		default:
-			continue
-		}
-		quantile, err = strconv.ParseFloat(rawQuantile, 64)
-		if err != nil {
-			continue
-		}
-		metric = append(metric, KubeletMetric{operation, method, quantile, time.Duration(int64(latency)) * time.Microsecond})
-	}
-	return metric, nil
+	var ingester kubeletMetricIngester
+	err := extraction.Processor004.ProcessSingle(strings.NewReader(metricsBlob), &ingester, &extraction.ProcessOptions{})
+	return ingester, err
 }
 
 // HighLatencyKubeletOperations logs and counts the high latency metrics exported by the kubelet server via /metrics.

test/e2e/util.go

@@ -49,6 +49,8 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 
 	"github.com/davecgh/go-spew/spew"
+	"github.com/prometheus/client_golang/extraction"
+	"github.com/prometheus/client_golang/model"
 	"golang.org/x/crypto/ssh"
 
 	. "github.com/onsi/ginkgo"
@@ -1582,6 +1584,34 @@ type LatencyMetric struct {
 	Latency  time.Duration
 }
 
+// latencyMetricIngestor implements extraction.Ingester
+type latencyMetricIngester []LatencyMetric
+
+func (l *latencyMetricIngester) Ingest(samples model.Samples) error {
+	for _, sample := range samples {
+		// Example line:
+		// apiserver_request_latencies_summary{resource="namespaces",verb="LIST",quantile="0.99"} 908
+		if sample.Metric[model.MetricNameLabel] != "apiserver_request_latencies_summary" {
+			continue
+		}
+
+		resource := string(sample.Metric["resource"])
+		verb := string(sample.Metric["verb"])
+		latency := sample.Value
+		quantile, err := strconv.ParseFloat(string(sample.Metric[model.QuantileLabel]), 64)
+		if err != nil {
+			return err
+		}
+		*l = append(*l, LatencyMetric{
+			verb,
+			resource,
+			quantile,
+			time.Duration(int64(latency)) * time.Microsecond,
+		})
+	}
+	return nil
+}
+
 // LatencyMetricByLatency implements sort.Interface for []LatencyMetric based on
 // the latency field.
 type LatencyMetricByLatency []LatencyMetric
@@ -1595,34 +1625,9 @@ func ReadLatencyMetrics(c *client.Client) ([]LatencyMetric, error) {
 	if err != nil {
 		return nil, err
 	}
-	metrics := make([]LatencyMetric, 0)
-	for _, line := range strings.Split(string(body), "\n") {
-		if strings.HasPrefix(line, "apiserver_request_latencies_summary{") {
-			// Example line:
-			// apiserver_request_latencies_summary{resource="namespaces",verb="LIST",quantile="0.99"} 908
-			// TODO: This parsing code is long and not readable. We should improve it.
-			keyVal := strings.Split(line, " ")
-			if len(keyVal) != 2 {
-				return nil, fmt.Errorf("Error parsing metric %q", line)
-			}
-			keyElems := strings.Split(line, "\"")
-			if len(keyElems) != 7 {
-				return nil, fmt.Errorf("Error parsing metric %q", line)
-			}
-			resource := keyElems[1]
-			verb := keyElems[3]
-			quantile, err := strconv.ParseFloat(keyElems[5], 64)
-			if err != nil {
-				return nil, fmt.Errorf("Error parsing metric %q", line)
-			}
-			latency, err := strconv.ParseFloat(keyVal[1], 64)
-			if err != nil {
-				return nil, fmt.Errorf("Error parsing metric %q", line)
-			}
-			metrics = append(metrics, LatencyMetric{verb, resource, quantile, time.Duration(int64(latency)) * time.Microsecond})
-		}
-	}
-	return metrics, nil
+	var ingester latencyMetricIngester
+	err = extraction.Processor004.ProcessSingle(strings.NewReader(body), &ingester, &extraction.ProcessOptions{})
+	return ingester, err
 }
 
 // Prints summary metrics for request types with latency above threshold