diff --git a/pkg/api/service.go b/pkg/api/service.go index e953a1cb..97c9c203 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -175,7 +175,7 @@ type ContainerSpec struct { // Each mount references a volume defined in ServiceSpec.Volumes. VolumeMounts []VolumeMount // Volumes is list of data volumes to mount into the container. - // TODO: replace with []VolumeMounts + // TODO(lhf): replace with []VolumeMounts Volumes []string } diff --git a/pkg/client/compose/deploy.go b/pkg/client/compose/deploy.go index e446c476..1bb4660d 100644 --- a/pkg/client/compose/deploy.go +++ b/pkg/client/compose/deploy.go @@ -4,11 +4,14 @@ import ( "context" "errors" "fmt" + "slices" + "strings" "github.com/compose-spec/compose-go/v2/graph" "github.com/compose-spec/compose-go/v2/types" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client/deploy" + "github.com/psviderski/uncloud/pkg/client/deploy/scheduler" ) type Client interface { @@ -45,63 +48,128 @@ func (d *Deployment) Plan(ctx context.Context) (deploy.SequenceOperation, error) if d.plan != nil { return *d.plan, nil } - plan := deploy.SequenceOperation{} + + // Generate service specs for all services in the project. + var serviceSpecs []api.ServiceSpec err := graph.InDependencyOrder(ctx, d.Project, func(ctx context.Context, name string, _ types.ServiceConfig) error { spec, err := d.ServiceSpec(name) if err != nil { - return fmt.Errorf("convert compose service '%s' to service spec: %w", name, err) - } - - // TODO: properly handle depends_on conditions in the service deployment plan as the first operation. - deployment := deploy.NewDeployment(d.Client, spec, nil) - if err != nil { - return fmt.Errorf("create deployment for service '%s': %w", name, err) - } - - servicePlan, err := deployment.Plan(ctx) - if err != nil { - return fmt.Errorf("create deployment plan for service '%s': %w", name, err) - } - - // Skip no-op (up-to-date) service plans. - if len(servicePlan.Operations) > 0 { - plan.Operations = append(plan.Operations, &servicePlan) + return err } + serviceSpecs = append(serviceSpecs, spec) return nil }) if err != nil { - d.plan = &plan + return plan, err } - return plan, err + // Check external volumes and plan the creation of missing volumes before deploying services. + volumeOps, err := d.planVolumes(ctx, serviceSpecs) + if err != nil { + return plan, err + } + for _, op := range volumeOps { + plan.Operations = append(plan.Operations, op) + } + + for _, spec := range serviceSpecs { + // TODO: properly handle depends_on conditions in the service deployment plan as the first operation. + deployment := deploy.NewDeployment(d.Client, spec, nil) + servicePlan, err := deployment.Plan(ctx) + if err != nil { + return plan, fmt.Errorf("create deployment plan for service '%s': %w", spec.Name, err) + } + + // Skip no-op (up-to-date) service plans. + if len(servicePlan.Operations) > 0 { + plan.Operations = append(plan.Operations, &servicePlan) + } + } + + d.plan = &plan + return plan, nil } // ServiceSpec returns the service specification for the given compose service that is ready for deployment. func (d *Deployment) ServiceSpec(name string) (api.ServiceSpec, error) { - service, err := d.Project.GetService(name) - if err != nil { - return api.ServiceSpec{}, fmt.Errorf("get config for compose service '%s': %w", name, err) - } - - spec, err := ServiceSpecFromCompose(name, service) + spec, err := ServiceSpecFromCompose(d.Project, name) if err != nil { return spec, fmt.Errorf("convert compose service '%s' to service spec: %w", name, err) } - // TODO: resolve the image to a digest and supported platforms using an image resolver that broadcasts requests - // to all machines in the cluster. If service.PullPolicy is "missing": - // - Broadcast request if any machine contains a particular image and resolve it to image@digest. - // - If not found, broadcast request to resolve an image using a registry, and resolve it to image@digest. - // TODO: configure placement filter based on the supported platforms of the image. - - // TODO: maybe instantiate ImageResolver here based on PullPolicy of each service? - return spec, nil } +// PlanVolumes checks if the external volumes exist and plans the creation of missing volumes. +func (d *Deployment) planVolumes( + ctx context.Context, serviceSpecs []api.ServiceSpec, +) ([]*deploy.CreateVolumeOperation, error) { + if len(d.Project.Volumes) == 0 { + // No volumes to check or create. + return nil, nil + } + + if err := d.checkExternalVolumesExist(ctx); err != nil { + return nil, err + } + + // TODO: The scheduler should ideally work with the resolved service specs to correctly identify eligible machines. + // Figure out where the best place to resolve the specs is. + volumeScheduler, err := scheduler.NewVolumeSchedulerWithClient(ctx, d.Client, serviceSpecs) + if err != nil { + return nil, fmt.Errorf("init volume scheduler: %w", err) + } + scheduledVolumes, err := volumeScheduler.Schedule() + if err != nil { + return nil, fmt.Errorf("schedule volumes: %w", err) + } + + // Generate operations to create scheduled missing volumes. + var ops []*deploy.CreateVolumeOperation + for machineID, volumes := range scheduledVolumes { + for _, v := range volumes { + ops = append(ops, &deploy.CreateVolumeOperation{ + MachineID: machineID, + VolumeSpec: v, + }) + } + } + + return ops, nil +} + +// checkExternalVolumesExist checks that all external volumes exist in the cluster. +func (d *Deployment) checkExternalVolumesExist(ctx context.Context) error { + var externalNames []string + for _, v := range d.Project.Volumes { + if v.External { + externalNames = append(externalNames, v.Name) + } + } + + volumes, err := d.Client.ListVolumes(ctx, &api.VolumeFilter{Names: externalNames}) + if err != nil { + return fmt.Errorf("list volumes: %w", err) + } + + var notFound []string + for _, name := range externalNames { + if !slices.ContainsFunc(volumes, func(vol api.MachineVolume) bool { + return vol.Volume.Name == name + }) { + notFound = append(notFound, fmt.Sprintf("'%s'", name)) + } + } + + if len(notFound) > 0 { + return fmt.Errorf("external volumes not found: %s", strings.Join(notFound, ", ")) + } + return nil +} + func (d *Deployment) Run(ctx context.Context) error { plan, err := d.Plan(ctx) if err != nil { diff --git a/pkg/client/compose/project.go b/pkg/client/compose/project.go index 5b8d3b1e..1f2c401c 100644 --- a/pkg/client/compose/project.go +++ b/pkg/client/compose/project.go @@ -5,13 +5,18 @@ package compose import ( "context" "fmt" + composecli "github.com/compose-spec/compose-go/v2/cli" "github.com/compose-spec/compose-go/v2/types" ) +// FakeProjectName is a placeholder name for the project to be able to strip it from the resource names used as prefix. +const FakeProjectName = "f-a-k-e" + func LoadProject(ctx context.Context, paths []string) (*types.Project, error) { options, err := composecli.NewProjectOptions( paths, + composecli.WithName(FakeProjectName), // First apply os.Environment, always wins. composecli.WithOsEnv, // Read dot env file to populate project environment. diff --git a/pkg/client/compose/service.go b/pkg/client/compose/service.go index 312a92c0..cf4792cf 100644 --- a/pkg/client/compose/service.go +++ b/pkg/client/compose/service.go @@ -2,12 +2,23 @@ package compose import ( "fmt" + "maps" + "os" + "slices" + "strings" "github.com/compose-spec/compose-go/v2/types" + "github.com/docker/docker/api/types/mount" + "github.com/opencontainers/go-digest" "github.com/psviderski/uncloud/pkg/api" ) -func ServiceSpecFromCompose(name string, service types.ServiceConfig) (api.ServiceSpec, error) { +func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.ServiceSpec, error) { + service, err := project.GetService(serviceName) + if err != nil { + return api.ServiceSpec{}, fmt.Errorf("get config for compose service '%s': %w", serviceName, err) + } + pullPolicy := "" switch service.PullPolicy { case types.PullPolicyAlways: @@ -37,9 +48,8 @@ func ServiceSpecFromCompose(name string, service types.ServiceConfig) (api.Servi Image: service.Image, Init: service.Init, PullPolicy: pullPolicy, - // TODO: volumes }, - Name: name, + Name: serviceName, } if ports, ok := service.Extensions[PortsExtensionKey].([]api.PortSpec); ok { @@ -56,9 +66,131 @@ func ServiceSpecFromCompose(name string, service types.ServiceConfig) (api.Servi spec.Replicas = uint(*service.Deploy.Replicas) } default: - return spec, fmt.Errorf("unsupported deploy mode: %s", service.Deploy.Mode) + return spec, fmt.Errorf("unsupported deploy mode: '%s'", service.Deploy.Mode) } } + // TODO: can service.tmpfs be handled as tmpfs volume mounts as well? + volumeSpecs, volumeMounts, err := volumeSpecsFromCompose(project.Volumes, service.Volumes) + if err != nil { + return spec, err + } + + spec.Volumes = volumeSpecs + spec.Container.VolumeMounts = volumeMounts + return spec, nil } + +func volumeSpecsFromCompose( + volumes types.Volumes, serviceVolumes []types.ServiceVolumeConfig, +) ([]api.VolumeSpec, []api.VolumeMount, error) { + volumeSpecs := make(map[string]api.VolumeSpec) + var volumeMounts []api.VolumeMount + + for _, serviceVolume := range serviceVolumes { + var volSpec api.VolumeSpec + + switch serviceVolume.Type { + case types.VolumeTypeBind: + volSpec = bindVolumeSpecFromCompose(serviceVolume) + case types.VolumeTypeVolume: + volSpec = dockerVolumeSpecFromCompose(serviceVolume, volumes[serviceVolume.Source]) + case types.VolumeTypeTmpfs: + volSpec = tmpfsVolumeSpecFromCompose(serviceVolume) + default: + return nil, nil, fmt.Errorf("unsupported volume type: '%s'", serviceVolume.Type) + } + + if existing, ok := volumeSpecs[volSpec.Name]; ok { + if !existing.Equals(volSpec) { + return nil, nil, fmt.Errorf("volume '%s' is used multiple times with different options", volSpec.Name) + } + } else { + volumeSpecs[volSpec.Name] = volSpec + } + + volumeMounts = append(volumeMounts, api.VolumeMount{ + VolumeName: volSpec.Name, + ContainerPath: serviceVolume.Target, + ReadOnly: serviceVolume.ReadOnly, + }) + } + + return slices.Collect(maps.Values(volumeSpecs)), volumeMounts, nil +} + +func bindVolumeSpecFromCompose(serviceVolume types.ServiceVolumeConfig) api.VolumeSpec { + // compose-go parser deduplicates volumes by the target path so it's safe to use it as the unique name. + name := "bind-" + digest.SHA256.FromString(serviceVolume.Target).Encoded() + spec := api.VolumeSpec{ + Name: name, + Type: api.VolumeTypeBind, + BindOptions: &api.BindOptions{ + HostPath: serviceVolume.Source, + }, + } + if serviceVolume.Bind != nil { + spec.BindOptions.CreateHostPath = serviceVolume.Bind.CreateHostPath + spec.BindOptions.Propagation = mount.Propagation(serviceVolume.Bind.Propagation) + spec.BindOptions.Recursive = serviceVolume.Bind.Recursive + } + + return spec +} + +func dockerVolumeSpecFromCompose(serviceVolume types.ServiceVolumeConfig, volume types.VolumeConfig) api.VolumeSpec { + spec := api.VolumeSpec{ + Name: serviceVolume.Source, + Type: api.VolumeTypeVolume, + VolumeOptions: &api.VolumeOptions{ + Name: strings.TrimPrefix(volume.Name, FakeProjectName+"_"), + }, + } + + if serviceVolume.Volume != nil { + spec.VolumeOptions.NoCopy = serviceVolume.Volume.NoCopy + spec.VolumeOptions.SubPath = serviceVolume.Volume.Subpath + } + + if !volume.External { + if volume.Driver != "" { + spec.VolumeOptions.Driver = &mount.Driver{ + Name: volume.Driver, + Options: volume.DriverOpts, + } + } + + labels := mergeLabels(volume.Labels, volume.CustomLabels) + if len(labels) > 0 { + spec.VolumeOptions.Labels = labels + } + } + + return spec +} + +func mergeLabels(labels ...types.Labels) types.Labels { + merged := types.Labels{} + for _, l := range labels { + for k, v := range l { + merged[k] = v + } + } + return merged +} + +func tmpfsVolumeSpecFromCompose(serviceVolume types.ServiceVolumeConfig) api.VolumeSpec { + // compose-go parser deduplicates volumes by the target path so it's safe to use it as the unique name. + name := "tmpfs-" + digest.SHA256.FromString(serviceVolume.Target).Encoded() + spec := api.VolumeSpec{ + Name: name, + Type: api.VolumeTypeTmpfs, + TmpfsOptions: &mount.TmpfsOptions{ + SizeBytes: int64(serviceVolume.Tmpfs.Size), + Mode: os.FileMode(serviceVolume.Tmpfs.Mode), + }, + } + + return spec +} diff --git a/pkg/client/deploy/operation.go b/pkg/client/deploy/operation.go index cf2ae337..cf13e8f3 100644 --- a/pkg/client/deploy/operation.go +++ b/pkg/client/deploy/operation.go @@ -3,9 +3,11 @@ package deploy import ( "context" "fmt" - "github.com/docker/docker/api/types/container" - "github.com/psviderski/uncloud/pkg/api" "strings" + + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/volume" + "github.com/psviderski/uncloud/pkg/api" ) // Operation represents a single atomic operation in a deployment process. @@ -109,6 +111,45 @@ func (o *RemoveContainerOperation) String() string { o.ServiceID, o.ContainerID, o.MachineID) } +// CreateVolumeOperation creates a volume on a specific machine. +type CreateVolumeOperation struct { + VolumeSpec api.VolumeSpec + MachineID string +} + +func (o *CreateVolumeOperation) Execute(ctx context.Context, cli Client) error { + if o.VolumeSpec.Type != api.VolumeTypeVolume { + return fmt.Errorf("invalid volume type: '%s', expected '%s'", o.VolumeSpec.Type, api.VolumeTypeVolume) + } + + opts := volume.CreateOptions{ + Name: o.VolumeSpec.DockerVolumeName(), + } + if o.VolumeSpec.VolumeOptions != nil { + if o.VolumeSpec.VolumeOptions.Driver != nil { + opts.Driver = o.VolumeSpec.VolumeOptions.Driver.Name + opts.DriverOpts = o.VolumeSpec.VolumeOptions.Driver.Options + } + opts.Labels = o.VolumeSpec.VolumeOptions.Labels + } + + if _, err := cli.CreateVolume(ctx, o.MachineID, opts); err != nil { + return fmt.Errorf("create volume: %w", err) + } + + return nil +} + +func (o *CreateVolumeOperation) Format(resolver NameResolver) string { + machineName := resolver.MachineName(o.MachineID) + return fmt.Sprintf("%s: Create volume [name=%s]", machineName, o.VolumeSpec.DockerVolumeName()) +} + +func (o *CreateVolumeOperation) String() string { + return fmt.Sprintf("CreateVolumeOperation[volume=%s, machine_id=%s]", + o.VolumeSpec.DockerVolumeName(), o.MachineID) +} + // SequenceOperation is a composite operation that executes a sequence of operations in order. type SequenceOperation struct { Operations []Operation diff --git a/pkg/client/deploy/resolver.go b/pkg/client/deploy/resolver.go index 336c962b..21e6b37c 100644 --- a/pkg/client/deploy/resolver.go +++ b/pkg/client/deploy/resolver.go @@ -150,6 +150,8 @@ type ImageResolverClient interface { api.MachineClient } +// TODO(lhf): as of April 2025, ImageDigestResolver is not used in the codebase and considered more harmful +// than helpful. It's safe to remove it. type ImageDigestResolver struct { Ctx context.Context Client ImageResolverClient diff --git a/pkg/client/deploy/scheduler/constraint.go b/pkg/client/deploy/scheduler/constraint.go index 58564f5f..d93c358d 100644 --- a/pkg/client/deploy/scheduler/constraint.go +++ b/pkg/client/deploy/scheduler/constraint.go @@ -21,6 +21,9 @@ type Constraint interface { func constraintsFromSpec(spec api.ServiceSpec) []Constraint { var constraints []Constraint + // TODO: add placement constraint based on the supported platforms of the image. + // TODO: add placement constraint to limit machines with the image if pull policy is never. + if len(spec.Placement.Machines) > 0 { constraints = append(constraints, &PlacementConstraint{ Machines: spec.Placement.Machines,