chore: split deploy operations into separate files in operation package

cmd/uncloud/deploy.go

@@ -14,6 +14,7 @@ import (
 	"github.com/psviderski/uncloud/pkg/client"
 	"github.com/psviderski/uncloud/pkg/client/compose"
 	"github.com/psviderski/uncloud/pkg/client/deploy"
+	"github.com/psviderski/uncloud/pkg/client/deploy/operation"
 	"github.com/spf13/cobra"
 )
 
@@ -198,7 +199,7 @@ func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error {
 	}, uncli.ProgressOut(), "Deploying services")
 }
 
-func printPlan(ctx context.Context, cli *client.Client, plan deploy.SequenceOperation) error {
+func printPlan(ctx context.Context, cli *client.Client, plan operation.SequenceOperation) error {
 	for _, op := range plan.Operations {
 		svcPlan, ok := op.(*deploy.Plan)
 		if !ok {

pkg/client/compose/deploy.go

@@ -13,6 +13,7 @@ import (
 	"github.com/docker/docker/api/types/volume"
 	"github.com/psviderski/uncloud/pkg/api"
 	"github.com/psviderski/uncloud/pkg/client/deploy"
+	"github.com/psviderski/uncloud/pkg/client/deploy/operation"
 	"github.com/psviderski/uncloud/pkg/client/deploy/scheduler"
 )
 
@@ -27,7 +28,7 @@ type Deployment struct {
 	SpecResolver *deploy.ServiceSpecResolver
 	Strategy     deploy.Strategy
 	state        *scheduler.ClusterState
-	plan         *deploy.SequenceOperation
+	plan         *operation.SequenceOperation
 }
 
 func NewDeployment(ctx context.Context, cli Client, project *types.Project) (*Deployment, error) {
@@ -58,11 +59,11 @@ func NewDeploymentWithStrategy(ctx context.Context, cli Client, project *types.P
 	}, nil
 }
 
-func (d *Deployment) Plan(ctx context.Context) (deploy.SequenceOperation, error) {
+func (d *Deployment) Plan(ctx context.Context) (operation.SequenceOperation, error) {
 	if d.plan != nil {
 		return *d.plan, nil
 	}
-	plan := deploy.SequenceOperation{}
+	plan := operation.SequenceOperation{}
 
 	// Generate service specs for all services in the project.
 	var serviceSpecs []api.ServiceSpec
@@ -123,7 +124,7 @@ func (d *Deployment) ServiceSpec(name string) (api.ServiceSpec, error) {
 }
 
 // PlanVolumes checks if the external volumes exist and plans the creation of missing volumes.
-func (d *Deployment) planVolumes(serviceSpecs []api.ServiceSpec) ([]*deploy.CreateVolumeOperation, error) {
+func (d *Deployment) planVolumes(serviceSpecs []api.ServiceSpec) ([]*operation.CreateVolumeOperation, error) {
 	if len(d.Project.Volumes) == 0 {
 		// No volumes to check or create.
 		return nil, nil
@@ -145,7 +146,7 @@ func (d *Deployment) planVolumes(serviceSpecs []api.ServiceSpec) ([]*deploy.Crea
 	}
 
 	// Generate operations to create scheduled missing volumes.
-	var ops []*deploy.CreateVolumeOperation
+	var ops []*operation.CreateVolumeOperation
 	for machineID, volumes := range scheduledVolumes {
 		for _, v := range volumes {
 			machineName := machineID
@@ -153,7 +154,7 @@ func (d *Deployment) planVolumes(serviceSpecs []api.ServiceSpec) ([]*deploy.Crea
 				machineName = m.Info.Name
 			}
 
-			ops = append(ops, &deploy.CreateVolumeOperation{
+			ops = append(ops, &operation.CreateVolumeOperation{
 				MachineID:   machineID,
 				MachineName: machineName,
 				VolumeSpec:  v,

pkg/client/deploy/deploy.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/psviderski/uncloud/pkg/client/deploy/operation"
 	"github.com/psviderski/uncloud/pkg/client/deploy/scheduler"
 )
 
@@ -33,7 +34,7 @@ type Deployment struct {
 type Plan struct {
 	ServiceID   string
 	ServiceName string
-	SequenceOperation
+	operation.SequenceOperation
 }
 
 // NewDeployment creates a new deployment for the given service specification.

pkg/client/deploy/operation/container.go

@@ -1,36 +1,13 @@
-package deploy
+package operation
 
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/docker/docker/api/types/container"
-	"github.com/docker/docker/api/types/volume"
 	"github.com/psviderski/uncloud/pkg/api"
 )
 
-// Operation represents a single atomic operation in a deployment process.
-// Operations can be composed to form complex deployment strategies.
-type Operation interface {
-	// Execute performs the operation using the provided client.
-	// TODO: Encapsulate the client in the operation as otherwise it gives an impression that different clients
-	//  can be provided. But in reality, the operation is tightly coupled with the client that was used to create it.
-	Execute(ctx context.Context, cli Client) error
-	// Format returns a human-readable representation of the operation.
-	// TODO: get rid of the resolver and assign the required names for formatting in the operation itself.
-	Format(resolver NameResolver) string
-	String() string
-}
-
-// NameResolver resolves machine and container IDs to their names.
-type NameResolver interface {
-	MachineName(machineID string) string
-	ContainerName(containerID string) string
-}
-
-// TODO: pass api.ServiceContainer to operations to simplify operation formatting in the plan.
-
 // RunContainerOperation creates and starts a new container on a specific machine.
 type RunContainerOperation struct {
 	ServiceID string
@@ -118,46 +95,6 @@ func (o *RemoveContainerOperation) String() string {
 		o.MachineID, o.Container.ServiceID(), o.Container.ID)
 }
 
-// CreateVolumeOperation creates a volume on a specific machine.
-type CreateVolumeOperation struct {
-	VolumeSpec api.VolumeSpec
-	MachineID  string
-	// MachineName is used for formatting the operation output only.
-	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(_ NameResolver) string {
-	return fmt.Sprintf("%s: Create volume [name=%s]", o.MachineName, o.VolumeSpec.DockerVolumeName())
-}
-
-func (o *CreateVolumeOperation) String() string {
-	return fmt.Sprintf("CreateVolumeOperation[machine_id=%s volume=%s]",
-		o.MachineID, o.VolumeSpec.DockerVolumeName())
-}
-
 // 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.
@@ -216,35 +153,3 @@ 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)
 }
-
-// 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(resolver NameResolver) string {
-	ops := make([]string, len(o.Operations))
-	for i, op := range o.Operations {
-		ops[i] = "- " + op.Format(resolver)
-	}
-
-	return strings.Join(ops, "\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, ", "))
-}

pkg/client/deploy/operation/operation.go

@@ -0,0 +1,34 @@
+package operation
+
+import (
+	"context"
+
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// Operation represents a single atomic operation in a deployment process.
+// Operations can be composed to form complex deployment strategies.
+type Operation interface {
+	// Execute performs the operation using the provided client.
+	// TODO: Encapsulate the client in the operation as otherwise it gives an impression that different clients
+	//  can be provided. But in reality, the operation is tightly coupled with the client that was used to create it.
+	Execute(ctx context.Context, cli Client) error
+	// Format returns a human-readable representation of the operation.
+	// TODO: get rid of the resolver and assign the required names for formatting in the operation itself.
+	Format(resolver NameResolver) string
+	String() string
+}
+
+// NameResolver resolves machine and container IDs to their names.
+type NameResolver interface {
+	MachineName(machineID string) string
+	ContainerName(containerID string) string
+}
+
+// TODO: pass api.ServiceContainer to operations to simplify operation formatting in the plan.
+
+// Client defines the interface required to execute deployment operations.
+type Client interface {
+	api.ContainerClient
+	api.VolumeClient
+}

pkg/client/deploy/operation/sequence.go

@@ -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(resolver NameResolver) string {
+	ops := make([]string, len(o.Operations))
+	for i, op := range o.Operations {
+		ops[i] = "- " + op.Format(resolver)
+	}
+
+	return strings.Join(ops, "\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, ", "))
+}

pkg/client/deploy/operation/volume.go

@@ -0,0 +1,49 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/docker/api/types/volume"
+	"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 output only.
+	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(_ NameResolver) string {
+	return fmt.Sprintf("%s: Create volume [name=%s]", o.MachineName, o.VolumeSpec.DockerVolumeName())
+}
+
+func (o *CreateVolumeOperation) String() string {
+	return fmt.Sprintf("CreateVolumeOperation[machine_id=%s volume=%s]",
+		o.MachineID, o.VolumeSpec.DockerVolumeName())
+}

pkg/client/deploy/strategy.go

@@ -8,6 +8,7 @@ import (
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/internal/secret"
 	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/psviderski/uncloud/pkg/client/deploy/operation"
 	"github.com/psviderski/uncloud/pkg/client/deploy/scheduler"
 )
 
@@ -144,7 +145,7 @@ func (s *RollingStrategy) planReplicated(svc *api.Service, spec api.ServiceSpec)
 
 		if len(containers) == 0 {
 			// No more existing containers on this machine, create a new one.
-			plan.Operations = append(plan.Operations, &RunContainerOperation{
+			plan.Operations = append(plan.Operations, &operation.RunContainerOperation{
 				ServiceID: plan.ServiceID,
 				Spec:      spec,
 				MachineID: m.Id,
@@ -164,7 +165,7 @@ func (s *RollingStrategy) planReplicated(svc *api.Service, spec api.ServiceSpec)
 
 		// Replace the old container with a new one.
 		order := determineUpdateOrder(ctr, spec)
-		plan.Operations = append(plan.Operations, &ReplaceContainerOperation{
+		plan.Operations = append(plan.Operations, &operation.ReplaceContainerOperation{
 			ServiceID:    plan.ServiceID,
 			Spec:         spec,
 			MachineID:    m.Id,
@@ -176,7 +177,7 @@ func (s *RollingStrategy) planReplicated(svc *api.Service, spec api.ServiceSpec)
 	// Remove any remaining containers that are not needed.
 	for mid, containers := range containersOnMachine {
 		for _, c := range containers {
-			plan.Operations = append(plan.Operations, &RemoveContainerOperation{
+			plan.Operations = append(plan.Operations, &operation.RemoveContainerOperation{
 				MachineID: mid,
 				Container: c,
 			})
@@ -227,7 +228,7 @@ func (s *RollingStrategy) planGlobal(svc *api.Service, spec api.ServiceSpec) (Pl
 	// Remove any remaining containers on machines that don't match the new placement constraints.
 	for _, containers := range containersOnMachine {
 		for _, c := range containers {
-			plan.Operations = append(plan.Operations, &RemoveContainerOperation{
+			plan.Operations = append(plan.Operations, &operation.RemoveContainerOperation{
 				MachineID: c.MachineID,
 				Container: c.Container,
 			})
@@ -242,12 +243,12 @@ func (s *RollingStrategy) planGlobal(svc *api.Service, spec api.ServiceSpec) (Pl
 // removing old ones. If there is a host port conflict, it stops the old container before starting a new one.
 func reconcileGlobalContainer(
 	containers []api.MachineServiceContainer, spec api.ServiceSpec, serviceID, machineID string, forceRecreate bool,
-) ([]Operation, error) {
-	var ops []Operation
+) ([]operation.Operation, error) {
+	var ops []operation.Operation
 
 	if len(containers) == 0 {
 		// No containers on this machine, create a new one.
-		ops = append(ops, &RunContainerOperation{
+		ops = append(ops, &operation.RunContainerOperation{
 			ServiceID: serviceID,
 			Spec:      spec,
 			MachineID: machineID,
@@ -277,7 +278,7 @@ func reconcileGlobalContainer(
 				if i == j {
 					continue
 				}
-				ops = append(ops, &RemoveContainerOperation{
+				ops = append(ops, &operation.RemoveContainerOperation{
 					MachineID: old.MachineID,
 					Container: old.Container,
 				})
@@ -310,7 +311,7 @@ func reconcileGlobalContainer(
 			}
 			conflictingPorts, err := c.Container.ConflictingServicePorts(spec.Ports)
 			if err != nil || len(conflictingPorts) > 0 {
-				ops = append(ops, &StopContainerOperation{
+				ops = append(ops, &operation.StopContainerOperation{
 					ServiceID:   serviceID,
 					ContainerID: c.Container.ID,
 					MachineID:   machineID,
@@ -320,7 +321,7 @@ func reconcileGlobalContainer(
 
 		// Replace the running container with a new one.
 		order := determineUpdateOrder(containerToReplace.Container, spec)
-		ops = append(ops, &ReplaceContainerOperation{
+		ops = append(ops, &operation.ReplaceContainerOperation{
 			ServiceID:    serviceID,
 			Spec:         spec,
 			MachineID:    machineID,
@@ -333,20 +334,20 @@ func reconcileGlobalContainer(
 			if c.Container.ID == containerToReplace.Container.ID {
 				continue
 			}
-			ops = append(ops, &RemoveContainerOperation{
+			ops = append(ops, &operation.RemoveContainerOperation{
 				MachineID: c.MachineID,
 				Container: c.Container,
 			})
 		}
 	} else {
 		// No running containers, create a new one and remove all stopped containers.
-		ops = append(ops, &RunContainerOperation{
+		ops = append(ops, &operation.RunContainerOperation{
 			ServiceID: serviceID,
 			Spec:      spec,
 			MachineID: machineID,
 		})
 		for _, c := range containers {
-			ops = append(ops, &RemoveContainerOperation{
+			ops = append(ops, &operation.RemoveContainerOperation{
 				MachineID: c.MachineID,
 				Container: c.Container,
 			})

pkg/client/deploy/strategy_test.go

@@ -7,6 +7,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/psviderski/uncloud/pkg/client/deploy/operation"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -304,7 +305,7 @@ func TestReconcileGlobalContainer(t *testing.T) {
 		containers    []api.MachineServiceContainer
 		spec          api.ServiceSpec
 		forceRecreate bool
-		expectedOps   []Operation
+		expectedOps   []operation.Operation
 	}{
 		{
 			name:       "no containers creates new",
@@ -312,8 +313,8 @@ func TestReconcileGlobalContainer(t *testing.T) {
 			spec: api.ServiceSpec{
 				Container: api.ContainerSpec{Image: "nginx:latest"},
 			},
-			expectedOps: []Operation{
-				&RunContainerOperation{
+			expectedOps: []operation.Operation{
+				&operation.RunContainerOperation{
 					ServiceID: "service-1",
 					MachineID: "machine-1",
 				},
@@ -330,8 +331,8 @@ func TestReconcileGlobalContainer(t *testing.T) {
 					{ContainerPort: 8080, PublishedPort: 8080, Protocol: "tcp", Mode: api.PortModeHost},
 				},
 			},
-			expectedOps: []Operation{
-				&ReplaceContainerOperation{
+			expectedOps: []operation.Operation{
+				&operation.ReplaceContainerOperation{
 					ServiceID:    "service-1",
 					MachineID:    "machine-1",
 					OldContainer: container1,
@@ -352,19 +353,19 @@ func TestReconcileGlobalContainer(t *testing.T) {
 					{ContainerPort: 9090, PublishedPort: 9090, Protocol: "tcp", Mode: api.PortModeHost},
 				},
 			},
-			expectedOps: []Operation{
-				&StopContainerOperation{
+			expectedOps: []operation.Operation{
+				&operation.StopContainerOperation{
 					ServiceID:   "service-1",
 					ContainerID: "container-2",
 					MachineID:   "machine-1",
 				},
-				&ReplaceContainerOperation{
+				&operation.ReplaceContainerOperation{
 					ServiceID:    "service-1",
 					MachineID:    "machine-1",
 					OldContainer: container1,
 					Order:        api.UpdateOrderStopFirst,
 				},
-				&RemoveContainerOperation{
+				&operation.RemoveContainerOperation{
 					MachineID: "machine-1",
 					Container: container2WithPort9090,
 				},
@@ -383,14 +384,14 @@ func TestReconcileGlobalContainer(t *testing.T) {
 				},
 			},
 			// Container-2 has no conflicting ports, so no StopContainerOperation for it.
-			expectedOps: []Operation{
-				&ReplaceContainerOperation{
+			expectedOps: []operation.Operation{
+				&operation.ReplaceContainerOperation{
 					ServiceID:    "service-1",
 					MachineID:    "machine-1",
 					OldContainer: container1,
 					Order:        api.UpdateOrderStopFirst,
 				},
-				&RemoveContainerOperation{
+				&operation.RemoveContainerOperation{
 					MachineID: "machine-1",
 					Container: container2WithPort3000,
 				},
@@ -409,11 +410,11 @@ func TestReconcileGlobalContainer(t *testing.T) {
 
 // assertOperationsEqual compares expected and actual operations, ignoring the Spec field
 // which is passed separately to the function and not the focus of these tests.
-func assertOperationsEqual(t *testing.T, expected, actual []Operation) {
+func assertOperationsEqual(t *testing.T, expected, actual []operation.Operation) {
 	t.Helper()
 	opts := cmp.Options{
-		cmpopts.IgnoreFields(RunContainerOperation{}, "Spec"),
-		cmpopts.IgnoreFields(ReplaceContainerOperation{}, "Spec"),
+		cmpopts.IgnoreFields(operation.RunContainerOperation{}, "Spec"),
+		cmpopts.IgnoreFields(operation.ReplaceContainerOperation{}, "Spec"),
 		cmpopts.IgnoreUnexported(api.Container{}),
 	}
 	if diff := cmp.Diff(expected, actual, opts); diff != "" {