dom/telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Create a template system for the graphite serializer

Browse Source 
closes #925
closes #879
This commit is contained in:
Cameron Sparr
2016-04-08 16:04:45 -06:00
parent 27fe4f7062
commit f5878eafb9
12 changed files with 401 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
plugins/outputs/graphite/README.md
View File

@@ -1,13 +1,34 @@
# Graphite Output Plugin
This plugin writes to [Graphite](http://graphite.readthedocs.org/en/latest/index.html) via raw TCP.
This plugin writes to [Graphite](http://graphite.readthedocs.org/en/latest/index.html)
via raw TCP.
## Configuration:
```toml
# Configuration for Graphite server to send metrics to
[[outputs.graphite]]
## TCP endpoint for your graphite instance.
servers = ["localhost:2003"]
## Prefix metrics name
prefix = ""
## Graphite output template
## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
template = "host.tags.measurement.field"
## timeout in seconds for the write connection to graphite
timeout = 2
```
Parameters:
Servers []string
Prefix string
Timeout int
Servers []string
Prefix string
Timeout int
Template string
* `servers`: List of strings, ["mygraphiteserver:2003"].
* `prefix`: String use to prefix all sent metrics.
* `timeout`: Connection timeout in second.
* `timeout`: Connection timeout in seconds.
* `template`: Template for graphite output format, see
https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
for more details.

14
plugins/outputs/graphite/graphite.go
View File

@@ -16,10 +16,11 @@ import (
type Graphite struct {
// URL is only for backwards compatability
Servers []string
Prefix string
Timeout int
conns []net.Conn
Servers []string
Prefix string
Template string
Timeout int
conns []net.Conn
}
var sampleConfig = `
@@ -27,6 +28,9 @@ var sampleConfig = `
servers = ["localhost:2003"]
## Prefix metrics name
prefix = ""
## Graphite output template
## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
template = "host.tags.measurement.field"
## timeout in seconds for the write connection to graphite
timeout = 2
`
@@ -72,7 +76,7 @@ func (g *Graphite) Description() string {
func (g *Graphite) Write(metrics []telegraf.Metric) error {
// Prepare data
var bp []string
s, err := serializers.NewGraphiteSerializer(g.Prefix)
s, err := serializers.NewGraphiteSerializer(g.Prefix, g.Template)
if err != nil {
return err
}

8
plugins/outputs/graphite/graphite_test.go
View File

@@ -54,7 +54,7 @@ func TestGraphiteOK(t *testing.T) {
m1, _ := telegraf.NewMetric(
"mymeasurement",
map[string]string{"host": "192.168.0.1"},
map[string]interface{}{"mymeasurement": float64(3.14)},
map[string]interface{}{"myfield": float64(3.14)},
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
)
m2, _ := telegraf.NewMetric(
@@ -90,10 +90,10 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup) {
reader := bufio.NewReader(conn)
tp := textproto.NewReader(reader)
data1, _ := tp.ReadLine()
assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement 3.14 1289430000", data1)
assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement.myfield 3.14 1289430000", data1)
data2, _ := tp.ReadLine()
assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement.value 3.14 1289430000", data2)
assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement 3.14 1289430000", data2)
data3, _ := tp.ReadLine()
assert.Equal(t, "my.prefix.192_168_0_1.my_measurement.value 3.14 1289430000", data3)
assert.Equal(t, "my.prefix.192_168_0_1.my_measurement 3.14 1289430000", data3)
conn.Close()
}

25
plugins/outputs/librato/librato.go
View File

@@ -21,6 +21,7 @@ type Librato struct {
NameFromTags bool
SourceTag string
Timeout internal.Duration
Template string
apiUrl string
client *http.Client
@@ -29,22 +30,20 @@ type Librato struct {
var sampleConfig = `
## Librator API Docs
## http://dev.librato.com/v1/metrics-authentication
## Librato API user
api_user = "telegraf@influxdb.com" # required.
## Librato API token
api_token = "my-secret-token" # required.
### Debug
## Debug
# debug = false
### Tag Field to populate source attribute (optional)
### This is typically the _hostname_ from which the metric was obtained.
## Tag Field to populate source attribute (optional)
## This is typically the _hostname_ from which the metric was obtained.
source_tag = "host"
## Connection timeout.
# timeout = "5s"
## Output Name Template (same as graphite buckets)
## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md#graphite
template = "host.tags.measurement.field"
`
type LMetrics struct {
@@ -152,17 +151,13 @@ func (l *Librato) Description() string {
return "Configuration for Librato API to send metrics to."
}
func (l *Librato) buildGaugeName(m telegraf.Metric, fieldName string) string {
// Use the GraphiteSerializer
graphiteSerializer := graphite.GraphiteSerializer{}
return graphiteSerializer.SerializeBucketName(m, fieldName)
}
func (l *Librato) buildGauges(m telegraf.Metric) ([]*Gauge, error) {
gauges := []*Gauge{}
serializer := graphite.GraphiteSerializer{Template: l.Template}
bucket := serializer.SerializeBucketName(m.Name(), m.Tags())
for fieldName, value := range m.Fields() {
gauge := &Gauge{
Name: l.buildGaugeName(m, fieldName),
Name: graphite.InsertField(bucket, fieldName),
MeasureTime: m.Time().Unix(),
}
if !gauge.verifyValue(value) {

16
plugins/outputs/librato/librato_test.go
View File

@@ -86,7 +86,7 @@ func TestBuildGauge(t *testing.T) {
{
testutil.TestMetric(0.0, "test1"),
&Gauge{
Name: "value1.test1.value",
Name: "value1.test1",
MeasureTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 0.0,
},
@@ -95,7 +95,7 @@ func TestBuildGauge(t *testing.T) {
{
testutil.TestMetric(1.0, "test2"),
&Gauge{
Name: "value1.test2.value",
Name: "value1.test2",
MeasureTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 1.0,
},
@@ -104,7 +104,7 @@ func TestBuildGauge(t *testing.T) {
{
testutil.TestMetric(10, "test3"),
&Gauge{
Name: "value1.test3.value",
Name: "value1.test3",
MeasureTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 10.0,
},
@@ -113,7 +113,7 @@ func TestBuildGauge(t *testing.T) {
{
testutil.TestMetric(int32(112345), "test4"),
&Gauge{
Name: "value1.test4.value",
Name: "value1.test4",
MeasureTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 112345.0,
},
@@ -122,7 +122,7 @@ func TestBuildGauge(t *testing.T) {
{
testutil.TestMetric(int64(112345), "test5"),
&Gauge{
Name: "value1.test5.value",
Name: "value1.test5",
MeasureTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 112345.0,
},
@@ -131,7 +131,7 @@ func TestBuildGauge(t *testing.T) {
{
testutil.TestMetric(float32(11234.5), "test6"),
&Gauge{
Name: "value1.test6.value",
Name: "value1.test6",
MeasureTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 11234.5,
},
@@ -189,7 +189,7 @@ func TestBuildGaugeWithSource(t *testing.T) {
{
pt1,
&Gauge{
Name: "192_168_0_1.value1.test1.value",
Name: "192_168_0_1.value1.test1",
MeasureTime: time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 0.0,
Source: "192.168.0.1",
@@ -199,7 +199,7 @@ func TestBuildGaugeWithSource(t *testing.T) {
{
pt2,
&Gauge{
Name: "192_168_0_1.value1.test1.value",
Name: "192_168_0_1.value1.test1",
MeasureTime: time.Date(2010, time.December, 10, 23, 0, 0, 0, time.UTC).Unix(),
Value: 1.0,
},

127
plugins/serializers/graphite/graphite.go
View File

@@ -8,11 +8,16 @@ import (
"github.com/influxdata/telegraf"
)
const DEFAULT_TEMPLATE = "host.tags.measurement.field"
var fieldDeleter = strings.NewReplacer(".FIELDNAME", "", "FIELDNAME.", "")
type GraphiteSerializer struct {
Prefix string
Prefix string
Template string
}
var sanitizedChars = strings.NewReplacer("/", "-", "@", "-", " ", "_")
var sanitizedChars = strings.NewReplacer("/", "-", "@", "-", " ", "_", "..", ".")
func (s *GraphiteSerializer) Serialize(metric telegraf.Metric) ([]string, error) {
out := []string{}
@@ -20,65 +25,95 @@ func (s *GraphiteSerializer) Serialize(metric telegraf.Metric) ([]string, error)
// Convert UnixNano to Unix timestamps
timestamp := metric.UnixNano() / 1000000000
for field_name, value := range metric.Fields() {
// Convert value
value_str := fmt.Sprintf("%#v", value)
// Write graphite metric
var graphitePoint string
graphitePoint = fmt.Sprintf("%s %s %d",
s.SerializeBucketName(metric, field_name),
value_str,
bucket := s.SerializeBucketName(metric.Name(), metric.Tags())
for fieldName, value := range metric.Fields() {
// Convert value to string
valueS := fmt.Sprintf("%#v", value)
point := fmt.Sprintf("%s %s %d",
// insert "field" section of template
InsertField(bucket, fieldName),
valueS,
timestamp)
out = append(out, graphitePoint)
out = append(out, point)
}
return out, nil
}
func (s *GraphiteSerializer) SerializeBucketName(metric telegraf.Metric, field_name string) string {
// Get the metric name
name := metric.Name()
// Convert UnixNano to Unix timestamps
tag_str := buildTags(metric)
// Write graphite metric
var serializedBucketName string
if name == field_name {
serializedBucketName = fmt.Sprintf("%s.%s",
tag_str,
strings.Replace(name, ".", "_", -1))
} else {
serializedBucketName = fmt.Sprintf("%s.%s.%s",
tag_str,
strings.Replace(name, ".", "_", -1),
strings.Replace(field_name, ".", "_", -1))
// SerializeBucketName will take the given measurement name and tags and
// produce a graphite bucket. It will use the GraphiteSerializer.Template
// to generate this, or DEFAULT_TEMPLATE.
//
// NOTE: SerializeBucketName replaces the "field" portion of the template with
// FIELDNAME. It is up to the user to replace this. This is so that
// SerializeBucketName can be called just once per measurement, rather than
// once per field. See GraphiteSerializer.InsertField() function.
func (s *GraphiteSerializer) SerializeBucketName(
measurement string,
tags map[string]string,
) string {
if s.Template == "" {
s.Template = DEFAULT_TEMPLATE
}
if s.Prefix != "" {
serializedBucketName = fmt.Sprintf("%s.%s", s.Prefix, serializedBucketName)
tagsCopy := make(map[string]string)
for k, v := range tags {
tagsCopy[k] = v
}
return serializedBucketName
var out []string
templateParts := strings.Split(s.Template, ".")
for _, templatePart := range templateParts {
switch templatePart {
case "measurement":
out = append(out, measurement)
case "tags":
// we will replace this later
out = append(out, "TAGS")
case "field":
// user of SerializeBucketName needs to replace this
out = append(out, "FIELDNAME")
default:
// This is a tag being applied
if tagvalue, ok := tagsCopy[templatePart]; ok {
out = append(out, strings.Replace(tagvalue, ".", "_", -1))
delete(tagsCopy, templatePart)
}
}
}
// insert remaining tags into output name
for i, templatePart := range out {
if templatePart == "TAGS" {
out[i] = buildTags(tagsCopy)
break
}
}
if s.Prefix == "" {
return sanitizedChars.Replace(strings.Join(out, "."))
}
return sanitizedChars.Replace(s.Prefix + "." + strings.Join(out, "."))
}
func buildTags(metric telegraf.Metric) string {
// InsertField takes the bucket string from SerializeBucketName and replaces the
// FIELDNAME portion. If fieldName == "value", it will simply delete the
// FIELDNAME portion.
func InsertField(bucket, fieldName string) string {
// if the field name is "value", then dont use it
if fieldName == "value" {
return fieldDeleter.Replace(bucket)
}
return strings.Replace(bucket, "FIELDNAME", fieldName, 1)
}
func buildTags(tags map[string]string) string {
var keys []string
tags := metric.Tags()
for k := range tags {
if k == "host" {
continue
}
keys = append(keys, k)
}
sort.Strings(keys)
var tag_str string
if host, ok := tags["host"]; ok {
if len(keys) > 0 {
tag_str = strings.Replace(host, ".", "_", -1) + "."
} else {
tag_str = strings.Replace(host, ".", "_", -1)
}
}
for i, k := range keys {
tag_value := strings.Replace(tags[k], ".", "_", -1)
if i == 0 {
@@ -87,5 +122,5 @@ func buildTags(metric telegraf.Metric) string {
tag_str += "." + tag_value
}
}
return sanitizedChars.Replace(tag_str)
return tag_str
}

217
plugins/serializers/graphite/graphite_test.go
View File

@@ -11,6 +11,23 @@ import (
"github.com/influxdata/telegraf"
)
var defaultTags = map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
const (
template1 = "tags.measurement.field"
template2 = "host.measurement.field"
template3 = "host.tags.field"
template4 = "host.tags.measurement"
// this template explicitly uses all tag keys, so "tags" should be empty
template5 = "host.datacenter.cpu.tags.measurement.field"
// this template has non-existent tag keys
template6 = "foo.host.cpu.bar.tags.measurement.field"
)
func TestGraphiteTags(t *testing.T) {
m1, _ := telegraf.NewMetric(
"mymeasurement",
@@ -31,12 +48,12 @@ func TestGraphiteTags(t *testing.T) {
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
)
tags1 := buildTags(m1)
tags2 := buildTags(m2)
tags3 := buildTags(m3)
tags1 := buildTags(m1.Tags())
tags2 := buildTags(m2.Tags())
tags3 := buildTags(m3.Tags())
assert.Equal(t, "192_168_0_1", tags1)
assert.Equal(t, "192_168_0_1.first.second", tags2)
assert.Equal(t, "first.second.192_168_0_1", tags2)
assert.Equal(t, "first.second", tags3)
}
@@ -93,6 +110,82 @@ func TestSerializeMetricHost(t *testing.T) {
assert.Equal(t, expS, mS)
}
// test that a field named "value" gets ignored.
func TestSerializeValueField(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"value": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{}
mS, err := s.Serialize(m)
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("localhost.cpu0.us-west-2.cpu 91.5 %d", now.Unix()),
}
assert.Equal(t, expS, mS)
}
// test that a field named "value" gets ignored in middle of template.
func TestSerializeValueField2(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"value": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
Template: "host.field.tags.measurement",
}
mS, err := s.Serialize(m)
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("localhost.cpu0.us-west-2.cpu 91.5 %d", now.Unix()),
}
assert.Equal(t, expS, mS)
}
// test that a field named "value" gets ignored at beginning of template.
func TestSerializeValueField3(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"value": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
Template: "field.host.tags.measurement",
}
mS, err := s.Serialize(m)
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("localhost.cpu0.us-west-2.cpu 91.5 %d", now.Unix()),
}
assert.Equal(t, expS, mS)
}
func TestSerializeMetricPrefix(t *testing.T) {
now := time.Now()
tags := map[string]string{
@@ -133,48 +226,128 @@ func TestSerializeBucketNameNoHost(t *testing.T) {
assert.NoError(t, err)
s := GraphiteSerializer{}
mS := s.SerializeBucketName(m, "usage_idle")
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := fmt.Sprintf("cpu0.us-west-2.cpu.usage_idle")
expS := "cpu0.us-west-2.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestSerializeBucketNameHost(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", tags, fields, now)
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{}
mS := s.SerializeBucketName(m, "usage_idle")
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := fmt.Sprintf("localhost.cpu0.us-west-2.cpu.usage_idle")
expS := "localhost.cpu0.us-west-2.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestSerializeBucketNamePrefix(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", tags, fields, now)
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Prefix: "prefix"}
mS := s.SerializeBucketName(m, "usage_idle")
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := fmt.Sprintf("prefix.localhost.cpu0.us-west-2.cpu.usage_idle")
expS := "prefix.localhost.cpu0.us-west-2.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestTemplate1(t *testing.T) {
now := time.Now()
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Template: template1}
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := "cpu0.us-west-2.localhost.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestTemplate2(t *testing.T) {
now := time.Now()
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Template: template2}
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := "localhost.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestTemplate3(t *testing.T) {
now := time.Now()
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Template: template3}
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := "localhost.cpu0.us-west-2.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestTemplate4(t *testing.T) {
now := time.Now()
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Template: template4}
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := "localhost.cpu0.us-west-2.cpu"
assert.Equal(t, expS, mS)
}
func TestTemplate5(t *testing.T) {
now := time.Now()
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Template: template5}
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := "localhost.us-west-2.cpu0.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}
func TestTemplate6(t *testing.T) {
now := time.Now()
fields := map[string]interface{}{
"usage_idle": float64(91.5),
}
m, err := telegraf.NewMetric("cpu", defaultTags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{Template: template6}
mS := s.SerializeBucketName(m.Name(), m.Tags())
expS := "localhost.cpu0.us-west-2.cpu.FIELDNAME"
assert.Equal(t, expS, mS)
}

11
plugins/serializers/registry.go
View File

@@ -30,6 +30,10 @@ type Config struct {
// Prefix to add to all measurements, only supports Graphite
Prefix string
// Template for converting telegraf metrics into Graphite
// only supports Graphite
Template string
}
// NewSerializer a Serializer interface based on the given config.
@@ -40,7 +44,7 @@ func NewSerializer(config *Config) (Serializer, error) {
case "influx":
serializer, err = NewInfluxSerializer()
case "graphite":
serializer, err = NewGraphiteSerializer(config.Prefix)
serializer, err = NewGraphiteSerializer(config.Prefix, config.Template)
case "json":
serializer, err = NewJsonSerializer()
}
@@ -55,8 +59,9 @@ func NewInfluxSerializer() (Serializer, error) {
return &influx.InfluxSerializer{}, nil
}
func NewGraphiteSerializer(prefix string) (Serializer, error) {
func NewGraphiteSerializer(prefix, template string) (Serializer, error) {
return &graphite.GraphiteSerializer{
Prefix: prefix,
Prefix: prefix,
Template: template,
}, nil
}