From 948201cb6fe419e3ee3d67a58f66d2e3e7aef184 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Sun, 20 Apr 2025 11:13:30 +1000 Subject: [PATCH 1/3] add VolumeScheduler with one flaky test --- pkg/api/service.go | 12 + pkg/api/volume.go | 32 +- pkg/client/deploy/scheduler/service.go | 28 +- pkg/client/deploy/scheduler/volume.go | 330 +++++++ pkg/client/deploy/scheduler/volume_test.go | 968 +++++++++++++++++++++ pkg/client/deploy/strategy.go | 8 +- pkg/client/service.go | 3 +- 7 files changed, 1363 insertions(+), 18 deletions(-) create mode 100644 pkg/client/deploy/scheduler/volume.go create mode 100644 pkg/client/deploy/scheduler/volume_test.go 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..0436bf84 --- /dev/null +++ b/pkg/client/deploy/scheduler/volume.go @@ -0,0 +1,330 @@ +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. +// TODO: add rules that if a volume exists, it should be used instead of creating a new one. +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 + // 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) { + // TODO: validate specs before scheduling, need to update tests to use helper functions to create specs with images. + var specsWithVolumes []api.ServiceSpec + // Docker volume name -> VolumeSpec. + volumeSpecs := make(map[string]api.VolumeSpec) + // Volume name -> set of machine IDs where the volume is located. + volumeMachines := 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 _, 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 { + if seenVolume, ok := volumeSpecs[v.DockerVolumeName()]; ok { + if !seenVolume.Equals(v) { + return nil, fmt.Errorf("volume '%s' is defined multiple times with different options", + v.DockerVolumeName()) + } + } else { + v = v.SetDefaults() + v.Name = v.DockerVolumeName() // Reset any aliases in a service spec to the actual Docker volume name. + volumeSpecs[v.Name] = v + } + } + } + + // 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 := volumeMachines[vol.Name]; !setInitialised { + volumeMachines[vol.Name] = mapset.NewSet[string]() + } + volumeMachines[vol.Name].Add(machine.Info.Id) + } + } + } + + return &VolumeScheduler{ + machines: machines, + serviceSpecs: specsWithVolumes, + volumeSpecs: volumeSpecs, + existingVolumeMachines: volumeMachines, + }, 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. + volumeServices := make(map[string][]string) + // 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 + + // Populate volumeServices with Docker volumes used by this service. + for _, v := range spec.MountedDockerVolumes() { + volumeName := v.DockerVolumeName() + volumeServices[volumeName] = append(volumeServices[volumeName], spec.Name) + } + } + + // 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 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 + } + } + + for serviceName, eligibleMachines := range serviceEligibleMachines { + fmt.Printf("### Service '%s' can be scheduled on machines: %v\n", serviceName, eligibleMachines.ToSlice()) + } + + // For each missing volume, intersect the eligible machines for all services using the volume + // and choose the first machine in the sorted intersection to create the volume on. + scheduledVolumes := make(map[string][]api.VolumeSpec) + for missingVolumeName, missingVolumeSpec := range s.volumeSpecs { + if _, ok := s.existingVolumeMachines[missingVolumeName]; ok { + // This volume already exists, no need to create it. + continue + } + + var eligibleMachines mapset.Set[string] + var quotedServiceNames []string // Used to format the error message. + for i, serviceName := range volumeServices[missingVolumeName] { + if i == 0 { + eligibleMachines = serviceEligibleMachines[serviceName] + } else { + eligibleMachines = serviceEligibleMachines[serviceName].Intersect(eligibleMachines) + } + quotedServiceNames = append(quotedServiceNames, fmt.Sprintf("'%s'", serviceName)) + } + + if eligibleMachines == nil { + return nil, fmt.Errorf("bug detected: no services using volume '%s'", missingVolumeName) + } + if eligibleMachines.Cardinality() == 0 { + return nil, 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, ", "), missingVolumeName) + } + + // Choose the first machine in the sorted eligible machines to create the volume on. + // Sort the eligible machines to ensure deterministic behavior. + // TODO: the first machine might not be the optimal one. Ideally, we need to do the intersection for all volumes + // multiple times until they converge. Then picking any machine is fine. + sortedEligibleMachines := eligibleMachines.ToSlice() + slices.Sort(sortedEligibleMachines) + machineID := sortedEligibleMachines[0] + scheduledVolumes[machineID] = append(scheduledVolumes[machineID], missingVolumeSpec) + // Update the eligible machines to the chosen machine for all services using this volume. + eligibleMachines = mapset.NewSet(machineID) + for _, serviceName := range volumeServices[missingVolumeName] { + serviceEligibleMachines[serviceName] = eligibleMachines + } + } + + return scheduledVolumes, nil + + //// For each missing volume (should be created only on one machine) + //machineToVolumeSpecs := make(map[string][]api.VolumeSpec) + //for volumeName, serviceNames := range missingVolumes { + // // Get the intersection of candidate machines for all services using this volume + // var intersection []string + // for i, serviceName := range serviceNames { + // if i == 0 { + // intersection = serviceEligibleMachines[serviceName] + // } else { + // intersection = s.intersectMachines(intersection, serviceEligibleMachines[serviceName]) + // } + // } + // + // if len(intersection) == 0 { + // return nil, fmt.Errorf("unable to find a machine where services %v can be placed together to share the missing volume %s", + // serviceNames, volumeName) + // } + // + // // Sort the intersection to ensure deterministic behavior + // sortedIntersection := make([]string, len(intersection)) + // copy(sortedIntersection, intersection) + // slices.Sort(sortedIntersection) + // + // // Choose the first machine in the sorted intersection to create the volume on + // machineID := sortedIntersection[0] + // + // // Get the volume spec for this volume name + // volumeSpec := volumeSpecs[volumeName] + // + // // Add the volume spec to the machine's list + // machineToVolumeSpecs[machineID] = append(machineToVolumeSpecs[machineID], volumeSpec) + // + // // Update the candidate machines for all services using this volume + // for _, serviceName := range serviceNames { + // serviceEligibleMachines[serviceName] = intersection + // } + //} + // + //return machineToVolumeSpecs, 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 +} + +// getAllVolumesAndSpecs returns a map of all volume names used by services and a map of volume names to their specs. +func (s *VolumeScheduler) getAllVolumesAndSpecs() (map[string]struct{}, map[string]api.VolumeSpec) { + volumes := make(map[string]struct{}) + volumeSpecs := make(map[string]api.VolumeSpec) + + for _, serviceSpec := range s.serviceSpecs { + for _, mount := range serviceSpec.Container.VolumeMounts { + if v, ok := serviceSpec.Volume(mount.VolumeName); ok && v.Type == api.VolumeTypeVolume { + volumeName := v.DockerVolumeName() + volumes[volumeName] = struct{}{} + volumeSpecs[volumeName] = v + } + } + } + + return volumes, volumeSpecs +} + +// getVolumeLocations returns a map of volume names to the list of machine IDs where they exist. +func (s *VolumeScheduler) getVolumeLocations(allVolumes map[string]struct{}) map[string][]string { + volumeLocations := make(map[string][]string) + + // Initialize the map with empty slices for all volumes + for volumeName := range allVolumes { + volumeLocations[volumeName] = []string{} + } + + // Populate the map with machine IDs where each volume exists + for _, machine := range s.machines { + for _, vol := range machine.Volumes { + if _, ok := volumeLocations[vol.Name]; ok { + volumeLocations[vol.Name] = append(volumeLocations[vol.Name], machine.Info.Id) + } + } + } + + return volumeLocations +} + +// getVolumeServices returns a map of volume names to the list of service names that use them. +func (s *VolumeScheduler) getVolumeServices(allVolumes map[string]struct{}) map[string][]string { + volumeServices := make(map[string][]string) + + // Initialize the map with empty slices for all volumes + for volumeName := range allVolumes { + volumeServices[volumeName] = []string{} + } + + for _, serviceSpec := range s.serviceSpecs { + serviceName := serviceSpec.Name + for _, mount := range serviceSpec.Container.VolumeMounts { + if v, ok := serviceSpec.Volume(mount.VolumeName); ok && v.Type == api.VolumeTypeVolume { + volumeName := v.DockerVolumeName() + if _, ok := allVolumes[volumeName]; ok { + volumeServices[volumeName] = append(volumeServices[volumeName], serviceName) + } + } + } + } + + return volumeServices +} + +// intersectMachines returns the intersection of two slices of machine IDs. +func (s *VolumeScheduler) intersectMachines(a, b []string) []string { + var result []string + for _, id := range a { + if slices.Contains(b, id) { + result = append(result, id) + } + } + return result +} diff --git a/pkg/client/deploy/scheduler/volume_test.go b/pkg/client/deploy/scheduler/volume_test.go new file mode 100644 index 00000000..0e239c5e --- /dev/null +++ b/pkg/client/deploy/scheduler/volume_test.go @@ -0,0 +1,968 @@ +package scheduler + +import ( + "testing" + + "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" +) + +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{ + 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{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service2", + Container: api.ContainerSpec{ + 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{ + 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{ + 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{ + 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{ + 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{ + 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{ + 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{ + 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{ + 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{ + 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{ + 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{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + { + VolumeName: "vol4", + ContainerPath: "/data4", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + // TODO: use vol4-alias name and Docker name in VolumeOptions + Name: "vol4", + Type: api.VolumeTypeVolume, + }, + }, + }, + { + Name: "service4", + Container: api.ContainerSpec{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + { + VolumeName: "vol5", + ContainerPath: "/data5", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + { + 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{ + 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{ + 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{ + 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{ + 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) + assert.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) + } + } + }) + } +} + +func TestVolumeScheduler_getAllVolumesAndSpecs(t *testing.T) { + serviceSpecs := map[string]api.ServiceSpec{ + "service1": { + Name: "service1", + Container: api.ContainerSpec{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + "service2": { + Name: "service2", + Container: api.ContainerSpec{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol2", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + }, + }, + } + + // Convert map to slice + specsList := make([]api.ServiceSpec, 0, len(serviceSpecs)) + for _, spec := range serviceSpecs { + specsList = append(specsList, spec) + } + scheduler, err := NewVolumeSchedulerWithMachines(nil, specsList) + assert.NoError(t, err) + volumes, specs := scheduler.getAllVolumesAndSpecs() + + assert.Len(t, volumes, 2) + assert.Contains(t, volumes, "vol1") + assert.Contains(t, volumes, "vol2") + + assert.Len(t, specs, 2) + assert.Equal(t, api.VolumeSpec{Name: "vol1", Type: api.VolumeTypeVolume}, specs["vol1"]) + assert.Equal(t, api.VolumeSpec{Name: "vol2", Type: api.VolumeTypeVolume}, specs["vol2"]) +} + +func TestVolumeScheduler_getVolumeLocations(t *testing.T) { + machines := []*Machine{ + { + Info: &pb.MachineInfo{ + Id: "machine1", + Name: "machine1", + }, + Volumes: []volume.Volume{ + { + Name: "vol1", + }, + }, + }, + { + Info: &pb.MachineInfo{ + Id: "machine2", + Name: "machine2", + }, + Volumes: []volume.Volume{ + { + Name: "vol2", + }, + }, + }, + } + + allVolumes := map[string]struct{}{ + "vol1": {}, + "vol2": {}, + "vol3": {}, + } + + scheduler, err := NewVolumeSchedulerWithMachines(machines, nil) + assert.NoError(t, err) + locations := scheduler.getVolumeLocations(allVolumes) + + assert.Len(t, locations, 3) + assert.Equal(t, []string{"machine1"}, locations["vol1"]) + assert.Equal(t, []string{"machine2"}, locations["vol2"]) + assert.Empty(t, locations["vol3"]) +} + +func TestVolumeScheduler_getVolumeServices(t *testing.T) { + serviceSpecs := map[string]api.ServiceSpec{ + "service1": { + Name: "service1", + Container: api.ContainerSpec{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + }, + }, + "service2": { + Name: "service2", + Container: api.ContainerSpec{ + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "vol1", + ContainerPath: "/data", + }, + { + VolumeName: "vol2", + ContainerPath: "/data2", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: "vol1", + Type: api.VolumeTypeVolume, + }, + { + Name: "vol2", + Type: api.VolumeTypeVolume, + }, + }, + }, + } + + allVolumes := map[string]struct{}{ + "vol1": {}, + "vol2": {}, + } + + // Convert map to slice + specsList := make([]api.ServiceSpec, 0, len(serviceSpecs)) + for _, spec := range serviceSpecs { + specsList = append(specsList, spec) + } + scheduler, err := NewVolumeSchedulerWithMachines(nil, specsList) + assert.NoError(t, err) + services := scheduler.getVolumeServices(allVolumes) + + assert.Len(t, services, 2) + assert.ElementsMatch(t, []string{"service1", "service2"}, services["vol1"]) + assert.ElementsMatch(t, []string{"service2"}, services["vol2"]) +} + +func TestVolumeScheduler_intersectMachines(t *testing.T) { + scheduler := &VolumeScheduler{} + + a := []string{"machine1", "machine2", "machine3"} + b := []string{"machine2", "machine3", "machine4"} + + result := scheduler.intersectMachines(a, b) + assert.ElementsMatch(t, []string{"machine2", "machine3"}, result) + + // Empty intersection + c := []string{"machine5", "machine6"} + result = scheduler.intersectMachines(a, c) + assert.Empty(t, result) + + // One empty slice + result = scheduler.intersectMachines(a, []string{}) + assert.Empty(t, result) + result = scheduler.intersectMachines([]string{}, b) + assert.Empty(t, result) +} 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 From 8ee551ca8fef6bf56aa66f15370d3c4c1810c784 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Sun, 20 Apr 2025 11:15:07 +1000 Subject: [PATCH 2/3] cleanup --- pkg/client/deploy/scheduler/volume.go | 114 ------------- pkg/client/deploy/scheduler/volume_test.go | 184 --------------------- test/e2e/service_test.go | 59 +++++++ 3 files changed, 59 insertions(+), 298 deletions(-) diff --git a/pkg/client/deploy/scheduler/volume.go b/pkg/client/deploy/scheduler/volume.go index 0436bf84..e9305989 100644 --- a/pkg/client/deploy/scheduler/volume.go +++ b/pkg/client/deploy/scheduler/volume.go @@ -193,46 +193,6 @@ func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) { } return scheduledVolumes, nil - - //// For each missing volume (should be created only on one machine) - //machineToVolumeSpecs := make(map[string][]api.VolumeSpec) - //for volumeName, serviceNames := range missingVolumes { - // // Get the intersection of candidate machines for all services using this volume - // var intersection []string - // for i, serviceName := range serviceNames { - // if i == 0 { - // intersection = serviceEligibleMachines[serviceName] - // } else { - // intersection = s.intersectMachines(intersection, serviceEligibleMachines[serviceName]) - // } - // } - // - // if len(intersection) == 0 { - // return nil, fmt.Errorf("unable to find a machine where services %v can be placed together to share the missing volume %s", - // serviceNames, volumeName) - // } - // - // // Sort the intersection to ensure deterministic behavior - // sortedIntersection := make([]string, len(intersection)) - // copy(sortedIntersection, intersection) - // slices.Sort(sortedIntersection) - // - // // Choose the first machine in the sorted intersection to create the volume on - // machineID := sortedIntersection[0] - // - // // Get the volume spec for this volume name - // volumeSpec := volumeSpecs[volumeName] - // - // // Add the volume spec to the machine's list - // machineToVolumeSpecs[machineID] = append(machineToVolumeSpecs[machineID], volumeSpec) - // - // // Update the candidate machines for all services using this volume - // for _, serviceName := range serviceNames { - // serviceEligibleMachines[serviceName] = intersection - // } - //} - // - //return machineToVolumeSpecs, nil } // serviceEligibleMachinesWithoutVolumes returns a set of machine IDs where the service can be scheduled @@ -254,77 +214,3 @@ func (s *VolumeScheduler) serviceEligibleMachinesWithoutVolumes(spec api.Service return machineIDs, nil } - -// getAllVolumesAndSpecs returns a map of all volume names used by services and a map of volume names to their specs. -func (s *VolumeScheduler) getAllVolumesAndSpecs() (map[string]struct{}, map[string]api.VolumeSpec) { - volumes := make(map[string]struct{}) - volumeSpecs := make(map[string]api.VolumeSpec) - - for _, serviceSpec := range s.serviceSpecs { - for _, mount := range serviceSpec.Container.VolumeMounts { - if v, ok := serviceSpec.Volume(mount.VolumeName); ok && v.Type == api.VolumeTypeVolume { - volumeName := v.DockerVolumeName() - volumes[volumeName] = struct{}{} - volumeSpecs[volumeName] = v - } - } - } - - return volumes, volumeSpecs -} - -// getVolumeLocations returns a map of volume names to the list of machine IDs where they exist. -func (s *VolumeScheduler) getVolumeLocations(allVolumes map[string]struct{}) map[string][]string { - volumeLocations := make(map[string][]string) - - // Initialize the map with empty slices for all volumes - for volumeName := range allVolumes { - volumeLocations[volumeName] = []string{} - } - - // Populate the map with machine IDs where each volume exists - for _, machine := range s.machines { - for _, vol := range machine.Volumes { - if _, ok := volumeLocations[vol.Name]; ok { - volumeLocations[vol.Name] = append(volumeLocations[vol.Name], machine.Info.Id) - } - } - } - - return volumeLocations -} - -// getVolumeServices returns a map of volume names to the list of service names that use them. -func (s *VolumeScheduler) getVolumeServices(allVolumes map[string]struct{}) map[string][]string { - volumeServices := make(map[string][]string) - - // Initialize the map with empty slices for all volumes - for volumeName := range allVolumes { - volumeServices[volumeName] = []string{} - } - - for _, serviceSpec := range s.serviceSpecs { - serviceName := serviceSpec.Name - for _, mount := range serviceSpec.Container.VolumeMounts { - if v, ok := serviceSpec.Volume(mount.VolumeName); ok && v.Type == api.VolumeTypeVolume { - volumeName := v.DockerVolumeName() - if _, ok := allVolumes[volumeName]; ok { - volumeServices[volumeName] = append(volumeServices[volumeName], serviceName) - } - } - } - } - - return volumeServices -} - -// intersectMachines returns the intersection of two slices of machine IDs. -func (s *VolumeScheduler) intersectMachines(a, b []string) []string { - var result []string - for _, id := range a { - if slices.Contains(b, id) { - result = append(result, id) - } - } - return result -} diff --git a/pkg/client/deploy/scheduler/volume_test.go b/pkg/client/deploy/scheduler/volume_test.go index 0e239c5e..df970898 100644 --- a/pkg/client/deploy/scheduler/volume_test.go +++ b/pkg/client/deploy/scheduler/volume_test.go @@ -782,187 +782,3 @@ func TestVolumeScheduler_Schedule(t *testing.T) { }) } } - -func TestVolumeScheduler_getAllVolumesAndSpecs(t *testing.T) { - serviceSpecs := map[string]api.ServiceSpec{ - "service1": { - Name: "service1", - Container: api.ContainerSpec{ - VolumeMounts: []api.VolumeMount{ - { - VolumeName: "vol1", - ContainerPath: "/data", - }, - }, - }, - Volumes: []api.VolumeSpec{ - { - Name: "vol1", - Type: api.VolumeTypeVolume, - }, - }, - }, - "service2": { - Name: "service2", - Container: api.ContainerSpec{ - VolumeMounts: []api.VolumeMount{ - { - VolumeName: "vol2", - ContainerPath: "/data", - }, - }, - }, - Volumes: []api.VolumeSpec{ - { - Name: "vol2", - Type: api.VolumeTypeVolume, - }, - }, - }, - } - - // Convert map to slice - specsList := make([]api.ServiceSpec, 0, len(serviceSpecs)) - for _, spec := range serviceSpecs { - specsList = append(specsList, spec) - } - scheduler, err := NewVolumeSchedulerWithMachines(nil, specsList) - assert.NoError(t, err) - volumes, specs := scheduler.getAllVolumesAndSpecs() - - assert.Len(t, volumes, 2) - assert.Contains(t, volumes, "vol1") - assert.Contains(t, volumes, "vol2") - - assert.Len(t, specs, 2) - assert.Equal(t, api.VolumeSpec{Name: "vol1", Type: api.VolumeTypeVolume}, specs["vol1"]) - assert.Equal(t, api.VolumeSpec{Name: "vol2", Type: api.VolumeTypeVolume}, specs["vol2"]) -} - -func TestVolumeScheduler_getVolumeLocations(t *testing.T) { - machines := []*Machine{ - { - Info: &pb.MachineInfo{ - Id: "machine1", - Name: "machine1", - }, - Volumes: []volume.Volume{ - { - Name: "vol1", - }, - }, - }, - { - Info: &pb.MachineInfo{ - Id: "machine2", - Name: "machine2", - }, - Volumes: []volume.Volume{ - { - Name: "vol2", - }, - }, - }, - } - - allVolumes := map[string]struct{}{ - "vol1": {}, - "vol2": {}, - "vol3": {}, - } - - scheduler, err := NewVolumeSchedulerWithMachines(machines, nil) - assert.NoError(t, err) - locations := scheduler.getVolumeLocations(allVolumes) - - assert.Len(t, locations, 3) - assert.Equal(t, []string{"machine1"}, locations["vol1"]) - assert.Equal(t, []string{"machine2"}, locations["vol2"]) - assert.Empty(t, locations["vol3"]) -} - -func TestVolumeScheduler_getVolumeServices(t *testing.T) { - serviceSpecs := map[string]api.ServiceSpec{ - "service1": { - Name: "service1", - Container: api.ContainerSpec{ - VolumeMounts: []api.VolumeMount{ - { - VolumeName: "vol1", - ContainerPath: "/data", - }, - }, - }, - Volumes: []api.VolumeSpec{ - { - Name: "vol1", - Type: api.VolumeTypeVolume, - }, - }, - }, - "service2": { - Name: "service2", - Container: api.ContainerSpec{ - VolumeMounts: []api.VolumeMount{ - { - VolumeName: "vol1", - ContainerPath: "/data", - }, - { - VolumeName: "vol2", - ContainerPath: "/data2", - }, - }, - }, - Volumes: []api.VolumeSpec{ - { - Name: "vol1", - Type: api.VolumeTypeVolume, - }, - { - Name: "vol2", - Type: api.VolumeTypeVolume, - }, - }, - }, - } - - allVolumes := map[string]struct{}{ - "vol1": {}, - "vol2": {}, - } - - // Convert map to slice - specsList := make([]api.ServiceSpec, 0, len(serviceSpecs)) - for _, spec := range serviceSpecs { - specsList = append(specsList, spec) - } - scheduler, err := NewVolumeSchedulerWithMachines(nil, specsList) - assert.NoError(t, err) - services := scheduler.getVolumeServices(allVolumes) - - assert.Len(t, services, 2) - assert.ElementsMatch(t, []string{"service1", "service2"}, services["vol1"]) - assert.ElementsMatch(t, []string{"service2"}, services["vol2"]) -} - -func TestVolumeScheduler_intersectMachines(t *testing.T) { - scheduler := &VolumeScheduler{} - - a := []string{"machine1", "machine2", "machine3"} - b := []string{"machine2", "machine3", "machine4"} - - result := scheduler.intersectMachines(a, b) - assert.ElementsMatch(t, []string{"machine2", "machine3"}, result) - - // Empty intersection - c := []string{"machine5", "machine6"} - result = scheduler.intersectMachines(a, c) - assert.Empty(t, result) - - // One empty slice - result = scheduler.intersectMachines(a, []string{}) - assert.Empty(t, result) - result = scheduler.intersectMachines([]string{}, b) - assert.Empty(t, result) -} diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 2fe0fb1d..a872f311 100644 --- a/test/e2e/service_test.go +++ b/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, + Mode: api.ServiceModeReplicated, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: volumeName, + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: volumeName, + Type: api.VolumeTypeVolume, + }, + }, + } + 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() From 343357adf9fa5661d4dcf063a5b79f0f5e7f839c Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Sun, 20 Apr 2025 18:53:18 +1000 Subject: [PATCH 3/3] find optimal solution in VolumeScheduler --- pkg/client/deploy/scheduler/volume.go | 174 +++++++++++++++------ pkg/client/deploy/scheduler/volume_test.go | 44 +++++- 2 files changed, 165 insertions(+), 53 deletions(-) diff --git a/pkg/client/deploy/scheduler/volume.go b/pkg/client/deploy/scheduler/volume.go index e9305989..ab6daefd 100644 --- a/pkg/client/deploy/scheduler/volume.go +++ b/pkg/client/deploy/scheduler/volume.go @@ -11,7 +11,12 @@ import ( ) // VolumeScheduler determines what missing volumes should be created and where for a multi-service deployment. -// TODO: add rules that if a volume exists, it should be used instead of creating a new one. +// 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 @@ -19,6 +24,8 @@ type VolumeScheduler struct { 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] @@ -37,16 +44,21 @@ func NewVolumeSchedulerWithClient(ctx context.Context, cli Client, specs []api.S // NewVolumeSchedulerWithMachines creates a new VolumeScheduler with the given cluster machines // and service specifications. func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec) (*VolumeScheduler, error) { - // TODO: validate specs before scheduling, need to update tests to use helper functions to create specs with images. 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. - volumeMachines := make(map[string]mapset.Set[string]) + 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) } @@ -59,16 +71,19 @@ func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec specsWithVolumes = append(specsWithVolumes, spec) for _, v := range mountedVolumes { - if seenVolume, ok := volumeSpecs[v.DockerVolumeName()]; ok { + 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.DockerVolumeName()) + return nil, fmt.Errorf("volume '%s' is defined multiple times with different options", v.Name) } } else { - v = v.SetDefaults() - v.Name = v.DockerVolumeName() // Reset any aliases in a service spec to the actual Docker volume name. volumeSpecs[v.Name] = v } + + volumeServices[v.Name] = append(volumeServices[v.Name], spec.Name) } } @@ -82,10 +97,10 @@ func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec "on machine '%s'", vol.Name, machine.Info.Name) } - if _, setInitialised := volumeMachines[vol.Name]; !setInitialised { - volumeMachines[vol.Name] = mapset.NewSet[string]() + if _, setInitialised := existingVolumeMachines[vol.Name]; !setInitialised { + existingVolumeMachines[vol.Name] = mapset.NewSet[string]() } - volumeMachines[vol.Name].Add(machine.Info.Id) + existingVolumeMachines[vol.Name].Add(machine.Info.Id) } } } @@ -94,7 +109,8 @@ func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec machines: machines, serviceSpecs: specsWithVolumes, volumeSpecs: volumeSpecs, - existingVolumeMachines: volumeMachines, + volumeServices: volumeServices, + existingVolumeMachines: existingVolumeMachines, }, nil } @@ -110,7 +126,6 @@ func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) { // 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. - volumeServices := make(map[string][]string) // Get eligible machines for each service without considering its volume mounts. for _, spec := range s.serviceSpecs { machineIDs, err := s.serviceEligibleMachinesWithoutVolumes(spec) @@ -118,12 +133,6 @@ func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) { return nil, err } serviceEligibleMachines[spec.Name] = machineIDs - - // Populate volumeServices with Docker volumes used by this service. - for _, v := range spec.MountedDockerVolumes() { - volumeName := v.DockerVolumeName() - volumeServices[volumeName] = append(volumeServices[volumeName], spec.Name) - } } // For each volume that exists on any machine(s) (which shouldn't be created), intersect each service's @@ -131,7 +140,7 @@ func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) { // 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 volumeServices[volumeName] { + for _, serviceName := range s.volumeServices[volumeName] { quotedServiceVolumes[serviceName] = append(quotedServiceVolumes[serviceName], fmt.Sprintf("'%s'", volumeName)) newEligibleMachines := serviceEligibleMachines[serviceName].Intersect(volumeMachines) @@ -144,51 +153,52 @@ func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) { } } - for serviceName, eligibleMachines := range serviceEligibleMachines { - fmt.Printf("### Service '%s' can be scheduled on machines: %v\n", serviceName, eligibleMachines.ToSlice()) + // 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{}{} } - // For each missing volume, intersect the eligible machines for all services using the volume - // and choose the first machine in the sorted intersection to create the volume on. + 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 { - // This volume already exists, no need to create it. continue } - var eligibleMachines mapset.Set[string] - var quotedServiceNames []string // Used to format the error message. - for i, serviceName := range volumeServices[missingVolumeName] { - if i == 0 { - eligibleMachines = serviceEligibleMachines[serviceName] - } else { - eligibleMachines = serviceEligibleMachines[serviceName].Intersect(eligibleMachines) - } - quotedServiceNames = append(quotedServiceNames, fmt.Sprintf("'%s'", serviceName)) - } - - if eligibleMachines == nil { + 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("unable to find a machine that satisfies placement constraints "+ - "for services %s that must be placed together to share volume '%s'", - strings.Join(quotedServiceNames, ", "), missingVolumeName) + return nil, fmt.Errorf("bug detected: no eligible machines for volume '%s'", missingVolumeName) } - // Choose the first machine in the sorted eligible machines to create the volume on. + // Choose the first machine in the sorted eligible machines to schedule the volume on. // Sort the eligible machines to ensure deterministic behavior. - // TODO: the first machine might not be the optimal one. Ideally, we need to do the intersection for all volumes - // multiple times until they converge. Then picking any machine is fine. 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) - // Update the eligible machines to the chosen machine for all services using this volume. - eligibleMachines = mapset.NewSet(machineID) - for _, serviceName := range volumeServices[missingVolumeName] { - serviceEligibleMachines[serviceName] = eligibleMachines + + // 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) } } @@ -214,3 +224,73 @@ func (s *VolumeScheduler) serviceEligibleMachinesWithoutVolumes(spec api.Service 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 index df970898..171f0eed 100644 --- a/pkg/client/deploy/scheduler/volume_test.go +++ b/pkg/client/deploy/scheduler/volume_test.go @@ -3,10 +3,12 @@ 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) { @@ -30,6 +32,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service1", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -74,6 +77,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service1", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -91,6 +95,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service2", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -138,6 +143,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service1", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -178,6 +184,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Machines: []string{"machine2"}, }, Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -225,6 +232,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Machines: []string{"machine1"}, }, Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -245,6 +253,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Machines: []string{"machine2"}, }, Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -290,6 +299,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Machines: []string{"machine2"}, }, Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -328,6 +338,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service1", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -353,6 +364,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service2", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol2", @@ -378,6 +390,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service3", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol3", @@ -458,6 +471,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service1", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol3", @@ -483,6 +497,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service2", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -516,13 +531,14 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service3", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol2", ContainerPath: "/data2", }, { - VolumeName: "vol4", + VolumeName: "vol4-alias", ContainerPath: "/data4", }, }, @@ -533,18 +549,24 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Type: api.VolumeTypeVolume, }, { - // TODO: use vol4-alias name and Docker name in VolumeOptions - Name: "vol4", + 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", + VolumeName: "vol2-alias", ContainerPath: "/data2", }, { @@ -555,8 +577,14 @@ func TestVolumeScheduler_Schedule(t *testing.T) { }, Volumes: []api.VolumeSpec{ { - Name: "vol2", + Name: "vol2-alias", Type: api.VolumeTypeVolume, + VolumeOptions: &api.VolumeOptions{ + Name: "vol2", + Driver: &mount.Driver{ + Name: api.VolumeDriverLocal, + }, + }, }, { Name: "vol5", @@ -607,6 +635,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Machines: []string{"machine1"}, }, Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -635,6 +664,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { Machines: []string{"machine2"}, }, Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol2", @@ -689,6 +719,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service1", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol1", @@ -714,6 +745,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { { Name: "service2", Container: api.ContainerSpec{ + Image: "portainer/pause:latest", VolumeMounts: []api.VolumeMount{ { VolumeName: "vol3", @@ -757,7 +789,7 @@ func TestVolumeScheduler_Schedule(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { scheduler, err := NewVolumeSchedulerWithMachines(tt.machines, tt.serviceSpecs) - assert.NoError(t, err) + require.NoError(t, err) result, err := scheduler.Schedule() if tt.wantErr != "" {