Support for remote ES output

Makefile

@@ -297,12 +297,13 @@ export $(shell sed 's/=.*//' ./dev-tools/integration/.env)
 # Start ES with docker without waiting
 .PHONY: int-docker-start-async
 int-docker-start-async:
-	@docker compose -f ./dev-tools/integration/docker-compose.yml --env-file ./dev-tools/integration/.env up  -d --remove-orphans elasticsearch
+	@docker compose -f ./dev-tools/integration/docker-compose.yml --env-file ./dev-tools/integration/.env up  -d --remove-orphans elasticsearch elasticsearch-remote
 
 # Wait for ES to be ready
 .PHONY: int-docker-wait
 int-docker-wait:
 	@./dev-tools/integration/wait-for-elasticsearch.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS}
+	@./dev-tools/integration/wait-for-elasticsearch.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_REMOTE_ELASTICSEARCH_HOST}
 
 # Start integration docker setup with wait for when the ES is ready
 .PHONY: int-docker-start
@@ -331,7 +332,8 @@ test-int: prepare-test-context  ## - Run integration tests with full setup (slow
 .PHONY: test-int-set
 test-int-set: ## - Run integration tests without setup
 	# Initialize indices one before running all the tests
-	ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS}) \
+	ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS} "fleet-server") \
+	REMOTE_ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_REMOTE_ELASTICSEARCH_HOST} "fleet-server-remote") \
 	ELASTICSEARCH_HOSTS=${TEST_ELASTICSEARCH_HOSTS} ELASTICSEARCH_USERNAME=${ELASTICSEARCH_USERNAME} ELASTICSEARCH_PASSWORD=${ELASTICSEARCH_PASSWORD} \
 	go test -v -tags=integration -count=1 -race -p 1 ./...
 
@@ -372,7 +374,7 @@ test-e2e: docker-cover-e2e-binaries build-e2e-agent-image e2e-certs build-docker
 .PHONY: test-e2e-set
 test-e2e-set: ## - Run the blackbox end to end tests without setup.
 	cd testing; \
-	ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS}) \
+	ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS} "fleet-server") \
 	ELASTICSEARCH_HOSTS=${TEST_ELASTICSEARCH_HOSTS} ELASTICSEARCH_USERNAME=${ELASTICSEARCH_USERNAME} ELASTICSEARCH_PASSWORD=${ELASTICSEARCH_PASSWORD} \
 	AGENT_E2E_IMAGE=$(shell cat "build/e2e-image") \
 	STANDALONE_E2E_IMAGE=$(DOCKER_IMAGE):$(DOCKER_IMAGE_TAG)$(if $(DEV),-dev,) \

changelog/fragments/1698660041-remote-es-output.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: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Support for remote elasticsearch output
+
+# 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:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component:
+
+# 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: 3051
+
+# 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

dev-tools/integration/.env

@@ -2,3 +2,4 @@ ELASTICSEARCH_VERSION=8.12.0-d6c4164b-SNAPSHOT
 ELASTICSEARCH_USERNAME=elastic
 ELASTICSEARCH_PASSWORD=changeme
 TEST_ELASTICSEARCH_HOSTS=localhost:9200
+TEST_REMOTE_ELASTICSEARCH_HOST=localhost:9201

dev-tools/integration/docker-compose.yml

@@ -23,3 +23,27 @@ services:
       - ./elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml
     ports:
       - 127.0.0.1:9200:9200
+
+  elasticsearch-remote:
+    image: "docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION}-amd64"
+    container_name: elasticsearch-remote
+    environment:
+      - node.name=es02
+      - cluster.name=es-docker-cluster2
+      - discovery.seed_hosts=elasticsearch
+      - bootstrap.memory_lock=true
+      - "ES_JAVA_OPTS=-Xms1G -Xmx1G"
+      - "ELASTIC_USERNAME=${ELASTICSEARCH_USERNAME}"
+      - "ELASTIC_PASSWORD=${ELASTICSEARCH_PASSWORD}"
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+      nofile:
+          soft: 65536
+          hard: 65536
+    volumes:
+      - ./elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml
+    ports:
+      - 127.0.0.1:9201:9200
+

dev-tools/integration/get-elasticsearch-servicetoken.sh

@@ -1,8 +1,9 @@
 #!/bin/bash
 
 host="$1"
+account="$2"
 
-jsonBody="$(curl -sSL -XPOST "$host/_security/service/elastic/fleet-server/credential/token/token1")"
+jsonBody="$(curl -sSL -XPOST "$host/_security/service/elastic/$account/credential/token/token1")"
 
 # use grep and sed to get the service token value as we may not have jq or a similar tool on the instance
 token=$(echo ${jsonBody} |  grep -Eo '"value"[^}]*' | grep -Eo ':.*' | sed -r "s/://" | sed -r 's/"//g')
