diff --git a/pkg/api/service.go b/pkg/api/service.go index 005fb606..abc6bb73 100644 --- a/pkg/api/service.go +++ b/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() diff --git a/pkg/api/volume.go b/pkg/api/volume.go index e82231bb..b93ac39e 100644 --- a/pkg/api/volume.go +++ b/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() diff --git a/pkg/client/deploy/scheduler/constraint.go b/pkg/client/deploy/scheduler/constraint.go index f2722529..58564f5f 100644 --- a/pkg/client/deploy/scheduler/constraint.go +++ b/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, ", ") +} diff --git a/pkg/client/deploy/scheduler/service.go b/pkg/client/deploy/scheduler/service.go index b7b76b6d..0dac7cf3 100644 --- a/pkg/client/deploy/scheduler/service.go +++ b/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") } diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index a73f652d..8a209b5a 100644 --- a/test/e2e/service_test.go +++ b/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. }