diff --git a/pkg/api/service.go b/pkg/api/service.go index abc6bb73..986a08aa 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -61,6 +61,18 @@ func (s *ServiceSpec) Volume(name string) (VolumeSpec, bool) { return VolumeSpec{}, false } +// MountedDockerVolumes returns the list of volumes of VolumeTypeVolume type that are mounted into the container. +func (s *ServiceSpec) MountedDockerVolumes() []VolumeSpec { + volumes := make(map[string]VolumeSpec) + for _, m := range s.Container.VolumeMounts { + if v, ok := s.Volume(m.VolumeName); ok && v.Type == VolumeTypeVolume { + volumes[v.Name] = v + } + } + + return slices.Collect(maps.Values(volumes)) +} + func (s *ServiceSpec) SetDefaults() ServiceSpec { spec := s.Clone() diff --git a/pkg/api/volume.go b/pkg/api/volume.go index b93ac39e..76ba0714 100644 --- a/pkg/api/volume.go +++ b/pkg/api/volume.go @@ -18,6 +18,9 @@ const ( VolumeTypeVolume = "volume" // VolumeTypeTmpfs is the type for mounting a temporary file system stored in the host memory. VolumeTypeTmpfs = "tmpfs" + + // VolumeDriverLocal is the default volume driver for local named Docker volumes. + VolumeDriverLocal = "local" ) // VolumeSpec defines a volume mount specification. As of April 2025, the volume must be created before deploying @@ -76,7 +79,7 @@ func (v *VolumeSpec) SetDefaults() VolumeSpec { spec.VolumeOptions = &VolumeOptions{} } if spec.VolumeOptions.Driver == nil { - spec.VolumeOptions.Driver = &mount.Driver{Name: "local"} + spec.VolumeOptions.Driver = &mount.Driver{Name: VolumeDriverLocal} } if spec.VolumeOptions.Name == "" { spec.VolumeOptions.Name = spec.Name @@ -113,6 +116,33 @@ func (v *VolumeSpec) Equals(other VolumeSpec) bool { return reflect.DeepEqual(vol, other) } +// MatchesDockerVolume checks if this VolumeSpec is compatible with the given named Docker volume. +// In other words, it checks if the spec could be used to create the volume. +func (v *VolumeSpec) MatchesDockerVolume(vol volume.Volume) bool { + if v.Type != VolumeTypeVolume { + return false + } + spec := v.SetDefaults() + + if spec.DockerVolumeName() != vol.Name { + return false + } + + volDriver := vol.Driver + if volDriver == "" { + volDriver = VolumeDriverLocal + } + if spec.VolumeOptions.Driver.Name != volDriver { + return false + } + + if !reflect.DeepEqual(spec.VolumeOptions.Driver.Options, vol.Options) { + return false + } + + return true +} + func (v *VolumeSpec) Clone() VolumeSpec { spec := *v diff --git a/pkg/client/deploy/scheduler/service.go b/pkg/client/deploy/scheduler/service.go index 0dac7cf3..76b7fd7d 100644 --- a/pkg/client/deploy/scheduler/service.go +++ b/pkg/client/deploy/scheduler/service.go @@ -10,29 +10,35 @@ import ( ) type ServiceScheduler struct { - Machines []*Machine - Spec api.ServiceSpec - Constraints []Constraint + machines []*Machine + spec api.ServiceSpec + constraints []Constraint } -func NewServiceScheduler(ctx context.Context, cli Client, spec api.ServiceSpec) (*ServiceScheduler, error) { +func NewServiceSchedulerWithClient(ctx context.Context, cli Client, spec api.ServiceSpec) (*ServiceScheduler, error) { machines, err := InspectMachines(ctx, cli) if err != nil { return nil, fmt.Errorf("inspect machines: %w", err) } + return NewServiceSchedulerWithMachines(machines, spec), nil +} + +// NewServiceSchedulerWithMachines creates a new ServiceScheduler with the given machines and service specification. +func NewServiceSchedulerWithMachines(machines []*Machine, spec api.ServiceSpec) *ServiceScheduler { constraints := constraintsFromSpec(spec) return &ServiceScheduler{ - Machines: machines, - Spec: spec, - Constraints: constraints, - }, nil + machines: machines, + spec: spec, + constraints: constraints, + } } -func (s *ServiceScheduler) AvailableMachines() ([]*Machine, error) { +// EligibleMachines returns a list of machines that satisfy all constraints for the next scheduled container. +func (s *ServiceScheduler) EligibleMachines() ([]*Machine, error) { var available []*Machine - for _, machine := range s.Machines { + for _, machine := range s.machines { if s.evaluateConstraints(machine) { available = append(available, machine) } @@ -44,7 +50,7 @@ func (s *ServiceScheduler) AvailableMachines() ([]*Machine, error) { } func (s *ServiceScheduler) evaluateConstraints(machine *Machine) bool { - for _, c := range s.Constraints { + for _, c := range s.constraints { if !c.Evaluate(machine) { return false } diff --git a/pkg/client/deploy/scheduler/volume.go b/pkg/client/deploy/scheduler/volume.go new file mode 100644 index 00000000..ab6daefd --- /dev/null +++ b/pkg/client/deploy/scheduler/volume.go @@ -0,0 +1,296 @@ +package scheduler + +import ( + "context" + "fmt" + "slices" + "strings" + + mapset "github.com/deckarep/golang-set/v2" + "github.com/psviderski/uncloud/pkg/api" +) + +// VolumeScheduler determines what missing volumes should be created and where for a multi-service deployment. +// It must satisfy 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. +// - If a volume already exists on a machine, it must be used instead of creating a new one. +// - A missing volume must only be created on one machine. +type VolumeScheduler struct { + // machines is a list of available machines in the cluster. + machines []*Machine + // serviceSpecs is a list of service specifications included in the deployment. + serviceSpecs []api.ServiceSpec + // 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. + 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. + existingVolumeMachines map[string]mapset.Set[string] +} + +// NewVolumeSchedulerWithClient creates a new VolumeScheduler with the given cluster client and service specifications. +func NewVolumeSchedulerWithClient(ctx context.Context, cli Client, specs []api.ServiceSpec) (*VolumeScheduler, error) { + machines, err := InspectMachines(ctx, cli) + if err != nil { + return nil, fmt.Errorf("inspect machines: %w", err) + } + + return NewVolumeSchedulerWithMachines(machines, specs) +} + +// NewVolumeSchedulerWithMachines creates a new VolumeScheduler with the given cluster machines +// and service specifications. +func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec) (*VolumeScheduler, error) { + var specsWithVolumes []api.ServiceSpec + // Docker volume name -> VolumeSpec. + volumeSpecs := make(map[string]api.VolumeSpec) + // Volume name -> list of service names that use the volume. + volumeServices := make(map[string][]string) + // Volume name -> set of machine IDs where the volume is located. + existingVolumeMachines := make(map[string]mapset.Set[string]) + + // Validate all service names are unique to avoid scheduling conflicts. + serviceNames := make(map[string]struct{}, len(specs)) + for _, spec := range specs { + if err := spec.Validate(); err != nil { + return nil, fmt.Errorf("invalid service spec: %w", err) + } + + if _, exists := serviceNames[spec.Name]; exists { + return nil, fmt.Errorf("duplicate service name: '%s'", spec.Name) + } + serviceNames[spec.Name] = struct{}{} + + mountedVolumes := spec.MountedDockerVolumes() + if len(mountedVolumes) == 0 { + continue + } + specsWithVolumes = append(specsWithVolumes, spec) + + for _, v := range mountedVolumes { + v = v.SetDefaults() + // Reset any aliases in a service spec to the actual Docker volume name. + v.Name = v.DockerVolumeName() + + if seenVolume, ok := volumeSpecs[v.Name]; ok { + if !seenVolume.Equals(v) { + return nil, fmt.Errorf("volume '%s' is defined multiple times with different options", v.Name) + } + } else { + volumeSpecs[v.Name] = v + } + + volumeServices[v.Name] = append(volumeServices[v.Name], spec.Name) + } + } + + // Validate the configurations of existing volumes on machines don't conflict with the volume specs, for example, + // a volume and a spec with the same name don't have different drivers. + for _, machine := range machines { + for _, vol := range machine.Volumes { + if spec, ok := volumeSpecs[vol.Name]; ok { + if !spec.MatchesDockerVolume(vol) { + return nil, fmt.Errorf("volume '%s' specification does not match the existing volume "+ + "on machine '%s'", vol.Name, machine.Info.Name) + } + + if _, setInitialised := existingVolumeMachines[vol.Name]; !setInitialised { + existingVolumeMachines[vol.Name] = mapset.NewSet[string]() + } + existingVolumeMachines[vol.Name].Add(machine.Info.Id) + } + } + } + + return &VolumeScheduler{ + machines: machines, + serviceSpecs: specsWithVolumes, + volumeSpecs: volumeSpecs, + volumeServices: volumeServices, + existingVolumeMachines: existingVolumeMachines, + }, nil +} + +// Schedule determines what missing volumes should be created and where for services in the multi-service deployment. +// It returns a map of machine IDs to a list of api.VolumeSpec that should be created on that machine, +// or an error if services can't be scheduled due to scheduling constraints. +func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) { + if len(s.serviceSpecs) == 0 { + // No services with volume mounts, nothing to schedule. + return nil, nil + } + + // Service name -> set of machine IDs where the service can be scheduled. + serviceEligibleMachines := make(map[string]mapset.Set[string]) + // Volume name -> list of service names that use the volume. + // Get eligible machines for each service without considering its volume mounts. + for _, spec := range s.serviceSpecs { + machineIDs, err := s.serviceEligibleMachinesWithoutVolumes(spec) + if err != nil { + return nil, err + } + serviceEligibleMachines[spec.Name] = machineIDs + } + + // For each volume that exists on any machine(s) (which shouldn't be created), intersect each service's + // eligible machines that use the volume with the machines the volume is located on. + // Service name -> list of processed volume names (quoted) to format the error message. + quotedServiceVolumes := make(map[string][]string) + for volumeName, volumeMachines := range s.existingVolumeMachines { + for _, serviceName := range s.volumeServices[volumeName] { + quotedServiceVolumes[serviceName] = append(quotedServiceVolumes[serviceName], + fmt.Sprintf("'%s'", volumeName)) + newEligibleMachines := serviceEligibleMachines[serviceName].Intersect(volumeMachines) + if newEligibleMachines.Cardinality() == 0 { + volumes := strings.Join(quotedServiceVolumes[serviceName], ", ") + return nil, fmt.Errorf("unable to find a machine that satisfies service '%s' "+ + "placement constraints and has all required volumes: %s", serviceName, volumes) + } + serviceEligibleMachines[serviceName] = newEligibleMachines + } + } + + // Skip constraints propagation for volumes that already exist on machines as the propagation only works + // for missing volumes. + placedVolumes := make(map[string]struct{}) + for volumeName := range s.existingVolumeMachines { + placedVolumes[volumeName] = struct{}{} + } + + if err := s.propagateConstraintsUntilConvergence(serviceEligibleMachines, placedVolumes); err != nil { + return nil, err + } + + // Schedule each missing volume on one of its eligible machines. + scheduledVolumes := make(map[string][]api.VolumeSpec) + for missingVolumeName, missingVolumeSpec := range s.volumeSpecs { + // Skip volumes that already exist on machines. + if _, ok := s.existingVolumeMachines[missingVolumeName]; ok { + continue + } + + serviceNames := s.volumeServices[missingVolumeName] + if len(serviceNames) == 0 { + return nil, fmt.Errorf("bug detected: no services using volume '%s'", missingVolumeName) + } + + // Get the current eligible machines (any service using the volume will have the same set after convergence). + eligibleMachines := serviceEligibleMachines[serviceNames[0]] + if eligibleMachines.Cardinality() == 0 { + return nil, fmt.Errorf("bug detected: no eligible machines for volume '%s'", missingVolumeName) + } + + // Choose the first machine in the sorted eligible machines to schedule the volume on. + // Sort the eligible machines to ensure deterministic behavior. + sortedEligibleMachines := eligibleMachines.ToSlice() + slices.Sort(sortedEligibleMachines) + machineID := sortedEligibleMachines[0] + // Update constraints for all services that use this volume to be placed on the selected machine. + for _, serviceName := range serviceNames { + serviceEligibleMachines[serviceName] = mapset.NewSet(machineID) + } + placedVolumes[missingVolumeName] = struct{}{} + scheduledVolumes[machineID] = append(scheduledVolumes[machineID], missingVolumeSpec) + + // Propagate the updated constraints. + if err := s.propagateConstraintsUntilConvergence(serviceEligibleMachines, placedVolumes); err != nil { + return nil, fmt.Errorf("unexpected error while propagating constraints after "+ + "scheduling volume '%s' on machine '%s': %w", missingVolumeName, machineID, err) + } + } + + return scheduledVolumes, nil +} + +// serviceEligibleMachinesWithoutVolumes returns a set of machine IDs where the service can be scheduled +// without considering its volume mounts. +func (s *VolumeScheduler) serviceEligibleMachinesWithoutVolumes(spec api.ServiceSpec) (mapset.Set[string], error) { + specWithoutVolumes := spec.Clone() + specWithoutVolumes.Container.VolumeMounts = nil + + scheduler := NewServiceSchedulerWithMachines(s.machines, specWithoutVolumes) + machines, err := scheduler.EligibleMachines() + if err != nil { + return nil, fmt.Errorf("schedule service '%s': %w", spec.Name, err) + } + + machineIDs := mapset.NewSetWithSize[string](len(machines)) + for _, m := range machines { + machineIDs.Add(m.Info.Id) + } + + return machineIDs, nil +} + +// propagateConstraintsUntilConvergence iteratively narrows down eligible machines for services by propagating +// constraints through shared volumes until convergence. It only processes volumes that need to be created (not +// existing volumes) and ensures services sharing a volume converge to the same set of eligible machines. +// If skipVolumes is provided, those volumes are excluded from constraint propagation. +// Returns an error if any services have no eligible machines after constraint propagation. +func (s *VolumeScheduler) propagateConstraintsUntilConvergence( + serviceEligibleMachines map[string]mapset.Set[string], + skipVolumes map[string]struct{}, +) error { + changed := true + // Loop until no more changes occur. + for changed { + changed = false + + // For each volume, find the intersection of eligible machines for all services using that volume and update + // their eligible machines with the intersection. This will narrow down the machines where the volume can + // be created. + for volumeName, serviceNames := range s.volumeServices { + if skipVolumes != nil { + if _, ok := skipVolumes[volumeName]; ok { + continue + } + } + // Skip if there are no services using this volume (shouldn't happen). + if len(serviceNames) == 0 { + continue + } + + // Find the intersection of eligible machines for all services using this volume. + var eligibleMachinesForVolume mapset.Set[string] + first := true + for _, serviceName := range serviceNames { + if first { + // First service: initialise the intersection. + eligibleMachinesForVolume = serviceEligibleMachines[serviceName].Clone() + first = false + } else { + eligibleMachinesForVolume = serviceEligibleMachines[serviceName].Intersect( + eligibleMachinesForVolume) + } + } + + // If no machines are eligible for this volume, we have a constraint violation. + //goland:noinspection GoDfaNilDereference + if eligibleMachinesForVolume.Cardinality() == 0 { + var quotedServiceNames []string // Used to format the error message. + for _, svcName := range serviceNames { + quotedServiceNames = append(quotedServiceNames, fmt.Sprintf("'%s'", svcName)) + } + return fmt.Errorf("unable to find a machine that satisfies placement constraints "+ + "for services %s that must be placed together to share volume '%s'", + strings.Join(quotedServiceNames, ", "), volumeName) + } + + // Update eligible machines for all services using this volume. + newCount := eligibleMachinesForVolume.Cardinality() + for _, serviceName := range serviceNames { + oldCount := serviceEligibleMachines[serviceName].Cardinality() + serviceEligibleMachines[serviceName] = eligibleMachinesForVolume + + if oldCount != newCount { + changed = true + } + } + } + } + + return nil +} diff --git a/pkg/client/deploy/scheduler/volume_test.go b/pkg/client/deploy/scheduler/volume_test.go new file mode 100644 index 00000000..171f0eed --- /dev/null +++ b/pkg/client/deploy/scheduler/volume_test.go @@ -0,0 +1,816 @@ +package scheduler + +import ( + "testing" + + "github.com/docker/docker/api/types/mount" + "github.com/docker/docker/api/types/volume" + "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestVolumeScheduler_Schedule(t *testing.T) { + tests := []struct { + name string + machines []*Machine + serviceSpecs []api.ServiceSpec + want map[string][]api.VolumeSpec + wantErr string + }{ + { + name: "single service with missing volume", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{ + "machine1": { + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + { + name: "multiple services sharing a missing volume", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{}, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{}, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{ + "machine1": { + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + { + name: "service with existing volume", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{ + { + Name: "vol1", + }, + }, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{}, + }, + { + name: "service with placement constraint and missing volume", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{}, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{}, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Placement: api.Placement{ + Machines: []string{"machine2"}, + }, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{ + "machine2": { + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + { + name: "services with conflicting placement constraints", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{}, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{}, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Placement: api.Placement{ + Machines: []string{"machine1"}, + }, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Placement: api.Placement{ + Machines: []string{"machine2"}, + }, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + wantErr: "unable to find a machine that satisfies placement constraints for services " + + "'service1', 'service2' that must be placed together to share volume 'vol1'", + }, + { + name: "service with existing volume on wrong machine", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{ + { + Name: "vol1", + }, + }, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{}, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Placement: api.Placement{ + Machines: []string{"machine2"}, + }, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + wantErr: "unable to find a machine that satisfies service 'service1' placement constraints " + + "and has all required volumes: 'vol1'", + }, + { + name: "multiple services with multiple volumes, some shared", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{}, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{}, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data1", + }, + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + { + VolumeName: "vol3", + ContainerPath: "/data3", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service3", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol3", + ContainerPath: "/data3", + }, + { + VolumeName: "vol4", + ContainerPath: "/data4", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{ + "machine1": { + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + { + name: "multiple services with multiple volumes, some shared and existing", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{ + { + Name: "vol2", + }, + }, + }, + { + Info: &pb.MachineInfo{ + Id: "machine3", + }, + Volumes: []volume.Volume{ + { + Name: "vol1", + }, + { + Name: "vol2", + }, + }, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol3", + ContainerPath: "/data3", + }, + { + VolumeName: "vol4", + ContainerPath: "/data4", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data1", + }, + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + { + VolumeName: "vol3", + ContainerPath: "/data3", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service3", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + { + VolumeName: "vol4-alias", + ContainerPath: "/data4", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol4-alias", + Type: api.VolumeTypeVolume, + VolumeOptions: &api.VolumeOptions{ + Name: "vol4", + Driver: &mount.Driver{ + Name: api.VolumeDriverLocal, + }, + }, + }, + }, + }, + { + Name: "service4", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2-alias", + ContainerPath: "/data2", + }, + { + VolumeName: "vol5", + ContainerPath: "/data5", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2-alias", + Type: api.VolumeTypeVolume, + VolumeOptions: &api.VolumeOptions{ + Name: "vol2", + Driver: &mount.Driver{ + Name: api.VolumeDriverLocal, + }, + }, + }, + { + Name: "vol5", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{ + "machine2": { + { + Name: "vol5", + Type: api.VolumeTypeVolume, + }, + }, + "machine3": { + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + { + name: "multiple services with multiple volumes, some shared, with conflicting placement constraints", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{}, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{}, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Placement: api.Placement{ + Machines: []string{"machine1"}, + }, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data1", + }, + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Placement: api.Placement{ + Machines: []string{"machine2"}, + }, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + { + VolumeName: "vol3", + ContainerPath: "/data3", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + wantErr: "unable to find a machine that satisfies placement constraints for services " + + "'service1', 'service2' that must be placed together to share volume 'vol2'", + }, + { + name: "multiple services with multiple volumes, no shared", + machines: []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + }, + Volumes: []volume.Volume{ + { + Name: "vol1", + }, + }, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + }, + Volumes: []volume.Volume{ + { + Name: "vol3", + }, + }, + }, + }, + serviceSpecs: []api.ServiceSpec{ + { + Name: "service1", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data1", + }, + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol3", + ContainerPath: "/data3", + }, + { + VolumeName: "vol4", + ContainerPath: "/data4", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol3", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + want: map[string][]api.VolumeSpec{ + "machine1": { + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + }, + "machine2": { + { + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + scheduler, err := NewVolumeSchedulerWithMachines(tt.machines, tt.serviceSpecs) + require.NoError(t, err) + result, err := scheduler.Schedule() + + if tt.wantErr != "" { + assert.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + } else { + assert.NoError(t, err) + + assert.Len(t, result, len(tt.want), "Number of machines with volumes to create should match") + for machineID, expectedVolumes := range tt.want { + // Transform the expected volumes to the canonical form with defaults set. + for i := range expectedVolumes { + expectedVolumes[i] = expectedVolumes[i].SetDefaults() + } + + actualVolumes, ok := result[machineID] + assert.True(t, ok, "Machine %s should be in the result", machineID) + assert.ElementsMatch(t, expectedVolumes, actualVolumes, + "Volumes for machine %s should match", machineID) + } + } + }) + } +} diff --git a/pkg/client/deploy/strategy.go b/pkg/client/deploy/strategy.go index f65bbea5..af8c0abf 100644 --- a/pkg/client/deploy/strategy.go +++ b/pkg/client/deploy/strategy.go @@ -56,12 +56,12 @@ func (s *RollingStrategy) planReplicated( return plan, err } - sched, err := scheduler.NewServiceScheduler(ctx, cli, spec) + sched, err := scheduler.NewServiceSchedulerWithClient(ctx, cli, spec) if err != nil { return plan, err } // TODO: return a detailed report on required constraints and which ones are satisfied? - availableMachines, err := sched.AvailableMachines() + availableMachines, err := sched.EligibleMachines() if err != nil { return plan, err } @@ -216,12 +216,12 @@ func (s *RollingStrategy) planGlobal( } } - sched, err := scheduler.NewServiceScheduler(ctx, cli, spec) + sched, err := scheduler.NewServiceSchedulerWithClient(ctx, cli, spec) if err != nil { return plan, err } - availableMachines, err := sched.AvailableMachines() + availableMachines, err := sched.EligibleMachines() if err != nil { return plan, err } diff --git a/pkg/client/service.go b/pkg/client/service.go index 4942c647..ac0706d8 100644 --- a/pkg/client/service.go +++ b/pkg/client/service.go @@ -11,7 +11,6 @@ import ( "github.com/docker/docker/api/types/container" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" - "github.com/psviderski/uncloud/pkg/client/deploy" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" @@ -40,7 +39,7 @@ func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunSer } } - deployment := cli.NewDeployment(spec, &deploy.RollingStrategy{}) + deployment := cli.NewDeployment(spec, nil) plan, err := deployment.Run(ctx) if err != nil { return resp, err