From 948201cb6fe419e3ee3d67a58f66d2e3e7aef184 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Sun, 20 Apr 2025 11:13:30 +1000 Subject: [PATCH] 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