[PRODENG-3633] Add overall apply/reset deadline and surface retry errors (#668)

Option A from the ticket: a Manager-level timeout wrapper rather than
threading context.Context through all 30+ phase implementations.

phase.Manager gains a Deadline time.Duration field. Run races each
phase's Run() against the deadline in a goroutine; on timeout it returns
an error naming the in-progress phase instead of blocking forever, and
does not start the next phase. The phase's goroutine is left running on
timeout rather than force-stopped -- launchpad is a short-lived CLI
process, so it exits shortly after and reclaims it. This does not make
individual waits (WaitGroup.Wait, channel ops) cancellable, only bounds
the total time Run can spend.

Wired through a new Product.SetTimeout(time.Duration) method rather than
adding a parameter to Apply/Reset, so none of the existing callers
(cmd/apply.go, cmd/reset.go, test/integration, test/smoke/*) need to
change signatures; only cmd/apply.go and cmd/reset.go call it, from a new
--timeout flag defaulting to 90m (matching the longest existing smoke
test timeout in the Makefile). --timeout 0 disables the deadline.

Separately, pkg/product/mke/config/cluster_spec.go's pingHost (the MKE
health-check retry loop) only logged "waiting for MKE ... to become
healthy" on every attempt and never the actual failure reason mid-retry,
which is why diagnosing the original PRODENG-3594 deadlock cost a full
50-minute CI run. Added retry.OnRetry to log the real error on each
failed attempt. The final returned error already carried every attempt's
error via retry-go's own Error type; nothing was actually being
discarded, it just wasn't visible while waiting.

Audited every other sync.WaitGroup use in the codebase (connect.go,
disconnect.go, run_hooks.go) for the double-send-on-channel pattern that
caused the original deadlock: none have it, all use defer wg.Done() with
mutex-protected shared state instead of a fixed-capacity channel.

Signed-off-by: James Nesbitt <jnesbitt@mirantis.com>

Author: james-nesbitt

cmd/apply.go

@@ -54,6 +54,11 @@ func NewApplyCommand() *cli.Command {
 				Usage: "force upgrade to run on compatible components, even if it doesn't look necessary",
 				Value: false,
 			},
+			&cli.DurationFlag{
+				Name:  "timeout",
+				Usage: "Overall deadline for apply; fails naming the in-progress phase if exceeded instead of blocking forever (0 disables)",
+				Value: 90 * time.Minute,
+			},
 		}...),
 		Before: actions(initLogger, startUpgradeCheck, initAnalytics, checkLicense, initExec),
 		After:  actions(closeAnalytics, upgradeCheckResult),
@@ -72,6 +77,8 @@ func NewApplyCommand() *cli.Command {
 				return fmt.Errorf("failed to load product config: %w", err)
 			}
 
+			product.SetTimeout(ctx.Duration("timeout"))
+
 			defer func() {
 				if err != nil && logFile != nil {
 					log.Infof("See %s for more logs ", logFile.Name())

cmd/reset.go

@@ -28,6 +28,11 @@ func NewResetCommand() *cli.Command {
 				Usage:   "Don't ask for confirmation",
 				Aliases: []string{"f"},
 			},
+			&cli.DurationFlag{
+				Name:  "timeout",
+				Usage: "Overall deadline for reset; fails naming the in-progress phase if exceeded instead of blocking forever (0 disables)",
+				Value: 90 * time.Minute,
+			},
 		}...),
 		Before: actions(initLogger, initAnalytics, checkLicense, initExec, requireForce),
 		After:  actions(closeAnalytics),
@@ -39,6 +44,8 @@ func NewResetCommand() *cli.Command {
 				return fmt.Errorf("failed to load product config: %w", err)
 			}
 
+			product.SetTimeout(ctx.Duration("timeout"))
+
 			err = product.Reset()
 			if err != nil {
 				analytics.TrackEvent("Cluster Reset Failed", nil)

pkg/phase/manager.go

@@ -38,6 +38,13 @@ type Manager struct {
 	config       interface{}
 	IgnoreErrors bool
 	SkipCleanup  bool
+	// Deadline bounds the total wall-clock time Run may spend across all
+	// phases. Zero means no deadline. A phase that is still running when the
+	// deadline elapses is abandoned (its goroutine is not force-stopped) and
+	// Run returns an error naming that phase; the caller is expected to be a
+	// short-lived CLI process that exits shortly after, taking the abandoned
+	// goroutine down with it.
+	Deadline time.Duration
 }
 
 // NewManager constructs new phase manager.
@@ -59,8 +66,16 @@ func (m *Manager) AddPhase(p phase) {
 	m.phases = append(m.phases, p)
 }
 
-// Run executes all the added Phases in order.
+// Run executes all the added Phases in order. If Deadline is set, the total
+// time spent across all phases is bounded; a phase still running when the
+// deadline elapses causes Run to return an error naming that phase instead
+// of blocking forever.
 func (m *Manager) Run() error {
+	var deadlineAt time.Time
+	if m.Deadline > 0 {
+		deadlineAt = time.Now().Add(m.Deadline)
+	}
+
 	for _, phase := range m.phases {
 		title := phase.Title()
 
@@ -89,7 +104,10 @@ func (m *Manager) Run() error {
 		log.Infof(text, title)
 		start := time.Now()
 
-		result := phase.Run()
+		timedOut, result := m.runPhase(phase, title, deadlineAt)
+		if timedOut {
+			return fmt.Errorf("exceeded overall deadline of %s while running phase %q", m.Deadline, title)
+		}
 
 		duration := time.Since(start)
 		log.Debugf("phase '%s' took %s", title, duration.Truncate(time.Minute))
@@ -126,3 +144,42 @@ func (m *Manager) Run() error {
 
 	return nil
 }
+
+// runPhase runs a single phase, optionally racing it against deadlineAt. It
+// returns the phase's error and whether the deadline elapsed first. On
+// timeout the phase's goroutine is left running; the caller (a short-lived
+// CLI process) is expected to exit shortly after, which reclaims it. Cleanup
+// hooks are not invoked on timeout since the phase never signaled it stopped
+// touching shared state.
+func (m *Manager) runPhase(target phase, title string, deadlineAt time.Time) (timedOut bool, result error) {
+	if deadlineAt.IsZero() {
+		if err := target.Run(); err != nil {
+			return false, fmt.Errorf("%w", err)
+		}
+		return false, nil
+	}
+
+	remaining := time.Until(deadlineAt)
+	if remaining <= 0 {
+		return true, nil
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- target.Run()
+	}()
+
+	timer := time.NewTimer(remaining)
+	defer timer.Stop()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			return false, fmt.Errorf("%w", err)
+		}
+		return false, nil
+	case <-timer.C:
+		log.Errorf("phase '%s' exceeded the overall deadline of %s", title, m.Deadline)
+		return true, nil
+	}
+}

pkg/phase/manager_test.go

@@ -0,0 +1,88 @@
+package phase
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type fakePhase struct {
+	title string
+	run   func() error
+}
+
+func (f *fakePhase) Title() string { return f.title }
+func (f *fakePhase) Run() error    { return f.run() }
+
+func TestManagerRun_NoDeadlineRunsToCompletion(t *testing.T) {
+	m := NewManager(struct{}{})
+	m.AddPhases(
+		&fakePhase{title: "a", run: func() error { return nil }},
+		&fakePhase{title: "b", run: func() error { return nil }},
+	)
+
+	require.NoError(t, m.Run())
+}
+
+func TestManagerRun_DeadlineNotExceededRunsToCompletion(t *testing.T) {
+	m := NewManager(struct{}{})
+	m.Deadline = time.Second
+	m.AddPhases(
+		&fakePhase{title: "fast", run: func() error { return nil }},
+	)
+
+	require.NoError(t, m.Run())
+}
+
+func TestManagerRun_DeadlineExceededNamesTheHungPhase(t *testing.T) {
+	m := NewManager(struct{}{})
+	m.Deadline = 20 * time.Millisecond
+	m.AddPhases(
+		&fakePhase{title: "quick", run: func() error { return nil }},
+		&fakePhase{title: "hangs-forever", run: func() error {
+			select {} // block forever, simulating an unbounded wait
+		}},
+	)
+
+	err := m.Run()
+	require.Error(t, err)
+	assert.True(t, strings.Contains(err.Error(), "hangs-forever"), "error should name the hung phase, got: %s", err)
+	assert.True(t, strings.Contains(err.Error(), "deadline"), "error should mention the deadline, got: %s", err)
+}
+
+func TestManagerRun_DeadlineExceededBeforeStartingNextPhase(t *testing.T) {
+	m := NewManager(struct{}{})
+	m.Deadline = 10 * time.Millisecond
+	started := false
+	m.AddPhases(
+		&fakePhase{title: "slow", run: func() error {
+			time.Sleep(50 * time.Millisecond)
+			return nil
+		}},
+		&fakePhase{title: "never-reached", run: func() error {
+			started = true
+			return nil
+		}},
+	)
+
+	err := m.Run()
+	require.Error(t, err)
+	assert.False(t, started, "the second phase must not start once the deadline has already elapsed")
+}
+
+func TestManagerRun_PhaseErrorStillPropagatesWithDeadlineSet(t *testing.T) {
+	m := NewManager(struct{}{})
+	m.Deadline = time.Second
+	wantErr := errors.New("boom")
+	m.AddPhases(
+		&fakePhase{title: "fails", run: func() error { return wantErr }},
+	)
+
+	err := m.Run()
+	require.Error(t, err)
+	assert.ErrorIs(t, err, wantErr)
+}

pkg/product/mke/apply.go

@@ -15,6 +15,7 @@ import (
 func (p *MKE) Apply(disableCleanup, force bool, concurrency int, forceUpgrade bool) error {
 	phaseManager := phase.NewManager(&p.ClusterConfig)
 	phaseManager.SkipCleanup = disableCleanup
+	phaseManager.Deadline = p.Timeout
 
 	phaseManager.AddPhases(
 		&mke.UpgradeCheck{},

pkg/product/mke/config/cluster_spec.go

@@ -277,6 +277,9 @@ func pingHost(host *Host, address string, waitgroup *sync.WaitGroup, errCh chan<
 			}
 			return nil
 		},
+		retry.OnRetry(func(n uint, err error) {
+			log.Errorf("%s: MKE health check attempt %d of 10 failed: %s", host, n+1, err.Error())
+		}),
 		retry.MaxJitter(time.Second*3),
 		retry.Delay(time.Second*30),
 		retry.DelayType(retry.FixedDelay),

pkg/product/mke/mke.go

@@ -2,6 +2,7 @@ package mke
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/Mirantis/launchpad/pkg/product/mke/config"
 	"gopkg.in/yaml.v2"
@@ -10,6 +11,15 @@ import (
 // MKE is the product.
 type MKE struct {
 	ClusterConfig config.ClusterConfig
+	// Timeout bounds the total wall-clock time Apply or Reset may spend
+	// across all phases. Zero (the default) means no deadline. Set via
+	// SetTimeout.
+	Timeout time.Duration
+}
+
+// SetTimeout sets the overall deadline for Apply and Reset.
+func (p *MKE) SetTimeout(timeout time.Duration) {
+	p.Timeout = timeout
 }
 
 // ClusterName returns the cluster name.

pkg/product/mke/reset.go

@@ -11,6 +11,7 @@ import (
 // Reset uninstalls a Docker Enterprise cluster.
 func (p *MKE) Reset() error {
 	phaseManager := phase.NewManager(&p.ClusterConfig)
+	phaseManager.Deadline = p.Timeout
 
 	phaseManager.AddPhases(
 		&common.Connect{},

pkg/product/product.go

@@ -1,5 +1,7 @@
 package product
 
+import "time"
+
 // Product is an interface that represents a product that launchpad can manage.
 type Product interface {
 	Apply(disableCleanup, force bool, concurrency int, forceUpgrade bool) error
@@ -8,4 +10,7 @@ type Product interface {
 	ClientConfig() error
 	Exec(target []string, interactive, first, all, parallel bool, role, os, cmd string) error
 	ClusterName() string
+	// SetTimeout bounds the total wall-clock time Apply or Reset may spend
+	// across all phases. Zero (the default) means no deadline.
+	SetTimeout(timeout time.Duration)
 }