Enhancement/5235 upgrade cleanup after error

.mockery.yaml

@@ -37,3 +37,17 @@ packages:
       installationModifier:
         config:
           mockname: "InstallationModifier"
+  github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade:
+    config:
+      inpackage: true
+      dir: internal/pkg/agent/application/upgrade
+      mockname: "mock_{{.InterfaceName}}"
+    interfaces:
+      artifactDownloader:
+      unpacker:
+      relinker:
+      watcher:
+      agentDirectoryCopier:
+      upgradeCleaner:
+      upgradeExecutor:
+

changelog/fragments/1753958476-return-disk-space-error-message-when-agent-runs-out-of-space-while-downloading-upgrade-artifact.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: Return disk space error message when agent runs out of space while downloading upgrade artifact.
+
+# 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: 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/elastic/elastic-agent/pull/9122
+
+# 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/elastic/elastic-agent/issues/5235

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

@@ -32,6 +32,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/reexec"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
+	upgradeErrors "github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/configuration"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/transpiler"
@@ -359,6 +360,7 @@ type Coordinator struct {
 	// run a ticker that checks to see if we have a new PID.
 	componentPIDTicker         *time.Ticker
 	componentPidRequiresUpdate *atomic.Bool
+	detailsProvider            func(string, details.State, string) *details.Details
 }
 
 // The channels Coordinator reads to receive updates from the various managers.
@@ -478,7 +480,8 @@ func New(
 		componentPIDTicker:         time.NewTicker(time.Second * 30),
 		componentPidRequiresUpdate: &atomic.Bool{},
 
-		fleetAcker: fleetAcker,
+		fleetAcker:      fleetAcker,
+		detailsProvider: details.NewDetails,
 	}
 	// Setup communication channels for any non-nil components. This pattern
 	// lets us transparently accept nil managers / simulated events during
@@ -734,7 +737,7 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 	if action != nil {
 		actionID = action.ActionID
 	}
-	det := details.NewDetails(version, details.StateRequested, actionID)
+	det := c.detailsProvider(version, details.StateRequested, actionID)
 	det.RegisterObserver(c.SetUpgradeDetails)
 
 	cb, err := c.upgradeMgr.Upgrade(ctx, version, sourceURI, action, det, skipVerifyOverride, skipDefaultPgp, pgpBytes...)
@@ -745,6 +748,11 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 			det.SetState(details.StateCompleted)
 			return c.upgradeMgr.AckAction(ctx, c.fleetAcker, action)
 		}
+
+		if errors.Is(err, upgradeErrors.ErrInsufficientDiskSpace) {
+			err = upgradeErrors.ErrInsufficientDiskSpace
+		}
+
 		det.Fail(err)
 		return err
 	}

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

@@ -15,16 +15,19 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
 	"github.com/elastic/elastic-agent/internal/pkg/testutils"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -41,10 +44,12 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/reload"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/reexec"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/secret"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/artifact"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
+	upgradeErrors "github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/configuration"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/storage"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/transpiler"
@@ -1500,6 +1505,7 @@ func TestCoordinatorInitiatesUpgrade(t *testing.T) {
 		overrideStateChan:  overrideStateChan,
 		upgradeDetailsChan: upgradeDetailsChan,
 		upgradeMgr:         upgradeMgr,
+		detailsProvider:    details.NewDetails,
 		logger:             logp.NewLogger("testing"),
 	}
 
@@ -1959,3 +1965,151 @@ func TestHasEndpoint(t *testing.T) {
 		})
 	}
 }
