[beatreceiver] Add support for elasticsearch.parameters

changelog/fragments/1769767252-add-support-for-elasticsearch.parameters-for-beatreceiver.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: enhancement
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: add support for elasticsearch.parameters for beatreceiver
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

internal/pkg/otel/translate/output_elasticsearch.go

@@ -112,22 +112,10 @@ func ToOTelConfig(output *config.C, logger *logp.Logger) (map[string]any, error)
 		return nil, err
 	}
 
-	// Create url using host name, protocol and path
-	outputHosts, err := outputs.ReadHostList(output)
+	hosts, err := getURL(escfg, output)
 	if err != nil {
-		return nil, fmt.Errorf("error reading host list: %w", err)
+		return nil, fmt.Errorf("error creating hosts:%w", err)
 	}
-	hosts := []string{}
-	for _, h := range outputHosts {
-		esURL, err := common.MakeURL(escfg.Protocol, escfg.Path, h, 9200)
-		if err != nil {
-			return nil, fmt.Errorf("cannot generate ES URL from host %w", err)
-		}
-		if !slices.Contains(hosts, esURL) {
-			hosts = append(hosts, esURL)
-		}
-	}
-
 	otelYAMLCfg := map[string]any{
 		"endpoints": hosts, // hosts, protocol, path, port
 
@@ -210,12 +198,45 @@ func getTotalNumWorkers(cfg *config.C) int {
 	return len(hostList)
 }
 
+func getURL(escfg esToOTelOptions, output *config.C) ([]string, error) {
+	// Create url using host name, protocol and path
+	outputHosts, err := outputs.ReadHostList(output)
+	if err != nil {
+		return nil, fmt.Errorf("error reading host list: %w", err)
+	}
+
+	hosts := []string{}
+	for _, h := range outputHosts {
+		esURL, err := common.MakeURL(escfg.Protocol, escfg.Path, h, 9200)
+
+		if err != nil {
+			return nil, fmt.Errorf("cannot generate ES URL from host %w", err)
+		}
+		if !slices.Contains(hosts, esURL) {
+			hosts = append(hosts, esURL)
+		}
+	}
+
+	if len(escfg.Params) != 0 {
+		// convert params to map[string][]string
+		var params = make(map[string][]string, 0)
+		for key, value := range escfg.Params {
+			params[key] = []string{value}
+		}
+
+		decodedParam := url.Values(params)
+		// It is enough to add params as encoded query to any one host
+		// Elasticsearch exporter will make sure to add these for every outgoing request
+		hosts[0] = strings.Join([]string{hosts[0], decodedParam.Encode()}, "?")
+	}
+
+	return hosts, nil
+}
+
 // log warning for unsupported config
 func checkUnsupportedConfig(cfg *config.C) error {
 	if cfg.HasField("indices") {
 		return fmt.Errorf("indices is currently not supported: %w", errors.ErrUnsupported)
-	} else if cfg.HasField("parameters") {
-		return fmt.Errorf("parameters is currently not supported: %w", errors.ErrUnsupported)
 	} else if value, err := cfg.Bool("allow_older_versions", -1); err == nil && !value {
 		return fmt.Errorf("allow_older_versions:false is currently not supported: %w", errors.ErrUnsupported)
 	} else if value, err := cfg.Bool("loadbalance", -1); err == nil && !value {

internal/pkg/otel/translate/output_elasticsearch_test.go

@@ -155,16 +155,18 @@ compression_params:
 		compareAndAssert(t, expOutput, confmap.NewFromStringMap(got))
 	})
 
-	t.Run("test hosts can be a string", func(t *testing.T) {
+	t.Run("test hosts can be a string and parameters is respected", func(t *testing.T) {
 		beatCfg := `
 hosts: "localhost:9200"
 index: "some-index"
 api_key: "TiNAGG4BaaMdaH1tRfuU:KnR6yE41RrSowb0kQ0HWoA"
+parameters:
+  somekey : somevalue
 `
 
 		OTelCfg := `
 endpoints:
-  - http://localhost:9200
+  - http://localhost:9200?somekey=somevalue
 logs_index: some-index
 logs_dynamic_pipeline:
   enabled: true
@@ -667,7 +669,6 @@ func TestToOTelConfig_CheckUnsupported(t *testing.T) {
 		wantErrContains string
 	}{
 		{"indices", map[string]any{"indices": []any{"i"}}, "indices is currently not supported"},
-		{"parameters", map[string]any{"parameters": map[string]any{"x": "y"}}, "parameters is currently not supported"},
 		{"allow_older_versions_false", map[string]any{"allow_older_versions": false}, "allow_older_versions:false is currently not supported"},
 		{"loadbalance_false", map[string]any{"loadbalance": false}, "ladbalance:false is currently not supported"},
 		{"non_indexable_policy", map[string]any{"non_indexable_policy": "x"}, "non_indexable_policy is currently not supported"},