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

Add support for Graphite 1.1.x tags (#4165)

Browse Source
This commit is contained in:
Pavel Boev
2018-05-22 01:59:56 +03:00
committed by Daniel Nelson
parent 703be4f124
commit 7660315e45
8 changed files with 581 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,8 +30,9 @@ var (
)
type GraphiteSerializer struct {
Prefix string
Template string
Prefix string
Template string
TagSupport bool
}
func (s *GraphiteSerializer) Serialize(metric telegraf.Metric) ([]byte, error) {
@@ -40,23 +41,42 @@ func (s *GraphiteSerializer) Serialize(metric telegraf.Metric) ([]byte, error) {
// Convert UnixNano to Unix timestamps
timestamp := metric.Time().UnixNano() / 1000000000
bucket := SerializeBucketName(metric.Name(), metric.Tags(), s.Template, s.Prefix)
if bucket == "" {
return out, nil
}
for fieldName, value := range metric.Fields() {
fieldValue := formatValue(value)
if fieldValue == "" {
continue
switch s.TagSupport {
case true:
for fieldName, value := range metric.Fields() {
fieldValue := formatValue(value)
if fieldValue == "" {
continue
}
bucket := SerializeBucketNameWithTags(metric.Name(), metric.Tags(), s.Prefix, fieldName)
metricString := fmt.Sprintf("%s %s %d\n",
// insert "field" section of template
bucket,
//bucket,
fieldValue,
timestamp)
point := []byte(metricString)
out = append(out, point...)
}
default:
bucket := SerializeBucketName(metric.Name(), metric.Tags(), s.Template, s.Prefix)
if bucket == "" {
return out, nil
}
for fieldName, value := range metric.Fields() {
fieldValue := formatValue(value)
if fieldValue == "" {
continue
}
metricString := fmt.Sprintf("%s %s %d\n",
// insert "field" section of template
sanitize(InsertField(bucket, fieldName)),
fieldValue,
timestamp)
point := []byte(metricString)
out = append(out, point...)
}
metricString := fmt.Sprintf("%s %s %d\n",
// insert "field" section of template
sanitize(InsertField(bucket, fieldName)),
fieldValue,
timestamp)
point := []byte(metricString)
out = append(out, point...)
}
return out, nil
}
@@ -165,6 +185,44 @@ func SerializeBucketName(
return prefix + "." + strings.Join(out, ".")
}
// SerializeBucketNameWithTags will take the given measurement name and tags and
// produce a graphite bucket. It will use the Graphite11Serializer.
// http://graphite.readthedocs.io/en/latest/tags.html
func SerializeBucketNameWithTags(
measurement string,
tags map[string]string,
prefix string,
field string,
) string {
var out string
var tagsCopy []string
for k, v := range tags {
if k == "name" {
k = "_name"
}
tagsCopy = append(tagsCopy, sanitize(k+"="+v))
}
sort.Strings(tagsCopy)
if prefix != "" {
out = prefix + "."
}
out += measurement
if field != "value" {
out += "." + field
}
out = sanitize(out)
if len(tagsCopy) > 0 {
out += ";" + strings.Join(tagsCopy, ";")
}
return out
}
// InsertField takes the bucket string from SerializeBucketName and replaces the
// FIELDNAME portion. If fieldName == "value", it will simply delete the
// FIELDNAME portion.

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

@@ -87,6 +87,35 @@ func TestSerializeMetricNoHost(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeMetricNoHostWithTagSupport(t *testing.T) {
now := time.Now()
tags := map[string]string{
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"usage_idle": float64(91.5),
"usage_busy": float64(8.5),
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("cpu.usage_idle;cpu=cpu0;datacenter=us-west-2 91.5 %d", now.Unix()),
fmt.Sprintf("cpu.usage_busy;cpu=cpu0;datacenter=us-west-2 8.5 %d", now.Unix()),
}
sort.Strings(mS)
sort.Strings(expS)
assert.Equal(t, expS, mS)
}
func TestSerializeMetricHost(t *testing.T) {
now := time.Now()
tags := map[string]string{
@@ -115,6 +144,36 @@ func TestSerializeMetricHost(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeMetricHostWithTagSupport(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),
"usage_busy": float64(8.5),
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("cpu.usage_idle;cpu=cpu0;datacenter=us-west-2;host=localhost 91.5 %d", now.Unix()),
fmt.Sprintf("cpu.usage_busy;cpu=cpu0;datacenter=us-west-2;host=localhost 8.5 %d", now.Unix()),
}
sort.Strings(mS)
sort.Strings(expS)
assert.Equal(t, expS, mS)
}
// test that a field named "value" gets ignored.
func TestSerializeValueField(t *testing.T) {
now := time.Now()
@@ -140,6 +199,32 @@ func TestSerializeValueField(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeValueFieldWithTagSupport(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 := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("cpu;cpu=cpu0;datacenter=us-west-2;host=localhost 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()
@@ -189,6 +274,28 @@ func TestSerializeValueString(t *testing.T) {
assert.Equal(t, "", mS[0])
}
func TestSerializeValueStringWithTagSupport(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"value": "asdasd",
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
assert.Equal(t, "", mS[0])
}
func TestSerializeValueBoolean(t *testing.T) {
now := time.Now()
tags := map[string]string{
@@ -219,6 +326,36 @@ func TestSerializeValueBoolean(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeValueBooleanWithTagSupport(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
"enabled": true,
"disabled": false,
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("cpu.enabled;cpu=cpu0;datacenter=us-west-2;host=localhost 1 %d", now.Unix()),
fmt.Sprintf("cpu.disabled;cpu=cpu0;datacenter=us-west-2;host=localhost 0 %d", now.Unix()),
}
sort.Strings(mS)
sort.Strings(expS)
assert.Equal(t, expS, mS)
}
func TestSerializeValueUnsigned(t *testing.T) {
now := time.Unix(0, 0)
tags := map[string]string{}
@@ -262,6 +399,32 @@ func TestSerializeFieldWithSpaces(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeFieldWithSpacesWithTagSupport(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": "cpu0",
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
`field\ with\ spaces`: float64(91.5),
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("cpu.field_with_spaces;cpu=cpu0;datacenter=us-west-2;host=localhost 91.5 %d", now.Unix()),
}
assert.Equal(t, expS, mS)
}
// test that tags with spaces get fixed.
func TestSerializeTagWithSpaces(t *testing.T) {
now := time.Now()
@@ -289,6 +452,32 @@ func TestSerializeTagWithSpaces(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeTagWithSpacesWithTagSupport(t *testing.T) {
now := time.Now()
tags := map[string]string{
"host": "localhost",
"cpu": `cpu\ 0`,
"datacenter": "us-west-2",
}
fields := map[string]interface{}{
`field_with_spaces`: float64(91.5),
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("cpu.field_with_spaces;cpu=cpu_0;datacenter=us-west-2;host=localhost 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()
@@ -371,6 +560,37 @@ func TestSerializeMetricPrefix(t *testing.T) {
assert.Equal(t, expS, mS)
}
func TestSerializeMetricPrefixWithTagSupport(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),
"usage_busy": float64(8.5),
}
m, err := metric.New("cpu", tags, fields, now)
assert.NoError(t, err)
s := GraphiteSerializer{
Prefix: "prefix",
TagSupport: true,
}
buf, _ := s.Serialize(m)
mS := strings.Split(strings.TrimSpace(string(buf)), "\n")
assert.NoError(t, err)
expS := []string{
fmt.Sprintf("prefix.cpu.usage_idle;cpu=cpu0;datacenter=us-west-2;host=localhost 91.5 %d", now.Unix()),
fmt.Sprintf("prefix.cpu.usage_busy;cpu=cpu0;datacenter=us-west-2;host=localhost 8.5 %d", now.Unix()),
}
sort.Strings(mS)
sort.Strings(expS)
assert.Equal(t, expS, mS)
}
func TestSerializeBucketNameNoHost(t *testing.T) {
now := time.Now()
tags := map[string]string{
@@ -579,6 +799,100 @@ func TestClean(t *testing.T) {
}
}
func TestCleanWithTagsSupport(t *testing.T) {
now := time.Unix(1234567890, 0)
tests := []struct {
name string
metric_name string
tags map[string]string
fields map[string]interface{}
expected string
}{
{
"Base metric",
"cpu",
map[string]string{"host": "localhost"},
map[string]interface{}{"usage_busy": float64(8.5)},
"cpu.usage_busy;host=localhost 8.5 1234567890\n",
},
{
"Dot and whitespace in tags",
"cpu",
map[string]string{"host": "localhost", "label.dot and space": "value with.dot"},
map[string]interface{}{"usage_busy": float64(8.5)},
"cpu.usage_busy;host=localhost;label.dot_and_space=value_with.dot 8.5 1234567890\n",
},
{
"Field with space",
"system",
map[string]string{"host": "localhost"},
map[string]interface{}{"uptime_format": "20 days, 23:26"},
"", // yes nothing. graphite don't serialize string fields
},
{
"Allowed punct",
"cpu",
map[string]string{"host": "localhost", "tag": "-_:="},
map[string]interface{}{"usage_busy": float64(10)},
"cpu.usage_busy;host=localhost;tag=-_:= 10 1234567890\n",
},
{
"Special conversions to hyphen",
"cpu",
map[string]string{"host": "localhost", "tag": "/@*"},
map[string]interface{}{"usage_busy": float64(10)},
"cpu.usage_busy;host=localhost;tag=--- 10 1234567890\n",
},
{
"Special drop chars",
"cpu",
map[string]string{"host": "localhost", "tag": `\no slash`},
map[string]interface{}{"usage_busy": float64(10)},
"cpu.usage_busy;host=localhost;tag=no_slash 10 1234567890\n",
},
{
"Empty tag & value field",
"cpu",
map[string]string{"host": "localhost"},
map[string]interface{}{"value": float64(10)},
"cpu;host=localhost 10 1234567890\n",
},
{
"Unicode Letters allowed",
"cpu",
map[string]string{"host": "localhost", "tag": "μnicodε_letters"},
map[string]interface{}{"value": float64(10)},
"cpu;host=localhost;tag=μnicodε_letters 10 1234567890\n",
},
{
"Other Unicode not allowed",
"cpu",
map[string]string{"host": "localhost", "tag": "“☢”"},
map[string]interface{}{"value": float64(10)},
"cpu;host=localhost;tag=___ 10 1234567890\n",
},
{
"Newline in tags",
"cpu",
map[string]string{"host": "localhost", "label": "some\nthing\nwith\nnewline"},
map[string]interface{}{"usage_busy": float64(8.5)},
"cpu.usage_busy;host=localhost;label=some_thing_with_newline 8.5 1234567890\n",
},
}
s := GraphiteSerializer{
TagSupport: true,
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
m, err := metric.New(tt.metric_name, tt.tags, tt.fields, now)
assert.NoError(t, err)
actual, _ := s.Serialize(m)
require.Equal(t, tt.expected, string(actual))
})
}
}
func TestSerializeBatch(t *testing.T) {
now := time.Unix(1234567890, 0)
tests := []struct {
@@ -607,3 +921,34 @@ func TestSerializeBatch(t *testing.T) {
})
}
}
func TestSerializeBatchWithTagsSupport(t *testing.T) {
now := time.Unix(1234567890, 0)
tests := []struct {
name string
metric_name string
tags map[string]string
fields map[string]interface{}
expected string
}{
{
"Base metric",
"cpu",
map[string]string{"host": "localhost"},
map[string]interface{}{"usage_busy": float64(8.5)},
"cpu.usage_busy;host=localhost 8.5 1234567890\ncpu.usage_busy;host=localhost 8.5 1234567890\n",
},
}
s := GraphiteSerializer{
TagSupport: true,
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
m, err := metric.New(tt.metric_name, tt.tags, tt.fields, now)
assert.NoError(t, err)
actual, _ := s.SerializeBatch([]telegraf.Metric{m, m})
require.Equal(t, tt.expected, string(actual))
})
}
}

12
plugins/serializers/registry.go
View File

@@ -38,6 +38,9 @@ type Config struct {
// Dataformat can be one of: influx, graphite, or json
DataFormat string
// Support tags in graphite protocol
GraphiteTagSupport bool
// Maximum line length in bytes; influx format only
InfluxMaxLineBytes int
@@ -67,7 +70,7 @@ func NewSerializer(config *Config) (Serializer, error) {
case "influx":
serializer, err = NewInfluxSerializerConfig(config)
case "graphite":
serializer, err = NewGraphiteSerializer(config.Prefix, config.Template)
serializer, err = NewGraphiteSerializer(config.Prefix, config.Template, config.GraphiteTagSupport)
case "json":
serializer, err = NewJsonSerializer(config.TimestampUnits)
default:
@@ -102,9 +105,10 @@ func NewInfluxSerializer() (Serializer, error) {
return influx.NewSerializer(), nil
}
func NewGraphiteSerializer(prefix, template string) (Serializer, error) {
func NewGraphiteSerializer(prefix, template string, tag_support bool) (Serializer, error) {
return &graphite.GraphiteSerializer{
Prefix: prefix,
Template: template,
Prefix: prefix,
Template: template,
TagSupport: tag_support,
}, nil
}