Prepare OTel Elasticsearch exporter with the credentials of the output (#5469)

When an OTel exporter is found in the configuration, and it is an Elasticsearch
exporter, it looks for a prepared output with the same name, and copies its
credentials.

Author: jsoriano

changelog/fragments/1758147274-Add-credentials-to-Elasticsearch-exporters.yaml

@@ -0,0 +1,32 @@
+# 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
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add credentials to OTel Elasticsearch exporters
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: When a policy includes OTel configuration with Elasticsearch exporters, it configures their credentials using the credentials in the Elasticsearch output.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: fleet-server
+
+# PR URL; optional; the PR number 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/elastic/fleet-server/pull/5469
+
+# 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/api/handleCheckin.go

@@ -9,6 +9,7 @@ import (
 	"compress/flate"
 	"compress/gzip"
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -17,6 +18,7 @@ import (
 	"net/http"
 	"reflect"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -876,6 +878,11 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a
 				policyOutput.Name, err)
 		}
 	}
+	// Prepare OTel exporters from the information in outputs.
+	if err := prepareOTelExporters(data.Outputs, data.Exporters); err != nil {
+		return nil, fmt.Errorf("failed to prepare OTel exporters: %w", err)
+	}
+
 	// Add replace inputs with agent prepared version.
 	data.Inputs = pp.Inputs
 
@@ -911,6 +918,49 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a
 	return &resp, nil
 }
 
