chore: implement VolumesConstraint to only schedule on machines with required volumes

pkg/api/service.go

@@ -52,6 +52,15 @@ type ServiceSpec struct {
 	Volumes []VolumeSpec
 }
 
+func (s *ServiceSpec) Volume(name string) (VolumeSpec, bool) {
+	for _, v := range s.Volumes {
+		if v.Name == name {
+			return v, true
+		}
+	}
+	return VolumeSpec{}, false
+}
+
 func (s *ServiceSpec) SetDefaults() ServiceSpec {
 	spec := s.Clone()
 

pkg/api/volume.go

@@ -14,7 +14,7 @@ import (
 const (
 	// VolumeTypeBind is the type for mounting a host path.
 	VolumeTypeBind = "bind"
-	// VolumeTypeVolume is the type for mounting a managed volume.
+	// VolumeTypeVolume is the type for mounting a named Docker volume.
 	VolumeTypeVolume = "volume"
 	// VolumeTypeTmpfs is the type for mounting a temporary file system stored in the host memory.
 	VolumeTypeTmpfs = "tmpfs"
@@ -42,7 +42,7 @@ type BindOptions struct {
 	Recursive      string            `json:",omitempty"`
 }
 
-// VolumeOptions represents options for a managed volume.
+// VolumeOptions represents options for a named Docker volume.
 type VolumeOptions struct {
 	// Driver specifies the volume driver and its options for volume creation.
 	// TODO: It seems we don't really need Driver and Labels if we only support externally managed volumes.
@@ -50,7 +50,7 @@ type VolumeOptions struct {
 	Driver *mount.Driver `json:",omitempty"`
 	// Labels are key-value metadata to apply to the volume if creating a new volume.
 	Labels map[string]string `json:",omitempty"`
-	// Name of the managed volume to use. If not specified, defaults to the VolumeSpec.Name.
+	// Name of the named Docker volume to use. If not specified, defaults to the VolumeSpec.Name.
 	Name string `json:",omitempty"`
 	// NoCopy prevents automatic copying of data from the container mount path to the volume.
 	NoCopy bool `json:",omitempty"`
@@ -58,6 +58,16 @@ type VolumeOptions struct {
 	SubPath string `json:",omitempty"`
 }
 
+func (v *VolumeSpec) DockerVolumeName() string {
+	if v.Type != VolumeTypeVolume {
+		return ""
+	}
+	if v.VolumeOptions != nil && v.VolumeOptions.Name != "" {
+		return v.VolumeOptions.Name
+	}
+	return v.Name
+}
+
 func (v *VolumeSpec) SetDefaults() VolumeSpec {
 	spec := v.Clone()
 

pkg/client/deploy/scheduler/constraint.go

@@ -1,8 +1,10 @@
 package scheduler
 
 import (
+	"slices"
 	"strings"
 
+	"github.com/docker/docker/api/types/volume"
 	"github.com/psviderski/uncloud/pkg/api"
 )
 
@@ -15,6 +17,7 @@ type Constraint interface {
 	Description() string
 }
 
+// constraintsFromSpec derives scheduling constraints from the service specification.
 func constraintsFromSpec(spec api.ServiceSpec) []Constraint {
 	var constraints []Constraint
 
@@ -24,7 +27,18 @@ func constraintsFromSpec(spec api.ServiceSpec) []Constraint {
 		})
 	}
 
-	// TODO: inspect and add VolumeConstraint.
+	// Add a VolumesConstraint for named Docker volumes that are mounted in the container.
+	var volumes []api.VolumeSpec
+	for _, m := range spec.Container.VolumeMounts {
+		if v, ok := spec.Volume(m.VolumeName); ok && v.Type == api.VolumeTypeVolume {
+			volumes = append(volumes, v)
+		}
+	}
+	if len(volumes) > 0 {
+		constraints = append(constraints, &VolumesConstraint{
+			Volumes: volumes,
+		})
+	}
 
 	return constraints
 }
@@ -45,5 +59,51 @@ func (c *PlacementConstraint) Evaluate(machine *Machine) bool {
 }
 
 func (c *PlacementConstraint) Description() string {
+	slices.Sort(c.Machines)
 	return "Placement constraint by machines: " + strings.Join(c.Machines, ", ")
 }
+
+// VolumesConstraint restricts container placement to machines that have the required named Docker volumes.
+type VolumesConstraint struct {
+	// Volumes is a list of named Docker volumes of type api.VolumeTypeVolume that must exist on the machine.
+	Volumes []api.VolumeSpec
+}
+
+// Evaluate determines if a machine has all the required volumes.
+// Returns true if all required volumes exist on the machine or if there are no required volumes.
+func (c *VolumesConstraint) Evaluate(machine *Machine) bool {
+	if len(c.Volumes) == 0 {
+		return true
+	}
+
+	for _, v := range c.Volumes {
+		if v.Type != api.VolumeTypeVolume {
+			continue
+		}
+
+		// TODO: should we check the volume driver to be local or any matched volume by name is ok?
+		if !slices.ContainsFunc(machine.Volumes, func(vol volume.Volume) bool {
+			return vol.Name == v.DockerVolumeName()
+		}) {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (c *VolumesConstraint) Description() string {
+	volumeNames := make([]string, 0, len(c.Volumes))
+	for _, v := range c.Volumes {
+		if v.Type == api.VolumeTypeVolume {
+			volumeNames = append(volumeNames, v.DockerVolumeName())
+		}
+	}
+	slices.Sort(volumeNames)
+
+	if len(volumeNames) == 0 {
+		return "No volumes constraint"
+	}
+
+	return "Volumes: " + strings.Join(volumeNames, ", ")
+}

pkg/client/deploy/scheduler/service.go

@@ -53,5 +53,7 @@ func (s *ServiceScheduler) evaluateConstraints(machine *Machine) bool {
 }
 
 func (s *ServiceScheduler) ScheduleContainer() ([]*pb.MachineInfo, error) {
+	// TODO: organise machines in a heap and supply a sort function from the strategy. Each scheduled container
+	//  should update the machine and reorder it in the heap.
 	return nil, errors.New("not implemented")
 }

test/e2e/service_test.go

@@ -582,9 +582,211 @@ func TestDeployment(t *testing.T) {
 		d := deploy.NewDeployment(cli, spec, nil)
 		_, err := d.Run(ctx)
 		require.Error(t, err, "Deployment should fail when volume doesn't exist")
-		require.Contains(t, err.Error(), "volume 'non-existent-volume' not found")
+		require.Contains(t, err.Error(), "no machines available")
+		// TODO: implement and check for more details about the failed constraints.
+		//require.Contains(t, err.Error(), "volume 'non-existent-volume' not found")
 	})
 
+	// Tests that when a volume exists on a single machine, all requested replicas will be deployed to that machine,
+	// regardless of how many replicas are requested.
+	t.Run("replicated with volume on single machine", func(t *testing.T) {
+		t.Parallel()
+
+		serviceName := "test-replicated-with-volume-single-machine"
+		volumeName := serviceName
+		t.Cleanup(func() {
+			err := cli.RemoveService(ctx, serviceName)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+
+			err = cli.RemoveVolume(ctx, c.Machines[1].Name, volumeName, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+		})
+
+		vol, err := cli.CreateVolume(ctx, c.Machines[1].Name, volume.CreateOptions{Name: volumeName})
+		require.NoError(t, err, "Failed to create test volume")
+
+		spec := api.ServiceSpec{
+			Name: serviceName,
+			Container: api.ContainerSpec{
+				Image: "portainer/pause:latest",
+				VolumeMounts: []api.VolumeMount{
+					{
+						VolumeName:    volumeName,
+						ContainerPath: "/data",
+					},
+				},
+			},
+			Volumes: []api.VolumeSpec{
+				{
+					Name: volumeName,
+					Type: api.VolumeTypeVolume,
+				},
+			},
+			Replicas: 3,
+		}
+
+		d := deploy.NewDeployment(cli, spec, nil)
+		_, err = d.Run(ctx)
+		require.NoError(t, err)
+
+		svc, err := cli.InspectService(ctx, serviceName)
+		require.NoError(t, err)
+		assertServiceMatchesSpec(t, svc, spec)
+
+		for _, ctr := range svc.Containers {
+			assert.Equal(t, vol.MachineID, ctr.MachineID,
+				"All containers should be on the machine where the volume is located")
+		}
+	})
+
+	// Tests replica distribution across machines that have the same volume.
+	t.Run("replicated with volume on multiple machines", func(t *testing.T) {
+		t.Parallel()
+
+		serviceName := "test-replicated-with-volume-multi-machine"
+		volumeName := serviceName
+		t.Cleanup(func() {
+			err := cli.RemoveService(ctx, serviceName)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+
+			err = cli.RemoveVolume(ctx, c.Machines[0].Name, volumeName, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+			err = cli.RemoveVolume(ctx, c.Machines[1].Name, volumeName, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+		})
+
+		vol1, err := cli.CreateVolume(ctx, c.Machines[0].Name, volume.CreateOptions{Name: volumeName})
+		require.NoError(t, err, "Failed to create volume on first machine")
+		vol2, err := cli.CreateVolume(ctx, c.Machines[1].Name, volume.CreateOptions{Name: volumeName})
+		require.NoError(t, err, "Failed to create volume on second machine")
+
+		spec := api.ServiceSpec{
+			Name: serviceName,
+			Container: api.ContainerSpec{
+				Image: "portainer/pause:latest",
+				VolumeMounts: []api.VolumeMount{
+					{
+						VolumeName:    volumeName,
+						ContainerPath: "/data",
+					},
+				},
+			},
+			Volumes: []api.VolumeSpec{
+				{
+					Name: volumeName,
+					Type: api.VolumeTypeVolume,
+				},
+			},
+			Replicas: 3,
+		}
+
+		d := deploy.NewDeployment(cli, spec, nil)
+		_, err = d.Run(ctx)
+		require.NoError(t, err)
+
+		svc, err := cli.InspectService(ctx, serviceName)
+		require.NoError(t, err)
+		assertServiceMatchesSpec(t, svc, spec)
+
+		machines := serviceMachines(svc)
+		assert.ElementsMatch(t, machines.ToSlice(), []string{vol1.MachineID, vol2.MachineID},
+			"Containers should be distributed across machines with the same volume")
+	})
+
+	// Tests that a service requiring multiple volumes is correctly deployed to a machine that has all of those volumes.
+	t.Run("replicated with multiple volumes on single machine", func(t *testing.T) {
+		t.Parallel()
+
+		serviceName := "test-replicated-with-multi-volume-single-machine"
+		vol1Name := serviceName + "1"
+		vol2Name := serviceName + "2"
+
+		t.Cleanup(func() {
+			err := cli.RemoveService(ctx, serviceName)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+
+			err = cli.RemoveVolume(ctx, c.Machines[0].Name, vol1Name, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+			err = cli.RemoveVolume(ctx, c.Machines[1].Name, vol1Name, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+			err = cli.RemoveVolume(ctx, c.Machines[1].Name, vol2Name, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+			err = cli.RemoveVolume(ctx, c.Machines[2].Name, vol2Name, false)
+			if !errors.Is(err, api.ErrNotFound) {
+				assert.NoError(t, err)
+			}
+		})
+
+		_, err := cli.CreateVolume(ctx, c.Machines[0].Name, volume.CreateOptions{Name: vol1Name})
+		require.NoError(t, err)
+		_, err = cli.CreateVolume(ctx, c.Machines[1].Name, volume.CreateOptions{Name: vol1Name})
+		require.NoError(t, err)
+		_, err = cli.CreateVolume(ctx, c.Machines[1].Name, volume.CreateOptions{Name: vol2Name})
+		require.NoError(t, err)
+		_, err = cli.CreateVolume(ctx, c.Machines[2].Name, volume.CreateOptions{Name: vol2Name})
+		require.NoError(t, err)
+
+		spec := api.ServiceSpec{
+			Name: serviceName,
+			Container: api.ContainerSpec{
+				Image: "portainer/pause:latest",
+				VolumeMounts: []api.VolumeMount{
+					{
+						VolumeName:    vol1Name,
+						ContainerPath: "/data1",
+					},
+					{
+						VolumeName:    vol2Name,
+						ContainerPath: "/data2",
+					},
+				},
+			},
+			Volumes: []api.VolumeSpec{
+				{
+					Name: vol1Name,
+					Type: api.VolumeTypeVolume,
+				},
+				{
+					Name: vol2Name,
+					Type: api.VolumeTypeVolume,
+				},
+			},
+			Replicas: 3,
+		}
+
+		d := deploy.NewDeployment(cli, spec, nil)
+		_, err = d.Run(ctx)
+		require.NoError(t, err)
+
+		svc, err := cli.InspectService(ctx, serviceName)
+		require.NoError(t, err)
+		assertServiceMatchesSpec(t, svc, spec)
+
+		machines := serviceMachines(svc)
+		assert.ElementsMatch(t, machines.ToSlice(), []string{c.Machines[1].ID},
+			"Containers should be deployed to the machine that has both volumes")
+	})
+
+	// TODO: global deployment with a volume
+
 	// TODO: test deployments with unreachable machines. See https://github.com/psviderski/uncloud/issues/29.
 }