diff --git a/cmd/uncloud/service/run.go b/cmd/uncloud/service/run.go index 0afbcd65..7bab51da 100644 --- a/cmd/uncloud/service/run.go +++ b/cmd/uncloud/service/run.go @@ -4,11 +4,9 @@ import ( "context" "fmt" "os" - "slices" "strings" "github.com/docker/compose/v2/pkg/progress" - "github.com/docker/docker/api/types/volume" "github.com/docker/docker/daemon/names" "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/secret" @@ -113,26 +111,8 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { } defer clusterClient.Close() - machineIDForVolumes, missingVolumes, err := selectMachineForVolumes( - ctx, - clusterClient, - spec.Volumes, - spec.Placement.Machines, - ) - if err != nil { - return err - } - var resp client.RunServiceResponse err = progress.RunWithTitle(ctx, func(ctx context.Context) error { - // Create missing volumes on the selected machine. - for _, v := range missingVolumes { - _, err = clusterClient.CreateVolume(ctx, machineIDForVolumes, volume.CreateOptions{Name: v.Name}) - if err != nil { - return fmt.Errorf("create volume '%s': %w", v.Name, err) - } - } - resp, err = clusterClient.RunService(ctx, spec) if err != nil { return fmt.Errorf("run service: %w", err) @@ -358,76 +338,3 @@ func parseVolumeFlagValue(volume string) (api.VolumeSpec, api.VolumeMount, error return spec, mount, nil } - -// selectMachineForVolumes selects a machine to run a service with the given volumes on and determines which volumes -// need to be created on the selected machine. An empty machineID is returned if no named volumes are specified. -func selectMachineForVolumes( - ctx context.Context, clusterClient *client.Client, volumes []api.VolumeSpec, machinesFilter []string, -) (machineID string, missingVolumes []api.VolumeSpec, err error) { - var volumeNames []string - for _, v := range volumes { - if v.Type == api.VolumeTypeVolume { - volumeNames = append(volumeNames, v.Name) - } - } - if len(volumeNames) == 0 { - return "", nil, nil - } - - vfilter := &api.VolumeFilter{ - Machines: machinesFilter, - Names: volumeNames, - } - vols, err := clusterClient.ListVolumes(ctx, vfilter) - if err != nil { - return "", nil, fmt.Errorf("list volumes: %w", err) - } - - if len(vols) > 0 { - // Some volumes have been found on the machines matching the machinesFilter. - // Pick the machine with the most volumes to create fewer duplicate volumes. - volumesCountOnMachines := make(map[string]int) - for _, vol := range vols { - volumesCountOnMachines[vol.MachineID]++ - } - - maxCount := 0 - for mid, count := range volumesCountOnMachines { - if count > maxCount { - machineID = mid - maxCount = count - } - } - } else { - // No volumes found on the machines matching the machinesFilter. - // Pick the first available machine to create the volumes on. - mfilter := &api.MachineFilter{ - Available: true, - NamesOrIDs: machinesFilter, - } - availableMachines, err := clusterClient.ListMachines(ctx, mfilter) - if err != nil { - return "", nil, fmt.Errorf("list machines: %w", err) - } - - if len(availableMachines) == 0 { - return "", nil, fmt.Errorf("no available machines to create the volume(s) on") - } - machineID = availableMachines[0].Machine.Id - } - - // Find missing volumes that need to be created on the selected machine. - for _, v := range volumes { - if v.Type != api.VolumeTypeVolume { - continue - } - - if !slices.ContainsFunc(vols, func(mv api.MachineVolume) bool { - return mv.Volume.Name == v.Name && mv.MachineID == machineID - }) { - missingVolumes = append(missingVolumes, v) - } - } - - return machineID, missingVolumes, nil -} diff --git a/pkg/client/deploy/scheduler/volume.go b/pkg/client/deploy/scheduler/volume.go index ab6daefd..33f3678e 100644 --- a/pkg/client/deploy/scheduler/volume.go +++ b/pkg/client/deploy/scheduler/volume.go @@ -11,7 +11,7 @@ import ( ) // VolumeScheduler determines what missing volumes should be created and where for a multi-service deployment. -// It must satisfy the following constraints: +// It satisfies 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. @@ -25,6 +25,7 @@ type VolumeScheduler struct { // 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. + // TODO: not all service spec may contain the service name, so use a slice of int indexes instead of names. 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. diff --git a/pkg/client/service.go b/pkg/client/service.go index ac0706d8..f3d611ee 100644 --- a/pkg/client/service.go +++ b/pkg/client/service.go @@ -9,8 +9,10 @@ import ( "sync" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/volume" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" + "github.com/psviderski/uncloud/pkg/client/deploy/scheduler" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" @@ -39,6 +41,33 @@ func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunSer } } + // Create missing named Docker volumes for the service. + if len(spec.MountedDockerVolumes()) > 0 { + volumeScheduler, err := scheduler.NewVolumeSchedulerWithClient(ctx, cli, []api.ServiceSpec{spec}) + if err != nil { + return resp, fmt.Errorf("init volume scheduler: %w", err) + } + + scheduledVolumes, err := volumeScheduler.Schedule() + if err != nil { + return resp, fmt.Errorf("schedule volumes: %w", err) + } + + // Create the missing volumes on the scheduled machines. + for machineID, volumes := range scheduledVolumes { + for _, v := range volumes { + _, err = cli.CreateVolume(ctx, machineID, volume.CreateOptions{ + Name: v.Name, + Driver: v.VolumeOptions.Driver.Name, + DriverOpts: v.VolumeOptions.Driver.Options, + }) + if err != nil { + return resp, fmt.Errorf("create volume '%s': %w", v.Name, err) + } + } + } + } + deployment := cli.NewDeployment(spec, nil) plan, err := deployment.Run(ctx) if err != nil { diff --git a/test/e2e/assert.go b/test/e2e/assert.go index f22757b0..5f8ac2c5 100644 --- a/test/e2e/assert.go +++ b/test/e2e/assert.go @@ -19,10 +19,11 @@ import ( ) func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpec) { + spec = spec.SetDefaults() assert.Equal(t, spec.Name, svc.Name) if svc.Mode == api.ServiceModeReplicated { - assert.Contains(t, []string{"", api.ServiceModeReplicated}, spec.Mode) + assert.Equal(t, api.ServiceModeReplicated, spec.Mode) assert.Len(t, svc.Containers, int(spec.Replicas), "Expected %d replicas", spec.Replicas) } else { assert.Equal(t, spec.Mode, svc.Mode) @@ -34,10 +35,10 @@ func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpe } func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api.ServiceSpec) { + spec = spec.SetDefaults() status := deploy.EvalContainerSpecChange(ctr.ServiceSpec, spec) assert.Equal(t, deploy.ContainerUpToDate, status) - spec = spec.SetDefaults() // Verify labels. assert.True(t, api.ValidateServiceID(ctr.Config.Labels[api.LabelServiceID])) assert.Equal(t, spec.Name, ctr.Config.Labels[api.LabelServiceName]) diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 2fe0fb1d..0c45fd45 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, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + VolumeMounts: []api.VolumeMount{ + { + VolumeName: volumeName, + ContainerPath: "/data", + }, + }, + }, + Volumes: []api.VolumeSpec{ + { + Name: volumeName, + Type: api.VolumeTypeVolume, + }, + }, + Replicas: 3, + } + 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()