2015-09-10 18:34:36 +00:00
|
|
|
package opentsdb
|
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
|
|
|
"net"
|
2016-07-18 19:01:36 +00:00
|
|
|
"net/url"
|
2015-09-10 18:34:36 +00:00
|
|
|
"sort"
|
|
|
|
"strconv"
|
|
|
|
"strings"
|
|
|
|
|
2016-01-27 21:21:36 +00:00
|
|
|
"github.com/influxdata/telegraf"
|
2016-01-27 23:15:14 +00:00
|
|
|
"github.com/influxdata/telegraf/plugins/outputs"
|
2015-09-10 18:34:36 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
type OpenTSDB struct {
|
|
|
|
Prefix string
|
|
|
|
|
|
|
|
Host string
|
|
|
|
Port int
|
2015-09-14 10:28:10 +00:00
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
HttpBatchSize int
|
|
|
|
|
2015-09-14 10:28:10 +00:00
|
|
|
Debug bool
|
2015-09-10 18:34:36 +00:00
|
|
|
}
|
|
|
|
|
2016-04-28 23:00:06 +00:00
|
|
|
var sanitizedChars = strings.NewReplacer("@", "-", "*", "-", " ", "_",
|
2016-05-25 10:37:48 +00:00
|
|
|
`%`, "-", "#", "-", "$", "-", ":", "_")
|
2016-04-28 23:00:06 +00:00
|
|
|
|
2015-09-10 18:34:36 +00:00
|
|
|
var sampleConfig = `
|
2016-02-18 21:26:51 +00:00
|
|
|
## prefix for metrics keys
|
2015-10-15 21:53:29 +00:00
|
|
|
prefix = "my.specific.prefix."
|
2015-09-10 18:34:36 +00:00
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
## DNS name of the OpenTSDB server
|
|
|
|
## Using "opentsdb.example.com" or "tcp://opentsdb.example.com" will use the
|
|
|
|
## telnet API. "http://opentsdb.example.com" will use the Http API.
|
2015-10-15 21:53:29 +00:00
|
|
|
host = "opentsdb.example.com"
|
2015-09-10 18:34:36 +00:00
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
## Port of the OpenTSDB server
|
2015-10-15 21:53:29 +00:00
|
|
|
port = 4242
|
2015-09-14 10:28:10 +00:00
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
## Number of data points to send to OpenTSDB in Http requests.
|
|
|
|
## Not used with telnet API.
|
|
|
|
httpBatchSize = 50
|
|
|
|
|
2016-02-18 21:26:51 +00:00
|
|
|
## Debug true - Prints OpenTSDB communication
|
2015-10-15 21:53:29 +00:00
|
|
|
debug = false
|
2015-09-10 18:34:36 +00:00
|
|
|
`
|
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
func ToLineFormat(tags map[string]string) string {
|
|
|
|
tagsArray := make([]string, len(tags))
|
|
|
|
index := 0
|
|
|
|
for k, v := range tags {
|
|
|
|
tagsArray[index] = fmt.Sprintf("%s=%s", k, v)
|
|
|
|
index++
|
|
|
|
}
|
|
|
|
sort.Strings(tagsArray)
|
|
|
|
return strings.Join(tagsArray, " ")
|
2015-09-10 18:34:36 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (o *OpenTSDB) Connect() error {
|
|
|
|
// Test Connection to OpenTSDB Server
|
2016-07-18 19:01:36 +00:00
|
|
|
u, err := url.Parse(o.Host)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("Error in parsing host url: %s", err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
uri := fmt.Sprintf("%s:%d", u.Host, o.Port)
|
2015-09-10 18:34:36 +00:00
|
|
|
tcpAddr, err := net.ResolveTCPAddr("tcp", uri)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("OpenTSDB: TCP address cannot be resolved")
|
|
|
|
}
|
|
|
|
connection, err := net.DialTCP("tcp", nil, tcpAddr)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("OpenTSDB: Telnet connect fail")
|
|
|
|
}
|
2015-10-16 22:13:32 +00:00
|
|
|
defer connection.Close()
|
2015-09-10 18:34:36 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-01-27 23:15:14 +00:00
|
|
|
func (o *OpenTSDB) Write(metrics []telegraf.Metric) error {
|
|
|
|
if len(metrics) == 0 {
|
2015-09-10 18:34:36 +00:00
|
|
|
return nil
|
|
|
|
}
|
2015-12-19 21:30:37 +00:00
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
u, err := url.Parse(o.Host)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("Error in parsing host url: %s", err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
if u.Scheme == "" || u.Scheme == "tcp" {
|
|
|
|
return o.WriteTelnet(metrics, u)
|
|
|
|
} else if u.Scheme == "http" {
|
|
|
|
return o.WriteHttp(metrics, u)
|
|
|
|
} else {
|
|
|
|
return fmt.Errorf("Unknown scheme in host parameter.")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (o *OpenTSDB) WriteHttp(metrics []telegraf.Metric, u *url.URL) error {
|
|
|
|
http := openTSDBHttp{
|
|
|
|
Host: u.Host,
|
|
|
|
Port: o.Port,
|
|
|
|
BatchSize: o.HttpBatchSize,
|
|
|
|
Debug: o.Debug,
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, m := range metrics {
|
|
|
|
now := m.UnixNano() / 1000000000
|
|
|
|
tags := cleanTags(m.Tags())
|
|
|
|
|
|
|
|
for fieldName, value := range m.Fields() {
|
2016-09-20 18:26:08 +00:00
|
|
|
switch value.(type) {
|
|
|
|
case int64:
|
|
|
|
case uint64:
|
|
|
|
case float64:
|
|
|
|
default:
|
|
|
|
fmt.Printf("OpenTSDB does not support metric value: [%s] of type [%T].\n", value, value)
|
2016-07-18 19:01:36 +00:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
metric := &HttpMetric{
|
|
|
|
Metric: sanitizedChars.Replace(fmt.Sprintf("%s%s_%s",
|
|
|
|
o.Prefix, m.Name(), fieldName)),
|
|
|
|
Tags: tags,
|
|
|
|
Timestamp: now,
|
2016-09-20 18:26:08 +00:00
|
|
|
Value: value,
|
2016-07-18 19:01:36 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
if err := http.sendDataPoint(metric); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := http.flush(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (o *OpenTSDB) WriteTelnet(metrics []telegraf.Metric, u *url.URL) error {
|
2015-09-10 18:34:36 +00:00
|
|
|
// Send Data with telnet / socket communication
|
2016-07-18 19:01:36 +00:00
|
|
|
uri := fmt.Sprintf("%s:%d", u.Host, o.Port)
|
2015-09-10 18:34:36 +00:00
|
|
|
tcpAddr, _ := net.ResolveTCPAddr("tcp", uri)
|
|
|
|
connection, err := net.DialTCP("tcp", nil, tcpAddr)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("OpenTSDB: Telnet connect fail")
|
|
|
|
}
|
2015-12-19 21:30:37 +00:00
|
|
|
defer connection.Close()
|
2015-09-10 18:34:36 +00:00
|
|
|
|
2016-01-27 23:15:14 +00:00
|
|
|
for _, m := range metrics {
|
2016-07-18 19:01:36 +00:00
|
|
|
now := m.UnixNano() / 1000000000
|
|
|
|
tags := ToLineFormat(cleanTags(m.Tags()))
|
|
|
|
|
|
|
|
for fieldName, value := range m.Fields() {
|
|
|
|
metricValue, buildError := buildValue(value)
|
|
|
|
if buildError != nil {
|
|
|
|
fmt.Printf("OpenTSDB: %s\n", buildError.Error())
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2015-12-19 21:30:37 +00:00
|
|
|
messageLine := fmt.Sprintf("put %s %v %s %s\n",
|
2016-07-18 19:01:36 +00:00
|
|
|
sanitizedChars.Replace(fmt.Sprintf("%s%s_%s", o.Prefix, m.Name(), fieldName)),
|
|
|
|
now, metricValue, tags)
|
|
|
|
|
2015-12-19 21:30:37 +00:00
|
|
|
if o.Debug {
|
|
|
|
fmt.Print(messageLine)
|
|
|
|
}
|
|
|
|
_, err := connection.Write([]byte(messageLine))
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("OpenTSDB: Telnet writing error %s", err.Error())
|
|
|
|
}
|
2015-09-10 18:34:36 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-07-18 19:01:36 +00:00
|
|
|
func cleanTags(tags map[string]string) map[string]string {
|
|
|
|
tagSet := make(map[string]string, len(tags))
|
|
|
|
for k, v := range tags {
|
|
|
|
tagSet[sanitizedChars.Replace(k)] = sanitizedChars.Replace(v)
|
2015-12-19 21:30:37 +00:00
|
|
|
}
|
2016-07-18 19:01:36 +00:00
|
|
|
return tagSet
|
2015-12-19 21:30:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func buildValue(v interface{}) (string, error) {
|
2015-09-10 18:34:36 +00:00
|
|
|
var retv string
|
|
|
|
switch p := v.(type) {
|
|
|
|
case int64:
|
|
|
|
retv = IntToString(int64(p))
|
|
|
|
case uint64:
|
|
|
|
retv = UIntToString(uint64(p))
|
|
|
|
case float64:
|
|
|
|
retv = FloatToString(float64(p))
|
|
|
|
default:
|
2015-09-14 10:28:10 +00:00
|
|
|
return retv, fmt.Errorf("unexpected type %T with value %v for OpenTSDB", v, v)
|
2015-09-10 18:34:36 +00:00
|
|
|
}
|
|
|
|
return retv, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func IntToString(input_num int64) string {
|
|
|
|
return strconv.FormatInt(input_num, 10)
|
|
|
|
}
|
|
|
|
|
|
|
|
func UIntToString(input_num uint64) string {
|
|
|
|
return strconv.FormatUint(input_num, 10)
|
|
|
|
}
|
|
|
|
|
|
|
|
func FloatToString(input_num float64) string {
|
|
|
|
return strconv.FormatFloat(input_num, 'f', 6, 64)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (o *OpenTSDB) SampleConfig() string {
|
|
|
|
return sampleConfig
|
|
|
|
}
|
|
|
|
|
|
|
|
func (o *OpenTSDB) Description() string {
|
|
|
|
return "Configuration for OpenTSDB server to send metrics to"
|
|
|
|
}
|
|
|
|
|
|
|
|
func (o *OpenTSDB) Close() error {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func init() {
|
2016-01-27 21:21:36 +00:00
|
|
|
outputs.Add("opentsdb", func() telegraf.Output {
|
2015-09-10 18:34:36 +00:00
|
|
|
return &OpenTSDB{}
|
|
|
|
})
|
|
|
|
}
|