parent
c549ab907a
commit
9d4b55be19
|
@ -8,6 +8,7 @@ import (
|
|||
"log"
|
||||
"math/rand"
|
||||
"net"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
@ -70,26 +71,27 @@ func (g *Graphite) Description() string {
|
|||
func (g *Graphite) Write(metrics []telegraf.Metric) error {
|
||||
// Prepare data
|
||||
var bp []string
|
||||
for _, point := range metrics {
|
||||
for _, metric := range metrics {
|
||||
// Get name
|
||||
name := point.Name()
|
||||
name := metric.Name()
|
||||
// Convert UnixNano to Unix timestamps
|
||||
timestamp := point.UnixNano() / 1000000000
|
||||
timestamp := metric.UnixNano() / 1000000000
|
||||
tag_str := buildTags(metric)
|
||||
|
||||
for field_name, value := range point.Fields() {
|
||||
for field_name, value := range metric.Fields() {
|
||||
// Convert value
|
||||
value_str := fmt.Sprintf("%#v", value)
|
||||
// Write graphite point
|
||||
// Write graphite metric
|
||||
var graphitePoint string
|
||||
if name == field_name {
|
||||
graphitePoint = fmt.Sprintf("%s.%s %s %d\n",
|
||||
strings.Replace(point.Tags()["host"], ".", "_", -1),
|
||||
tag_str,
|
||||
strings.Replace(name, ".", "_", -1),
|
||||
value_str,
|
||||
timestamp)
|
||||
} else {
|
||||
graphitePoint = fmt.Sprintf("%s.%s.%s %s %d\n",
|
||||
strings.Replace(point.Tags()["host"], ".", "_", -1),
|
||||
tag_str,
|
||||
strings.Replace(name, ".", "_", -1),
|
||||
strings.Replace(field_name, ".", "_", -1),
|
||||
value_str,
|
||||
|
@ -99,7 +101,6 @@ func (g *Graphite) Write(metrics []telegraf.Metric) error {
|
|||
graphitePoint = fmt.Sprintf("%s.%s", g.Prefix, graphitePoint)
|
||||
}
|
||||
bp = append(bp, graphitePoint)
|
||||
//fmt.Printf(graphitePoint)
|
||||
}
|
||||
}
|
||||
graphitePoints := strings.Join(bp, "")
|
||||
|
@ -127,6 +128,37 @@ func (g *Graphite) Write(metrics []telegraf.Metric) error {
|
|||
return err
|
||||
}
|
||||
|
||||
func buildTags(metric telegraf.Metric) 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 {
|
||||
tag_str += tag_value
|
||||
} else {
|
||||
tag_str += "." + tag_value
|
||||
}
|
||||
}
|
||||
return tag_str
|
||||
}
|
||||
|
||||
func init() {
|
||||
outputs.Add("graphite", func() telegraf.Output {
|
||||
return &Graphite{}
|
||||
|
|
|
@ -21,7 +21,7 @@ func TestGraphiteError(t *testing.T) {
|
|||
Prefix: "my.prefix",
|
||||
}
|
||||
// Init metrics
|
||||
pt1, _ := telegraf.NewMetric(
|
||||
m1, _ := telegraf.NewMetric(
|
||||
"mymeasurement",
|
||||
map[string]string{"host": "192.168.0.1"},
|
||||
map[string]interface{}{"mymeasurement": float64(3.14)},
|
||||
|
@ -29,7 +29,7 @@ func TestGraphiteError(t *testing.T) {
|
|||
)
|
||||
// Prepare point list
|
||||
var metrics []telegraf.Metric
|
||||
metrics = append(metrics, pt1)
|
||||
metrics = append(metrics, m1)
|
||||
// Error
|
||||
err1 := g.Connect()
|
||||
require.NoError(t, err1)
|
||||
|
@ -45,19 +45,19 @@ func TestGraphiteOK(t *testing.T) {
|
|||
Prefix: "my.prefix",
|
||||
}
|
||||
// Init metrics
|
||||
pt1, _ := telegraf.NewMetric(
|
||||
m1, _ := telegraf.NewMetric(
|
||||
"mymeasurement",
|
||||
map[string]string{"host": "192.168.0.1"},
|
||||
map[string]interface{}{"mymeasurement": float64(3.14)},
|
||||
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
|
||||
)
|
||||
pt2, _ := telegraf.NewMetric(
|
||||
m2, _ := telegraf.NewMetric(
|
||||
"mymeasurement",
|
||||
map[string]string{"host": "192.168.0.1"},
|
||||
map[string]interface{}{"value": float64(3.14)},
|
||||
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
|
||||
)
|
||||
pt3, _ := telegraf.NewMetric(
|
||||
m3, _ := telegraf.NewMetric(
|
||||
"my_measurement",
|
||||
map[string]string{"host": "192.168.0.1"},
|
||||
map[string]interface{}{"value": float64(3.14)},
|
||||
|
@ -65,9 +65,9 @@ func TestGraphiteOK(t *testing.T) {
|
|||
)
|
||||
// Prepare point list
|
||||
var metrics []telegraf.Metric
|
||||
metrics = append(metrics, pt1)
|
||||
metrics = append(metrics, pt2)
|
||||
metrics = append(metrics, pt3)
|
||||
metrics = append(metrics, m1)
|
||||
metrics = append(metrics, m2)
|
||||
metrics = append(metrics, m3)
|
||||
// Start TCP server
|
||||
wg.Add(1)
|
||||
go TCPServer(t, &wg)
|
||||
|
@ -102,3 +102,32 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup) {
|
|||
conn.Close()
|
||||
wg.Done()
|
||||
}
|
||||
|
||||
func TestGraphiteTags(t *testing.T) {
|
||||
m1, _ := telegraf.NewMetric(
|
||||
"mymeasurement",
|
||||
map[string]string{"host": "192.168.0.1"},
|
||||
map[string]interface{}{"value": float64(3.14)},
|
||||
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
|
||||
)
|
||||
m2, _ := telegraf.NewMetric(
|
||||
"mymeasurement",
|
||||
map[string]string{"host": "192.168.0.1", "afoo": "first", "bfoo": "second"},
|
||||
map[string]interface{}{"value": float64(3.14)},
|
||||
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
|
||||
)
|
||||
m3, _ := telegraf.NewMetric(
|
||||
"mymeasurement",
|
||||
map[string]string{"afoo": "first", "bfoo": "second"},
|
||||
map[string]interface{}{"value": float64(3.14)},
|
||||
time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
|
||||
)
|
||||
|
||||
tags1 := buildTags(m1)
|
||||
tags2 := buildTags(m2)
|
||||
tags3 := buildTags(m3)
|
||||
|
||||
assert.Equal(t, "192_168_0_1", tags1)
|
||||
assert.Equal(t, "192_168_0_1.first.second", tags2)
|
||||
assert.Equal(t, "first.second", tags3)
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue