Add SendGetBodyAs on elasticsearch

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
@@ -418,6 +428,7 @@ func (c *Configuration) getConfigOptions(logger *zap.Logger) ([]elastic.ClientOp
 	}
 	options = append(options, elastic.SetHttpClient(httpClient))
 	options = append(options, elastic.SetBasicAuth(c.Username, c.Password))
+	options = append(options, elastic.SetSendGetBodyAs(c.SendGetBodyAs))
 
 	options, err := addLoggerOptions(options, c.LogLevel)
 	if err != nil {

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          = "GET"
 )
 
 // 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,
+		"The Elasticsearch client SendGetBodyAs parameter ; allows to override GET verb for HTTP requests which contain a body.")
 
 	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)