Add SendGetBodyAs on elasticsearch (#3193)

Signed-off-by: Nathanael Marchand <nathanael.marchand@outlook.com>

pkg/es/config/config.go

@@ -75,6 +75,7 @@ type Configuration struct {
 	UseILM                         bool           `mapstructure:"use_ilm"`
 	Version                        uint           `mapstructure:"version"`
 	LogLevel                       string         `mapstructure:"log_level"`
+	SendGetBodyAs                  string         `mapstructure:"send_get_body_as"`
 }
 
 // TagsAsFields holds configuration for tag schema.
@@ -116,6 +117,7 @@ type ClientBuilder interface {
 	TagKeysAsFields() ([]string, error)
 	GetUseILM() bool
 	GetLogLevel() string
+	GetSendGetBodyAs() string
 }
 
 // NewClient creates a new ElasticSearch client
@@ -257,6 +259,9 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 	if c.LogLevel == "" {
 		c.LogLevel = source.LogLevel
 	}
+	if c.SendGetBodyAs == "" {
+		c.SendGetBodyAs = source.SendGetBodyAs
+	}
 }
 
 // GetRemoteReadClusters returns list of remote read clusters
@@ -356,6 +361,11 @@ func (c *Configuration) GetLogLevel() string {
 	return c.LogLevel
 }
 
+// GetSendGetBodyAs returns the SendGetBodyAs the ES client should use.
+func (c *Configuration) GetSendGetBodyAs() string {
+	return c.SendGetBodyAs
+}
+
 // GetTokenFilePath returns file path containing the bearer token
 func (c *Configuration) GetTokenFilePath() string {
 	return c.TokenFilePath
@@ -419,6 +429,10 @@ func (c *Configuration) getConfigOptions(logger *zap.Logger) ([]elastic.ClientOp
 	options = append(options, elastic.SetHttpClient(httpClient))
 	options = append(options, elastic.SetBasicAuth(c.Username, c.Password))
 
+	if c.SendGetBodyAs != "" {
+		options = append(options, elastic.SetSendGetBodyAs(c.SendGetBodyAs))
+	}
+
 	options, err := addLoggerOptions(options, c.LogLevel)
 	if err != nil {
 		return options, err

plugin/storage/es/options.go

@@ -59,6 +59,7 @@ const (
 	suffixVersion                        = ".version"
 	suffixMaxDocCount                    = ".max-doc-count"
 	suffixLogLevel                       = ".log-level"
+	suffixSendGetBodyAs                  = ".send-get-body-as"
 	// default number of documents to return from a query (elasticsearch allowed limit)
 	// see search.max_buckets and index.max_result_window
 	defaultMaxDocCount        = 10_000
@@ -68,6 +69,7 @@ const (
 	defaultIndexDateSeparator = "-"
 
 	defaultIndexRolloverFrequency = "day"
+	defaultSendGetBodyAs          = ""
 )
 
 // TODO this should be moved next to config.Configuration struct (maybe ./flags package)
@@ -110,6 +112,7 @@ func NewOptions(primaryNamespace string, otherNamespaces ...string) *Options {
 		RemoteReadClusters:   []string{},
 		MaxDocCount:          defaultMaxDocCount,
 		LogLevel:             "error",
+		SendGetBodyAs:        defaultSendGetBodyAs,
 	}
 	options := &Options{
 		Primary: namespaceConfig{
@@ -265,6 +268,10 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixLogLevel,
 		nsConfig.LogLevel,
 		"The Elasticsearch client log-level. Valid levels: [debug, info, error]")
+	flagSet.String(
+		nsConfig.namespace+suffixSendGetBodyAs,
+		nsConfig.SendGetBodyAs,
+		"HTTP verb for requests that contain a body [GET, POST].")
 
 	if nsConfig.namespace == archiveNamespace {
 		flagSet.Bool(
@@ -315,6 +322,7 @@ func initFromViper(cfg *namespaceConfig, v *viper.Viper) {
 	cfg.CreateIndexTemplates = v.GetBool(cfg.namespace + suffixCreateIndexTemplate)
 	cfg.Version = uint(v.GetInt(cfg.namespace + suffixVersion))
 	cfg.LogLevel = v.GetString(cfg.namespace + suffixLogLevel)
+	cfg.SendGetBodyAs = v.GetString(cfg.namespace + suffixSendGetBodyAs)
 
 	cfg.MaxDocCount = v.GetInt(cfg.namespace + suffixMaxDocCount)
 	cfg.UseILM = v.GetBool(cfg.namespace + suffixUseILM)

plugin/storage/es/options_test.go

@@ -74,6 +74,7 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--es.tags-as-fields.config-file=./file.txt",
 		"--es.tags-as-fields.dot-replacement=!",
 		"--es.use-ilm=true",
+		"--es.send-get-body-as=POST",
 	})
 	require.NoError(t, err)
 	opts.InitFromViper(v)
@@ -109,6 +110,7 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, "2006.01.02", aux.IndexDateLayoutServices)
 	assert.Equal(t, "2006.01.02.15", aux.IndexDateLayoutSpans)
 	assert.True(t, primary.UseILM)
+	assert.Equal(t, "POST", aux.SendGetBodyAs)
 }
 
 func TestEmptyRemoteReadClusters(t *testing.T) {