telegraf/plugins/outputs/elasticsearch/elasticsearch.go

342 lines
8.9 KiB
Go

package elasticsearch
import (
"context"
"fmt"
"log"
"net/http"
"strconv"
"strings"
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/outputs"
"gopkg.in/olivere/elastic.v5"
)
type Elasticsearch struct {
URLs []string `toml:"urls"`
IndexName string
Username string
Password string
EnableSniffer bool
Timeout internal.Duration
HealthCheckInterval internal.Duration
ManageTemplate bool
TemplateName string
OverwriteTemplate bool
SSLCA string `toml:"ssl_ca"` // Path to CA file
SSLCert string `toml:"ssl_cert"` // Path to host cert file
SSLKey string `toml:"ssl_key"` // Path to cert key file
InsecureSkipVerify bool // Use SSL but skip chain & host verification
Client *elastic.Client
}
var sampleConfig = `
## The full HTTP endpoint URL for your Elasticsearch 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.
urls = [ "http://node1.es.example.com:9200" ] # required.
## Elasticsearch client timeout, defaults to "5s" if not set.
timeout = "5s"
## Set to true to ask Elasticsearch a list of all cluster nodes,
## thus it is not necessary to list all nodes in the urls config option.
enable_sniffer = false
## Set the interval to check if the Elasticsearch nodes are available
## Setting to "0s" will disable the health check (not recommended in production)
health_check_interval = "10s"
## HTTP basic authentication details (eg. when using Shield)
# username = "telegraf"
# password = "mypassword"
## Index Config
## The target index for metrics (Elasticsearch will create if it not exists).
## You can use the date specifiers below to create indexes per time frame.
## The metric timestamp will be used to decide the destination index name
# %Y - year (2016)
# %y - last two digits of year (00..99)
# %m - month (01..12)
# %d - day of month (e.g., 01)
# %H - hour (00..23)
# %V - week of the year (ISO week) (01..53)
index_name = "telegraf-%Y.%m.%d" # required.
## Optional SSL Config
# ssl_ca = "/etc/telegraf/ca.pem"
# ssl_cert = "/etc/telegraf/cert.pem"
# ssl_key = "/etc/telegraf/key.pem"
## Use SSL but skip chain & host verification
# insecure_skip_verify = false
## Template Config
## Set to true if you want telegraf to manage its index template.
## If enabled it will create a recommended index template for telegraf indexes
manage_template = true
## The template name used for telegraf indexes
template_name = "telegraf"
## Set to true if you want telegraf to overwrite an existing template
overwrite_template = false
`
func (a *Elasticsearch) Connect() error {
if a.URLs == nil || a.IndexName == "" {
return fmt.Errorf("Elasticsearch urls or index_name is not defined")
}
ctx, cancel := context.WithTimeout(context.Background(), a.Timeout.Duration)
defer cancel()
var clientOptions []elastic.ClientOptionFunc
tlsCfg, err := internal.GetTLSConfig(a.SSLCert, a.SSLKey, a.SSLCA, a.InsecureSkipVerify)
if err != nil {
return err
}
tr := &http.Transport{
TLSClientConfig: tlsCfg,
}
httpclient := &http.Client{
Transport: tr,
Timeout: a.Timeout.Duration,
}
clientOptions = append(clientOptions,
elastic.SetHttpClient(httpclient),
elastic.SetSniff(a.EnableSniffer),
elastic.SetURL(a.URLs...),
elastic.SetHealthcheckInterval(a.HealthCheckInterval.Duration),
)
if a.Username != "" && a.Password != "" {
clientOptions = append(clientOptions,
elastic.SetBasicAuth(a.Username, a.Password),
)
}
if a.HealthCheckInterval.Duration == 0 {
clientOptions = append(clientOptions,
elastic.SetHealthcheck(false),
)
log.Printf("D! Elasticsearch output: disabling health check")
}
client, err := elastic.NewClient(clientOptions...)
if err != nil {
return err
}
// check for ES version on first node
esVersion, err := client.ElasticsearchVersion(a.URLs[0])
if err != nil {
return fmt.Errorf("Elasticsearch version check failed: %s", err)
}
// quit if ES version is not supported
i, err := strconv.Atoi(strings.Split(esVersion, ".")[0])
if err != nil || i < 5 {
return fmt.Errorf("Elasticsearch version not supported: %s", esVersion)
}
log.Println("I! Elasticsearch version: " + esVersion)
a.Client = client
if a.ManageTemplate {
err := a.manageTemplate(ctx)
if err != nil {
return err
}
}
return nil
}
func (a *Elasticsearch) Write(metrics []telegraf.Metric) error {
if len(metrics) == 0 {
return nil
}
bulkRequest := a.Client.Bulk()
for _, metric := range metrics {
var name = metric.Name()
// index name has to be re-evaluated each time for telegraf
// to send the metric to the correct time-based index
indexName := a.GetIndexName(a.IndexName, metric.Time())
m := make(map[string]interface{})
m["@timestamp"] = metric.Time()
m["measurement_name"] = name
m["tag"] = metric.Tags()
m[name] = metric.Fields()
bulkRequest.Add(elastic.NewBulkIndexRequest().
Index(indexName).
Type("metrics").
Doc(m))
}
ctx, cancel := context.WithTimeout(context.Background(), a.Timeout.Duration)
defer cancel()
res, err := bulkRequest.Do(ctx)
if err != nil {
return fmt.Errorf("Error sending bulk request to Elasticsearch: %s", err)
}
if res.Errors {
for id, err := range res.Failed() {
log.Printf("E! Elasticsearch indexing failure, id: %d, error: %s, caused by: %s, %s", id, err.Error.Reason, err.Error.CausedBy["reason"], err.Error.CausedBy["type"])
}
return fmt.Errorf("W! Elasticsearch failed to index %d metrics", len(res.Failed()))
}
return nil
}
func (a *Elasticsearch) manageTemplate(ctx context.Context) error {
if a.TemplateName == "" {
return fmt.Errorf("Elasticsearch template_name configuration not defined")
}
templateExists, errExists := a.Client.IndexTemplateExists(a.TemplateName).Do(ctx)
if errExists != nil {
return fmt.Errorf("Elasticsearch template check failed, template name: %s, error: %s", a.TemplateName, errExists)
}
templatePattern := a.IndexName + "*"
if strings.Contains(a.IndexName, "%") {
templatePattern = a.IndexName[0:strings.Index(a.IndexName, "%")] + "*"
}
if (a.OverwriteTemplate) || (!templateExists) {
// Create or update the template
tmpl := fmt.Sprintf(`
{
"template":"%s",
"settings": {
"index": {
"refresh_interval": "10s",
"mapping.total_fields.limit": 5000
}
},
"mappings" : {
"_default_" : {
"_all": { "enabled": false },
"properties" : {
"@timestamp" : { "type" : "date" },
"measurement_name" : { "type" : "keyword" }
},
"dynamic_templates": [
{
"tags": {
"match_mapping_type": "string",
"path_match": "tag.*",
"mapping": {
"ignore_above": 512,
"type": "keyword"
}
}
},
{
"metrics_long": {
"match_mapping_type": "long",
"mapping": {
"type": "float",
"index": false
}
}
},
{
"metrics_double": {
"match_mapping_type": "double",
"mapping": {
"type": "float",
"index": false
}
}
},
{
"text_fields": {
"match": "*",
"mapping": {
"norms": false
}
}
}
]
}
}
}`, templatePattern)
_, errCreateTemplate := a.Client.IndexPutTemplate(a.TemplateName).BodyString(tmpl).Do(ctx)
if errCreateTemplate != nil {
return fmt.Errorf("Elasticsearch failed to create index template %s : %s", a.TemplateName, errCreateTemplate)
}
log.Printf("D! Elasticsearch template %s created or updated\n", a.TemplateName)
} else {
log.Println("D! Found existing Elasticsearch template. Skipping template management")
}
return nil
}
func (a *Elasticsearch) GetIndexName(indexName string, eventTime time.Time) string {
if strings.Contains(indexName, "%") {
var dateReplacer = strings.NewReplacer(
"%Y", eventTime.UTC().Format("2006"),
"%y", eventTime.UTC().Format("06"),
"%m", eventTime.UTC().Format("01"),
"%d", eventTime.UTC().Format("02"),
"%H", eventTime.UTC().Format("15"),
"%V", getISOWeek(eventTime.UTC()),
)
indexName = dateReplacer.Replace(indexName)
}
return indexName
}
func getISOWeek(eventTime time.Time) string {
_, week := eventTime.ISOWeek()
return strconv.Itoa(week)
}
func (a *Elasticsearch) SampleConfig() string {
return sampleConfig
}
func (a *Elasticsearch) Description() string {
return "Configuration for Elasticsearch to send metrics to."
}
func (a *Elasticsearch) Close() error {
a.Client = nil
return nil
}
func init() {
outputs.Add("elasticsearch", func() telegraf.Output {
return &Elasticsearch{
Timeout: internal.Duration{Duration: time.Second * 5},
HealthCheckInterval: internal.Duration{Duration: time.Second * 10},
}
})
}