chore: refactor schedulers to work with cluster state snapshot, VolumeScheduler updates state with scheduled volumes

pkg/client/deploy/scheduler/constraint.go

@@ -1,9 +1,11 @@
 package scheduler
 
 import (
+	"reflect"
 	"slices"
 	"strings"
 
+	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/api/types/volume"
 	"github.com/psviderski/uncloud/pkg/api"
 )
@@ -73,7 +75,7 @@ type VolumesConstraint struct {
 }
 
 // Evaluate determines if a machine has all the required volumes.
-// Returns true if all required volumes exist on the machine or if there are no required volumes.
+// Returns true if all required volumes exist or scheduled on the machine or if there are no required volumes.
 func (c *VolumesConstraint) Evaluate(machine *Machine) bool {
 	if len(c.Volumes) == 0 {
 		return true
@@ -84,9 +86,35 @@ func (c *VolumesConstraint) Evaluate(machine *Machine) bool {
 			continue
 		}
 
-		// TODO: should we check the volume driver to be local or any matched volume by name is ok?
-		if !slices.ContainsFunc(machine.Volumes, func(vol volume.Volume) bool {
-			return vol.Name == v.DockerVolumeName()
+		// Check if the required volume already exists on the machine.
+		if slices.ContainsFunc(machine.Volumes, func(vol volume.Volume) bool {
+			if v.DockerVolumeName() == vol.Name {
+				return v.MatchesDockerVolume(vol)
+			}
+			return false
+		}) {
+			continue
+		}
+
+		// Check if the required volume has been scheduled on the machine. The driver names and options must match.
+		if !slices.ContainsFunc(machine.ScheduledVolumes, func(scheduled api.VolumeSpec) bool {
+			if v.DockerVolumeName() != scheduled.DockerVolumeName() {
+				return false
+			}
+
+			// The volume spec with an empty driver can mount the volume that matches the name no matter the driver.
+			if v.VolumeOptions.Driver == nil {
+				return true
+			}
+
+			// If the driver is specified in the spec, the spec's driver and options must match the volume's driver
+			// and options to successfully mount the volume.
+			scheduled = scheduled.SetDefaults()
+			scheduledDriver := scheduled.VolumeOptions.Driver
+			if scheduledDriver == nil {
+				scheduledDriver = &mount.Driver{Name: api.VolumeDriverLocal}
+			}
+			return reflect.DeepEqual(v.VolumeOptions.Driver, scheduledDriver)
 		}) {
 			return false
 		}

pkg/client/deploy/scheduler/service.go

@@ -1,35 +1,24 @@
 package scheduler
 
 import (
-	"context"
 	"errors"
-	"fmt"
 
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/pkg/api"
 )
 
 type ServiceScheduler struct {
-	machines    []*Machine
+	state       *ClusterState
 	spec        api.ServiceSpec
 	constraints []Constraint
 }
 
-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 {
+// NewServiceScheduler creates a new ServiceScheduler with the given cluster state and service specification.
+func NewServiceScheduler(state *ClusterState, spec api.ServiceSpec) *ServiceScheduler {
 	constraints := constraintsFromSpec(spec)
 
 	return &ServiceScheduler{
-		machines:    machines,
+		state:       state,
 		spec:        spec,
 		constraints: constraints,
 	}
@@ -38,7 +27,7 @@ func NewServiceSchedulerWithMachines(machines []*Machine, spec api.ServiceSpec)
 // 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.state.Machines {
 		if s.evaluateConstraints(machine) {
 			available = append(available, machine)
 		}

pkg/client/deploy/scheduler/state.go

@@ -9,19 +9,26 @@ import (
 	"github.com/psviderski/uncloud/pkg/api"
 )
 
+// ClusterState represents the current and planned state of machines and their resources in the cluster.
+type ClusterState struct {
+	Machines []*Machine
+}
+
+type Machine struct {
+	Info             *pb.MachineInfo
+	Volumes          []volume.Volume
+	ScheduledVolumes []api.VolumeSpec
+}
+
 type Client interface {
 	api.MachineClient
 	api.VolumeClient
 }
 
-type Machine struct {
-	Info    *pb.MachineInfo
-	Volumes []volume.Volume
-}
-
-// InspectMachines retrieves the list of available machines and their details required for scheduling purposes.
-// TODO: refactor to get all the details in one broadcast call to machine API.
-func InspectMachines(ctx context.Context, cli Client) ([]*Machine, error) {
+// InspectClusterState creates a new cluster state by inspecting the machines using the cluster client.
+func InspectClusterState(ctx context.Context, cli Client) (*ClusterState, error) {
+	// TODO: refactor to get all the details in one broadcast call to machine API,
+	//  e.g. InspectMachine with include options.
 	machineMembers, err := cli.ListMachines(ctx, &api.MachineFilter{Available: true})
 	if err != nil {
 		return nil, fmt.Errorf("list machines: %w", err)
@@ -46,5 +53,7 @@ func InspectMachines(ctx context.Context, cli Client) ([]*Machine, error) {
 		machines = append(machines, machine)
 	}
 
-	return machines, nil
+	return &ClusterState{
+		Machines: machines,
+	}, nil
 }

pkg/client/deploy/scheduler/volume.go

@@ -1,7 +1,6 @@
 package scheduler
 
 import (
-	"context"
 	"fmt"
 	"slices"
 	"strings"
@@ -18,8 +17,8 @@ import (
 //   - 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
+	// state is the current state of machines and their resources in the cluster.
+	state *ClusterState
 	// 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.
@@ -32,19 +31,8 @@ type VolumeScheduler struct {
 	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) {
+// NewVolumeScheduler creates a new VolumeScheduler with the given cluster state and service specifications.
+func NewVolumeScheduler(state *ClusterState, specs []api.ServiceSpec) (*VolumeScheduler, error) {
 	var specsWithVolumes []api.ServiceSpec
 	// Docker volume name -> VolumeSpec.
 	volumeSpecs := make(map[string]api.VolumeSpec)
@@ -90,7 +78,7 @@ func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec
 
 	// 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 _, machine := range state.Machines {
 		for _, vol := range machine.Volumes {
 			if spec, ok := volumeSpecs[vol.Name]; ok {
 				if !spec.MatchesDockerVolume(vol) {
@@ -107,7 +95,7 @@ func NewVolumeSchedulerWithMachines(machines []*Machine, specs []api.ServiceSpec
 	}
 
 	return &VolumeScheduler{
-		machines:               machines,
+		state:                  state,
 		serviceSpecs:           specsWithVolumes,
 		volumeSpecs:            volumeSpecs,
 		volumeServices:         volumeServices,
@@ -204,6 +192,16 @@ func (s *VolumeScheduler) Schedule() (map[string][]api.VolumeSpec, error) {
 		}
 	}
 
+	// Update the state of the machines with the scheduled volumes.
+	for machineID, volumes := range scheduledVolumes {
+		for _, m := range s.state.Machines {
+			if m.Info.Id == machineID {
+				m.ScheduledVolumes = append(m.ScheduledVolumes, volumes...)
+				break
+			}
+		}
+	}
+
 	return scheduledVolumes, nil
 }
 
@@ -213,7 +211,7 @@ func (s *VolumeScheduler) serviceEligibleMachinesWithoutVolumes(spec api.Service
 	specWithoutVolumes := spec.Clone()
 	specWithoutVolumes.Container.VolumeMounts = nil
 
-	scheduler := NewServiceSchedulerWithMachines(s.machines, specWithoutVolumes)
+	scheduler := NewServiceScheduler(s.state, specWithoutVolumes)
 	machines, err := scheduler.EligibleMachines()
 	if err != nil {
 		return nil, fmt.Errorf("schedule service '%s': %w", spec.Name, err)

pkg/client/deploy/scheduler/volume_test.go

@@ -806,7 +806,10 @@ 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)
+			state := &ClusterState{
+				Machines: tt.machines,
+			}
+			scheduler, err := NewVolumeScheduler(state, tt.serviceSpecs)
 			require.NoError(t, err)
 			result, err := scheduler.Schedule()