Migrate agent to a different cluster (#8014)

Author: michalpristas

changelog/fragments/1745923309-Handle-Migrate-action-in-agent.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: Handle Migrate action in agent
+
+# 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: Adds ability to re-enroll agent to different cluster using Fleet UI.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# 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/owner/repo/1234
+
+# 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/agent/application/actions/handlers/handler_action_migrate.go

@@ -0,0 +1,107 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package handlers
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
+	fleetgateway "github.com/elastic/elastic-agent/internal/pkg/agent/application/gateway/fleet"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/reexec"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
+	"github.com/elastic/elastic-agent/internal/pkg/core/backoff"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+	"github.com/elastic/elastic-agent/pkg/features"
+)
+
+const ()
+
+type migrateCoordinator interface {
+	Migrate(_ context.Context, _ *fleetapi.ActionMigrate, _ func(done <-chan struct{}) backoff.Backoff) error
+	ReExec(callback reexec.ShutdownCallbackFn, argOverrides ...string)
+	HasEndpoint() bool
+}
+
+// Migrate handles migrate change coming from fleet.
+type Migrate struct {
+	log       *logger.Logger
+	agentInfo info.Agent
+	coord     migrateCoordinator
+
+	tamperProtectionFn func() bool // allows to inject the flag for tests, defaults to features.TamperProtection
+}
+
+// NewMigrate creates a new Migrate handler.
+func NewMigrate(
+	log *logger.Logger,
+	agentInfo info.Agent,
+	coord migrateCoordinator,
+) *Migrate {
+	return &Migrate{
+		log:                log,
+		agentInfo:          agentInfo,
+		coord:              coord,
+		tamperProtectionFn: features.TamperProtection,
+	}
+}
+
+// Handle handles MIGRATE action.
+func (h *Migrate) Handle(ctx context.Context, a fleetapi.Action, ack acker.Acker) error {
+	h.log.Debugf("handlerMigrate: action '%+v' received", a)
+
+	action, ok := a.(*fleetapi.ActionMigrate)
+	if !ok {
+		return fmt.Errorf("invalid type, expected ActionMigrate and received %T", a)
+	}
+
+	// if endpoint is present do not proceed
+	if h.tamperProtectionFn() && h.coord.HasEndpoint() {
+		err := errors.New("unsupported action: tamper protected agent")
+		h.ackFailure(ctx, err, action, ack)
+		return err
+	}
+
+	if err := h.coord.Migrate(ctx, action, fleetgateway.RequestBackoff); err != nil {
+		// this should not happen, unmanaged agent should not receive the action
+		// defensive coding to avoid misbehavior
+		if errors.Is(err, coordinator.ErrNotManaged) {
+			return errors.New("unmanaged agent, use Enroll instead")
+		}
+
+		// ack failure
+		h.ackFailure(ctx, err, action, ack)
+
+		if errors.Is(err, coordinator.ErrFleetServer) {
+			return errors.New("action not available for agents running Fleet Server")
+		}
+
+		return fmt.Errorf("migration of agent to a new cluster failed: %w", err)
+
+	}
+
+	// reexec and load new config
+	h.coord.ReExec(nil)
+	return nil
+}
+
+func (h *Migrate) ackFailure(ctx context.Context, err error, action *fleetapi.ActionMigrate, acker acker.Acker) {
+	action.Err = err
+
+	if err := acker.Ack(ctx, action); err != nil {
+		h.log.Errorw("failed to ack migrate action",
+			"error.message", err,
+			"action", action)
+	}
+
+	if err := acker.Commit(ctx); err != nil {
+		h.log.Errorw("failed to commit migrate action",
+			"error.message", err,
+			"action", action)
+	}
+}

internal/pkg/agent/application/actions/handlers/handler_action_migrate_test.go

