feat(elasticsearch): add flag to enable gzip compression (#7072)

## Which problem is this PR solving?
Currently, Jaeger sends its traces to ElasticSearch as uncompressed
text. Since text is can be compressed quite well, enabling Gzip
compression can significantly reduce Jaeger's network traffic.
ElasticSearch has accepted compressed requests since version 5.0 and
since the same version it has already sent compressed responses by
default (cf.
elastic/elasticsearch@0a6f40c).

## Description of the changes
* 🛑 (breaking change) **Enable by default** the compression for requests
to ElasticSearch
* Add a new flag `--es.http-compression=true|false` that can be used to
opt-out of compression . The setting is already supported by both client
libraries that are used.

## How was this change tested?
I tested the change running a local ElasticSearch instance and
`SPAN_STORAGE_TYPE=elasticsearch ./cmd/collector/collector-linux-amd64
--es.http-compression=true`. I sent traces to Jaeger using `tracepusher`
and observed the network traffic between Jaeger and ElasticSearch using
`tcpdump` to verify that the traffic is indeed compressed.

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [x] I have added unit tests for the new functionality
- [x] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `npm run lint` and `npm run test`

---------

Signed-off-by: Timon Engelke <[email protected]>

Author: timonegk

internal/storage/elasticsearch/config/config.go

@@ -99,6 +99,8 @@ type Configuration struct {
 	SendGetBodyAs string `mapstructure:"send_get_body_as"`
 	// QueryTimeout contains the timeout used for queries. A timeout of zero means no timeout.
 	QueryTimeout time.Duration `mapstructure:"query_timeout"`
+	// HTTPCompression can be set to false to disable gzip compression for requests to ElasticSearch
+	HTTPCompression bool `mapstructure:"http_compression"`
 
 	// ---- elasticsearch client related configs ----
 	BulkProcessing BulkProcessing `mapstructure:"bulk_processing"`
@@ -319,6 +321,7 @@ func newElasticsearchV8(c *Configuration, logger *zap.Logger) (*esV8.Client, err
 	options.Username = c.Authentication.BasicAuthentication.Username
 	options.Password = c.Authentication.BasicAuthentication.Password
 	options.DiscoverNodesOnStart = c.Sniffing.Enabled
+	options.CompressRequestBody = c.HTTPCompression
 	transport, err := GetHTTPRoundTripper(c, logger)
 	if err != nil {
 		return nil, err
@@ -413,6 +416,9 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 	if c.SendGetBodyAs == "" {
 		c.SendGetBodyAs = source.SendGetBodyAs
 	}
+	if !c.HTTPCompression {
+		c.HTTPCompression = source.HTTPCompression
+	}
 }
 
 // RolloverFrequencyAsNegativeDuration returns the index rollover frequency duration for the given frequency string
@@ -486,6 +492,7 @@ func (c *Configuration) getConfigOptions(logger *zap.Logger) ([]elastic.ClientOp
 	if c.SendGetBodyAs != "" {
 		options = append(options, elastic.SetSendGetBodyAs(c.SendGetBodyAs))
 	}
+	options = append(options, elastic.SetGzip(c.HTTPCompression))
 
 	options, err := addLoggerOptions(options, c.LogLevel, logger)
 	if err != nil {

internal/storage/v1/elasticsearch/options.go

@@ -57,6 +57,7 @@ const (
 	suffixMaxDocCount                    = ".max-doc-count"
 	suffixLogLevel                       = ".log-level"
 	suffixSendGetBodyAs                  = ".send-get-body-as"
+	suffixHTTPCompression                = ".http-compression"
 	// default number of documents to return from a query (elasticsearch allowed limit)
 	// see search.max_buckets and index.max_result_window
 	defaultMaxDocCount        = 10_000
@@ -268,6 +269,10 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixSendGetBodyAs,
 		nsConfig.SendGetBodyAs,
 		"HTTP verb for requests that contain a body [GET, POST].")
+	flagSet.Bool(
+		nsConfig.namespace+suffixHTTPCompression,
+		nsConfig.HTTPCompression,
+		"Use gzip compression for requests to ElasticSearch.")
 	flagSet.Duration(
 		nsConfig.namespace+suffixAdaptiveSamplingLookback,
 		nsConfig.AdaptiveSamplingLookback,
@@ -339,6 +344,7 @@ func initFromViper(cfg *namespaceConfig, v *viper.Viper) {
 	cfg.Version = v.GetUint(cfg.namespace + suffixVersion)
 	cfg.LogLevel = v.GetString(cfg.namespace + suffixLogLevel)
 	cfg.SendGetBodyAs = v.GetString(cfg.namespace + suffixSendGetBodyAs)
+	cfg.HTTPCompression = v.GetBool(cfg.namespace + suffixHTTPCompression)
 
 	cfg.MaxDocCount = v.GetInt(cfg.namespace + suffixMaxDocCount)
 	cfg.UseILM = v.GetBool(cfg.namespace + suffixUseILM)
@@ -421,6 +427,7 @@ func DefaultConfig() config.Configuration {
 		MaxDocCount:          defaultMaxDocCount,
 		LogLevel:             "error",
 		SendGetBodyAs:        defaultSendGetBodyAs,
+		HTTPCompression:      true,
 		Indices: config.Indices{
 			Spans:        defaultIndexOptions,
 			Services:     defaultIndexOptions,

internal/storage/v1/elasticsearch/options_test.go

@@ -63,6 +63,7 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--es.tags-as-fields.dot-replacement=!",
 		"--es.use-ilm=true",
 		"--es.send-get-body-as=POST",
+		"--es.http-compression=true",
 	})
 	require.NoError(t, err)
 	opts.InitFromViper(v)
@@ -86,6 +87,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, "20060102", primary.Indices.Services.DateLayout)
 	assert.Equal(t, "2006010215", primary.Indices.Spans.DateLayout)
 	assert.True(t, primary.UseILM)
+	assert.True(t, primary.HTTPCompression)
 }
 
 func TestEmptyRemoteReadClusters(t *testing.T) {