forked from influxdata/telegraf
-
Notifications
You must be signed in to change notification settings - Fork 0
/
riemann.go
156 lines (129 loc) · 3.09 KB
/
riemann.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
package riemann
import (
"fmt"
"log"
"os"
"sort"
"strings"
"github.com/amir/raidman"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/outputs"
)
const deprecationMsg = "I! WARNING: this Riemann output plugin will be deprecated in a future release, see https://github.com/influxdata/telegraf/issues/1878 for more details & discussion."
type Riemann struct {
URL string
Transport string
Separator string
client *raidman.Client
}
var sampleConfig = `
## URL of server
url = "localhost:5555"
## transport protocol to use either tcp or udp
transport = "tcp"
## separator to use between input name and field name in Riemann service name
separator = " "
`
func (r *Riemann) Connect() error {
log.Printf(deprecationMsg)
c, err := raidman.Dial(r.Transport, r.URL)
if err != nil {
r.client = nil
return err
}
r.client = c
return nil
}
func (r *Riemann) Close() error {
if r.client == nil {
return nil
}
r.client.Close()
r.client = nil
return nil
}
func (r *Riemann) SampleConfig() string {
return sampleConfig
}
func (r *Riemann) Description() string {
return "Configuration for the Riemann server to send metrics to"
}
func (r *Riemann) Write(metrics []telegraf.Metric) error {
log.Printf(deprecationMsg)
if len(metrics) == 0 {
return nil
}
if r.client == nil {
err := r.Connect()
if err != nil {
return fmt.Errorf("FAILED to (re)connect to Riemann. Error: %s\n", err)
}
}
var events []*raidman.Event
for _, p := range metrics {
evs := buildEvents(p, r.Separator)
for _, ev := range evs {
events = append(events, ev)
}
}
var senderr = r.client.SendMulti(events)
if senderr != nil {
r.Close() // always retuns nil
return fmt.Errorf("FAILED to send riemann message (will try to reconnect). Error: %s\n",
senderr)
}
return nil
}
func buildEvents(p telegraf.Metric, s string) []*raidman.Event {
events := []*raidman.Event{}
for fieldName, value := range p.Fields() {
host, ok := p.Tags()["host"]
if !ok {
hostname, err := os.Hostname()
if err != nil {
host = "unknown"
} else {
host = hostname
}
}
event := &raidman.Event{
Host: host,
Service: serviceName(s, p.Name(), p.Tags(), fieldName),
}
switch value.(type) {
case string:
event.State = value.(string)
default:
event.Metric = value
}
events = append(events, event)
}
return events
}
func serviceName(s string, n string, t map[string]string, f string) string {
serviceStrings := []string{}
serviceStrings = append(serviceStrings, n)
// we'll skip the 'host' tag
tagStrings := []string{}
tagNames := []string{}
for tagName := range t {
tagNames = append(tagNames, tagName)
}
sort.Strings(tagNames)
for _, tagName := range tagNames {
if tagName != "host" {
tagStrings = append(tagStrings, t[tagName])
}
}
var tagString string = strings.Join(tagStrings, s)
if tagString != "" {
serviceStrings = append(serviceStrings, tagString)
}
serviceStrings = append(serviceStrings, f)
return strings.Join(serviceStrings, s)
}
func init() {
outputs.Add("riemann", func() telegraf.Output {
return &Riemann{}
})
}