refactor: 'service run' create missing volumes satisfying all placement constraints

cmd/uncloud/service/run.go

@@ -4,11 +4,9 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"slices"
 	"strings"
 
 	"github.com/docker/compose/v2/pkg/progress"
-	"github.com/docker/docker/api/types/volume"
 	"github.com/docker/docker/daemon/names"
 	"github.com/psviderski/uncloud/internal/cli"
 	"github.com/psviderski/uncloud/internal/secret"
@@ -113,26 +111,8 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error {
 	}
 	defer clusterClient.Close()
 
-	machineIDForVolumes, missingVolumes, err := selectMachineForVolumes(
-		ctx,
-		clusterClient,
-		spec.Volumes,
-		spec.Placement.Machines,
-	)
-	if err != nil {
-		return err
-	}
-
 	var resp client.RunServiceResponse
 	err = progress.RunWithTitle(ctx, func(ctx context.Context) error {
-		// Create missing volumes on the selected machine.
-		for _, v := range missingVolumes {
-			_, err = clusterClient.CreateVolume(ctx, machineIDForVolumes, volume.CreateOptions{Name: v.Name})
-			if err != nil {
-				return fmt.Errorf("create volume '%s': %w", v.Name, err)
-			}
-		}
-
 		resp, err = clusterClient.RunService(ctx, spec)
 		if err != nil {
 			return fmt.Errorf("run service: %w", err)
@@ -358,76 +338,3 @@ func parseVolumeFlagValue(volume string) (api.VolumeSpec, api.VolumeMount, error
 
 	return spec, mount, nil
 }
-
-// selectMachineForVolumes selects a machine to run a service with the given volumes on and determines which volumes
-// need to be created on the selected machine. An empty machineID is returned if no named volumes are specified.
-func selectMachineForVolumes(
-	ctx context.Context, clusterClient *client.Client, volumes []api.VolumeSpec, machinesFilter []string,
-) (machineID string, missingVolumes []api.VolumeSpec, err error) {
-	var volumeNames []string
-	for _, v := range volumes {
-		if v.Type == api.VolumeTypeVolume {
-			volumeNames = append(volumeNames, v.Name)
-		}
-	}
-	if len(volumeNames) == 0 {
-		return "", nil, nil
-	}
-
-	vfilter := &api.VolumeFilter{
-		Machines: machinesFilter,
-		Names:    volumeNames,
-	}
-	vols, err := clusterClient.ListVolumes(ctx, vfilter)
-	if err != nil {
-		return "", nil, fmt.Errorf("list volumes: %w", err)
-	}
-
-	if len(vols) > 0 {
-		// Some volumes have been found on the machines matching the machinesFilter.
-		// Pick the machine with the most volumes to create fewer duplicate volumes.
-		volumesCountOnMachines := make(map[string]int)
-		for _, vol := range vols {
-			volumesCountOnMachines[vol.MachineID]++
-		}
-
-		maxCount := 0
-		for mid, count := range volumesCountOnMachines {
-			if count > maxCount {
-				machineID = mid
-				maxCount = count
-			}
-		}
-	} else {
-		// No volumes found on the machines matching the machinesFilter.
-		// Pick the first available machine to create the volumes on.
-		mfilter := &api.MachineFilter{
-			Available:  true,
-			NamesOrIDs: machinesFilter,
-		}
-		availableMachines, err := clusterClient.ListMachines(ctx, mfilter)
-		if err != nil {
-			return "", nil, fmt.Errorf("list machines: %w", err)
-		}
-
-		if len(availableMachines) == 0 {
-			return "", nil, fmt.Errorf("no available machines to create the volume(s) on")
-		}
-		machineID = availableMachines[0].Machine.Id
-	}
-
-	// Find missing volumes that need to be created on the selected machine.
-	for _, v := range volumes {
-		if v.Type != api.VolumeTypeVolume {
-			continue
-		}
-
-		if !slices.ContainsFunc(vols, func(mv api.MachineVolume) bool {
-			return mv.Volume.Name == v.Name && mv.MachineID == machineID
-		}) {
-			missingVolumes = append(missingVolumes, v)
-		}
-	}
-
-	return machineID, missingVolumes, nil
-}

pkg/client/deploy/scheduler/volume.go

@@ -11,7 +11,7 @@ import (
 )
 
 // VolumeScheduler determines what missing volumes should be created and where for a multi-service deployment.
-// It must satisfy the following constraints:
+// It satisfies the following constraints:
 //   - Services that share a volume must be placed on the same machine where the volume is located.
 //     If the volume is located on multiple machines, services can be placed on any of them.
 //   - Services must respect their individual placement constraints.
@@ -25,6 +25,7 @@ type VolumeScheduler struct {
 	// volumeSpecs is a map of volume names to their specifications from the service specs in a canonical form.
 	volumeSpecs map[string]api.VolumeSpec
 	// volumeServices is a map of volume names to the list of service names that use the volume.
+	// TODO: not all service spec may contain the service name, so use a slice of int indexes instead of names.
 	volumeServices map[string][]string
 	// existingVolumeMachines is a map of volume names to the set of machine IDs where those volumes are located.
 	// Contains only volumes that are used by at least one service in serviceSpecs.

pkg/client/service.go

@@ -9,8 +9,10 @@ import (
 	"sync"
 
 	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/volume"
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/psviderski/uncloud/pkg/client/deploy/scheduler"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -39,6 +41,33 @@ func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunSer
 		}
 	}
 
+	// Create missing named Docker volumes for the service.
+	if len(spec.MountedDockerVolumes()) > 0 {
+		volumeScheduler, err := scheduler.NewVolumeSchedulerWithClient(ctx, cli, []api.ServiceSpec{spec})
+		if err != nil {
+			return resp, fmt.Errorf("init volume scheduler: %w", err)
+		}
+
+		scheduledVolumes, err := volumeScheduler.Schedule()
+		if err != nil {
+			return resp, fmt.Errorf("schedule volumes: %w", err)
+		}
+
+		// Create the missing volumes on the scheduled machines.
+		for machineID, volumes := range scheduledVolumes {
+			for _, v := range volumes {
+				_, err = cli.CreateVolume(ctx, machineID, volume.CreateOptions{
+					Name:       v.Name,
+					Driver:     v.VolumeOptions.Driver.Name,
+					DriverOpts: v.VolumeOptions.Driver.Options,
+				})
+				if err != nil {
+					return resp, fmt.Errorf("create volume '%s': %w", v.Name, err)
+				}
+			}
+		}
+	}
+
 	deployment := cli.NewDeployment(spec, nil)
 	plan, err := deployment.Run(ctx)
 	if err != nil {

test/e2e/assert.go

@@ -19,10 +19,11 @@ import (
 )
 
 func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpec) {
+	spec = spec.SetDefaults()
 	assert.Equal(t, spec.Name, svc.Name)
 
 	if svc.Mode == api.ServiceModeReplicated {
-		assert.Contains(t, []string{"", api.ServiceModeReplicated}, spec.Mode)
+		assert.Equal(t, api.ServiceModeReplicated, spec.Mode)
 		assert.Len(t, svc.Containers, int(spec.Replicas), "Expected %d replicas", spec.Replicas)
 	} else {
 		assert.Equal(t, spec.Mode, svc.Mode)
@@ -34,10 +35,10 @@ func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpe
 }
 
 func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api.ServiceSpec) {
+	spec = spec.SetDefaults()
 	status := deploy.EvalContainerSpecChange(ctr.ServiceSpec, spec)
 	assert.Equal(t, deploy.ContainerUpToDate, status)
 
-	spec = spec.SetDefaults()
 	// Verify labels.
 	assert.True(t, api.ValidateServiceID(ctr.Config.Labels[api.LabelServiceID]))
 	assert.Equal(t, spec.Name, ctr.Config.Labels[api.LabelServiceName])

test/e2e/service_test.go

@@ -1388,6 +1388,65 @@ func TestServiceLifecycle(t *testing.T) {
 		assert.Equal(t, spec.Ports, ports)
 	})
 
+	t.Run("3 replicas with volume auto-created", func(t *testing.T) {
+		t.Parallel()
+
+		name := "test-3-replicas-volume-auto-created"
+		volumeName := name
+		t.Cleanup(func() {
+			err := cli.RemoveService(ctx, name)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+
+			volumes, err := cli.ListVolumes(ctx, &api.VolumeFilter{Names: []string{volumeName}})
+			require.NoError(t, err)
+			for _, v := range volumes {
+				err = cli.RemoveVolume(ctx, v.MachineID, v.Volume.Name, false)
+				assert.NoError(t, err)
+			}
+		})
+
+		volumes, err := cli.ListVolumes(ctx, &api.VolumeFilter{Names: []string{volumeName}})
+		require.NoError(t, err)
+		assert.Len(t, volumes, 0, "Volume should not exist before service creation")
+
+		spec := api.ServiceSpec{
+			Name: name,
+			Container: api.ContainerSpec{
+				Image: "portainer/pause:latest",
+				VolumeMounts: []api.VolumeMount{
+					{
+						VolumeName:    volumeName,
+						ContainerPath: "/data",
+					},
+				},
+			},
+			Volumes: []api.VolumeSpec{
+				{
+					Name: volumeName,
+					Type: api.VolumeTypeVolume,
+				},
+			},
+			Replicas: 3,
+		}
+		resp, err := cli.RunService(ctx, spec)
+		require.NoError(t, err)
+
+		svc, err := cli.InspectService(ctx, resp.ID)
+		require.NoError(t, err)
+		assertServiceMatchesSpec(t, svc, spec)
+
+		volumes, err = cli.ListVolumes(ctx, &api.VolumeFilter{Names: []string{volumeName}})
+		require.NoError(t, err)
+		assert.Len(t, volumes, 1, "Volume should be created automatically")
+		assert.Equal(t, volumeName, volumes[0].Volume.Name)
+
+		machines := serviceMachines(svc)
+		assert.Equal(t, []string{volumes[0].MachineID}, machines.ToSlice(),
+			"Replicas should be on the same machine as the volume")
+	})
+
 	t.Run("global mode", func(t *testing.T) {
 		t.Parallel()