2015-08-11 20:02:04 +00:00
|
|
|
package influxdb
|
|
|
|
|
|
|
|
import (
|
2015-09-09 21:56:10 +00:00
|
|
|
"errors"
|
2015-08-11 20:02:04 +00:00
|
|
|
"fmt"
|
|
|
|
"log"
|
2015-09-09 21:56:10 +00:00
|
|
|
"math/rand"
|
2015-08-11 20:02:04 +00:00
|
|
|
"net/url"
|
|
|
|
"strings"
|
2015-12-15 17:08:13 +00:00
|
|
|
"time"
|
2015-08-11 20:02:04 +00:00
|
|
|
|
2016-01-27 23:15:14 +00:00
|
|
|
"github.com/influxdata/telegraf"
|
2016-01-20 18:57:35 +00:00
|
|
|
"github.com/influxdata/telegraf/internal"
|
|
|
|
"github.com/influxdata/telegraf/plugins/outputs"
|
2016-01-27 23:15:14 +00:00
|
|
|
|
|
|
|
"github.com/influxdata/influxdb/client/v2"
|
2015-08-11 20:02:04 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
type InfluxDB struct {
|
2015-09-09 21:56:10 +00:00
|
|
|
// URL is only for backwards compatability
|
2016-05-24 09:08:34 +00:00
|
|
|
URL string
|
|
|
|
URLs []string `toml:"urls"`
|
|
|
|
Username string
|
|
|
|
Password string
|
|
|
|
Database string
|
|
|
|
UserAgent string
|
|
|
|
RetentionPolicy string
|
|
|
|
WriteConsistency string
|
|
|
|
Timeout internal.Duration
|
|
|
|
UDPPayload int `toml:"udp_payload"`
|
2015-08-11 20:02:04 +00:00
|
|
|
|
2016-02-13 00:00:11 +00:00
|
|
|
// Path to CA file
|
|
|
|
SSLCA string `toml:"ssl_ca"`
|
|
|
|
// Path to host cert file
|
|
|
|
SSLCert string `toml:"ssl_cert"`
|
|
|
|
// Path to cert key file
|
|
|
|
SSLKey string `toml:"ssl_key"`
|
|
|
|
// Use SSL but skip chain & host verification
|
|
|
|
InsecureSkipVerify bool
|
|
|
|
|
2016-06-13 14:21:11 +00:00
|
|
|
// Precision is only here for legacy support. It will be ignored.
|
|
|
|
Precision string
|
|
|
|
|
2015-10-16 22:13:32 +00:00
|
|
|
conns []client.Client
|
2015-08-11 20:02:04 +00:00
|
|
|
}
|
|
|
|
|
2015-08-25 23:59:12 +00:00
|
|
|
var sampleConfig = `
|
2016-02-18 21:26:51 +00:00
|
|
|
## The full HTTP or UDP endpoint URL for your InfluxDB instance.
|
|
|
|
## Multiple urls can be specified as part of the same cluster,
|
|
|
|
## this means that only ONE of the urls will be written to each interval.
|
2015-11-10 21:19:32 +00:00
|
|
|
# urls = ["udp://localhost:8089"] # UDP endpoint example
|
2015-10-15 21:53:29 +00:00
|
|
|
urls = ["http://localhost:8086"] # required
|
2016-02-29 16:13:00 +00:00
|
|
|
## The target database for metrics (telegraf will create it if not exists).
|
2015-10-15 21:53:29 +00:00
|
|
|
database = "telegraf" # required
|
2015-09-24 18:06:11 +00:00
|
|
|
|
2016-06-16 11:18:08 +00:00
|
|
|
## Retention policy to write to. Empty string writes to the default rp.
|
|
|
|
retention_policy = ""
|
2016-05-24 09:08:34 +00:00
|
|
|
## Write consistency (clusters only), can be: "any", "one", "quorom", "all"
|
|
|
|
write_consistency = "any"
|
|
|
|
|
2016-02-19 21:46:03 +00:00
|
|
|
## Write timeout (for the InfluxDB client), formatted as a string.
|
|
|
|
## If not provided, will default to 5s. 0s means no timeout (not recommended).
|
|
|
|
timeout = "5s"
|
2015-10-15 21:53:29 +00:00
|
|
|
# username = "telegraf"
|
|
|
|
# password = "metricsmetricsmetricsmetrics"
|
2016-02-18 21:26:51 +00:00
|
|
|
## Set the user agent for HTTP POSTs (can be useful for log differentiation)
|
2015-10-15 21:53:29 +00:00
|
|
|
# user_agent = "telegraf"
|
2016-02-18 21:26:51 +00:00
|
|
|
## Set UDP payload size, defaults to InfluxDB UDP Client default (512 bytes)
|
2015-11-10 21:19:32 +00:00
|
|
|
# udp_payload = 512
|
2016-02-13 00:00:11 +00:00
|
|
|
|
2016-02-18 21:26:51 +00:00
|
|
|
## Optional SSL Config
|
2016-02-13 00:00:11 +00:00
|
|
|
# ssl_ca = "/etc/telegraf/ca.pem"
|
|
|
|
# ssl_cert = "/etc/telegraf/cert.pem"
|
|
|
|
# ssl_key = "/etc/telegraf/key.pem"
|
2016-02-18 21:26:51 +00:00
|
|
|
## Use SSL but skip chain & host verification
|
2016-02-13 00:00:11 +00:00
|
|
|
# insecure_skip_verify = false
|
2015-08-25 23:59:12 +00:00
|
|
|
`
|
|
|
|
|
2015-08-11 20:02:04 +00:00
|
|
|
func (i *InfluxDB) Connect() error {
|
2015-11-10 21:19:32 +00:00
|
|
|
var urls []string
|
|
|
|
for _, u := range i.URLs {
|
2015-09-09 21:56:10 +00:00
|
|
|
urls = append(urls, u)
|
2015-08-11 20:02:04 +00:00
|
|
|
}
|
|
|
|
|
2015-09-09 21:56:10 +00:00
|
|
|
// Backward-compatability with single Influx URL config files
|
|
|
|
// This could eventually be removed in favor of specifying the urls as a list
|
|
|
|
if i.URL != "" {
|
2015-11-10 21:19:32 +00:00
|
|
|
urls = append(urls, i.URL)
|
2015-09-09 21:56:10 +00:00
|
|
|
}
|
2015-08-11 20:02:04 +00:00
|
|
|
|
2016-02-13 00:00:11 +00:00
|
|
|
tlsCfg, err := internal.GetTLSConfig(
|
|
|
|
i.SSLCert, i.SSLKey, i.SSLCA, i.InsecureSkipVerify)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2015-10-16 22:13:32 +00:00
|
|
|
var conns []client.Client
|
2015-11-10 21:19:32 +00:00
|
|
|
for _, u := range urls {
|
|
|
|
switch {
|
|
|
|
case strings.HasPrefix(u, "udp"):
|
2015-11-12 21:54:43 +00:00
|
|
|
parsed_url, err := url.Parse(u)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2015-11-10 21:19:32 +00:00
|
|
|
if i.UDPPayload == 0 {
|
|
|
|
i.UDPPayload = client.UDPPayloadSize
|
|
|
|
}
|
|
|
|
c, err := client.NewUDPClient(client.UDPConfig{
|
|
|
|
Addr: parsed_url.Host,
|
|
|
|
PayloadSize: i.UDPPayload,
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
conns = append(conns, c)
|
|
|
|
default:
|
|
|
|
// If URL doesn't start with "udp", assume HTTP client
|
|
|
|
c, err := client.NewHTTPClient(client.HTTPConfig{
|
2015-11-12 21:54:43 +00:00
|
|
|
Addr: u,
|
2015-11-10 21:19:32 +00:00
|
|
|
Username: i.Username,
|
|
|
|
Password: i.Password,
|
|
|
|
UserAgent: i.UserAgent,
|
|
|
|
Timeout: i.Timeout.Duration,
|
2016-02-13 00:00:11 +00:00
|
|
|
TLSConfig: tlsCfg,
|
2015-11-10 21:19:32 +00:00
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2016-04-19 04:23:48 +00:00
|
|
|
err = createDatabase(c, i.Database)
|
|
|
|
if err != nil {
|
|
|
|
log.Println("Database creation failed: " + err.Error())
|
2016-02-26 16:26:43 +00:00
|
|
|
continue
|
2015-11-10 21:19:32 +00:00
|
|
|
}
|
2015-08-11 20:02:04 +00:00
|
|
|
|
2015-11-10 21:19:32 +00:00
|
|
|
conns = append(conns, c)
|
2015-09-09 21:56:10 +00:00
|
|
|
}
|
2015-08-11 20:02:04 +00:00
|
|
|
}
|
|
|
|
|
2015-09-09 21:56:10 +00:00
|
|
|
i.conns = conns
|
2015-12-15 17:08:13 +00:00
|
|
|
rand.Seed(time.Now().UnixNano())
|
2015-10-22 17:14:10 +00:00
|
|
|
return nil
|
2015-08-11 20:02:04 +00:00
|
|
|
}
|
|
|
|
|
2016-04-19 04:23:48 +00:00
|
|
|
func createDatabase(c client.Client, database string) error {
|
|
|
|
// Create Database if it doesn't exist
|
|
|
|
_, err := c.Query(client.Query{
|
|
|
|
Command: fmt.Sprintf("CREATE DATABASE IF NOT EXISTS \"%s\"", database),
|
|
|
|
})
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2015-08-12 17:04:25 +00:00
|
|
|
func (i *InfluxDB) Close() error {
|
2016-04-19 04:23:48 +00:00
|
|
|
var errS string
|
|
|
|
for j, _ := range i.conns {
|
|
|
|
if err := i.conns[j].Close(); err != nil {
|
|
|
|
errS += err.Error()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if errS != "" {
|
|
|
|
return fmt.Errorf("output influxdb close failed: %s", errS)
|
|
|
|
}
|
2015-08-12 17:04:25 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-08-25 23:59:12 +00:00
|
|
|
func (i *InfluxDB) SampleConfig() string {
|
|
|
|
return sampleConfig
|
|
|
|
}
|
|
|
|
|
|
|
|
func (i *InfluxDB) Description() string {
|
|
|
|
return "Configuration for influxdb server to send metrics to"
|
|
|
|
}
|
|
|
|
|
2015-09-09 21:56:10 +00:00
|
|
|
// Choose a random server in the cluster to write to until a successful write
|
|
|
|
// occurs, logging each unsuccessful. If all servers fail, return error.
|
2016-01-27 23:15:14 +00:00
|
|
|
func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
|
2016-02-26 16:26:43 +00:00
|
|
|
if len(i.conns) == 0 {
|
|
|
|
err := i.Connect()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
2016-02-18 19:21:20 +00:00
|
|
|
bp, err := client.NewBatchPoints(client.BatchPointsConfig{
|
2016-05-24 09:08:34 +00:00
|
|
|
Database: i.Database,
|
|
|
|
RetentionPolicy: i.RetentionPolicy,
|
|
|
|
WriteConsistency: i.WriteConsistency,
|
2015-10-16 22:13:32 +00:00
|
|
|
})
|
2016-02-18 19:21:20 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2015-10-16 22:13:32 +00:00
|
|
|
|
2016-01-27 23:15:14 +00:00
|
|
|
for _, metric := range metrics {
|
|
|
|
bp.AddPoint(metric.Point())
|
2015-10-16 22:13:32 +00:00
|
|
|
}
|
2015-09-09 21:56:10 +00:00
|
|
|
|
|
|
|
// This will get set to nil if a successful write occurs
|
2016-02-18 19:21:20 +00:00
|
|
|
err = errors.New("Could not write to any InfluxDB server in cluster")
|
2015-09-09 21:56:10 +00:00
|
|
|
|
|
|
|
p := rand.Perm(len(i.conns))
|
|
|
|
for _, n := range p {
|
2015-10-16 22:13:32 +00:00
|
|
|
if e := i.conns[n].Write(bp); e != nil {
|
2016-04-19 04:23:48 +00:00
|
|
|
// Log write failure
|
|
|
|
log.Printf("ERROR: %s", e)
|
|
|
|
// If the database was not found, try to recreate it
|
|
|
|
if strings.Contains(e.Error(), "database not found") {
|
|
|
|
if errc := createDatabase(i.conns[n], i.Database); errc != nil {
|
|
|
|
log.Printf("ERROR: Database %s not found and failed to recreate\n",
|
|
|
|
i.Database)
|
|
|
|
}
|
|
|
|
}
|
2015-09-09 21:56:10 +00:00
|
|
|
} else {
|
|
|
|
err = nil
|
|
|
|
break
|
|
|
|
}
|
2015-08-11 20:02:04 +00:00
|
|
|
}
|
2016-03-14 10:28:01 +00:00
|
|
|
|
2015-09-09 21:56:10 +00:00
|
|
|
return err
|
2015-08-11 20:02:04 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func init() {
|
2016-01-27 21:21:36 +00:00
|
|
|
outputs.Add("influxdb", func() telegraf.Output {
|
2016-02-19 21:46:03 +00:00
|
|
|
return &InfluxDB{
|
2016-07-20 11:33:21 +00:00
|
|
|
Timeout: internal.Duration{Duration: time.Second * 5},
|
|
|
|
Database: "telegraf",
|
2016-02-19 21:46:03 +00:00
|
|
|
}
|
2015-08-11 20:02:04 +00:00
|
|
|
})
|
|
|
|
}
|