From 041c5af905538cc6cdf0c1a88c46776b81fad0e8 Mon Sep 17 00:00:00 2001 From: Jimmi Dyson Date: Tue, 20 Sep 2016 12:46:15 +0100 Subject: [PATCH] Switch to Prometheus decoder --- collector/prometheus_collector.go | 151 ++++++++++++------------- collector/prometheus_collector_test.go | 69 +++++++---- info/v1/metric.go | 6 +- manager/manager_test.go | 3 +- 4 files changed, 123 insertions(+), 106 deletions(-) diff --git a/collector/prometheus_collector.go b/collector/prometheus_collector.go index 37ee44c8..a9a57259 100644 --- a/collector/prometheus_collector.go +++ b/collector/prometheus_collector.go @@ -17,13 +17,14 @@ package collector import ( "encoding/json" "fmt" - "io/ioutil" - "math" + "io" "net/http" - "strconv" - "strings" "time" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" + "github.com/google/cadvisor/container" "github.com/google/cadvisor/info/v1" ) @@ -100,62 +101,63 @@ func (collector *PrometheusCollector) Name() string { return collector.name } -func getMetricData(line string) string { - fields := strings.Fields(line) - data := fields[3] - if len(fields) > 4 { - for i := range fields { - if i > 3 { - data = data + "_" + fields[i] - } - } - } - return strings.TrimSpace(data) -} - func (collector *PrometheusCollector) GetSpec() []v1.MetricSpec { - specs := []v1.MetricSpec{} response, err := collector.httpClient.Get(collector.configFile.Endpoint.URL) if err != nil { - return specs + return nil } defer response.Body.Close() - pageContent, err := ioutil.ReadAll(response.Body) - if err != nil { - return specs + if response.StatusCode != http.StatusOK { + return nil } - lines := strings.Split(string(pageContent), "\n") - lineCount := len(lines) - for i, line := range lines { - if strings.HasPrefix(line, "# HELP") { - if i+2 >= lineCount { - break - } + dec := expfmt.NewDecoder(response.Body, expfmt.ResponseFormat(response.Header)) - stopIndex := strings.IndexAny(lines[i+2], "{ ") - if stopIndex == -1 { - continue - } + var specs []v1.MetricSpec - name := strings.TrimSpace(lines[i+2][0:stopIndex]) - if _, ok := collector.metricsSet[name]; collector.metricsSet != nil && !ok { - continue - } - spec := v1.MetricSpec{ - Name: name, - Type: v1.MetricType(getMetricData(lines[i+1])), - Format: "float", - Units: getMetricData(lines[i]), - } - specs = append(specs, spec) + for { + d := dto.MetricFamily{} + if err = dec.Decode(&d); err != nil { + break } + name := d.GetName() + if len(name) == 0 { + continue + } + if _, ok := collector.metricsSet[name]; collector.metricsSet != nil && !ok { + continue + } + spec := v1.MetricSpec{ + Name: name, + Type: metricType(d.GetType()), + Format: v1.FloatType, + } + specs = append(specs, spec) } + + if err != nil && err != io.EOF { + return nil + } + return specs } +// metricType converts Prometheus metric type to cadvisor metric type. +// If there is no mapping then just return the name of the Prometheus metric type. +func metricType(t dto.MetricType) v1.MetricType { + switch t { + case dto.MetricType_COUNTER: + return v1.MetricCumulative + case dto.MetricType_GAUGE: + return v1.MetricGauge + default: + return v1.MetricType(t.String()) + } + +} + //Returns collected metrics and the next collection time of the collector func (collector *PrometheusCollector) Collect(metrics map[string][]v1.MetricVal) (time.Time, map[string][]v1.MetricVal, error) { currentTime := time.Now() @@ -168,59 +170,54 @@ func (collector *PrometheusCollector) Collect(metrics map[string][]v1.MetricVal) } defer response.Body.Close() - pageContent, err := ioutil.ReadAll(response.Body) - if err != nil { - return nextCollectionTime, nil, err + if response.StatusCode != http.StatusOK { + return nextCollectionTime, nil, fmt.Errorf("server returned HTTP status %s", response.Status) } - var errorSlice []error - lines := strings.Split(string(pageContent), "\n") + sdec := expfmt.SampleDecoder{ + Dec: expfmt.NewDecoder(response.Body, expfmt.ResponseFormat(response.Header)), + Opts: &expfmt.DecodeOptions{ + Timestamp: model.TimeFromUnixNano(currentTime.UnixNano()), + }, + } - newMetrics := make(map[string][]v1.MetricVal) - - for _, line := range lines { - if line == "" { + var ( + decSamples = make(model.Vector, 0, 50) + newMetrics = make(map[string][]v1.MetricVal) + ) + for { + if err = sdec.Decode(&decSamples); err != nil { break } - if !strings.HasPrefix(line, "# HELP") && !strings.HasPrefix(line, "# TYPE") { - var metLabel string - startLabelIndex := strings.Index(line, "{") - spaceIndex := strings.Index(line, " ") - if startLabelIndex == -1 { - startLabelIndex = spaceIndex - } - metName := strings.TrimSpace(line[0:startLabelIndex]) + for _, sample := range decSamples { + metName := string(sample.Metric[model.MetricNameLabel]) + if len(metName) == 0 { + continue + } if _, ok := collector.metricsSet[metName]; collector.metricsSet != nil && !ok { continue } - if startLabelIndex+1 <= spaceIndex-1 { - metLabel = strings.TrimSpace(line[(startLabelIndex + 1):(spaceIndex - 1)]) - } - - metVal, err := strconv.ParseFloat(line[spaceIndex+1:], 64) - if err != nil { - errorSlice = append(errorSlice, err) - } - if math.IsNaN(metVal) { - metVal = 0 - } - metric := v1.MetricVal{ - Label: metLabel, - FloatValue: metVal, - Timestamp: currentTime, + FloatValue: float64(sample.Value), + Timestamp: sample.Timestamp.Time(), } newMetrics[metName] = append(newMetrics[metName], metric) if len(newMetrics) > collector.metricCountLimit { return nextCollectionTime, nil, fmt.Errorf("too many metrics to collect") } } + decSamples = decSamples[:0] } + + if err != nil && err != io.EOF { + return nextCollectionTime, nil, err + } + for key, val := range newMetrics { metrics[key] = append(metrics[key], val...) } - return nextCollectionTime, metrics, compileErrors(errorSlice) + return nextCollectionTime, metrics, nil } diff --git a/collector/prometheus_collector_test.go b/collector/prometheus_collector_test.go index a31cee66..4285a4dc 100644 --- a/collector/prometheus_collector_test.go +++ b/collector/prometheus_collector_test.go @@ -36,21 +36,26 @@ func TestPrometheus(t *testing.T) { containerHandler := containertest.NewMockContainerHandler("mockContainer") collector, err := NewPrometheusCollector("Prometheus", configFile, 100, containerHandler, http.DefaultClient) assert.NoError(err) - assert.Equal(collector.name, "Prometheus") - assert.Equal(collector.configFile.Endpoint.URL, "http://localhost:8080/metrics") + assert.Equal("Prometheus", collector.name) + assert.Equal("http://localhost:8080/metrics", collector.configFile.Endpoint.URL) tempServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - text := "# HELP go_gc_duration_seconds A summary of the GC invocation durations.\n" - text += "# TYPE go_gc_duration_seconds summary\n" - text += "go_gc_duration_seconds{quantile=\"0\"} 5.8348000000000004e-05\n" - text += "go_gc_duration_seconds{quantile=\"1\"} 0.000499764\n" - text += "# HELP go_goroutines Number of goroutines that currently exist.\n" - text += "# TYPE go_goroutines gauge\n" - text += "go_goroutines 16\n" - text += "# HELP empty_metric A metric without any values\n" - text += "# TYPE empty_metric counter\n" - text += "\n" + text := `# HELP go_gc_duration_seconds A summary of the GC invocation durations. +# TYPE go_gc_duration_seconds summary +go_gc_duration_seconds{quantile="0"} 5.8348000000000004e-05 +go_gc_duration_seconds{quantile="1"} 0.000499764 +go_gc_duration_seconds_sum 1.7560473e+07 +go_gc_duration_seconds_count 2693 +# HELP go_goroutines Number of goroutines that currently exist. +# TYPE go_goroutines gauge +go_goroutines 16 +# HELP empty_metric A metric without any values +# TYPE empty_metric counter +# HELP metric_with_spaces_in_label A metric with spaces in a label. +# TYPE metric_with_spaces_in_label gauge +metric_with_spaces_in_label{name="Network Agent"} 72 +` fmt.Fprintln(w, text) })) @@ -60,9 +65,17 @@ func TestPrometheus(t *testing.T) { var spec []v1.MetricSpec require.NotPanics(t, func() { spec = collector.GetSpec() }) - assert.Len(spec, 2) - assert.Equal(spec[0].Name, "go_gc_duration_seconds") - assert.Equal(spec[1].Name, "go_goroutines") + assert.Len(spec, 3) + specNames := make(map[string]struct{}, 3) + for _, s := range spec { + specNames[s.Name] = struct{}{} + } + expectedSpecNames := map[string]struct{}{ + "go_gc_duration_seconds": {}, + "go_goroutines": {}, + "metric_with_spaces_in_label": {}, + } + assert.Equal(expectedSpecNames, specNames) metrics := map[string][]v1.MetricVal{} _, metrics, errMetric := collector.Collect(metrics) @@ -70,11 +83,14 @@ func TestPrometheus(t *testing.T) { assert.NoError(errMetric) go_gc_duration := metrics["go_gc_duration_seconds"] - assert.Equal(go_gc_duration[0].FloatValue, 5.8348000000000004e-05) - assert.Equal(go_gc_duration[1].FloatValue, 0.000499764) + assert.Equal(5.8348000000000004e-05, go_gc_duration[0].FloatValue) + assert.Equal(0.000499764, go_gc_duration[1].FloatValue) goRoutines := metrics["go_goroutines"] - assert.Equal(goRoutines[0].FloatValue, 16) + assert.Equal(16, goRoutines[0].FloatValue) + + metricWithSpaces := metrics["metric_with_spaces_in_label"] + assert.Equal(72, metricWithSpaces[0].FloatValue) } func TestPrometheusEndpointConfig(t *testing.T) { @@ -158,13 +174,16 @@ func TestPrometheusFiltersMetrics(t *testing.T) { tempServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - text := "# HELP go_gc_duration_seconds A summary of the GC invocation durations.\n" - text += "# TYPE go_gc_duration_seconds summary\n" - text += "go_gc_duration_seconds{quantile=\"0\"} 5.8348000000000004e-05\n" - text += "go_gc_duration_seconds{quantile=\"1\"} 0.000499764\n" - text += "# HELP go_goroutines Number of goroutines that currently exist.\n" - text += "# TYPE go_goroutines gauge\n" - text += "go_goroutines 16" + text := `# HELP go_gc_duration_seconds A summary of the GC invocation durations. +# TYPE go_gc_duration_seconds summary +go_gc_duration_seconds{quantile="0"} 5.8348000000000004e-05 +go_gc_duration_seconds{quantile="1"} 0.000499764 +go_gc_duration_seconds_sum 1.7560473e+07 +go_gc_duration_seconds_count 2693 +# HELP go_goroutines Number of goroutines that currently exist. +# TYPE go_goroutines gauge +go_goroutines 16 +` fmt.Fprintln(w, text) })) diff --git a/info/v1/metric.go b/info/v1/metric.go index 90fd9e49..30c23ed1 100644 --- a/info/v1/metric.go +++ b/info/v1/metric.go @@ -26,10 +26,10 @@ const ( MetricGauge MetricType = "gauge" // A counter-like value that is only expected to increase. - MetricCumulative = "cumulative" + MetricCumulative MetricType = "cumulative" // Rate over a time period. - MetricDelta = "delta" + MetricDelta MetricType = "delta" ) // DataType for metric being exported. @@ -37,7 +37,7 @@ type DataType string const ( IntType DataType = "int" - FloatType = "float" + FloatType DataType = "float" ) // Spec for custom metric. diff --git a/manager/manager_test.go b/manager/manager_test.go index 2229de9b..efa33816 100644 --- a/manager/manager_test.go +++ b/manager/manager_test.go @@ -23,6 +23,8 @@ import ( "testing" "time" + "net/http" + "github.com/google/cadvisor/cache/memory" "github.com/google/cadvisor/collector" "github.com/google/cadvisor/container" @@ -33,7 +35,6 @@ import ( "github.com/google/cadvisor/info/v2" "github.com/google/cadvisor/utils/sysfs/fakesysfs" "github.com/stretchr/testify/assert" - "net/http" ) // TODO(vmarmol): Refactor these tests.