package elasticsearch import ( "context" "fmt" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/internal" "github.com/influxdata/telegraf/plugins/outputs" "gopkg.in/olivere/elastic.v5" "log" "net/http" "strconv" "strings" "time" ) 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) 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"), ) indexName = dateReplacer.Replace(indexName) } return indexName } 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}, } }) }