+
+type mockUpgradeMgrUpgradeErrorHandlingTest struct {
+	upgradeErr error
+}
+
+func (m *mockUpgradeMgrUpgradeErrorHandlingTest) Upgradeable() bool {
+	return true
+}
+
+func (m *mockUpgradeMgrUpgradeErrorHandlingTest) Reload(rawConfig *config.Config) error {
+	return nil
+}
+
+func (m *mockUpgradeMgrUpgradeErrorHandlingTest) Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+	return nil, m.upgradeErr
+}
+
+func (m *mockUpgradeMgrUpgradeErrorHandlingTest) Ack(ctx context.Context, acker acker.Acker) error {
+	return nil
+}
+
+func (m *mockUpgradeMgrUpgradeErrorHandlingTest) AckAction(ctx context.Context, acker acker.Acker, action fleetapi.Action) error {
+	return errors.New("ack action error")
+}
+
+func (m *mockUpgradeMgrUpgradeErrorHandlingTest) MarkerWatcher() upgrade.MarkerWatcher {
+	return nil
+}
+
+type testDetail struct {
+	initialState  details.State
+	expectedState details.State
+	failedState   details.State
+	errorMsg      string
+}
+
+func TestCoordinatorUpgradeErrorHandling(t *testing.T) {
+	testCases := map[string]struct {
+		upgradeErr    error
+		expectedError error
+		detail        testDetail
+	}{
+		"insufficient disk space": {
+			upgradeErr:    upgradeErrors.ErrInsufficientDiskSpace,
+			expectedError: upgradeErrors.ErrInsufficientDiskSpace,
+			detail: testDetail{
+				initialState:  details.StateRequested,
+				expectedState: details.StateFailed,
+				failedState:   details.StateRequested,
+				errorMsg:      upgradeErrors.ErrInsufficientDiskSpace.Error(),
+			},
+		},
+		"wrapped insufficient disk space": {
+			upgradeErr:    fmt.Errorf("wrapped: %w", upgradeErrors.ErrInsufficientDiskSpace),
+			expectedError: upgradeErrors.ErrInsufficientDiskSpace,
+			detail: testDetail{
+				initialState:  details.StateRequested,
+				expectedState: details.StateFailed,
+				failedState:   details.StateRequested,
+				errorMsg:      upgradeErrors.ErrInsufficientDiskSpace.Error(),
+			},
+		},
+		"same version error": {
+			upgradeErr:    upgrade.ErrUpgradeSameVersion,
+			expectedError: errors.New("ack action error"),
+			detail: testDetail{
+				initialState:  details.StateRequested,
+				expectedState: details.StateCompleted,
+				failedState:   "",
+				errorMsg:      "",
+			},
+		},
+		"generic error": {
+			upgradeErr:    errors.New("test error"),
+			expectedError: errors.New("test error"),
+			detail: testDetail{
+				initialState:  details.StateRequested,
+				expectedState: details.StateFailed,
+				failedState:   details.StateRequested,
+				errorMsg:      "test error",
+			},
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			upgradeMgr := &mockUpgradeMgrUpgradeErrorHandlingTest{}
+			upgradeMgr.upgradeErr = tc.upgradeErr
+
+			det := details.NewDetails("1.0.0", tc.detail.initialState, "test-action-id")
+			detailsProvider := func(version string, state details.State, actionID string) *details.Details {
+				return det
+			}
+
+			coord := &Coordinator{
+				upgradeMgr:         upgradeMgr,
+				detailsProvider:    detailsProvider,
+				stateBroadcaster:   broadcaster.New(State{State: agentclient.Healthy, Message: "Running"}, 64, 32),
+				overrideStateChan:  make(chan *coordinatorOverrideState),
+				upgradeDetailsChan: make(chan *details.Details, 2),
+			}
+
+			go func() {
+				state1 := <-coord.overrideStateChan
+				assert.Equal(t, agentclient.Upgrading, state1.state)
+
+				state2 := <-coord.overrideStateChan
+				assert.Nil(t, state2)
+			}()
+
+			err := coord.Upgrade(t.Context(), "mockversion", "mockuri", nil, false, false)
+			require.Error(t, err)
+			require.Equal(t, err, tc.expectedError)
+
+			require.Equal(t, tc.detail.expectedState, det.State, "State mismatch")
+			require.Equal(t, tc.detail.failedState, det.Metadata.FailedState, "FailedState mismatch")
+			require.Equal(t, tc.detail.errorMsg, det.Metadata.ErrorMsg, "ErrorMsg mismatch")
+		})
+	}
+
+}
+
+func TestCoordinatorNew(t *testing.T) {
+	t.Run("correctly sets details provider", func(t *testing.T) {
+		coord := New(
+			nil,
+			nil,
+			logp.InfoLevel,
+			nil,
+			component.RuntimeSpecs{},
+			nil,
+			nil,
+			nil,
+			nil,
+			nil,
+			nil,
+			nil,
+			false,
+			nil,
+			nil,
+		)
+
+		assert.NotNil(t, coord.detailsProvider)
+		detailsProviderPtr := reflect.ValueOf(coord.detailsProvider).Pointer()
+		newDetailsPtr := reflect.ValueOf(details.NewDetails).Pointer()
+		assert.Equal(t, newDetailsPtr, detailsProviderPtr, "detailsProvider should be details.NewDetails")
+	})
+}

internal/pkg/agent/application/paths/common.go

@@ -197,7 +197,7 @@ func DataFrom(topDirPath string) string {
 
 // Run returns the run directory for Agent
 func Run() string {
-	return filepath.Join(Home(), "run")
+	return filepath.Join(HomePath(), "run")
 }
 
 // Components returns the component directory for Agent

internal/pkg/agent/application/paths/files.go

@@ -78,19 +78,21 @@ func AgentCapabilitiesPath() string {
 	return filepath.Join(Config(), defaultAgentCapabilitiesFile)
 }
 
+var HomePath = Home // used only for mocking Home() for testing purposes
+
 // AgentActionStoreFile is the file that contains the action that can be replayed after restart.
 func AgentActionStoreFile() string {
-	return filepath.Join(Home(), defaultAgentActionStoreFile)
+	return filepath.Join(HomePath(), defaultAgentActionStoreFile)
 }
 
 // AgentStateStoreYmlFile is the file that contains the persisted state of the agent including the action that can be replayed after restart.
 func AgentStateStoreYmlFile() string {
-	return filepath.Join(Home(), defaultAgentStateStoreYmlFile)
+	return filepath.Join(HomePath(), defaultAgentStateStoreYmlFile)
 }
 
 // AgentStateStoreFile is the file that contains the persisted state of the agent including the action that can be replayed after restart encrypted.
 func AgentStateStoreFile() string {
-	return filepath.Join(Home(), defaultAgentStateStoreFile)
+	return filepath.Join(HomePath(), defaultAgentStateStoreFile)
 }
 
 // AgentInputsDPath is directory that contains the fragment of inputs yaml for K8s deployment.