CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(outputs.elasticsearch): Allow custom template index settings (#15471)

This commit is contained in:
Dane Strandboge 2024-06-11 03:40:33 -05:00 committed by GitHub
parent 28f6396348
commit 6eef8aa15c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 132 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
plugins/outputs/elasticsearch/README.md
View File

@ -328,6 +328,16 @@ to use them.
# To pass custom HTTP headers please define it in a given below section
# [outputs.elasticsearch.headers]
# "X-Custom-Header" = "custom-value"
## Template Index Settings
## Overrides the template settings.index section with any provided options.
## Defaults provided here in the config
# template_index_settings = {
# refresh_interval = "10s",
# mapping.total_fields.limit = 5000,
# auto_expand_replicas = "0-1",
# codec = "best_compression"
# }
```
### Permissions

100
plugins/outputs/elasticsearch/elasticsearch.go
View File

@ -6,6 +6,7 @@ import (
"context"
"crypto/sha256"
_ "embed"
"encoding/json"
"errors"
"fmt"
"math"
@ -28,27 +29,28 @@ import (
var sampleConfig string
type Elasticsearch struct {
AuthBearerToken config.Secret `toml:"auth_bearer_token"`
DefaultPipeline string `toml:"default_pipeline"`
DefaultTagValue string `toml:"default_tag_value"`
EnableGzip bool `toml:"enable_gzip"`
EnableSniffer bool `toml:"enable_sniffer"`
FloatHandling string `toml:"float_handling"`
FloatReplacement float64 `toml:"float_replacement_value"`
ForceDocumentID bool `toml:"force_document_id"`
HealthCheckInterval config.Duration `toml:"health_check_interval"`
HealthCheckTimeout config.Duration `toml:"health_check_timeout"`
IndexName string `toml:"index_name"`
ManageTemplate bool `toml:"manage_template"`
OverwriteTemplate bool `toml:"overwrite_template"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
TemplateName string `toml:"template_name"`
Timeout config.Duration `toml:"timeout"`
URLs []string `toml:"urls"`
UsePipeline string `toml:"use_pipeline"`
Headers map[string]string `toml:"headers"`
Log telegraf.Logger `toml:"-"`
AuthBearerToken config.Secret `toml:"auth_bearer_token"`
DefaultPipeline string `toml:"default_pipeline"`
DefaultTagValue string `toml:"default_tag_value"`
EnableGzip bool `toml:"enable_gzip"`
EnableSniffer bool `toml:"enable_sniffer"`
FloatHandling string `toml:"float_handling"`
FloatReplacement float64 `toml:"float_replacement_value"`
ForceDocumentID bool `toml:"force_document_id"`
HealthCheckInterval config.Duration `toml:"health_check_interval"`
HealthCheckTimeout config.Duration `toml:"health_check_timeout"`
IndexName string `toml:"index_name"`
IndexTemplate map[string]interface{} `toml:"template_index_settings"`
ManageTemplate bool `toml:"manage_template"`
OverwriteTemplate bool `toml:"overwrite_template"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
TemplateName string `toml:"template_name"`
Timeout config.Duration `toml:"timeout"`
URLs []string `toml:"urls"`
UsePipeline string `toml:"use_pipeline"`
Headers map[string]string `toml:"headers"`
Log telegraf.Logger `toml:"-"`
majorReleaseNumber int
pipelineName string
pipelineTagKeys []string
@ -66,12 +68,7 @@ const telegrafTemplate = `
"index_patterns" : [ "{{.TemplatePattern}}" ],
{{ end }}
"settings": {
"index": {
"refresh_interval": "10s",
"mapping.total_fields.limit": 5000,
"auto_expand_replicas" : "0-1",
"codec" : "best_compression"
}
"index": {{.IndexTemplate}}
},
"mappings" : {
{{ if (lt .Version 7) }}
@ -128,9 +125,18 @@ const telegrafTemplate = `
}
}`
const defaultTemplateIndexSettings = `
{
"refresh_interval": "10s",
"mapping.total_fields.limit": 5000,
"auto_expand_replicas": "0-1",
"codec": "best_compression"
}`
type templatePart struct {
TemplatePattern string
Version int
IndexTemplate string
}
func (*Elasticsearch) SampleConfig() string {
@ -369,19 +375,12 @@ func (a *Elasticsearch) manageTemplate(ctx context.Context) error {
}
if (a.OverwriteTemplate) || (!templateExists) || (templatePattern != "") {
tp := templatePart{
TemplatePattern: templatePattern + "*",
Version: a.majorReleaseNumber,
}
t := template.Must(template.New("template").Parse(telegrafTemplate))
var tmpl bytes.Buffer
if err := t.Execute(&tmpl, tp); err != nil {
data, err := a.createNewTemplate(templatePattern)
if err != nil {
return err
}
_, errCreateTemplate := a.Client.IndexPutTemplate(a.TemplateName).BodyString(tmpl.String()).Do(ctx)
_, errCreateTemplate := a.Client.IndexPutTemplate(a.TemplateName).BodyString(data.String()).Do(ctx)
if errCreateTemplate != nil {
return fmt.Errorf("elasticsearch failed to create index template %s: %w", a.TemplateName, errCreateTemplate)
}
@ -393,6 +392,33 @@ func (a *Elasticsearch) manageTemplate(ctx context.Context) error {
return nil
}
func (a *Elasticsearch) createNewTemplate(templatePattern string) (*bytes.Buffer, error) {
var indexTemplate string
if a.IndexTemplate != nil {
data, err := json.Marshal(&a.IndexTemplate)
if err != nil {
return nil, fmt.Errorf("elasticsearch failed to create index settings for template %s: %w", a.TemplateName, err)
}
indexTemplate = string(data)
} else {
indexTemplate = defaultTemplateIndexSettings
}
tp := templatePart{
TemplatePattern: templatePattern + "*",
Version: a.majorReleaseNumber,
IndexTemplate: indexTemplate,
}
t := template.Must(template.New("template").Parse(telegrafTemplate))
var tmpl bytes.Buffer
if err := t.Execute(&tmpl, tp); err != nil {
return nil, err
}
return &tmpl, nil
}
func (a *Elasticsearch) GetTagKeys(indexName string) (string, []string) {
tagKeys := []string{}
startTag := strings.Index(indexName, "{{")

49
plugins/outputs/elasticsearch/elasticsearch_test.go
View File

@ -2,6 +2,7 @@ package elasticsearch
import (
"context"
"encoding/json"
"fmt"
"math"
"net/http"
@ -755,3 +756,51 @@ func TestAuthorizationHeaderWhenBearerTokenIsPresent(t *testing.T) {
err = e.Write(testutil.MockMetrics())
require.NoError(t, err)
}
func TestStandardIndexSettings(t *testing.T) {
e := &Elasticsearch{
TemplateName: "test",
IndexName: "telegraf-%Y.%m.%d",
Log: testutil.Logger{},
}
buf, err := e.createNewTemplate("test")
require.NoError(t, err)
var jsonData esTemplate
err = json.Unmarshal(buf.Bytes(), &jsonData)
require.NoError(t, err)
index := jsonData.Settings.Index
require.Equal(t, "10s", index["refresh_interval"])
require.Equal(t, float64(5000), index["mapping.total_fields.limit"])
require.Equal(t, "0-1", index["auto_expand_replicas"])
require.Equal(t, "best_compression", index["codec"])
}
func TestDifferentIndexSettings(t *testing.T) {
e := &Elasticsearch{
TemplateName: "test",
IndexName: "telegraf-%Y.%m.%d",
IndexTemplate: map[string]interface{}{
"refresh_interval": "20s",
"mapping.total_fields.limit": 1000,
"codec": "best_compression",
},
Log: testutil.Logger{},
}
buf, err := e.createNewTemplate("test")
require.NoError(t, err)
var jsonData esTemplate
err = json.Unmarshal(buf.Bytes(), &jsonData)
require.NoError(t, err)
index := jsonData.Settings.Index
require.Equal(t, "20s", index["refresh_interval"])
require.Equal(t, float64(1000), index["mapping.total_fields.limit"])
require.Equal(t, "best_compression", index["codec"])
}
type esTemplate struct {
Settings esSettings `json:"settings"`
}
type esSettings struct {
Index map[string]interface{} `json:"index"`
}

10
plugins/outputs/elasticsearch/sample.conf
View File

@ -82,3 +82,13 @@
# To pass custom HTTP headers please define it in a given below section
# [outputs.elasticsearch.headers]
# "X-Custom-Header" = "custom-value"
## Template Index Settings
## Overrides the template settings.index section with any provided options.
## Defaults provided here in the config
# template_index_settings = {
# refresh_interval = "10s",
# mapping.total_fields.limit = 5000,
# auto_expand_replicas = "0-1",
# codec = "best_compression"
# }