feat(volumes): create and mount data volumes when running service with 'uc run' command

cmd/uncloud/service/run.go

@@ -7,9 +7,14 @@ import (
 	"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/machine/api/pb"
+	"github.com/psviderski/uncloud/internal/secret"
 	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/psviderski/uncloud/pkg/client"
 	"github.com/psviderski/uncloud/pkg/client/deploy"
 	"github.com/spf13/cobra"
 )
@@ -81,8 +86,13 @@ func NewRunCommand() *cobra.Command {
 	cmd.Flags().UintVar(&opts.replicas, "replicas", 1,
 		"Number of containers to run for the service. Only valid for a replicated service.")
 	cmd.Flags().StringSliceVarP(&opts.volumes, "volume", "v", nil,
-		"Bind mount a host file or directory into a service container using the format "+
-			"/host/path:/container/path[:ro]. Can be specified multiple times.")
+		"Mount a data volume or host path into service containers. Service containers will be scheduled on the machine(s) where\n"+
+			"the volume is located. Can be specified multiple times.\n"+
+			"Format: volume_name:/container/path[:ro|volume-nocopy] or /host/path:/container/path[:ro]\n"+
+			"Examples:\n"+
+			"  -v postgres-data:/var/lib/postgresql/data  Mount volume 'postgres-data' to /var/lib/postgresql/data in container\n"+
+			"  -v /data/uploads:/app/uploads         	 Bind mount /data/uploads host directory to /app/uploads in container\n"+
+			"  -v /host/path:/container/path:ro 		 Bind mount a host directory or file as read-only")
 
 	cmd.Flags().StringVarP(
 		&opts.cluster, "context", "c", "",
@@ -93,92 +103,59 @@ func NewRunCommand() *cobra.Command {
 }
 
 func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error {
-	env, err := parseEnv(opts.env)
+	spec, err := prepareServiceSpec(opts)
 	if err != nil {
 		return err
 	}
 
-	switch opts.mode {
-	case api.ServiceModeReplicated, api.ServiceModeGlobal:
-	default:
-		return fmt.Errorf("invalid replication mode: '%s'", opts.mode)
-	}
-
-	switch opts.pull {
-	case api.PullPolicyAlways, api.PullPolicyMissing, api.PullPolicyNever:
-	default:
-		return fmt.Errorf("invalid pull policy: '%s'", opts.pull)
-	}
-
-	var machineFilter deploy.MachineFilter
-	if len(opts.machines) > 0 {
-		var machines []string
-		for _, value := range opts.machines {
-			if value == "" {
-				continue
-			}
-
-			mlist := strings.Split(value, ",")
-			for _, m := range mlist {
-				if m = strings.TrimSpace(m); m != "" {
-					machines = append(machines, m)
-				}
-			}
-		}
-
-		if len(machines) > 0 {
-			machineFilter = func(m *pb.MachineInfo) bool {
-				return slices.Contains(machines, m.Name)
-			}
-		}
-	}
-
-	ports := make([]api.PortSpec, len(opts.publish))
-	for i, publishPort := range opts.publish {
-		port, err := api.ParsePortSpec(publishPort)
-		if err != nil {
-			return fmt.Errorf("invalid service port '%s': %w", publishPort, err)
-		}
-		ports[i] = port
-	}
-	// TODO: parse and validate opts.volumes to fail fast if invalid.
-
-	spec := api.ServiceSpec{
-		Container: api.ContainerSpec{
-			Command:    opts.command,
-			Env:        env,
-			Image:      opts.image,
-			PullPolicy: opts.pull,
-			Volumes:    opts.volumes,
-		},
-		Mode:     opts.mode,
-		Name:     opts.name,
-		Ports:    ports,
-		Replicas: opts.replicas,
-	}
-
-	// Overwrite the default ENTRYPOINT of the image or reset it if an empty string is passed.
-	if opts.entrypoint != "" {
-		spec.Container.Entrypoint = []string{opts.entrypoint}
-	} else if opts.entrypointChanged {
-		spec.Container.Entrypoint = []string{""}
-	}
-
-	if err := spec.Validate(); err != nil {
-		return fmt.Errorf("invalid service configuration: %w", err)
-	}
-
 	clusterClient, err := uncli.ConnectCluster(ctx, opts.cluster)
 	if err != nil {
 		return fmt.Errorf("connect to cluster: %w", err)
 	}
 	defer clusterClient.Close()
 
-	resp, err := clusterClient.RunService(ctx, spec, machineFilter)
+	var deployFilter deploy.MachineFilter
+	machines := cli.ExpandCommaSeparatedValues(opts.machines)
+	if len(machines) > 0 {
+		deployFilter = func(m *pb.MachineInfo) bool {
+			return slices.Contains(machines, m.Name)
+		}
+	}
+
+	machineIDForVolumes, missingVolumes, err := selectMachineForVolumes(ctx, clusterClient, spec.Volumes, machines)
+	if err != nil {
+		return err
+	}
+	// machineIDForVolumes is not empty if the spec includes named volumes.
+	if machineIDForVolumes != "" {
+		// The service must be deployed on the machine where the existing volumes are located and the missing ones
+		// will be created.
+		deployFilter = func(m *pb.MachineInfo) bool {
+			return m.Id == machineIDForVolumes
+		}
+	}
+
+	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, deployFilter)
 		if err != nil {
 			return fmt.Errorf("run service: %w", err)
 		}
 
+		return nil
+	}, uncli.ProgressOut(), fmt.Sprintf("Running service %s (%s mode)", spec.Name, spec.Mode))
+	if err != nil {
+		return err
+	}
+
 	svc, err := clusterClient.InspectService(ctx, resp.ID)
 	if err != nil {
 		return fmt.Errorf("inspect service: %w", err)
@@ -196,6 +173,77 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error {
 	return nil
 }
 
+func prepareServiceSpec(opts runOptions) (api.ServiceSpec, error) {
+	var spec api.ServiceSpec
+
+	env, err := parseEnv(opts.env)
+	if err != nil {
+		return spec, err
+	}
+
+	switch opts.mode {
+	case api.ServiceModeReplicated, api.ServiceModeGlobal:
+	default:
+		return spec, fmt.Errorf("invalid replication mode: '%s'", opts.mode)
+	}
+
+	switch opts.pull {
+	case api.PullPolicyAlways, api.PullPolicyMissing, api.PullPolicyNever:
+	default:
+		return spec, fmt.Errorf("invalid pull policy: '%s'", opts.pull)
+	}
+
+	ports := make([]api.PortSpec, len(opts.publish))
+	for i, publishPort := range opts.publish {
+		port, err := api.ParsePortSpec(publishPort)
+		if err != nil {
+			return spec, fmt.Errorf("invalid service port '%s': %w", publishPort, err)
+		}
+		ports[i] = port
+	}
+
+	volumes, mounts, err := parseVolumeFlags(opts.volumes)
+	if err != nil {
+		return spec, err
+	}
+
+	spec = api.ServiceSpec{
+		Container: api.ContainerSpec{
+			Command:      opts.command,
+			Env:          env,
+			Image:        opts.image,
+			PullPolicy:   opts.pull,
+			VolumeMounts: mounts,
+		},
+		Mode:     opts.mode,
+		Name:     opts.name,
+		Ports:    ports,
+		Replicas: opts.replicas,
+		Volumes:  volumes,
+	}
+
+	// Overwrite the default ENTRYPOINT of the image or reset it if an empty string is passed.
+	if opts.entrypoint != "" {
+		spec.Container.Entrypoint = []string{opts.entrypoint}
+	} else if opts.entrypointChanged {
+		spec.Container.Entrypoint = []string{""}
+	}
+
+	if err = spec.Validate(); err != nil {
+		return spec, fmt.Errorf("invalid service configuration: %w", err)
+	}
+
+	// Generate a service name if not specified to be able to include it in the progress title.
+	if spec.Name == "" {
+		spec.Name, err = deploy.GenerateServiceName(spec.Container.Image)
+		if err != nil {
+			return spec, fmt.Errorf("generate service name: %w", err)
+		}
+	}
+
+	return spec, err
+}
+
 // parseEnv parses the environment variables from the command line arguments.
 // It supports two formats: "VAR=value" or just "VAR" to use the value from the local environment.
 func parseEnv(env []string) (api.EnvVars, error) {
@@ -217,3 +265,176 @@ func parseEnv(env []string) (api.EnvVars, error) {
 
 	return envVars, nil
 }
+
+// parseVolumeFlags parses volume flag values in Docker CLI format and returns VolumeSpecs and VolumeMounts.
+// It handles both named volumes (volume_name:/container/path[:ro|volume-nocopy])
+// and bind mounts (/host/path:/container/path[:ro]).
+func parseVolumeFlags(volumes []string) ([]api.VolumeSpec, []api.VolumeMount, error) {
+	specs := make([]api.VolumeSpec, 0, len(volumes))
+	mounts := make([]api.VolumeMount, 0, len(volumes))
+
+	// Track volume names to avoid duplicate specs.
+	seenVolumes := make(map[string]struct{})
+
+	for _, vol := range volumes {
+		spec, mount, err := parseVolumeFlagValue(vol)
+		if err != nil {
+			return nil, nil, fmt.Errorf("invalid volume mount '%s': %w", vol, err)
+		}
+
+		if _, ok := seenVolumes[spec.Name]; !ok {
+			specs = append(specs, spec)
+			seenVolumes[spec.Name] = struct{}{}
+		}
+
+		mounts = append(mounts, mount)
+	}
+
+	return specs, mounts, nil
+}
+
+func parseVolumeFlagValue(volume string) (api.VolumeSpec, api.VolumeMount, error) {
+	var spec api.VolumeSpec
+	var mount api.VolumeMount
+
+	parts := strings.Split(volume, ":")
+	switch len(parts) {
+	case 1:
+		return spec, mount, fmt.Errorf("invalid format, must contain at least one separator ':'", volume)
+	case 2, 3:
+		// Format: (volume_name|/host/path):/container/path[:opts]
+		if !strings.HasPrefix(parts[1], "/") {
+			return spec, mount, fmt.Errorf("invalid container mount path: '%s', must be absolute path", parts[1])
+		}
+
+		mount.ContainerPath = parts[1]
+		volumeNoCopy := false
+
+		if len(parts) == 3 {
+			opts := strings.Split(parts[2], ",")
+			for _, opt := range opts {
+				switch opt {
+				case "ro", "readonly":
+					mount.ReadOnly = true
+				case "volume-nocopy":
+					volumeNoCopy = true
+				default:
+					return spec, mount, fmt.Errorf("invalid option: '%s'", opt)
+				}
+			}
+		}
+
+		if strings.HasPrefix(parts[0], "/") {
+			// Host path bind mount: /host/path:/container/path
+			suffix, err := secret.RandomAlphaNumeric(4)
+			if err != nil {
+				return spec, mount, fmt.Errorf("generate random suffix: %w", err)
+			}
+
+			spec = api.VolumeSpec{
+				Name: "bind-" + suffix,
+				Type: api.VolumeTypeBind,
+				BindOptions: &api.BindOptions{
+					HostPath:       parts[0],
+					CreateHostPath: true,
+				},
+			}
+		} else {
+			// Named volume mount: volume_name:/container/path
+			volumeName := parts[0]
+			if !names.RestrictedNamePattern.MatchString(volumeName) {
+				return spec, mount, fmt.Errorf("volume name '%s' includes invalid characters, only '%s' are allowed. "+
+					"If you intended to pass a host directory or file, use absolute path",
+					volumeName, names.RestrictedNameChars)
+			}
+
+			spec = api.VolumeSpec{
+				Name: volumeName,
+				Type: api.VolumeTypeVolume,
+				VolumeOptions: &api.VolumeOptions{
+					Name:   volumeName,
+					NoCopy: volumeNoCopy,
+				},
+			}
+		}
+
+		mount.VolumeName = spec.Name
+	default:
+		return spec, mount, fmt.Errorf("invalid format, must container at most 2 separators ':'")
+	}
+
+	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 _, volume := range volumes {
+		if volume.Type == api.VolumeTypeVolume {
+			volumeNames = append(volumeNames, volume.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 _, volume := range volumes {
+		if volume.Type != api.VolumeTypeVolume {
+			continue
+		}
+
+		if !slices.ContainsFunc(vols, func(v api.MachineVolume) bool {
+			return v.Volume.Name == volume.Name && v.MachineID == machineID
+		}) {
+			missingVolumes = append(missingVolumes, volume)
+		}
+	}
+
+	return machineID, missingVolumes, nil
+}

pkg/api/service.go

@@ -168,7 +168,7 @@ func (s *ContainerSpec) SetDefaults() ContainerSpec {
 
 func (s *ContainerSpec) Validate() error {
 	if _, err := reference.ParseDockerRef(s.Image); err != nil {
-		return fmt.Errorf("invalid image: %w", err)
+		return fmt.Errorf("invalid image '%s': %w", s.Image, err)
 	}
 
 	for _, m := range s.VolumeMounts {

pkg/client/container.go

@@ -61,8 +61,8 @@ func (cli *Client) CreateContainer(
 			return resp, fmt.Errorf("unsupported pull policy: '%s'", spec.Container.PullPolicy)
 		}
 
-		// Not found error is expected if the image is missing.
-		if !dockerclient.IsErrNotFound(err) {
+		// NotFound (No such image) error is expected if the image is missing.
+		if !dockerclient.IsErrNotFound(err) || !strings.Contains(err.Error(), "No such image") {
 			return resp, err
 		}
 

pkg/client/deploy/resolver.go

@@ -67,25 +67,9 @@ func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error {
 		return nil
 	}
 
-	// Generate a random service name from the image when not provided.
-	img, err := reference.ParseDockerRef(spec.Container.Image)
-	if err != nil {
-		return fmt.Errorf("invalid image: %w", err)
-	}
-	// Get the image name without the repository and tag/digest parts.
-	imageName := reference.FamiliarName(img)
-	// Get the last part of the image name (path), e.g. "nginx" from "bitnami/nginx".
-	if i := strings.LastIndex(imageName, "/"); i != -1 {
-		imageName = imageName[i+1:]
-	}
-	// Append a random suffix to the image name to generate an optimistically unique service name.
-	suffix, err := secret.RandomAlphaNumeric(4)
-	if err != nil {
-		return fmt.Errorf("generate random suffix: %w", err)
-	}
-	spec.Name = fmt.Sprintf("%s-%s", imageName, suffix)
-
-	return nil
+	var err error
+	spec.Name, err = GenerateServiceName(spec.Container.Image)
+	return err
 }
 
 // expandIngressPorts processes HTTP(S) ingress ports in a service spec by:
@@ -142,6 +126,25 @@ func (r *ServiceSpecResolver) resolveImageDigest(spec *api.ServiceSpec) error {
 	return nil
 }
 
+func GenerateServiceName(image string) (string, error) {
+	img, err := reference.ParseDockerRef(image)
+	if err != nil {
+		return "", fmt.Errorf("invalid image '%s': %w", image, err)
+	}
+	// Get the image name without the repository and tag/digest parts.
+	imageName := reference.FamiliarName(img)
+	// Get the last part of the image name (path), e.g. "nginx" from "bitnami/nginx".
+	if i := strings.LastIndex(imageName, "/"); i != -1 {
+		imageName = imageName[i+1:]
+	}
+	// Append a random suffix to the image name to generate an optimistically unique service name.
+	suffix, err := secret.RandomAlphaNumeric(4)
+	if err != nil {
+		return "", fmt.Errorf("generate random suffix: %w", err)
+	}
+	return fmt.Sprintf("%s-%s", imageName, suffix), nil
+}
+
 type ImageResolverClient interface {
 	api.ImageClient
 	api.MachineClient

pkg/client/service.go

@@ -8,7 +8,6 @@ import (
 	"slices"
 	"sync"
 
-	"github.com/docker/compose/v2/pkg/progress"
 	"github.com/docker/docker/api/types/container"
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	"github.com/psviderski/uncloud/pkg/api"
@@ -44,23 +43,14 @@ func (cli *Client) RunService(
 	}
 
 	deployment := cli.NewDeployment(spec, &deploy.RollingStrategy{MachineFilter: filter})
-	plan, err := deployment.Plan(ctx)
+	plan, err := deployment.Run(ctx)
 	if err != nil {
-		return resp, fmt.Errorf("plan deployment: %w", err)
-	}
-
-	err = progress.RunWithTitle(ctx, func(ctx context.Context) error {
-		_, err = deployment.Run(ctx)
-		if err != nil {
-			return err
+		return resp, err
 	}
 
 	resp.ID = plan.ServiceID
 	resp.Name = plan.ServiceName
 
-		return nil
-	}, cli.progressOut(), fmt.Sprintf("Running service %s (%s mode)", plan.ServiceName, spec.Mode))
-
 	return resp, err
 }