Support for remote ES output

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

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,16 +568,26 @@ 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))
+	output := ""
 	for _, k := range toRetireAPIKeyIDs {
 		if k.ID == skip || k.ID == "" {
 			continue
 		}
 		ids = append(ids, k.ID)
+		output = k.Output
+	}
+	// using remote es bulker to invalidate api key - supposing all retire api key ids belong to the same remote es
+	bulk := ack.bulk
+	if output != "" {
+		outputBulk := ack.bulk.GetBulker(output)
+		if outputBulk != nil {
+			bulk = outputBulk
+		}
 	}
 
 	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 {
+		if err := bulk.APIKeyInvalidate(ctx, ids...); err != nil {
 			zlog.Info().Err(err).Strs("ids", ids).Msg("Failed to invalidate API keys")
 		}
 	}

internal/pkg/api/handleAck_test.go

@@ -571,6 +571,46 @@ func TestInvalidateAPIKeys(t *testing.T) {
 	}
 }
 
+func TestInvalidateAPIKeysRemoteOutput(t *testing.T) {
+	toRetire1 := []model.ToRetireAPIKeyIdsItems{{
+		ID:     "toRetire1",
+		Output: "remote1",
+	}}
+
+	wants := map[string][]string{
+		"1": {"toRetire1"},
+	}
+
+	agent := model.Agent{
+		Outputs: map[string]*model.PolicyOutput{
+			"1": {ToRetireAPIKeyIds: toRetire1},
+		},
+	}
+
+	for i, out := range agent.Outputs {
+		want := wants[i]
+
+		bulker := ftesting.NewMockBulk()
+		remoteBulker := ftesting.NewMockBulk()
+		bulker.On("GetBulker", mock.AnythingOfType("string")).Return(remoteBulker)
+		if len(want) > 0 {
+			remoteBulker.On("APIKeyInvalidate",
+				context.Background(), mock.MatchedBy(func(ids []string) bool {
+					// if A contains B and B contains A => A = B
+					return assert.Subset(t, ids, want) &&
+						assert.Subset(t, want, ids)
+				})).
+				Return(nil)
+		}
+
+		logger := testlog.SetLogger(t)
+		ack := &AckT{bulk: bulker}
+		ack.invalidateAPIKeys(context.Background(), logger, out.ToRetireAPIKeyIds, "")
+
+		bulker.AssertExpectations(t)
+	}
+}
+
 func TestAckHandleUpgrade(t *testing.T) {
 	tests := []struct {
 		name   string

internal/pkg/bulk/bulk_remote_output_test.go

@@ -0,0 +1,146 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+//go:build !integration
+
+package bulk
+
+import (
+	"testing"
+
+	testlog "github.com/elastic/fleet-server/v7/internal/pkg/testing/log"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_CheckRemoteOutputChanged(t *testing.T) {
+	testcases := []struct {
+		name    string
+		cfg     map[string]interface{}
+		newCfg  map[string]interface{}
+		changed bool
+	}{
+		{
+			name: "initial nil",
+			cfg:  nil,
+			newCfg: map[string]interface{}{
+				"type":          "remote_elasticsearch",
+				"hosts":         []string{"https://remote-es:443"},
+				"service_token": "token1",
+			},
+			changed: false,
+		},
+		{
+			name: "no changes",
+			cfg: map[string]interface{}{
+				"type":          "remote_elasticsearch",
+				"hosts":         []string{"https://remote-es:443"},
+				"service_token": "token1",
+			},
+			newCfg: map[string]interface{}{
+				"type":          "remote_elasticsearch",
+				"hosts":         []string{"https://remote-es:443"},
+				"service_token": "token1",
+			},
+			changed: false,
+		},
+		{
+			name: "change to service token",
+			cfg: map[string]interface{}{
+				"type":          "remote_elasticsearch",
+				"hosts":         []string{"https://remote-es:443"},
+				"service_token": "token1",
+			},
+			newCfg: map[string]interface{}{
+				"type":          "remote_elasticsearch",
+				"hosts":         []string{"https://remote-es:443"},
+				"service_token": "token2",
+			},
+			changed: true,
+		},
+		{
+			name: "change to advanced config",
+			cfg: map[string]interface{}{
+				"type":                "remote_elasticsearch",
+				"hosts":               []string{"https://remote-es:443"},
+				"service_token":       "token1",
+				"server.memory_limit": "4",
+			},
+			newCfg: map[string]interface{}{
+				"type":                "remote_elasticsearch",
+				"hosts":               []string{"https://remote-es:443"},
+				"service_token":       "token1",
+				"server.memory_limit": "5",
+			},
+			changed: true,
+		}}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			log := testlog.SetLogger(t)
+			bulker := NewBulker(nil, nil)
+			bulker.remoteOutputConfigMap["remote1"] = tc.cfg
+			hasChanged := bulker.CheckRemoteOutputChanged(log, "remote1", tc.newCfg)
+			assert.Equal(t, tc.changed, hasChanged)
+		})
+	}
+}
+
+func Test_CreateAndGetBulkerNew(t *testing.T) {
+	log := testlog.SetLogger(t)
+	bulker := NewBulker(nil, nil)
+	outputMap := make(map[string]map[string]interface{})
+	outputMap["remote1"] = map[string]interface{}{
+		"type":          "remote_elasticsearch",
+		"hosts":         []interface{}{"https://remote-es:443"},
+		"service_token": "token1",
+	}
+	newBulker, hasChanged, err := bulker.CreateAndGetBulker(log, "remote1", "token1", outputMap)
+	assert.NotNil(t, newBulker)
+	assert.Equal(t, false, hasChanged)
+	assert.Nil(t, err)
+}
+
+func Test_CreateAndGetBulkerExisting(t *testing.T) {
+	log := testlog.SetLogger(t)
+	bulker := NewBulker(nil, nil)
+	outputBulker := NewBulker(nil, nil)
+	bulker.bulkerMap["remote1"] = outputBulker
+	outputMap := make(map[string]map[string]interface{})
+	cfg := map[string]interface{}{
+		"type":          "remote_elasticsearch",
+		"hosts":         []interface{}{"https://remote-es:443"},
+		"service_token": "token1",
+	}
+	bulker.remoteOutputConfigMap["remote1"] = cfg
+	outputMap["remote1"] = cfg
+	newBulker, hasChanged, err := bulker.CreateAndGetBulker(log, "remote1", "token1", outputMap)
+	assert.Equal(t, outputBulker, newBulker)
+	assert.Equal(t, false, hasChanged)
+	assert.Nil(t, err)
+}
+
+func Test_CreateAndGetBulkerChanged(t *testing.T) {
+	log := testlog.SetLogger(t)
+	bulker := NewBulker(nil, nil)
+	outputBulker := NewBulker(nil, nil)
+	bulker.bulkerMap["remote1"] = outputBulker
+	outputMap := make(map[string]map[string]interface{})
+	bulker.remoteOutputConfigMap["remote1"] = map[string]interface{}{
+		"type":          "remote_elasticsearch",
+		"hosts":         []interface{}{"https://remote-es:443"},
+		"service_token": "token1",
+	}
+	outputMap["remote1"] = map[string]interface{}{
+		"type":          "remote_elasticsearch",
+		"hosts":         []interface{}{"https://remote-es:443"},
+		"service_token": "token2",
+	}
+	cancelFnCalled := false
+	outputBulker.cancelFn = func() { cancelFnCalled = true }
+	newBulker, hasChanged, err := bulker.CreateAndGetBulker(log, "remote1", "token2", outputMap)
+	assert.NotEqual(t, outputBulker, newBulker)
+	assert.Equal(t, true, hasChanged)
+	assert.Nil(t, err)
+	assert.Equal(t, true, cancelFnCalled)
+}