[9.1] (backport #9392) Enhancement/5235 use disk space error to set upgrade detail in coordinator (#10057)

* Enhancement/5235 use disk space error to set upgrade detail in coordinator (#9392)

* enhancement(5235): added insufficinet disk space error handling in the coordinator

* enhancement(5235): added coordinator tests for insufficient disk space error

enhancement(5235): updated error in test

enhancement(5235): fix coordinator test

* enhancement(5235): added changelog fragment

(cherry picked from commit a7a76f6)

# Conflicts:
#	internal/pkg/agent/application/coordinator/coordinator_unit_test.go

* resolved merge conflicts

---------

Co-authored-by: Kaan Yalti <kaan.yalti@elastic.co>

Author: mergify[bot]

changelog/fragments/1757010848-agent-responds-with-clean-insufficient-disk-space-error-message.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: when there is a disk space error during an upgrade, agent responds with clean insufficient disk space error message
+
+# 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/9392
+
+# 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

@@ -29,6 +29,7 @@ import (
 	"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/application/upgrade"
+	upgradeErrors "github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/artifact/download/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/configuration"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/transpiler"
@@ -671,6 +672,15 @@ func (c *Coordinator) Upgrade(ctx context.Context, version string, sourceURI str
 			det.SetState(details.StateCompleted)
 			return c.upgradeMgr.AckAction(ctx, c.fleetAcker, action)
 		}
+
+		c.logger.Errorw("upgrade failed", "error", logp.Error(err))
+		// If ErrInsufficientDiskSpace is in the error chain, we want to set the
+		// the error to ErrInsufficientDiskSpace so that the error message is
+		// more concise and clear.
+		if errors.Is(err, upgradeErrors.ErrInsufficientDiskSpace) {
+			err = upgradeErrors.ErrInsufficientDiskSpace
+		}
+
 		det.Fail(err)
 		return err
 	}

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

@@ -15,10 +15,13 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
 	"github.com/elastic/elastic-agent/internal/pkg/otel/translate"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -32,8 +35,10 @@ import (
 	"github.com/elastic/elastic-agent-libs/logp"
 	"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/reexec"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/artifact"
+	upgradeErrors "github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/artifact/download/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/transpiler"
 	"github.com/elastic/elastic-agent/internal/pkg/config"
@@ -1650,3 +1655,100 @@ func (fs *fakeMonitoringServer) Reset() {
 func (fs *fakeMonitoringServer) Addr() net.Addr {
 	return nil
 }
+
+type mockUpgradeManager struct {
+	upgradeErr error
+}
+
+func (m *mockUpgradeManager) Upgradeable() bool {
+	return true
+}
+
+func (m *mockUpgradeManager) Reload(cfg *config.Config) error {
+	return nil
+}
+
+func (m *mockUpgradeManager) Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (_ reexec.ShutdownCallbackFn, err error) {
+	return nil, m.upgradeErr
+}
+
+func (m *mockUpgradeManager) Ack(ctx context.Context, acker acker.Acker) error {
+	return nil
+}
+
+func (m *mockUpgradeManager) AckAction(ctx context.Context, acker acker.Acker, action fleetapi.Action) error {
+	return nil
+}
+
+func (m *mockUpgradeManager) MarkerWatcher() upgrade.MarkerWatcher {
+	return nil
+}
+
+func TestCoordinator_Upgrade_InsufficientDiskSpaceError(t *testing.T) {
+	log, _ := loggertest.New("coordinator-insufficient-disk-space-test")
+
+	mockUpgradeManager := &mockUpgradeManager{
+		upgradeErr: fmt.Errorf("wrapped: %w", upgradeErrors.ErrInsufficientDiskSpace),
+	}
+
+	initialState := State{
+		CoordinatorState:   agentclient.Healthy,
+		CoordinatorMessage: "Running",
+	}
+
+	coord := &Coordinator{
+		state:              initialState,
+		logger:             log,
+		upgradeMgr:         mockUpgradeManager,
+		stateBroadcaster:   broadcaster.New(initialState, 64, 32),
+		overrideStateChan:  make(chan *coordinatorOverrideState),
+		upgradeDetailsChan: make(chan *details.Details),
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	overrideStates := []agentclient.State{}
+	go func() {
+		state1 := <-coord.overrideStateChan
+		overrideStates = append(overrideStates, state1.state)
+
+		state2 := <-coord.overrideStateChan
+		if state2 != nil {
+			overrideStates = append(overrideStates, state2.state)
+		}
+
+		wg.Done()
+	}()
+
+	upgradeDetails := []*details.Details{}
+	go func() {
+		upgradeDetails = append(upgradeDetails, <-coord.upgradeDetailsChan)
+		upgradeDetails = append(upgradeDetails, <-coord.upgradeDetailsChan)
+		wg.Done()
+	}()
+
+	err := coord.Upgrade(t.Context(), "", "", nil)
+	require.Error(t, err)
+	require.Equal(t, err, upgradeErrors.ErrInsufficientDiskSpace)
+
+	wg.Wait()
+
+	require.Equal(t, []agentclient.State{agentclient.Upgrading}, overrideStates)
+
+	require.Equal(t, []*details.Details{
+		{
+			TargetVersion: "",
+			State:         details.StateRequested,
+			ActionID:      "",
+		},
+		{
+			TargetVersion: "",
+			State:         details.StateFailed,
+			Metadata: details.Metadata{
+				FailedState: details.StateRequested,
+				ErrorMsg:    upgradeErrors.ErrInsufficientDiskSpace.Error(),
+			},
+		},
+	}, upgradeDetails)
+}