Add compression to elasticsearch output (redpanda-data#853)

* add compression option to elasticsearch writer * generate docs with `make docs` * rename compression -> gzip_compression
codingconcepts · Aug 28, 2021 · 5c14ca1 · 5c14ca1
1 parent a3cad03
commit 5c14ca1
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 19 deletions.
diff --git a/config/elasticsearch.yaml b/config/elasticsearch.yaml
@@ -64,6 +64,7 @@ output:
         token: ""
         role: ""
         role_external_id: ""
+    gzip_compression: false
 logger:
   level: INFO
   format: json

diff --git a/lib/output/elasticsearch.go b/lib/output/elasticsearch.go
@@ -54,6 +54,7 @@ false for connections to succeed.`,
 					docs.FieldCommon("enabled", "Whether to connect to Amazon Elastic Service."),
 				}.Merge(sess.FieldSpecs())...,
 			),
+			docs.FieldAdvanced("gzip_compression", "Whether to enable gzip compression on the request side"),
 		),
 		Categories: []Category{
 			CategoryServices,

diff --git a/lib/output/writer/elasticsearch.go b/lib/output/writer/elasticsearch.go
@@ -37,22 +37,23 @@ type OptionalAWSConfig struct {
 // ElasticsearchConfig contains configuration fields for the Elasticsearch
 // output type.
 type ElasticsearchConfig struct {
-	URLs           []string             `json:"urls" yaml:"urls"`
-	Sniff          bool                 `json:"sniff" yaml:"sniff"`
-	Healthcheck    bool                 `json:"healthcheck" yaml:"healthcheck"`
-	ID             string               `json:"id" yaml:"id"`
-	Action         string               `json:"action" yaml:"action"`
-	Index          string               `json:"index" yaml:"index"`
-	Pipeline       string               `json:"pipeline" yaml:"pipeline"`
-	Routing        string               `json:"routing" yaml:"routing"`
-	Type           string               `json:"type" yaml:"type"`
-	Timeout        string               `json:"timeout" yaml:"timeout"`
-	TLS            btls.Config          `json:"tls" yaml:"tls"`
-	Auth           auth.BasicAuthConfig `json:"basic_auth" yaml:"basic_auth"`
-	AWS            OptionalAWSConfig    `json:"aws" yaml:"aws"`
-	MaxInFlight    int                  `json:"max_in_flight" yaml:"max_in_flight"`
-	retries.Config `json:",inline" yaml:",inline"`
-	Batching       batch.PolicyConfig `json:"batching" yaml:"batching"`
+	URLs            []string             `json:"urls" yaml:"urls"`
+	Sniff           bool                 `json:"sniff" yaml:"sniff"`
+	Healthcheck     bool                 `json:"healthcheck" yaml:"healthcheck"`
+	ID              string               `json:"id" yaml:"id"`
+	Action          string               `json:"action" yaml:"action"`
+	Index           string               `json:"index" yaml:"index"`
+	Pipeline        string               `json:"pipeline" yaml:"pipeline"`
+	Routing         string               `json:"routing" yaml:"routing"`
+	Type            string               `json:"type" yaml:"type"`
+	Timeout         string               `json:"timeout" yaml:"timeout"`
+	TLS             btls.Config          `json:"tls" yaml:"tls"`
+	Auth            auth.BasicAuthConfig `json:"basic_auth" yaml:"basic_auth"`
+	AWS             OptionalAWSConfig    `json:"aws" yaml:"aws"`
+	GzipCompression bool                 `json:"gzip_compression" yaml:"gzip_compression"`
+	MaxInFlight     int                  `json:"max_in_flight" yaml:"max_in_flight"`
+	retries.Config  `json:",inline" yaml:",inline"`
+	Batching        batch.PolicyConfig `json:"batching" yaml:"batching"`
 }
 
 // NewElasticsearchConfig creates a new ElasticsearchConfig with default values.
@@ -79,9 +80,10 @@ func NewElasticsearchConfig() ElasticsearchConfig {
 			Enabled: false,
 			Config:  sess.NewConfig(),
 		},
-		MaxInFlight: 1,
-		Config:      rConf,
-		Batching:    batch.NewPolicyConfig(),
+		GzipCompression: false,
+		MaxInFlight:     1,
+		Config:          rConf,
+		Batching:        batch.NewPolicyConfig(),
 	}
 }
 
@@ -217,6 +219,10 @@ func (e *Elasticsearch) Connect() error {
 		opts = append(opts, elastic.SetHttpClient(signingClient))
 	}
 
+	if e.conf.GzipCompression {
+		opts = append(opts, elastic.SetGzip(true))
+	}
+
 	client, err := elastic.NewClient(opts...)
 	if err != nil {
 		return err

diff --git a/website/docs/components/outputs/elasticsearch.md b/website/docs/components/outputs/elasticsearch.md
@@ -98,6 +98,7 @@ output:
         token: ""
         role: ""
         role_external_id: ""
+    gzip_compression: false
 ```
 
 </TabItem>
@@ -593,4 +594,12 @@ An external ID to provide when assuming a role.
 Type: `string`  
 Default: `""`  
 
+### `gzip_compression`
+
+Whether to enable gzip compression on the request side
+
+
+Type: `bool`  
+Default: `false`  
+