feat(outputs.elasticsearch): Allow settings extra headers for elastic…

…search output
influxdata · Jun 8, 2024 · 043e20a · 043e20a
1 parent 94a5ab6
commit 043e20a
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 20 deletions.
diff --git a/plugins/outputs/elasticsearch/README.md b/plugins/outputs/elasticsearch/README.md
@@ -323,6 +323,11 @@ to use them.
   ## no pipeline is used for the metric.
   # use_pipeline = "{{es_pipeline}}"
   # default_pipeline = "my_pipeline"
+  #
+  # Custom HTTP headers
+  # To pass extra HTTP headers please define it in a given below section
+  # [outputs.elasticsearch.extra_headers]
+  #    "X-Custom-Header" = "custom-value"
 ```
 
 ### Permissions
@@ -389,6 +394,8 @@ the `default_tag_value` will be used instead.
   instead.
 * `default_pipeline`: If dynamic pipeline names the tag does not exist in a
   particular metric, this value will be used instead.
+* `extra_headers`: Extra HTTP headers, which are passed to Elasticsearch header
+  before each request.
 
 ## Known issues
 

diff --git a/plugins/outputs/elasticsearch/elasticsearch.go b/plugins/outputs/elasticsearch/elasticsearch.go
@@ -28,26 +28,27 @@ 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"`
-	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"`
+	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"`
+	ExtraHeaders        map[string]string `toml:"extra_headers"`
+	Log                 telegraf.Logger   `toml:"-"`
 	majorReleaseNumber  int
 	pipelineName        string
 	pipelineTagKeys     []string
@@ -183,6 +184,16 @@ func (a *Elasticsearch) Connect() error {
 		elastic.SetGzip(a.EnableGzip),
 	)
 
+	if len(a.ExtraHeaders) > 0 {
+		extraHeaders := http.Header{}
+		for k, vals := range a.ExtraHeaders {
+			for _, v := range strings.Split(vals, ",") {
+				extraHeaders.Add(k, v)
+			}
+		}
+		clientOptions = append(clientOptions, elastic.SetHeaders(extraHeaders))
+	}
+
 	authOptions, err := a.getAuthOptions()
 	if err != nil {
 		return err