+// prepareOTelExporters prepares OTel exporters by copying credentials and potentially other
+// settings from already prepared outputs.
+func prepareOTelExporters(outputs map[string]map[string]any, exporters map[string]any) error {
+	for id, c := range exporters {
+		var config map[string]any
+		if c == nil {
+			config = make(map[string]any)
+		} else if cmap, ok := c.(map[string]any); ok {
+			config = cmap
+		} else {
+			return fmt.Errorf("unexpected config type for %q, expected map, found %T", id, c)
+		}
+
+		exporterType, name, found := strings.Cut(id, "/")
+		if !found {
+			return fmt.Errorf("unexpected exporter id format %q", id)
+		}
+
+		output, found := outputs[name]
+		if !found {
+			return fmt.Errorf("output %q not found for exporter %q", name, id)
+		}
+
+		switch exporterType {
+		case policy.OTelExporterTypeElasticsearch:
+			ot, ok := output["type"].(string)
+			if !ok || (ot != policy.OutputTypeElasticsearch && ot != policy.OutputTypeRemoteElasticsearch) {
+				return fmt.Errorf("unexpected output type %q found for exporter %q", name, id)
+			}
+			apiKey, ok := output["api_key"].(string)
+			if !ok || apiKey == "" {
+				return fmt.Errorf("api key not found in output %q for exporter %q", name, id)
+			}
+			config["api_key"] = base64.StdEncoding.EncodeToString([]byte(apiKey))
+		default:
+			return fmt.Errorf("OTel exporter %q not supported", exporterType)
+		}
+
+		exporters[id] = config
+	}
+	return nil
+}
+
 func getAgentAndVerifyAPIKeyID(ctx context.Context, bulker bulk.Bulk, agentID string, apiKeyID string) (*model.Agent, error) {
 	span, ctx := apm.StartSpan(ctx, "getAgentAndVerifyAPIKeyID", "read")
 	defer span.End()

internal/pkg/policy/policy_output.go

@@ -29,6 +29,8 @@ const (
 	OutputTypeRemoteElasticsearch = "remote_elasticsearch"
 	OutputTypeLogstash            = "logstash"
 	OutputTypeKafka               = "kafka"
+
+	OTelExporterTypeElasticsearch = "elasticsearch"
 )
 
 var (

internal/pkg/server/fleet_integration_test.go

@@ -10,6 +10,7 @@ package server
 import (
 	"bytes"
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -26,6 +27,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/hashicorp/go-cleanhttp"
 	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
@@ -1675,8 +1677,7 @@ func Test_SmokeTest_AuditUnenroll(t *testing.T) {
 }
 
 func TestCheckinOTelColPolicy(t *testing.T) {
-	ctx, cancel := context.WithCancel(t.Context())
-	defer cancel()
+	ctx := t.Context()
 
 	idSuffix := uuid.Must(uuid.NewV4()).String()
 	componentID := func(id string) string {
@@ -1697,21 +1698,21 @@ func TestCheckinOTelColPolicy(t *testing.T) {
 			componentID("someprocessor"): map[string]any{},
 		},
 		Connectors: map[string]any{
-			componentID("forward"): map[string]any{},
+			"forward": map[string]any{},
 		},
 		Exporters: map[string]any{
-			componentID("someexporter"): map[string]any{},
+			"elasticsearch/default": map[string]any{},
 		},
 		Service: &model.Service{
 			Pipelines: map[string]*model.PipelinesItem{
 				componentID("metrics"): &model.PipelinesItem{
 					Receivers:  []string{componentID("somereceiver")},
 					Processors: []string{componentID("someprocessor")},
-					Exporters:  []string{componentID("forward")},
+					Exporters:  []string{"forward"},
 				},
 				"metrics": &model.PipelinesItem{
-					Receivers: []string{componentID("forward")},
-					Exporters: []string{componentID("someexporter")},
+					Receivers: []string{"forward"},
+					Exporters: []string{"elasticsearch/default"},
 				},
 			},
 		},
@@ -1737,18 +1738,16 @@ func TestCheckinOTelColPolicy(t *testing.T) {
 	t.Log("Agent enrollment successful")
 	p, _ := io.ReadAll(res.Body)
 	res.Body.Close()
-	var obj map[string]interface{}
-	err = json.Unmarshal(p, &obj)
+	var enrollResponse struct {
+		Item struct {
+			ID           string `json:"id"`
+			AccessApiKey string `json:"access_api_key"`
+		} `json:"item"`
+	}
+	err = json.Unmarshal(p, &enrollResponse)
 	require.NoError(t, err)
-
-	item := obj["item"]
-	mm, ok := item.(map[string]interface{})
-	require.True(t, ok, "expected attribute item to be an object")
-	agentID, ok := mm["id"].(string)
-	require.True(t, ok, "expected attribute id to be a string")
-
-	apiKey, ok := mm["access_api_key"].(string)
-	require.True(t, ok, "expected attribute apiKey to be a string")
+	agentID := enrollResponse.Item.ID
+	apiKey := enrollResponse.Item.AccessApiKey
 
 	// checkin
 	t.Logf("Fake a checkin for agent %s", agentID)
@@ -1766,19 +1765,40 @@ func TestCheckinOTelColPolicy(t *testing.T) {
 	p, err = io.ReadAll(res.Body)
 	require.NoError(t, err)
 
-	err = json.Unmarshal(p, &obj)
+	var checkinResponse struct {
+		Actions []struct {
+			AgentID string `json:"agent_id"`
+			ID      string `json:"id"`
+			Data    struct {
+				Policy struct {
+					Exporters map[string]struct {
+						ApiKey string `json:"api_key"`
+					} `json:"exporters"`
+					Outputs map[string]struct {
+						ApiKey string `json:"api_key"`
+						Type   string `json:"type"`
+					} `json:"outputs"`
+				} `json:"policy"`
+			} `json:"data"`
+		} `json:"actions"`
+	}
+	err = json.Unmarshal(p, &checkinResponse)
 	require.NoError(t, err)
 
-	actionsRaw, ok := obj["actions"]
-	require.True(t, ok, "expected actions is missing")
-	actions, ok := actionsRaw.([]interface{})
-	require.True(t, ok, "expected actions to be an array")
-	require.Greater(t, len(actions), 0, "expected at least 1 action")
-	action, ok := actions[0].(map[string]interface{})
-	require.True(t, ok, "expected action to be an object")
-	_, ok = action["id"].(string)
-	require.True(t, ok, "expected action id to be string")
-	aAgentID, ok := action["agent_id"].(string)
-	require.True(t, ok, "expected action agent_id to be string")
-	require.Equal(t, agentID, aAgentID)
+	require.Len(t, checkinResponse.Actions, 1, "expected 1 action")
+
+	action := checkinResponse.Actions[0]
+	assert.NotEmpty(t, action.ID)
+	assert.Equal(t, agentID, action.AgentID)
+
+	output, found := action.Data.Policy.Outputs["default"]
+	require.True(t, found, "default output not found")
+	require.Equal(t, "elasticsearch", output.Type)
+	require.NotEmpty(t, output.ApiKey)
+
+	exporter, found := action.Data.Policy.Exporters["elasticsearch/default"]
+	require.True(t, found, "default exporter not found")
+	encodedApiKey := base64.StdEncoding.EncodeToString([]byte(output.ApiKey))
+
+	assert.Equal(t, encodedApiKey, exporter.ApiKey)
 }