@@ -0,0 +1,128 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package handlers
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/reexec"
+	"github.com/elastic/elastic-agent/internal/pkg/core/backoff"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
+	"github.com/elastic/elastic-agent/pkg/core/logger/loggertest"
+	mockinfo "github.com/elastic/elastic-agent/testing/mocks/internal_/pkg/agent/application/info"
+)
+
+func TestActionMigratelHandler(t *testing.T) {
+	log, _ := loggertest.New("")
+	mockAgentInfo := mockinfo.NewAgent(t)
+	t.Run("wrong action type", func(t *testing.T) {
+		action := &fleetapi.ActionSettings{}
+		ack := &fakeAcker{}
+		ack.On("Ack", t.Context(), action).Return(nil)
+		ack.On("Commit", t.Context()).Return(nil)
+
+		coord := &fakeMigrateCoordinator{}
+		coord.On("Migrate", mock.Anything, mock.Anything).Return(nil)
+		coord.On("ReExec", mock.Anything, mock.Anything)
+
+		h := NewMigrate(log, mockAgentInfo, coord)
+		require.NotNil(t, h.Handle(t.Context(), action, ack))
+		coord.AssertNumberOfCalls(t, "Migrate", 0)
+		coord.AssertNumberOfCalls(t, "ReExec", 0)
+	})
+
+	t.Run("tamper protected agent", func(t *testing.T) {
+		action := &fleetapi.ActionMigrate{
+			ActionType: "MIGRATE",
+		}
+
+		ack := &fakeAcker{}
+		ack.On("Ack", t.Context(), action).Return(nil)
+		ack.On("Commit", t.Context()).Return(nil)
+
+		coord := &fakeMigrateCoordinator{}
+		coord.On("Migrate", mock.Anything, mock.Anything).Return(nil)
+		coord.On("ReExec", mock.Anything, mock.Anything)
+		coord.On("HasEndpoint").Return(true)
+
+		h := NewMigrate(log, mockAgentInfo, coord)
+		h.tamperProtectionFn = func() bool { return true }
+
+		require.NotNil(t, h.Handle(t.Context(), action, ack))
+		coord.AssertNumberOfCalls(t, "Migrate", 0)
+		ack.AssertCalled(t, "Ack", t.Context(), action)
+		ack.AssertCalled(t, "Commit", t.Context())
+		coord.AssertNumberOfCalls(t, "ReExec", 0)
+	})
+
+	t.Run("action propagated to coordinator", func(t *testing.T) {
+		action := &fleetapi.ActionMigrate{}
+
+		ack := &fakeAcker{}
+		ack.On("Ack", t.Context(), action).Return(nil)
+		ack.On("Commit", t.Context()).Return(nil)
+
+		coord := &fakeMigrateCoordinator{}
+		coord.On("Migrate", mock.Anything, mock.Anything).Return(nil)
+		coord.On("ReExec", mock.Anything, mock.Anything)
+
+		h := NewMigrate(log, mockAgentInfo, coord)
+		h.tamperProtectionFn = func() bool { return false }
+
+		require.Nil(t, h.Handle(t.Context(), action, ack))
+		coord.AssertNumberOfCalls(t, "Migrate", 1)
+
+		// ack delegated to migrate coordinator
+		ack.AssertNumberOfCalls(t, "Ack", 0)
+		ack.AssertNumberOfCalls(t, "Commit", 0)
+		coord.AssertCalled(t, "ReExec", mock.Anything, mock.Anything)
+	})
+
+	t.Run("fleet server", func(t *testing.T) {
+		action := &fleetapi.ActionMigrate{}
+
+		ack := &fakeAcker{}
+		ack.On("Ack", t.Context(), action).Return(nil)
+		ack.On("Commit", t.Context()).Return(nil)
+
+		coord := &fakeMigrateCoordinator{}
+		coord.On("Migrate", mock.Anything, mock.Anything).Return(coordinator.ErrFleetServer)
+		coord.On("ReExec", mock.Anything, mock.Anything)
+
+		h := NewMigrate(log, mockAgentInfo, coord)
+		h.tamperProtectionFn = func() bool { return false }
+
+		require.Error(t, coordinator.ErrFleetServer, h.Handle(t.Context(), action, ack))
+		coord.AssertNumberOfCalls(t, "Migrate", 1)
+
+		// ack not delegated to migrate coordinator, failure is reported
+		ack.AssertNumberOfCalls(t, "Ack", 1)
+		ack.AssertNumberOfCalls(t, "Commit", 1)
+		coord.AssertNotCalled(t, "ReExec", mock.Anything, mock.Anything)
+	})
+}
+
+type fakeMigrateCoordinator struct {
+	mock.Mock
+}
+
+func (f *fakeMigrateCoordinator) Migrate(ctx context.Context, a *fleetapi.ActionMigrate, _ func(done <-chan struct{}) backoff.Backoff) error {
+	args := f.Called(ctx, a)
+	return args.Error(0)
+}
+
+func (f *fakeMigrateCoordinator) ReExec(callback reexec.ShutdownCallbackFn, argOverrides ...string) {
+	f.Called(callback, argOverrides)
+}
+
+func (f *fakeMigrateCoordinator) HasEndpoint() bool {
+	args := f.Called()
+	return args.Bool(0)
+}

