From f8ebd94a796c47b2b5b85b1f445027836c7f36ab Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Fri, 11 Apr 2025 21:28:36 +1000 Subject: [PATCH] feat(volumes): create and mount data volumes when running service with 'uc run' command --- cmd/uncloud/service/run.go | 371 +++++++++++++++++++++++++++------- pkg/api/service.go | 2 +- pkg/client/container.go | 4 +- pkg/client/deploy/resolver.go | 41 ++-- pkg/client/service.go | 18 +- 5 files changed, 325 insertions(+), 111 deletions(-) diff --git a/cmd/uncloud/service/run.go b/cmd/uncloud/service/run.go index 06727a73..190de07b 100644 --- a/cmd/uncloud/service/run.go +++ b/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,90 +103,57 @@ 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 fmt.Errorf("run service: %w", err) + 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) @@ -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 +} diff --git a/pkg/api/service.go b/pkg/api/service.go index 4c7dfc31..72ff2fb8 100644 --- a/pkg/api/service.go +++ b/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 { diff --git a/pkg/client/container.go b/pkg/client/container.go index 8b6dbaef..c1e59981 100644 --- a/pkg/client/container.go +++ b/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 } diff --git a/pkg/client/deploy/resolver.go b/pkg/client/deploy/resolver.go index 7b0c9a35..336c962b 100644 --- a/pkg/client/deploy/resolver.go +++ b/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 diff --git a/pkg/client/service.go b/pkg/client/service.go index babfbc19..6c7d0edd 100644 --- a/pkg/client/service.go +++ b/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,22 +43,13 @@ 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) + return resp, err } - err = progress.RunWithTitle(ctx, func(ctx context.Context) error { - _, err = deployment.Run(ctx) - if err != nil { - return err - } - - resp.ID = plan.ServiceID - resp.Name = plan.ServiceName - - return nil - }, cli.progressOut(), fmt.Sprintf("Running service %s (%s mode)", plan.ServiceName, spec.Mode)) + resp.ID = plan.ServiceID + resp.Name = plan.ServiceName return resp, err }