Merge pull request #1197 from iksaif/master

Add support for remote storage on Graphite

cmd/prometheus/config.go

@@ -163,6 +163,18 @@ func init() {
 	)
 
 	// Remote storage.
+	cfg.fs.StringVar(
+		&cfg.remote.GraphiteAddress, "storage.remote.graphite-address", "",
+		"The host:port of the remote Graphite server to send samples to. None, if empty.",
+	)
+	cfg.fs.StringVar(
+		&cfg.remote.GraphiteTransport, "storage.remote.graphite-transport", "tcp",
+		"Transport protocol to use to communicate with Graphite. 'tcp', if empty.",
+	)
+	cfg.fs.StringVar(
+		&cfg.remote.GraphitePrefix, "storage.remote.graphite-prefix", "",
+		"The prefix to prepend to all metrics exported to Graphite. None, if empty.",
+	)
 	cfg.fs.StringVar(
 		&cfg.remote.OpentsdbURL, "storage.remote.opentsdb-url", "",
 		"The URL of the remote OpenTSDB server to send samples to. None, if empty.",

storage/remote/graphite/client.go

@@ -0,0 +1,107 @@
+// Copyright 2015 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 graphite
+
+import (
+	"bytes"
+	"fmt"
+	"math"
+	"net"
+	"sort"
+	"time"
+
+	"github.com/prometheus/common/log"
+	"github.com/prometheus/common/model"
+)
+
+// Client allows sending batches of Prometheus samples to Graphite.
+type Client struct {
+	address   string
+	transport string
+	timeout   time.Duration
+	prefix    string
+}
+
+// NewClient creates a new Client.
+func NewClient(address string, transport string, timeout time.Duration, prefix string) *Client {
+	return &Client{
+		address:   address,
+		transport: transport,
+		timeout:   timeout,
+		prefix:    prefix,
+	}
+}
+
+func pathFromMetric(m model.Metric, prefix string) string {
+	var buffer bytes.Buffer
+
+	buffer.WriteString(prefix)
+	buffer.WriteString(escape(m[model.MetricNameLabel]))
+
+	// We want to sort the labels.
+	labels := make(model.LabelNames, 0, len(m))
+	for l, _ := range m {
+		labels = append(labels, l)
+	}
+	sort.Sort(labels)
+
+	// For each label, in order, add ".<label>=<value>".
+	for _, l := range labels {
+		v := m[l]
+
+		if l == model.MetricNameLabel || len(l) == 0 {
+			continue
+		}
+		// Here we use '=' instead of '.' to be able
+		// to later read back the value correctly. Using '.' would
+		// not allow to distinguish labels from values.
+		buffer.WriteString(fmt.Sprintf(
+			".%s=%s", string(l), escape(v)))
+	}
+	return buffer.String()
+}
+
+// Store sends a batch of samples to Graphite.
+func (c *Client) Store(samples model.Samples) error {
+	conn, err := net.DialTimeout(c.transport, c.address, c.timeout)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	var buf bytes.Buffer
+	for _, s := range samples {
+		k := pathFromMetric(s.Metric, c.prefix)
+		t := float64(s.Timestamp.UnixNano()) / 1e9
+		v := float64(s.Value)
+		if math.IsNaN(v) || math.IsInf(v, 0) {
+			log.Warnf("cannot send value %f to Graphite,"+
+				"skipping sample %#v", v, s)
+			continue
+		}
+		fmt.Fprintf(&buf, "%s %f %f\n", k, v, t)
+	}
+
+	_, err = conn.Write(buf.Bytes())
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Name identifies the client as a Graphite client.
+func (c Client) Name() string {
+	return "graphite"
+}

storage/remote/graphite/client_test.go

@@ -0,0 +1,57 @@
+// Copyright 2015 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 graphite
+
+import (
+	"testing"
+
+	"github.com/prometheus/common/model"
+)
+
+var (
+	metric = model.Metric{
+		model.MetricNameLabel: "test:metric",
+		"testlabel":           "test:value",
+		"many_chars":          "abc!ABC:012-3!45ö67~89./(){},=.\"\\",
+	}
+)
+
+func TestEscape(t *testing.T) {
+	// Can we correctly keep and escape valid chars.
+	value := "abzABZ019(){},'\"\\"
+	expected := "abzABZ019\\(\\)\\{\\}\\,\\'\\\"\\\\"
+	actual := escape(model.LabelValue(value))
+	if expected != actual {
+		t.Errorf("Expected %s, got %s", expected, actual)
+	}
+
+	// Test percent-encoding.
+	value = "é/|_;:%."
+	expected = "%C3%A9%2F|_;:%25%2E"
+	actual = escape(model.LabelValue(value))
+	if expected != actual {
+		t.Errorf("Expected %s, got %s", expected, actual)
+	}
+}
+
+func TestPathFromMetric(t *testing.T) {
+	expected := ("prefix." +
+		"test:metric" +
+		".many_chars=abc!ABC:012-3!45%C3%B667~89%2E%2F\\(\\)\\{\\}\\,%3D%2E\\\"\\\\" +
+		".testlabel=test:value")
+	actual := pathFromMetric(metric, "prefix.")
+	if expected != actual {
+		t.Errorf("Expected %s, got %s", expected, actual)
+	}
+}

storage/remote/graphite/escape.go

@@ -0,0 +1,103 @@
+// Copyright 2015 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 graphite
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+
+	"github.com/prometheus/common/model"
+)
+
+const (
+	// From https://github.com/graphite-project/graphite-web/blob/master/webapp/graphite/render/grammar.py#L83
+	symbols    = "(){},=.'\"\\"
+	printables = ("0123456789abcdefghijklmnopqrstuvwxyz" +
+		"ABCDEFGHIJKLMNOPQRSTUVWXYZ" +
+		"!\"#$%&\\'()*+,-./:;<=>?@[\\]^_`{|}~")
+)
+
+// Graphite doesn't support tags, so label names and values must be
+// encoded into the metric path. The list of characters that are usable
+// with Graphite is rather fuzzy. One 'source of truth' might be the grammar
+// used to parse requests in the webapp:
+// https://github.com/graphite-project/graphite-web/blob/master/webapp/graphite/render/grammar.py#L83
+// The list of valid symbols is defined as:
+//   legal = printables - symbols + escaped(symbols)
+//
+// The default storage backend for Graphite (whisper) stores data
+// in filenames, so we also need to use only valid filename characters.
+// Fortunately on UNIX only '/' isn't, and Windows is completely unsupported
+// by Graphite: http://graphite.readthedocs.org/en/latest/install.html#windows-users
+
+// escape escapes a model.LabelValue into runes allowed in Graphite. The runes
+// allowed in Graphite are all single-byte. This function encodes the arbitrary
+// byte sequence found in this TagValue in way very similar to the traditional
+// percent-encoding (https://en.wikipedia.org/wiki/Percent-encoding):
+//
+// - The string that underlies TagValue is scanned byte by byte.
+//
+// - If a byte represents a legal Graphite rune with the exception of '%', '/',
+//   '=' and '.', that byte is directly copied to the resulting byte slice.
+//   % is used for percent-encoding of other bytes.
+//   / is not usable in filenames.
+//   = is used when generating the path to associate values to labels.
+//   . already means something for Graphite and thus can't be used in a value.
+//
+// - If the byte is any of (){},=.'"\, then a '\' will be prepended to it. We
+//   do not percent-encode them since they are explicitly usable in this
+//   way in Graphite.
+//
+// - All other bytes are replaced by '%' followed by two bytes containing the
+//   uppercase ASCII representation of their hexadecimal value.
+//
+// This encoding allows to save arbitrary Go strings in Graphite. That's
+// required because Prometheus label values can contain anything. Using
+// percent encoding makes it easy to unescape, even in javascript.
+//
+// Examples:
+//
+// "foo-bar-42" -> "foo-bar-42"
+//
+// "foo_bar%42" -> "foo_bar%2542"
+//
+// "http://example.org:8080" -> "http:%2F%2Fexample%2Eorg:8080"
+//
+// "Björn's email: bjoern@soundcloud.com" ->
+// "Bj%C3%B6rn's%20email:%20bjoern%40soundcloud.com"
+//
+// "日" -> "%E6%97%A5"
+func escape(tv model.LabelValue) string {
+	length := len(tv)
+	result := bytes.NewBuffer(make([]byte, 0, length))
+	for i := 0; i < length; i++ {
+		b := tv[i]
+		switch {
+		// . is reserved by graphite, % is used to escape other bytes.
+		case b == '.' || b == '%' || b == '/' || b == '=':
+			fmt.Fprintf(result, "%%%X", b)
+		// These symbols are ok only if backslash escaped.
+		case strings.IndexByte(symbols, b) != -1:
+			result.WriteString("\\" + string(b))
+		// These are all fine.
+		case strings.IndexByte(printables, b) != -1:
+			result.WriteByte(b)
+		// Defaults to percent-encoding.
+		default:
+			fmt.Fprintf(result, "%%%X", b)
+		}
+	}
+	return result.String()
+}

storage/remote/remote.go

@@ -24,6 +24,7 @@ import (
 	influx "github.com/influxdb/influxdb/client"
 
 	"github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/storage/remote/graphite"
 	"github.com/prometheus/prometheus/storage/remote/influxdb"
 	"github.com/prometheus/prometheus/storage/remote/opentsdb"
 )
@@ -48,6 +49,12 @@ func (s *Storage) ApplyConfig(conf *config.Config) bool {
 // New returns a new remote Storage.
 func New(o *Options) *Storage {
 	s := &Storage{}
+	if o.GraphiteAddress != "" {
+		c := graphite.NewClient(
+			o.GraphiteAddress, o.GraphiteTransport,
+			o.StorageTimeout, o.GraphitePrefix)
+		s.queues = append(s.queues, NewStorageQueueManager(c, 100*1024))
+	}
 	if o.OpentsdbURL != "" {
 		c := opentsdb.NewClient(o.OpentsdbURL, o.StorageTimeout)
 		s.queues = append(s.queues, NewStorageQueueManager(c, 100*1024))
@@ -78,6 +85,9 @@ type Options struct {
 	InfluxdbPassword        string
 	InfluxdbDatabase        string
 	OpentsdbURL             string
+	GraphiteAddress         string
+	GraphiteTransport       string
+	GraphitePrefix          string
 }
 
 // Run starts the background processing of the storage queues.