move deploy.operation package to operation

2026-11-10 12:44:16 +00:00 · 2026-10-26 15:19:38 +10:00
parent c16e452439
commit e50c7a0cf6
12 changed files with 9 additions and 9 deletions
@@ -0,0 +1,295 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/docker/compose/v2/pkg/progress"
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/pkg/stringid"
+	cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
+	"github.com/psviderski/uncloud/internal/cli/tui"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// ContainerHealthError indicates that a service container failed to become healthy during deployment.
+type ContainerHealthError struct {
+	error
+	ServiceName string
+	ContainerID string
+	MachineName string
+}
+
+func (e *ContainerHealthError) Unwrap() error {
+	return e.error
+}
+
+// RunContainerOperation creates and starts a new container on a specific machine.
+type RunContainerOperation struct {
+	ServiceID string
+	Spec      api.ServiceSpec
+	MachineID string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName string
+	// SkipHealthMonitor skips the monitoring period and health checks after starting a container.
+	SkipHealthMonitor bool
+}
+
+func (o *RunContainerOperation) Execute(ctx context.Context, cli Client) error {
+	resp, err := cli.CreateContainer(ctx, o.ServiceID, o.Spec, o.MachineID)
+	if err != nil {
+		return fmt.Errorf("create container: %w", err)
+	}
+	// Override event ID so StartContainer and WaitContainerHealthy update the same progress line as creation.
+	// TODO: This is a hack to work around the limitations of the compose progress library.
+	//  We likely need to fork or create our own to decouple event IDs from presentation layer.
+	ctx = cliprogress.WithEventID(ctx, cliprogress.NewContainerEventID(ctx, resp.Name, o.MachineName))
+	if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil {
+		return fmt.Errorf("start container: %w", err)
+	}
+
+	if o.SkipHealthMonitor {
+		return nil
+	}
+
+	opts := api.WaitContainerHealthyOptions{MonitorPeriod: o.Spec.UpdateConfig.MonitorPeriod}
+	if err = cli.WaitContainerHealthy(ctx, o.ServiceID, resp.ID, opts); err != nil {
+		return &ContainerHealthError{
+			error: fmt.Errorf("container '%s/%s' failed to become healthy: %w",
+				o.Spec.Name, stringid.TruncateID(resp.ID), err),
+			ServiceName: o.Spec.Name,
+			ContainerID: resp.ID,
+			MachineName: o.MachineName,
+		}
+	}
+
+	return nil
+}
+
+func (o *RunContainerOperation) Format() string {
+	return tui.BoldGreen.Render("+") + "   " +
+		tui.Faint.Render("run container") + " " +
+		o.Spec.Name + " " +
+		tui.Faint.Render("on") + " " +
+		o.MachineName
+}
+
+func (o *RunContainerOperation) String() string {
+	return fmt.Sprintf("RunContainerOperation[machine_id=%s service_id=%s image=%s]",
+		o.MachineID, o.ServiceID, o.Spec.Container.Image)
+}
+
+// StopContainerOperation stops a container on a specific machine.
+type StopContainerOperation struct {
+	ServiceID   string
+	ContainerID string
+	MachineID   string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName     string
+	StopGracePeriod *time.Duration
+}
+
+func (o *StopContainerOperation) Execute(ctx context.Context, cli Client) error {
+	if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, stopOptions(o.StopGracePeriod)); err != nil {
+		return fmt.Errorf("stop container: %w", err)
+	}
+	return nil
+}
+
+func (o *StopContainerOperation) Format() string {
+	// TODO: pass service name to format the display name consistently with other operations.
+	displayName := stringid.TruncateID(o.ContainerID)
+
+	return tui.BoldRed.Render("-") + "   " +
+		tui.Faint.Render("stop container") + " " +
+		displayName + " " +
+		tui.Faint.Render("on") + " " +
+		o.MachineName
+}
+
+func (o *StopContainerOperation) String() string {
+	return fmt.Sprintf("StopContainerOperation[machine_id=%s service_id=%s container_id=%s]",
+		o.MachineID, o.ServiceID, o.ContainerID)
+}
+
+// RemoveContainerOperation stops and removes a container from a specific machine.
+type RemoveContainerOperation struct {
+	MachineID string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName     string
+	Container       api.ServiceContainer
+	StopGracePeriod *time.Duration
+}
+
+func (o *RemoveContainerOperation) Execute(ctx context.Context, cli Client) error {
+	err := cli.StopContainer(ctx, o.Container.ServiceID(), o.Container.ID, stopOptions(o.StopGracePeriod))
+	if err != nil {
+		return fmt.Errorf("stop container: %w", err)
+	}
+
+	if err = cli.RemoveContainer(ctx, o.Container.ServiceID(), o.Container.ID, container.RemoveOptions{
+		// Remove anonymous volumes created by the container.
+		RemoveVolumes: true,
+	}); err != nil {
+		return fmt.Errorf("remove container: %w", err)
+	}
+
+	return nil
+}
+
+func (o *RemoveContainerOperation) Format() string {
+	displayName := o.Container.ServiceSpec.Name + tui.Faint.Render("/") + o.Container.ShortID()
+
+	return tui.BoldRed.Render("-") + "   " +
+		tui.Faint.Render("remove container") + " " +
+		displayName + " " +
+		tui.Faint.Render("on") + " " +
+		o.MachineName
+}
+
+func (o *RemoveContainerOperation) String() string {
+	return fmt.Sprintf("RemoveContainerOperation[machine_id=%s service_id=%s container_id=%s]",
+		o.MachineID, o.Container.ServiceID(), o.Container.ID)
+}
+
+// ReplaceContainerOperation replaces an old container with a new one based on the specified update order.
+// For start-first: starts new container, then removes old container.
+// For stop-first: stops old container, starts new container, then removes old container.
+type ReplaceContainerOperation struct {
+	ServiceID string
+	Spec      api.ServiceSpec
+	MachineID string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName  string
+	OldContainer api.ServiceContainer
+	// Order specifies the update order: "start-first" or "stop-first".
+	Order string
+	// SkipHealthMonitor skips the monitoring period and health checks after starting a new container.
+	SkipHealthMonitor bool
+	StopGracePeriod   *time.Duration
+}
+
+func (o *ReplaceContainerOperation) Execute(ctx context.Context, cli Client) error {
+	stopFirst := o.Order == api.UpdateOrderStopFirst
+
+	// Create the new container before stopping the old one. CreateContainer also pulls the image according to the
+	// service pull policy, so the stop-first downtime only includes stopping the old container and starting and
+	// monitoring the new one.
+	resp, err := cli.CreateContainer(ctx, o.ServiceID, o.Spec, o.MachineID)
+	if err != nil {
+		return fmt.Errorf("create new container: %w", err)
+	}
+	// Override event ID so StartContainer and WaitContainerHealthy update the same progress line as creation.
+	newCtx := cliprogress.WithEventID(ctx, cliprogress.NewContainerEventID(ctx, resp.Name, o.MachineName))
+
+	wasRunning := false
+	if stopFirst {
+		// Inspect the old container to remember its running state before stopping.
+		ctr, err := cli.InspectContainer(ctx, o.ServiceID, o.OldContainer.ID)
+		if err != nil {
+			return fmt.Errorf("inspect old container: %w", err)
+		}
+		wasRunning = ctr.Container.State.Running
+		if wasRunning {
+			err = cli.StopContainer(ctx, o.ServiceID, o.OldContainer.ID, stopOptions(o.StopGracePeriod))
+			if err != nil {
+				return fmt.Errorf("stop old container: %w", err)
+			}
+		}
+	}
+
+	if err = cli.StartContainer(newCtx, o.ServiceID, resp.ID); err != nil {
+		return fmt.Errorf("start new container: %w", err)
+	}
+
+	if !o.SkipHealthMonitor {
+		opts := api.WaitContainerHealthyOptions{MonitorPeriod: o.Spec.UpdateConfig.MonitorPeriod}
+		if err = cli.WaitContainerHealthy(newCtx, o.ServiceID, resp.ID, opts); err != nil {
+			// New container failed to become healthy. Stop it and roll back to the previous container.
+			// Don't remove the new stopped container to allow users to inspect logs and state.
+
+			// Use context without progress to not overwrite the container Unhealthy status with Stopped.
+			ctxWithoutProgress := progress.WithContextWriter(ctx, nil)
+			_ = cli.StopContainer(ctxWithoutProgress, o.ServiceID, resp.ID, stopOptions(o.StopGracePeriod))
+
+			newCtr := fmt.Sprintf("%s/%s", o.Spec.Name, stringid.TruncateID(resp.ID))
+			healthErr := fmt.Errorf(
+				"new container '%s' failed to become healthy: %w. "+
+					"It's stopped and available for inspection. View logs with 'uc logs %s'",
+				newCtr, err, newCtr,
+			)
+
+			finalErr := healthErr
+			if stopFirst && wasRunning {
+				// Restart the old container only if it was running before we stopped it.
+				oldCtr := fmt.Sprintf("%s/%s", o.OldContainer.ServiceSpec.Name, o.OldContainer.ShortID())
+				if rollbackErr := cli.StartContainer(ctx, o.ServiceID, o.OldContainer.ID); rollbackErr != nil {
+					finalErr = fmt.Errorf("%w. Rolled back to old container '%s' but failed to restart it: %w",
+						healthErr, oldCtr, rollbackErr)
+				} else {
+					finalErr = fmt.Errorf("%w. Rolled back to old container '%s'", healthErr, oldCtr)
+				}
+			}
+
+			return &ContainerHealthError{
+				error:       finalErr,
+				ServiceName: o.Spec.Name,
+				ContainerID: resp.ID,
+				MachineName: o.MachineName,
+			}
+		}
+	}
+
+	// For start-first, we need to stop before removing.
+	// For stop-first, the container is already stopped.
+	if !stopFirst {
+		// TODO: the new container is propagated to Caddy upstreams through the cluster store asynchronously.
+		//  There still might be a brief downtime (for a 1 replica service) when Caddy doesn't know about
+		//  the new container but we're stopping the old container. We should somehow ensure Caddy is updated
+		//  with the new container before we stop the old one to avoid this downtime.
+		if err = cli.StopContainer(ctx, o.ServiceID, o.OldContainer.ID, stopOptions(o.StopGracePeriod)); err != nil {
+			return fmt.Errorf("stop old container: %w", err)
+		}
+	}
+
+	if err = cli.RemoveContainer(ctx, o.ServiceID, o.OldContainer.ID, container.RemoveOptions{
+		RemoveVolumes: true,
+	}); err != nil {
+		return fmt.Errorf("remove old container: %w", err)
+	}
+
+	return nil
+}
+
+func (o *ReplaceContainerOperation) Format() string {
+	displayName := o.Spec.Name + tui.Faint.Render("/") + o.OldContainer.ShortID()
+
+	if o.Order == api.UpdateOrderStopFirst {
+		return tui.BoldYellow.Render("-") + tui.Yellow.Render("/") + tui.BoldYellow.Render("+") + " " +
+			tui.Faint.Render("replace container") + " " +
+			displayName + " " +
+			tui.Faint.Render("on") + " " +
+			o.MachineName + " " +
+			tui.Yellow.Render("(stop-first)")
+	}
+	return tui.BoldGreen.Render("+") + tui.Green.Render("/") + tui.BoldGreen.Render("-") + " " +
+		tui.Faint.Render("replace container") + " " +
+		displayName + " " +
+		tui.Faint.Render("on") + " " +
+		o.MachineName
+}
+
+func (o *ReplaceContainerOperation) String() string {
+	return fmt.Sprintf("ReplaceContainerOperation[machine_id=%s service_id=%s old_container_id=%s order=%s]",
+		o.MachineID, o.ServiceID, o.OldContainer.ID, o.Order)
+}
+
+// stopOptions converts a stop grace period duration to Docker container stop options.
+func stopOptions(gracePeriod *time.Duration) container.StopOptions {
+	if gracePeriod == nil {
+		return container.StopOptions{}
+	}
+	t := int(gracePeriod.Seconds())
+	return container.StopOptions{Timeout: &t}
+}
@@ -0,0 +1,155 @@
+package operation
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/volume"
+	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReplaceContainerOperation_StopFirstPreparation(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name      string
+		createErr error
+		wantErr   string
+		wantCalls []string
+	}{
+		{
+			name: "prepare before cutover",
+			wantCalls: []string{
+				"create:service",
+				"inspect:old",
+				"stop:old",
+				"start:new",
+				"wait:new",
+				"remove:old",
+			},
+		},
+		{
+			name:      "create failure does not interrupt old container",
+			createErr: errors.New("pull failed"),
+			wantErr:   "create new container: pull failed",
+			wantCalls: []string{"create:service"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			cli := &recordingClient{
+				oldContainerRunning: true,
+				createErr:           tt.createErr,
+			}
+			op := newStopFirstReplaceContainerOperation()
+
+			err := op.Execute(context.Background(), cli)
+			if tt.wantErr != "" {
+				require.ErrorContains(t, err, tt.wantErr)
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Equal(t, tt.wantCalls, cli.calls)
+		})
+	}
+}
+
+func newStopFirstReplaceContainerOperation() *ReplaceContainerOperation {
+	return &ReplaceContainerOperation{
+		ServiceID:    "service-id",
+		Spec:         api.ServiceSpec{Name: "service"},
+		MachineID:    "machine-id",
+		MachineName:  "machine",
+		OldContainer: api.ServiceContainer{Container: api.Container{InspectResponse: container.InspectResponse{ContainerJSONBase: &container.ContainerJSONBase{ID: "old"}}}},
+		Order:        api.UpdateOrderStopFirst,
+	}
+}
+
+type recordingClient struct {
+	calls               []string
+	oldContainerRunning bool
+	createErr           error
+}
+
+func (c *recordingClient) CreateContainer(
+	_ context.Context, _ string, spec api.ServiceSpec, _ string,
+) (api.CreateContainerResponse, error) {
+	c.calls = append(c.calls, "create:"+spec.Name)
+	return api.CreateContainerResponse{
+		CreateResponse: container.CreateResponse{ID: "new"},
+		Name:           "new",
+	}, c.createErr
+}
+
+func (c *recordingClient) InspectContainer(
+	_ context.Context, _, containerID string,
+) (api.MachineServiceContainer, error) {
+	c.calls = append(c.calls, "inspect:"+containerID)
+	return api.MachineServiceContainer{
+		Container: api.ServiceContainer{
+			Container: api.Container{InspectResponse: container.InspectResponse{
+				ContainerJSONBase: &container.ContainerJSONBase{
+					ID:    containerID,
+					State: &container.State{Running: c.oldContainerRunning},
+				},
+			}},
+		},
+	}, nil
+}
+
+func (c *recordingClient) StartContainer(_ context.Context, _, containerID string) error {
+	c.calls = append(c.calls, "start:"+containerID)
+	return nil
+}
+
+func (c *recordingClient) StopContainer(
+	_ context.Context, _, containerID string, _ container.StopOptions,
+) error {
+	c.calls = append(c.calls, "stop:"+containerID)
+	return nil
+}
+
+func (c *recordingClient) RemoveContainer(
+	_ context.Context, _, containerID string, _ container.RemoveOptions,
+) error {
+	c.calls = append(c.calls, "remove:"+containerID)
+	return nil
+}
+
+func (c *recordingClient) CreatePreDeployHookContainer(
+	context.Context, string, api.ServiceSpec, string,
+) (api.CreateContainerResponse, error) {
+	panic("unexpected CreatePreDeployHookContainer call")
+}
+
+func (c *recordingClient) ExecContainer(context.Context, string, string, api.ExecOptions) (int, error) {
+	panic("unexpected ExecContainer call")
+}
+
+func (c *recordingClient) WaitContainerHealthy(
+	_ context.Context, _, containerID string, _ api.WaitContainerHealthyOptions,
+) error {
+	c.calls = append(c.calls, "wait:"+containerID)
+	return nil
+}
+
+func (c *recordingClient) CreateVolume(
+	context.Context, string, volume.CreateOptions,
+) (api.MachineVolume, error) {
+	panic("unexpected CreateVolume call")
+}
+
+func (c *recordingClient) ListVolumes(context.Context, *api.VolumeFilter) ([]api.MachineVolume, error) {
+	panic("unexpected ListVolumes call")
+}
+
+func (c *recordingClient) RemoveVolume(context.Context, string, string, bool) error {
+	panic("unexpected RemoveVolume call")
+}
@@ -0,0 +1,23 @@
+package operation
+
+import (
+	"context"
+
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// Operation represents a single operation in the cluster that can be executed.
+// Operations can be composed to form complex workflows (plans).
+type Operation interface {
+	// Execute performs the operation using the provided client.
+	Execute(ctx context.Context, cli Client) error
+	// Format returns a human-readable representation of the operation.
+	Format() string
+	String() string
+}
+
+// Client defines the interface required to execute cluster operations.
+type Client interface {
+	api.ContainerClient
+	api.VolumeClient
+}
@@ -0,0 +1,222 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"charm.land/lipgloss/v2"
+	"github.com/charmbracelet/x/ansi"
+	"github.com/containerd/errdefs"
+	"github.com/docker/compose/v2/pkg/progress"
+	"github.com/docker/docker/api/types/container"
+	cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
+	"github.com/psviderski/uncloud/internal/cli/tui"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// PreDeployHookError indicates that a pre-deploy hook container exited with a non-zero code or timed out.
+type PreDeployHookError struct {
+	error
+	ServiceName string
+	ContainerID string
+	MachineName string
+}
+
+func (e *PreDeployHookError) Unwrap() error {
+	return e.error
+}
+
+// DefaultPreDeployTimeout is the maximum duration to wait for a pre-deploy hook container to complete.
+const DefaultPreDeployTimeout = 5 * time.Minute
+
+// StopPreDeployOperation stops a running pre-deploy hook container from a previous deployment.
+type StopPreDeployOperation struct {
+	MachineID string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName string
+	Container   api.ServiceContainer
+}
+
+func (o *StopPreDeployOperation) Execute(ctx context.Context, cli Client) error {
+	ctx = cliprogress.WithEventID(ctx,
+		cliprogress.OldPreDeployHookEventID(o.Container.ServiceName(), o.Container.ID, o.MachineName))
+	if err := cli.StopContainer(ctx, o.Container.ServiceID(), o.Container.ID, container.StopOptions{}); err != nil {
+		if !errdefs.IsNotFound(err) {
+			return fmt.Errorf("stop pre-deploy hook container '%s': %w", o.Container.ID, err)
+		}
+	}
+	return nil
+}
+
+func (o *StopPreDeployOperation) Format() string {
+	displayName := o.Container.ServiceSpec.Name + tui.Faint.Render("/") + o.Container.ShortID()
+	status, _ := o.Container.HumanState()
+
+	return tui.BoldRed.Render("⏹") + "   " +
+		tui.Faint.Render("stop pre-deploy hook") + " " +
+		displayName + " " +
+		tui.Faint.Render("("+status+")") + " " +
+		tui.Faint.Render("on") + " " +
+		o.MachineName
+}
+
+func (o *StopPreDeployOperation) String() string {
+	return fmt.Sprintf("StopPreDeployOperation[machine_id=%s container_id=%s]",
+		o.MachineID, o.Container.ID)
+}
+
+// RunPreDeployOperation runs a one-shot pre-deploy hook container before service deployment.
+type RunPreDeployOperation struct {
+	ServiceID string
+	Spec      api.ServiceSpec
+	MachineID string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName string
+	// OldContainerIDs are pre-deploy hook containers from previous deployments to remove.
+	OldContainerIDs []string
+}
+
+func (o *RunPreDeployOperation) Execute(ctx context.Context, cli Client) error {
+	// Remove old pre-deploy containers.
+	for _, id := range o.OldContainerIDs {
+		oldCtx := cliprogress.WithEventID(ctx, cliprogress.OldPreDeployHookEventID(o.Spec.Name, id, o.MachineName))
+		_ = cli.StopContainer(oldCtx, o.ServiceID, id, container.StopOptions{})
+		err := cli.RemoveContainer(oldCtx, o.ServiceID, id, container.RemoveOptions{RemoveVolumes: true})
+		if err != nil && !errdefs.IsNotFound(err) {
+			return fmt.Errorf("remove old pre-deploy hook container '%s': %w", id, err)
+		}
+	}
+
+	// Set the event ID override so all downstream client methods (create, start, stop) use the same
+	// pre-deploy hook progress event instead of the generic container one.
+	ctx = cliprogress.WithEventID(ctx, cliprogress.PreDeployHookEventID(o.Spec.Name, o.MachineName))
+
+	resp, err := cli.CreatePreDeployHookContainer(ctx, o.ServiceID, o.Spec, o.MachineID)
+	if err != nil {
+		return fmt.Errorf("create pre-deploy hook container: %w", err)
+	}
+	if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil {
+		return fmt.Errorf("start pre-deploy hook container: %w", err)
+	}
+
+	timeout := DefaultPreDeployTimeout
+	if o.Spec.PreDeploy.Timeout != nil {
+		timeout = *o.Spec.PreDeploy.Timeout
+	}
+
+	return o.waitForExit(ctx, cli, resp.ID, timeout)
+}
+
+// waitForExit polls the container state until it exits or the timeout is reached.
+func (o *RunPreDeployOperation) waitForExit(
+	ctx context.Context, cli Client, containerID string, timeout time.Duration,
+) error {
+	pw := progress.ContextWriter(ctx)
+	eventID := cliprogress.PreDeployHookEventID(o.Spec.Name, o.MachineName)
+	pw.Event(progress.Waiting(eventID))
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	var ctr *api.ServiceContainer
+	for {
+		select {
+		case <-timeoutCtx.Done():
+			// Stop the container on timeout or context cancellation.
+			_ = cli.StopContainer(ctx, o.ServiceID, containerID, container.StopOptions{})
+
+			ctrID := containerID
+			if ctr != nil {
+				ctrID = fmt.Sprintf("%s/%s", o.Spec.Name, ctr.ShortID())
+			}
+			if ctx.Err() != nil {
+				// The parent context has been cancelled before the timeout.
+				pw.Event(progress.NewEvent(eventID, progress.Error, "Cancelled"))
+				return fmt.Errorf("pre-deploy hook container '%s': %w", ctrID, ctx.Err())
+			}
+			pw.Event(progress.NewEvent(eventID, progress.Error, fmt.Sprintf("Timeout (%s)", timeout)))
+			return &PreDeployHookError{
+				ServiceName: o.Spec.Name,
+				ContainerID: containerID,
+				MachineName: o.MachineName,
+				error: fmt.Errorf("pre-deploy hook container '%s' timed out after %s. "+
+					"It's stopped and available for inspection. View logs with 'uc logs %s'",
+					ctrID, timeout, o.Spec.Name),
+			}
+
+		case <-ticker.C:
+			mc, err := cli.InspectContainer(ctx, o.ServiceID, containerID)
+			if err != nil {
+				return fmt.Errorf("inspect pre-deploy hook container: %w", err)
+			}
+			ctr = &mc.Container
+
+			if ctr.State.Running {
+				if state, err := ctr.Container.HumanState(); err == nil {
+					pw.Event(progress.NewEvent(eventID, progress.Working, fmt.Sprintf("Waiting (%s)", state)))
+				}
+				continue
+			}
+
+			// Hook container has exited successfully.
+			if ctr.State.ExitCode == 0 {
+				pw.Event(progress.Event{
+					ID:     eventID,
+					Status: progress.Done,
+				})
+				return nil
+			}
+
+			pw.Event(progress.ErrorEvent(eventID))
+			ctrID := fmt.Sprintf("%s/%s", o.Spec.Name, ctr.ShortID())
+			return &PreDeployHookError{
+				ServiceName: o.Spec.Name,
+				ContainerID: containerID,
+				MachineName: o.MachineName,
+				error: fmt.Errorf("pre-deploy hook container '%s' failed with exit code: %d. "+
+					"It's stopped and available for inspection. View logs with 'uc logs %s'",
+					ctrID, ctr.State.ExitCode, o.Spec.Name),
+			}
+		}
+	}
+}
+
+func (o *RunPreDeployOperation) Format() string {
+	cmd := strings.Join(o.Spec.PreDeploy.Command, " ")
+
+	timeout := DefaultPreDeployTimeout
+	if o.Spec.PreDeploy.Timeout != nil {
+		timeout = *o.Spec.PreDeploy.Timeout
+	}
+
+	prefix := tui.BoldGreen.Render("▶") + "   " +
+		tui.Faint.Render("run pre-deploy hook") + " " +
+		o.Spec.Name + " ["
+	suffix := "] " +
+		tui.Faint.Render("on") + " " +
+		o.MachineName + " " +
+		tui.Yellow.Render(fmt.Sprintf("(timeout %s)", timeout))
+
+	// Truncate the command to fit within the terminal width.
+	termWidth := tui.TerminalWidth()
+	if termWidth > 0 {
+		buffer := 10
+		maxCmdWidth := termWidth - lipgloss.Width(prefix) - lipgloss.Width(suffix) - buffer
+		if maxCmdWidth <= 0 {
+			maxCmdWidth = 10
+		}
+		cmd = ansi.Truncate(cmd, maxCmdWidth, "…")
+	}
+
+	return prefix + cmd + suffix
+}
+
+func (o *RunPreDeployOperation) String() string {
+	return fmt.Sprintf("RunPreDeployOperation[machine_id=%s service_id=%s cmd=%v]",
+		o.MachineID, o.ServiceID, o.Spec.PreDeploy.Command)
+}
@@ -0,0 +1,39 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+	"strings"
+)
+
+// SequenceOperation is a composite operation that executes a sequence of operations in order.
+type SequenceOperation struct {
+	Operations []Operation
+}
+
+func (o *SequenceOperation) Execute(ctx context.Context, cli Client) error {
+	for _, op := range o.Operations {
+		if err := op.Execute(ctx, cli); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (o *SequenceOperation) Format() string {
+	lines := make([]string, len(o.Operations))
+	for i, op := range o.Operations {
+		lines[i] = op.Format()
+	}
+
+	return strings.Join(lines, "\n")
+}
+
+func (o *SequenceOperation) String() string {
+	ops := make([]string, len(o.Operations))
+	for i, op := range o.Operations {
+		ops[i] = op.String()
+	}
+
+	return fmt.Sprintf("SequenceOperation[%s]", strings.Join(ops, ", "))
+}
@@ -0,0 +1,54 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/docker/api/types/volume"
+	"github.com/psviderski/uncloud/internal/cli/tui"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// CreateVolumeOperation creates a volume on a specific machine.
+type CreateVolumeOperation struct {
+	VolumeSpec api.VolumeSpec
+	MachineID  string
+	// MachineName is used for formatting the operation as part of the deployment plan.
+	MachineName string
+}
+
+func (o *CreateVolumeOperation) Execute(ctx context.Context, cli Client) error {
+	if o.VolumeSpec.Type != api.VolumeTypeVolume {
+		return fmt.Errorf("invalid volume type: '%s', expected '%s'", o.VolumeSpec.Type, api.VolumeTypeVolume)
+	}
+
+	opts := volume.CreateOptions{
+		Name: o.VolumeSpec.DockerVolumeName(),
+	}
+	if o.VolumeSpec.VolumeOptions != nil {
+		if o.VolumeSpec.VolumeOptions.Driver != nil {
+			opts.Driver = o.VolumeSpec.VolumeOptions.Driver.Name
+			opts.DriverOpts = o.VolumeSpec.VolumeOptions.Driver.Options
+		}
+		opts.Labels = o.VolumeSpec.VolumeOptions.Labels
+	}
+
+	if _, err := cli.CreateVolume(ctx, o.MachineID, opts); err != nil {
+		return fmt.Errorf("create volume: %w", err)
+	}
+
+	return nil
+}
+
+func (o *CreateVolumeOperation) Format() string {
+	return fmt.Sprintf("%s create volume %s %s %s",
+		tui.BoldGreen.Render("+"),
+		tui.NameStyle.Render(o.VolumeSpec.DockerVolumeName()),
+		tui.Faint.Render("on"),
+		o.MachineName)
+}
+
+func (o *CreateVolumeOperation) String() string {
+	return fmt.Sprintf("CreateVolumeOperation[machine_id=%s volume=%s]",
+		o.MachineID, o.VolumeSpec.DockerVolumeName())
+}