internal/pkg/agent/application/actions/handlers/handler_action_policy_change.go

@@ -10,9 +10,7 @@ import (
 	goerrors "errors"
 	"fmt"
 	"io"
-	"net/http"
 	"sort"
-	"time"
 
 	"gopkg.in/yaml.v2"
 
@@ -33,10 +31,6 @@ import (
 	"github.com/elastic/elastic-agent/pkg/core/logger"
 )
 
-const (
-	apiStatusTimeout = 15 * time.Second
-)
-
 // PolicyChangeHandler is a handler for POLICY_CHANGE action.
 type PolicyChangeHandler struct {
 	log                  *logger.Logger
@@ -171,28 +165,7 @@ func testFleetConfig(ctx context.Context, log *logger.Logger, clientConfig remot
 				clientConfig.Hosts, clientConfig.Host)))
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, apiStatusTimeout)
-	defer cancel()
-
-	// TODO: a HEAD should be enough as we need to test only the connectivity part
-	resp, err := fleetClient.Send(ctx, http.MethodGet, "/api/status", nil, nil, nil)
-	if err != nil {
-		return errors.New(
-			err, "fail to communicate with Fleet Server API client hosts",
-			errors.TypeNetwork, errors.M("hosts", clientConfig.Hosts))
-	}
-
-	if resp.StatusCode != http.StatusOK {
-		return errors.New(
-			err, fmt.Sprintf("fleet server ping returned a bad status code: %d", resp.StatusCode),
-			errors.TypeNetwork, errors.M("hosts", clientConfig.Hosts))
-	}
-
-	// discard body for proper cancellation and connection reuse
-	_, _ = io.Copy(io.Discard, resp.Body)
-	resp.Body.Close()
-
-	return nil
+	return client.CheckRemote(ctx, fleetClient)
 }
 
 // updateFleetConfig copies the relevant Fleet client settings from policyConfig on agentConfig. The destination struct is modified in-place

internal/pkg/agent/application/coordinator/config_operations.go

@@ -0,0 +1,61 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package coordinator
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/otiai10/copy"
+
+	"github.com/elastic/elastic-agent-libs/file"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
+)
+
+const (
+	backupSuffix = ".enroll.bak"
+)
+
+// RestoreConfig restores from backup if needed and signals restore was performed
+func RestoreConfig() error {
+	configFile := paths.AgentConfigFile()
+	backup := configFile + backupSuffix
+
+	// check backup exists
+	if _, err := os.Stat(backup); os.IsNotExist(err) {
+		return nil
+	}
+
+	if err := file.SafeFileRotate(configFile, backup); err != nil {
+		return fmt.Errorf("failed to safe rotate backup config file: %w", err)
+	}
+
+	return nil
+}
+
+// backupConfig creates a backup of currently used fleet config
+func backupConfig() error {
+	configFile := paths.AgentConfigFile()
+	backup := configFile + backupSuffix
+
+	err := copy.Copy(configFile, backup, copy.Options{
+		PermissionControl: copy.AddPermission(0600),
+	})
+	if err != nil {
+		return fmt.Errorf("failed to backup config file %s -> %s: %w", configFile, backup, err)
+	}
+
+	return nil
+}
+
+// cleanBackupConfig removes backup config file
+func cleanBackupConfig() error {
+	backup := paths.AgentConfigFile() + backupSuffix
+	if err := os.RemoveAll(backup); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+
+	return nil
+}