@@ -11,9 +12,9 @@ token=$(echo ${jsonBody} |  grep -Eo '"value"[^}]*' | grep -Eo ':.*' | sed -r "s
 # very useful during development, without recreating elasticsearch instance every time.
 if [ -z "$token" ]
 then
-    token=`cat .service_token` 
+    token=`cat .service_token_$account` 
 else
-    echo "$token" > .service_token
+    echo "$token" > .service_token_$account
 fi
 
 echo $token

internal/pkg/api/handleAck.go

@@ -416,15 +416,15 @@ func (ack *AckT) handlePolicyChange(ctx context.Context, zlog zerolog.Logger, ag
 		return nil
 	}
 
-	for _, output := range agent.Outputs {
+	for outputName, output := range agent.Outputs {
 		if output.Type != policy.OutputTypeElasticsearch {
 			continue
 		}
 
 		err := ack.updateAPIKey(ctx,
 			zlog,
 			agent.Id,
-			output.APIKeyID, output.PermissionsHash, output.ToRetireAPIKeyIds)
+			output.APIKeyID, output.PermissionsHash, output.ToRetireAPIKeyIds, outputName)
 		if err != nil {
 			return err
 		}
@@ -445,20 +445,31 @@ func (ack *AckT) updateAPIKey(ctx context.Context,
 	zlog zerolog.Logger,
 	agentID string,
 	apiKeyID, permissionHash string,
-	toRetireAPIKeyIDs []model.ToRetireAPIKeyIdsItems) error {
+	toRetireAPIKeyIDs []model.ToRetireAPIKeyIdsItems, outputName string) error {
+	bulk := ack.bulk
+	// use output bulker if exists
+	if outputName != "" {
+		outputBulk := ack.bulk.GetBulker(outputName)
+		if outputBulk != nil {
+			zlog.Debug().Str("outputName", outputName).Msg("Using output bulker in updateAPIKey")
+			bulk = outputBulk
+		}
+	}
 	if apiKeyID != "" {
-		res, err := ack.bulk.APIKeyRead(ctx, apiKeyID, true)
+		res, err := bulk.APIKeyRead(ctx, apiKeyID, true)
 		if err != nil {
 			if isAgentActive(ctx, zlog, ack.bulk, agentID) {
 				zlog.Error().
 					Err(err).
 					Str(LogAPIKeyID, apiKeyID).
+					Str("outputName", outputName).
 					Msg("Failed to read API Key roles")
 			} else {
 				// race when API key was invalidated before acking
 				zlog.Info().
 					Err(err).
 					Str(LogAPIKeyID, apiKeyID).
+					Str("outputName", outputName).
 					Msg("Failed to read invalidated API Key roles")
 
 				// prevents future checks
@@ -473,14 +484,15 @@ func (ack *AckT) updateAPIKey(ctx context.Context,
 					Str(LogAPIKeyID, apiKeyID).
 					Msg("Failed to cleanup roles")
 			} else if removedRolesCount > 0 {
-				if err := ack.bulk.APIKeyUpdate(ctx, apiKeyID, permissionHash, clean); err != nil {
-					zlog.Error().Err(err).RawJSON("roles", clean).Str(LogAPIKeyID, apiKeyID).Msg("Failed to update API Key")
+				if err := bulk.APIKeyUpdate(ctx, apiKeyID, permissionHash, clean); err != nil {
+					zlog.Error().Err(err).RawJSON("roles", clean).Str(LogAPIKeyID, apiKeyID).Str("outputName", outputName).Msg("Failed to update API Key")
 				} else {
 					zlog.Debug().
 						Str("hash.sha256", permissionHash).
 						Str(LogAPIKeyID, apiKeyID).
 						RawJSON("roles", clean).
 						Int("removedRoles", removedRolesCount).
+						Str("outputName", outputName).
 						Msg("Updating agent record to pick up reduced roles.")
 				}
 			}
@@ -556,19 +568,48 @@ func cleanRoles(roles json.RawMessage) (json.RawMessage, int, error) {
 
 func (ack *AckT) invalidateAPIKeys(ctx context.Context, zlog zerolog.Logger, toRetireAPIKeyIDs []model.ToRetireAPIKeyIdsItems, skip string) {
 	ids := make([]string, 0, len(toRetireAPIKeyIDs))
+	remoteIds := make(map[string][]string)
 	for _, k := range toRetireAPIKeyIDs {
 		if k.ID == skip || k.ID == "" {
 			continue
 		}
-		ids = append(ids, k.ID)
+		if k.Output != "" {
+			if remoteIds[k.Output] == nil {
+				remoteIds[k.Output] = make([]string, 0)
+			}
+			remoteIds[k.Output] = append(remoteIds[k.Output], k.ID)
+		} else {
+			ids = append(ids, k.ID)
+		}
 	}
-
 	if len(ids) > 0 {
 		zlog.Info().Strs("fleet.policy.apiKeyIDsToRetire", ids).Msg("Invalidate old API keys")
 		if err := ack.bulk.APIKeyInvalidate(ctx, ids...); err != nil {
 			zlog.Info().Err(err).Strs("ids", ids).Msg("Failed to invalidate API keys")
 		}
 	}
+	// using remote es bulker to invalidate api key
+	for outputName, outputIds := range remoteIds {
+		outputBulk := ack.bulk.GetBulker(outputName)
+
+		if outputBulk == nil {
+			// read output config from .fleet-policies, not filtering by policy id as agent could be reassigned
+			policy, err := dl.QueryOutputFromPolicy(ctx, ack.bulk, outputName)
+			if err != nil || policy == nil {
+				zlog.Warn().Str("outputName", outputName).Any("ids", outputIds).Msg("Output policy not found, API keys will be orphaned")
+			} else {
+				outputBulk, _, err = ack.bulk.CreateAndGetBulker(ctx, zlog, outputName, policy.Data.Outputs)
+				if err != nil {
+					zlog.Warn().Str("outputName", outputName).Any("ids", outputIds).Msg("Failed to recreate output bulker, API keys will be orphaned")
+				}
+			}
+		}
+		if outputBulk != nil {
+			if err := outputBulk.APIKeyInvalidate(ctx, outputIds...); err != nil {
+				zlog.Info().Err(err).Strs("ids", outputIds).Str("outputName", outputName).Msg("Failed to invalidate API keys")
+			}
+		}
+	}
 }
 
 func (ack *AckT) handleUnenroll(ctx context.Context, zlog zerolog.Logger, agent *model.Agent) error {

internal/pkg/api/handleAck_test.go

@@ -571,6 +571,74 @@ func TestInvalidateAPIKeys(t *testing.T) {
 	}
 }
 
+func TestInvalidateAPIKeysRemoteOutput(t *testing.T) {
+	toRetire := []model.ToRetireAPIKeyIdsItems{{
+		ID:     "toRetire1",
+		Output: "remote1",
+	}, {
+		ID:     "toRetire11",
+		Output: "remote1",
+	}, {
+		ID:     "toRetire2",
+		Output: "remote2",
+	}}
+
+	bulker := ftesting.NewMockBulk()
+	remoteBulker := ftesting.NewMockBulk()
+	remoteBulker2 := ftesting.NewMockBulk()
+	bulker.On("GetBulker", "remote1").Return(remoteBulker)
+	bulker.On("GetBulker", "remote2").Return(remoteBulker2)
+
+	remoteBulker.On("APIKeyInvalidate",
+		context.Background(), []string{"toRetire1", "toRetire11"}).
+		Return(nil)
+	remoteBulker2.On("APIKeyInvalidate",
+		context.Background(), []string{"toRetire2"}).
+		Return(nil)
+
+	logger := testlog.SetLogger(t)
+	ack := &AckT{bulk: bulker}
+	ack.invalidateAPIKeys(context.Background(), logger, toRetire, "")
+
+	bulker.AssertExpectations(t)
+	remoteBulker.AssertExpectations(t)
+	remoteBulker2.AssertExpectations(t)
+}
+
+func TestInvalidateAPIKeysRemoteOutputReadFromPolicies(t *testing.T) {
+	toRetire := []model.ToRetireAPIKeyIdsItems{{
+		ID:     "toRetire1",
+		Output: "remote1",
+	}}
+
+	remoteBulker := ftesting.NewMockBulk()
+	remoteBulker.On("APIKeyInvalidate",
+		context.Background(), []string{"toRetire1"}).
+		Return(nil)
+
+	bulkerFn := func(t *testing.T) *ftesting.MockBulk {
+		m := ftesting.NewMockBulk()
+		m.On("Search", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&es.ResultT{HitsT: es.HitsT{
+			Hits: []es.HitT{{
+				Source: []byte(`{"data":{"outputs":{"remote1":{}}}}`),
+			}},
+		}}, nil).Once()
+
+		m.On("CreateAndGetBulker", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(remoteBulker, false, nil)
+		m.On("GetBulker", "remote1").Return(nil)
+		return m
+	}
+
+	bulker := bulkerFn(t)
+
+	logger := testlog.SetLogger(t)
+	ack := &AckT{bulk: bulker}
+	ack.invalidateAPIKeys(context.Background(), logger, toRetire, "")
+
+	bulker.AssertExpectations(t)
+	remoteBulker.AssertExpectations(t)
+}
+
 func TestAckHandleUpgrade(t *testing.T) {
 	tests := []struct {
 		name   string

internal/pkg/api/handleCheckin.go

@@ -748,6 +748,13 @@ func processPolicy(ctx context.Context, zlog zerolog.Logger, bulker bulk.Bulk, a
 	}
 
 	data := model.ClonePolicyData(pp.Policy.Data)
+	for policyName, policyOutput := range data.Outputs {
+		err := policy.ProcessOutputSecret(ctx, policyOutput, bulker)
+		if err != nil {
+			return nil, fmt.Errorf("failed to process output secrets %q: %w",
+				policyName, err)
+		}
+	}
 	// Iterate through the policy outputs and prepare them
 	for _, policyOutput := range pp.Outputs {
 		err = policyOutput.Prepare(ctx, zlog, bulker, &agent, data.Outputs)