From f3cb6657aec8a4cfe48effc6a36cf68e2d4d7306 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Wed, 16 Apr 2025 22:04:37 +1000 Subject: [PATCH] chore: replace machine filter with placement constraint in service spec --- cmd/uncloud/caddy/deploy.go | 42 ++-- cmd/uncloud/deploy.go | 2 + cmd/uncloud/machine/add.go | 14 +- cmd/uncloud/machine/init.go | 3 +- cmd/uncloud/service/run.go | 59 +++--- pkg/api/placement.go | 8 + pkg/api/service.go | 2 + pkg/client/caddy.go | 28 ++- pkg/client/deploy/deploy.go | 1 + pkg/client/deploy/scheduler/constraint.go | 14 +- pkg/client/deploy/scheduler/service.go | 30 +-- pkg/client/deploy/strategy.go | 87 ++++---- pkg/client/service.go | 6 +- test/e2e/assert.go | 6 +- test/e2e/service_test.go | 230 ++++++++-------------- 15 files changed, 238 insertions(+), 294 deletions(-) create mode 100644 pkg/api/placement.go diff --git a/cmd/uncloud/caddy/deploy.go b/cmd/uncloud/caddy/deploy.go index b42d1a58..3ecd76a8 100644 --- a/cmd/uncloud/caddy/deploy.go +++ b/cmd/uncloud/caddy/deploy.go @@ -19,9 +19,9 @@ import ( ) type deployOptions struct { - image string - machine string - context string + image string + machines []string + context string } func NewDeployCommand() *cobra.Command { @@ -40,8 +40,9 @@ func NewDeployCommand() *cobra.Command { cmd.Flags().StringVar(&opts.image, "image", "", "Caddy Docker image to deploy. (default caddy:LATEST_VERSION)") - cmd.Flags().StringVarP(&opts.machine, "machine", "m", "", - "Machine names to deploy to (comma-separated). (default is all machines)") + cmd.Flags().StringSliceVarP(&opts.machines, "machine", "m", nil, + "Machine names to deploy to. Can be specified multiple times or as a comma-separated "+ + "list of machine names. (default is all machines)") cmd.Flags().StringVarP( &opts.context, "context", "c", "", "Name of the cluster context to deploy to. (default is the current context)", @@ -88,16 +89,10 @@ func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { fmt.Println() fmt.Println("Preparing a deployment plan...") - var filter deploy.MachineFilter - if opts.machine != "" { - machines := strings.Split(opts.machine, ",") - for i, m := range machines { - machines[i] = strings.TrimSpace(m) - } - filter = machineFilter(machines) + placement := api.Placement{ + Machines: cli.ExpandCommaSeparatedValues(opts.machines), } - - d, err := clusterClient.NewCaddyDeployment(opts.image, filter) + d, err := clusterClient.NewCaddyDeployment(opts.image, placement) if err != nil { return fmt.Errorf("create caddy deployment: %w", err) } @@ -108,31 +103,18 @@ func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { plan, err := d.Plan(ctx) if err != nil { - if errors.Is(err, deploy.ErrNoMatchingMachines) { - return fmt.Errorf("no machines found matching: %s", opts.machine) - } return fmt.Errorf("plan caddy deployment: %w", err) } if len(plan.Operations) == 0 { - if opts.machine != "" { - fmt.Printf("%s service is up to date on selected machines.\n", client.CaddyServiceName) - } else { - fmt.Printf("%s service is up to date.\n", client.CaddyServiceName) - } + fmt.Printf("%s service is up to date.\n", client.CaddyServiceName) } else { if svc.ID == "" { - if opts.machine != "" { - fmt.Println("This will run a Caddy container on selected machines.") + if len(opts.machines) > 0 { + fmt.Println("This will run a Caddy container on each selected machine.") } else { fmt.Println("This will run a Caddy container on each machine.") } - } else { - if opts.machine != "" { - fmt.Println("This will perform a rolling update of Caddy containers on selected machines.") - } else { - fmt.Println("This will perform a rolling update of Caddy containers on each machine.") - } } // Initialise a machine and container name resolver to properly format the plan output. diff --git a/cmd/uncloud/deploy.go b/cmd/uncloud/deploy.go index 444b0c58..187185e4 100644 --- a/cmd/uncloud/deploy.go +++ b/cmd/uncloud/deploy.go @@ -44,6 +44,8 @@ func NewDeployCommand() *cobra.Command { "One or more Compose files to deploy services from. (default compose-ports-long.yaml)") cmd.Flags().StringVarP(&opts.context, "context", "c", "", "Name of the cluster context to deploy to (default is the current context)") + // TODO: Consider adding a filter flag to specify which machines to deploy to but keep the rest running. + // Could be useful to test a new version on a subset of machines before rolling out to all. return cmd } diff --git a/cmd/uncloud/machine/add.go b/cmd/uncloud/machine/add.go index e6d7415d..b3954064 100644 --- a/cmd/uncloud/machine/add.go +++ b/cmd/uncloud/machine/add.go @@ -12,13 +12,11 @@ import ( "github.com/psviderski/uncloud/cmd/uncloud/caddy" "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli/config" - "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" "github.com/spf13/cobra" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" - "google.golang.org/protobuf/types/known/emptypb" ) type addOptions struct { @@ -105,14 +103,6 @@ func add(ctx context.Context, uncli *cli.CLI, remoteMachine cli.RemoteMachine, o return fmt.Errorf("wait for cluster to be initialised on machine: %w", err) } - // Inspect the added machine to get its ID to create a filter for the Caddy deployment. - minfo, err := machineClient.Inspect(ctx, &emptypb.Empty{}) - if err != nil { - return fmt.Errorf("inspect machine: %w", err) - } - filter := func(m *pb.MachineInfo) bool { - return m.Id == minfo.Id - } // Deploy a Caddy service container to the added machine. If caddy service is already deployed on other machines, // use the deployed image version. Otherwise, use the latest version. caddyImage := "" @@ -137,7 +127,9 @@ func add(ctx context.Context, uncli *cli.CLI, remoteMachine cli.RemoteMachine, o } } - d, err := machineClient.NewCaddyDeployment(caddyImage, filter) + // TODO: scale the existing Caddy service to the new machine instead of running a new deployment + // that may cause a small downtime. + d, err := machineClient.NewCaddyDeployment(caddyImage, api.Placement{}) if err != nil { return fmt.Errorf("create caddy deployment: %w", err) } diff --git a/cmd/uncloud/machine/init.go b/cmd/uncloud/machine/init.go index dc21e1e3..dd8d1a85 100644 --- a/cmd/uncloud/machine/init.go +++ b/cmd/uncloud/machine/init.go @@ -12,6 +12,7 @@ import ( "github.com/psviderski/uncloud/internal/cli/config" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/cluster" + "github.com/psviderski/uncloud/pkg/api" "github.com/spf13/cobra" ) @@ -134,7 +135,7 @@ func initCluster(ctx context.Context, uncli *cli.CLI, remoteMachine *cli.RemoteM } if !opts.noCaddy { - d, err := client.NewCaddyDeployment("", nil) + d, err := client.NewCaddyDeployment("", api.Placement{}) if err != nil { return fmt.Errorf("create caddy deployment: %w", err) } diff --git a/cmd/uncloud/service/run.go b/cmd/uncloud/service/run.go index 66cca336..0afbcd65 100644 --- a/cmd/uncloud/service/run.go +++ b/cmd/uncloud/service/run.go @@ -11,7 +11,6 @@ import ( "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" @@ -66,7 +65,7 @@ func NewRunCommand() *cobra.Command { "the machines) or '%s' (one container on every machine).", api.ServiceModeReplicated, api.ServiceModeGlobal)) cmd.Flags().StringSliceVarP(&opts.machines, "machine", "m", nil, - "Placement constraint by machine name, limiting which machines the service can run on. Can be specified "+ + "Placement constraint by machine names, limiting which machines the service can run on. Can be specified "+ "multiple times or as a comma-separated list of machine names. (default is any suitable machine)") cmd.Flags().StringVarP(&opts.name, "name", "n", "", "Assign a name to the service. A random name is generated if not specified.") @@ -114,26 +113,15 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { } defer clusterClient.Close() - 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) + machineIDForVolumes, missingVolumes, err := selectMachineForVolumes( + ctx, + clusterClient, + spec.Volumes, + spec.Placement.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 { @@ -145,7 +133,7 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { } } - resp, err = clusterClient.RunService(ctx, spec, deployFilter) + resp, err = clusterClient.RunService(ctx, spec) if err != nil { return fmt.Errorf("run service: %w", err) } @@ -207,6 +195,10 @@ func prepareServiceSpec(opts runOptions) (api.ServiceSpec, error) { return spec, err } + placement := api.Placement{ + Machines: cli.ExpandCommaSeparatedValues(opts.machines), + } + spec = api.ServiceSpec{ Container: api.ContainerSpec{ Command: opts.command, @@ -215,11 +207,12 @@ func prepareServiceSpec(opts runOptions) (api.ServiceSpec, error) { PullPolicy: opts.pull, VolumeMounts: mounts, }, - Mode: opts.mode, - Name: opts.name, - Ports: ports, - Replicas: opts.replicas, - Volumes: volumes, + Mode: opts.mode, + Name: opts.name, + Placement: placement, + Ports: ports, + Replicas: opts.replicas, + Volumes: volumes, } // Overwrite the default ENTRYPOINT of the image or reset it if an empty string is passed. @@ -372,9 +365,9 @@ 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) + for _, v := range volumes { + if v.Type == api.VolumeTypeVolume { + volumeNames = append(volumeNames, v.Name) } } if len(volumeNames) == 0 { @@ -424,15 +417,15 @@ func selectMachineForVolumes( } // Find missing volumes that need to be created on the selected machine. - for _, volume := range volumes { - if volume.Type != api.VolumeTypeVolume { + for _, v := range volumes { + if v.Type != api.VolumeTypeVolume { continue } - if !slices.ContainsFunc(vols, func(v api.MachineVolume) bool { - return v.Volume.Name == volume.Name && v.MachineID == machineID + if !slices.ContainsFunc(vols, func(mv api.MachineVolume) bool { + return mv.Volume.Name == v.Name && mv.MachineID == machineID }) { - missingVolumes = append(missingVolumes, volume) + missingVolumes = append(missingVolumes, v) } } diff --git a/pkg/api/placement.go b/pkg/api/placement.go new file mode 100644 index 00000000..df81bfcc --- /dev/null +++ b/pkg/api/placement.go @@ -0,0 +1,8 @@ +package api + +// Placement defines the placement constraints for service containers. +type Placement struct { + // Machines is a list of machine names or IDs where service containers are allowed to be deployed. + // If empty, containers can be deployed to any available machine in the cluster. + Machines []string `json:",omitempty"` +} diff --git a/pkg/api/service.go b/pkg/api/service.go index 72ff2fb8..005fb606 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -42,6 +42,8 @@ type ServiceSpec struct { // Mode is the replication mode of the service. Default is ServiceModeReplicated if empty. Mode string Name string + // Placement defines the placement constraints for the service. + Placement Placement // Ports defines what service ports to publish to make the service accessible outside the cluster. Ports []PortSpec // Replicas is the number of containers to run for the service. Only valid for a replicated service. diff --git a/pkg/client/caddy.go b/pkg/client/caddy.go index b9688998..4686a04d 100644 --- a/pkg/client/caddy.go +++ b/pkg/client/caddy.go @@ -2,13 +2,14 @@ package client import ( "fmt" + "regexp" + "github.com/Masterminds/semver" "github.com/distribution/reference" "github.com/google/go-containerregistry/pkg/name" "github.com/google/go-containerregistry/pkg/v1/remote" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client/deploy" - "regexp" ) const ( @@ -22,7 +23,7 @@ var caddyImageTagRegex = regexp.MustCompile(`^2\.\d+\.\d+$`) // NewCaddyDeployment creates a new deployment for a Caddy reverse proxy service. // The service is deployed in global mode to all machines in the cluster. If the image is not provided, the latest // version of the official Caddy Docker image is used. -func (cli *Client) NewCaddyDeployment(image string, filter deploy.MachineFilter) (*deploy.Deployment, error) { +func (cli *Client) NewCaddyDeployment(image string, placement api.Placement) (*deploy.Deployment, error) { latest, err := LatestCaddyImage() if err != nil { return nil, fmt.Errorf("look up latest Caddy image: %w", err) @@ -36,10 +37,16 @@ func (cli *Client) NewCaddyDeployment(image string, filter deploy.MachineFilter) Container: api.ContainerSpec{ Command: []string{"caddy", "run", "-c", "/config/caddy.json", "--watch"}, Image: image, - Volumes: []string{"/var/lib/uncloud/caddy:/config"}, + VolumeMounts: []api.VolumeMount{ + { + VolumeName: "config", + ContainerPath: "/config", + }, + }, }, - Mode: api.ServiceModeGlobal, - Name: CaddyServiceName, + Mode: api.ServiceModeGlobal, + Name: CaddyServiceName, + Placement: placement, Ports: []api.PortSpec{ { PublishedPort: 80, @@ -54,9 +61,18 @@ func (cli *Client) NewCaddyDeployment(image string, filter deploy.MachineFilter) Mode: api.PortModeHost, }, }, + Volumes: []api.VolumeSpec{ + { + Name: "config", + Type: api.VolumeTypeBind, + BindOptions: &api.BindOptions{ + HostPath: "/var/lib/uncloud/caddy", + }, + }, + }, } - return cli.NewDeployment(spec, &deploy.RollingStrategy{MachineFilter: filter}), nil + return cli.NewDeployment(spec, nil), nil } // LatestCaddyImage returns the latest image of the official Caddy Docker image on Docker Hub. diff --git a/pkg/client/deploy/deploy.go b/pkg/client/deploy/deploy.go index ca1f8a10..b96f8746 100644 --- a/pkg/client/deploy/deploy.go +++ b/pkg/client/deploy/deploy.go @@ -15,6 +15,7 @@ type Client interface { api.ImageClient api.MachineClient api.ServiceClient + api.VolumeClient } // Deployment manages the process of creating or updating a service to match a desired state. diff --git a/pkg/client/deploy/scheduler/constraint.go b/pkg/client/deploy/scheduler/constraint.go index 5bd9e1ea..f2722529 100644 --- a/pkg/client/deploy/scheduler/constraint.go +++ b/pkg/client/deploy/scheduler/constraint.go @@ -16,10 +16,22 @@ type Constraint interface { } func constraintsFromSpec(spec api.ServiceSpec) []Constraint { - return []Constraint{} + var constraints []Constraint + + if len(spec.Placement.Machines) > 0 { + constraints = append(constraints, &PlacementConstraint{ + Machines: spec.Placement.Machines, + }) + } + + // TODO: inspect and add VolumeConstraint. + + return constraints } type PlacementConstraint struct { + // Machines is a list of machine names or IDs where service containers are allowed to be deployed. + // If empty, containers can be deployed to any available machine in the cluster. Machines []string } diff --git a/pkg/client/deploy/scheduler/service.go b/pkg/client/deploy/scheduler/service.go index 981eca3c..b7b76b6d 100644 --- a/pkg/client/deploy/scheduler/service.go +++ b/pkg/client/deploy/scheduler/service.go @@ -1,32 +1,38 @@ 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 - spec api.ServiceSpec - constraints []Constraint + Machines []*Machine + Spec api.ServiceSpec + Constraints []Constraint } -func NewServiceScheduler(machines []*Machine, spec api.ServiceSpec, constraints []Constraint) *ServiceScheduler { - specConstraints := constraintsFromSpec(spec) - specConstraints = append(specConstraints, constraints...) +func NewServiceScheduler(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) + } + + constraints := constraintsFromSpec(spec) return &ServiceScheduler{ - machines: machines, - spec: spec, - constraints: specConstraints, - } + Machines: machines, + Spec: spec, + Constraints: constraints, + }, nil } func (s *ServiceScheduler) AvailableMachines() ([]*Machine, error) { var available []*Machine - for _, machine := range s.machines { + for _, machine := range s.Machines { if s.evaluateConstraints(machine) { available = append(available, machine) } @@ -38,7 +44,7 @@ func (s *ServiceScheduler) AvailableMachines() ([]*Machine, error) { } func (s *ServiceScheduler) evaluateConstraints(machine *Machine) bool { - for _, c := range s.constraints { + for _, c := range s.Constraints { if !c.Evaluate(machine) { return false } diff --git a/pkg/client/deploy/strategy.go b/pkg/client/deploy/strategy.go index 0fb5968d..f65bbea5 100644 --- a/pkg/client/deploy/strategy.go +++ b/pkg/client/deploy/strategy.go @@ -9,6 +9,7 @@ import ( "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/deploy/scheduler" ) // Strategy defines how a service should be deployed or updated. Different implementations can provide various @@ -18,22 +19,19 @@ type Strategy interface { Type() string // Plan returns the operation to reconcile the service to the desired state. // If the service does not exist (new deployment), svc will be nil. - Plan(ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec) (Plan, error) + Plan(ctx context.Context, cli scheduler.Client, svc *api.Service, spec api.ServiceSpec) (Plan, error) } // RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time // to minimize service disruption. -type RollingStrategy struct { - // MachineFilter optionally restricts which machines can be used for deployment. - MachineFilter MachineFilter -} +type RollingStrategy struct{} func (s *RollingStrategy) Type() string { return "rolling" } func (s *RollingStrategy) Plan( - ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec, + ctx context.Context, cli scheduler.Client, svc *api.Service, spec api.ServiceSpec, ) (Plan, error) { // We can assume that the spec is valid at this point because it has been validated by the deployment. switch spec.Mode { @@ -51,37 +49,29 @@ func (s *RollingStrategy) Plan( // in the cluster. // TODO: schedule containers only on machines that contain the image if pull policy is set to 'never'. func (s *RollingStrategy) planReplicated( - ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec, + ctx context.Context, cli scheduler.Client, svc *api.Service, spec api.ServiceSpec, ) (Plan, error) { plan, err := newEmptyPlan(svc, spec) if err != nil { return plan, err } - availableMachines, err := cli.ListMachines(ctx, &api.MachineFilter{Available: true}) + sched, err := scheduler.NewServiceScheduler(ctx, cli, spec) if err != nil { - return plan, fmt.Errorf("list machines: %w", err) + return plan, err + } + // TODO: return a detailed report on required constraints and which ones are satisfied? + availableMachines, err := sched.AvailableMachines() + if err != nil { + return plan, err } - // Filter machines that match the machine filter if provided. var matchedMachines []*pb.MachineInfo - var unmatchedMachines []*pb.MachineInfo for _, m := range availableMachines { - if s.MachineFilter == nil || s.MachineFilter(m.Machine) { - matchedMachines = append(matchedMachines, m.Machine) - } else { - unmatchedMachines = append(unmatchedMachines, m.Machine) - } + matchedMachines = append(matchedMachines, m.Info) } - if len(matchedMachines) == 0 { - if s.MachineFilter != nil { - return plan, ErrNoMatchingMachines - } - return plan, fmt.Errorf("no available machines to deploy service") - } - - // TODO: filter machines that contain the service volumes if the service uses any.s + // TODO: filter machines that contain the service volumes if the service uses any. // Randomise the order of machines to avoid always deploying to the same machines first. rand.Shuffle(len(matchedMachines), func(i, j int) { @@ -209,7 +199,7 @@ func (s *RollingStrategy) planReplicated( // It handles multiple containers per machine (though this should not occur in normal operation) and skips machines // that are down. func (s *RollingStrategy) planGlobal( - ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec, + ctx context.Context, cli scheduler.Client, svc *api.Service, spec api.ServiceSpec, ) (Plan, error) { plan, err := newEmptyPlan(svc, spec) if err != nil { @@ -226,39 +216,36 @@ func (s *RollingStrategy) planGlobal( } } - machines, err := cli.ListMachines(ctx, nil) + sched, err := scheduler.NewServiceScheduler(ctx, cli, spec) if err != nil { - return plan, fmt.Errorf("list machines: %w", err) - } - // Filter machines if a machine filter is provided. - // TODO: not sure this is the right behaviour to ignore other machines that might run service containers. - // Maybe there should be another filter to specify which machines to deploy to but keep the rest running. - // Could be useful to test a new version on a subset of machines before rolling out to all. - if s.MachineFilter != nil { - machines = slices.DeleteFunc(machines, func(m *pb.MachineMember) bool { - return !s.MachineFilter(m.Machine) - }) - if len(machines) == 0 { - return plan, ErrNoMatchingMachines - } + return plan, err } - // TODO: figure out how to return a warning if there are machines down. Embed the machinesDown in the plan? - var machinesDown []*pb.MachineInfo - for _, m := range machines { - // Skip machines that are down but collect them to report a warning later. - if m.State == pb.MachineMember_DOWN { - machinesDown = append(machinesDown, m.Machine) - fmt.Printf("WARNING: failed to run a service container on machine '%s' which is Down.\n", m.Machine.Id) - continue - } + availableMachines, err := sched.AvailableMachines() + if err != nil { + return plan, err + } - containers := containersOnMachine[m.Machine.Id] - ops, err := reconcileGlobalContainer(containers, spec, plan.ServiceID, m.Machine.Id) + for _, m := range availableMachines { + containers := containersOnMachine[m.Info.Id] + ops, err := reconcileGlobalContainer(containers, spec, plan.ServiceID, m.Info.Id) if err != nil { return plan, err } plan.Operations = append(plan.Operations, ops...) + + delete(containersOnMachine, m.Info.Id) + } + + // Remove any remaining containers on machines that don't match the new placement constraints. + for _, containers := range containersOnMachine { + for _, c := range containers { + plan.Operations = append(plan.Operations, &RemoveContainerOperation{ + ServiceID: plan.ServiceID, + ContainerID: c.Container.ID, + MachineID: c.MachineID, + }) + } } return plan, nil diff --git a/pkg/client/service.go b/pkg/client/service.go index 6c7d0edd..4942c647 100644 --- a/pkg/client/service.go +++ b/pkg/client/service.go @@ -22,9 +22,7 @@ type RunServiceResponse struct { Name string } -func (cli *Client) RunService( - ctx context.Context, spec api.ServiceSpec, filter deploy.MachineFilter, -) (RunServiceResponse, error) { +func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunServiceResponse, error) { var resp RunServiceResponse if err := spec.Validate(); err != nil { @@ -42,7 +40,7 @@ func (cli *Client) RunService( } } - deployment := cli.NewDeployment(spec, &deploy.RollingStrategy{MachineFilter: filter}) + deployment := cli.NewDeployment(spec, &deploy.RollingStrategy{}) plan, err := deployment.Run(ctx) if err != nil { return resp, err diff --git a/test/e2e/assert.go b/test/e2e/assert.go index 54518a8a..f22757b0 100644 --- a/test/e2e/assert.go +++ b/test/e2e/assert.go @@ -114,7 +114,7 @@ func sortMounts(mounts []mount.Mount) { } // serviceContainersByMachine returns a map of machine ID to service containers on that machine. -func serviceContainersByMachine(t *testing.T, svc api.Service) map[string][]api.ServiceContainer { +func serviceContainersByMachine(svc api.Service) map[string][]api.ServiceContainer { containers := make(map[string][]api.ServiceContainer) for _, c := range svc.Containers { containers[c.MachineID] = append(containers[c.MachineID], c.Container) @@ -122,7 +122,7 @@ func serviceContainersByMachine(t *testing.T, svc api.Service) map[string][]api. return containers } -func serviceMachines(t *testing.T, svc api.Service) mapset.Set[string] { +func serviceMachines(svc api.Service) mapset.Set[string] { machines := mapset.NewSet[string]() for _, c := range svc.Containers { machines.Add(c.MachineID) @@ -131,7 +131,7 @@ func serviceMachines(t *testing.T, svc api.Service) mapset.Set[string] { return machines } -func serviceContainerIDs(t *testing.T, svc api.Service) mapset.Set[string] { +func serviceContainerIDs(svc api.Service) mapset.Set[string] { ids := mapset.NewSet[string]() for _, c := range svc.Containers { ids.Add(c.Container.ID) diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index f7175923..a73f652d 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -9,7 +9,6 @@ import ( "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/mount" "github.com/docker/docker/api/types/volume" - "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/internal/ucind" "github.com/psviderski/uncloud/pkg/api" @@ -80,11 +79,11 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, spec) assert.Len(t, svc.Containers, 3) - machines := serviceMachines(t, svc) + machines := serviceMachines(svc) assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine") // Deploy a published port. - initialContainers := serviceContainerIDs(t, svc) + initialContainers := serviceContainerIDs(svc) specWithPort := api.ServiceSpec{ Name: name, @@ -115,9 +114,9 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, specWithPort) assert.Len(t, svc.Containers, 3) - machines = serviceMachines(t, svc) + machines = serviceMachines(svc) assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine") - containers := serviceContainerIDs(t, svc) + containers := serviceContainerIDs(svc) assert.Empty(t, initialContainers.Intersect(containers).ToSlice(), "All existing containers should be replaced") @@ -155,9 +154,9 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, specWithPortAndInit) assert.Len(t, svc.Containers, 3) - machines = serviceMachines(t, svc) + machines = serviceMachines(svc) assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine") - containers = serviceContainerIDs(t, svc) + containers = serviceContainerIDs(svc) assert.Empty(t, initialContainers.Intersect(containers).ToSlice(), "All existing containers should be replaced") @@ -176,14 +175,14 @@ func TestDeployment(t *testing.T) { svc, err = cli.InspectService(ctx, name) require.NoError(t, err) - containers = serviceContainerIDs(t, svc) + containers = serviceContainerIDs(svc) assert.ElementsMatch(t, initialContainers.ToSlice(), containers.ToSlice()) }) - t.Run("global with machine filter", func(t *testing.T) { + t.Run("global with machine placement", func(t *testing.T) { t.Parallel() - name := "global-deployment-filtered" + name := "test-global-deployment-machine-placement" t.Cleanup(func() { err := cli.RemoveService(ctx, name) if !errors.Is(err, api.ErrNotFound) { @@ -191,13 +190,16 @@ func TestDeployment(t *testing.T) { } }) - // First deploy globally without filter to get containers on all machines. + // First deploy globally to machines #0 and #1. spec := api.ServiceSpec{ Name: name, Mode: api.ServiceModeGlobal, Container: api.ContainerSpec{ Image: "portainer/pause:latest", }, + Placement: api.Placement{ + Machines: []string{c.Machines[0].Name, c.Machines[1].Name}, + }, } deployment := cli.NewDeployment(spec, nil) @@ -206,89 +208,57 @@ func TestDeployment(t *testing.T) { svc, err := cli.InspectService(ctx, name) require.NoError(t, err) - assert.Len(t, svc.Containers, 3, "expected 1 container on each machine") + assertServiceMatchesSpec(t, svc, spec) - // Store initial container IDs by machine. - initialContainers := make(map[string]string) // machineID -> containerID - for _, ctr := range svc.Containers { - initialContainers[ctr.MachineID] = ctr.Container.ID - } + assert.Len(t, svc.Containers, 2, "Expected 1 container on machines %s and %s", + c.Machines[0].Name, c.Machines[1].Name) + initialMachines := serviceMachines(svc) + assert.ElementsMatch(t, initialMachines.ToSlice(), []string{c.Machines[0].ID, c.Machines[1].ID}) + + initialContainers := serviceContainerIDs(svc) // Update spec with Init=true, but only deploy to machines #0 and #2. init := true specWithInit := spec specWithInit.Container.Init = &init + specWithInit.Placement.Machines = []string{c.Machines[0].Name, c.Machines[2].Name} - filter := func(m *pb.MachineInfo) bool { - return m.Name == c.Machines[0].Name || m.Name == c.Machines[2].Name - } - strategy := &deploy.RollingStrategy{MachineFilter: filter} - deployment = cli.NewDeployment(specWithInit, strategy) + deployment = cli.NewDeployment(specWithInit, nil) _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) require.NoError(t, err) - assert.Len(t, svc.Containers, 3, "still 1 container on each machine") + assertServiceMatchesSpec(t, svc, specWithInit) - // Verify: - // 1. Containers on machines #0 and #2 were updated (new IDs, init enabled) - // 2. Container on machine #1 remains unchanged (same ID, no init) - for _, ctr := range svc.Containers { - machine, err := cli.InspectMachine(ctx, ctr.MachineID) - require.NoError(t, err) + assert.Len(t, svc.Containers, 2, "Expected 1 container on machines %s and %s", + c.Machines[0].Name, c.Machines[2].Name) + machines := serviceMachines(svc) + assert.ElementsMatch(t, machines.ToSlice(), []string{c.Machines[0].ID, c.Machines[2].ID}) - oldContainerID := initialContainers[ctr.MachineID] - switch machine.Machine.Name { - case c.Machines[0].Name, c.Machines[2].Name: - // These containers should be updated. - assert.NotEqual(t, oldContainerID, ctr.Container.ID, - "Container on machine %s should have been updated", machine.Machine.Name) + containers := serviceContainerIDs(svc) + assert.Empty(t, initialContainers.Intersect(containers).ToSlice(), + "All initial containers should be replaced") - svcSpec := ctr.Container.ServiceSpec - require.NoError(t, err) - assert.NotNil(t, svcSpec.Container.Init) - assert.True(t, *svcSpec.Container.Init, - "Container on machine %s should have init enabled", machine.Machine.Name) - case c.Machines[1].Name: - // This container should remain unchanged. - assert.Equal(t, oldContainerID, ctr.Container.ID, - "Container on machine %s should not have been updated", machine.Machine.Name) - } - } + // Now deploy the same spec without a placement constraint. + initialContainers = containers // Reset container tracking. + specWithInit.Placement = api.Placement{} - // Now deploy another update without filter - should affect all machines. - init = false - specWithPort := spec - specWithPort.Ports = []api.PortSpec{ - { - PublishedPort: 8001, - ContainerPort: 8001, - Protocol: api.ProtocolTCP, - Mode: api.PortModeHost, - }, - } - deployment = cli.NewDeployment(specWithPort, nil) + deployment = cli.NewDeployment(specWithInit, nil) _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) require.NoError(t, err) + assertServiceMatchesSpec(t, svc, specWithInit) assert.Len(t, svc.Containers, 3) - // Verify all containers are updated with a published port. - for _, ctr := range svc.Containers { - svcSpec := ctr.Container.ServiceSpec - require.NoError(t, err) - assert.Nil(t, svcSpec.Container.Init, - "Container on machine %s should have init disabled", ctr.MachineID) + machines = serviceMachines(svc) + assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine") - ports, err := ctr.Container.ServicePorts() - require.NoError(t, err) - assert.Equal(t, specWithPort.Ports, ports, - "Container on machine %s should have updated port", ctr.MachineID) - } + containers = serviceContainerIDs(svc) + assert.True(t, initialContainers.IsSubset(containers), "Expected all initial containers to remain") }) t.Run("caddy", func(t *testing.T) { @@ -299,7 +269,7 @@ func TestDeployment(t *testing.T) { } }) - deployment, err := cli.NewCaddyDeployment("", nil) + deployment, err := cli.NewCaddyDeployment("", api.Placement{}) require.NoError(t, err) _, err = deployment.Run(ctx) @@ -338,7 +308,7 @@ func TestDeployment(t *testing.T) { }, ctr.HostConfig.RestartPolicy) }) - t.Run("caddy with machine filter", func(t *testing.T) { + t.Run("caddy with machine placement", func(t *testing.T) { t.Cleanup(func() { err := cli.RemoveService(ctx, client.CaddyServiceName) if !errors.Is(err, api.ErrNotFound) { @@ -346,12 +316,10 @@ func TestDeployment(t *testing.T) { } }) - // Deploy to machine #0 - filter := func(m *pb.MachineInfo) bool { - return m.Name == c.Machines[0].Name - } - - deployment, err := cli.NewCaddyDeployment("", filter) + // Deploy to machine #0. + deployment, err := cli.NewCaddyDeployment("", api.Placement{ + Machines: []string{c.Machines[0].Name}, + }) require.NoError(t, err) image := deployment.Spec.Container.Image @@ -361,17 +329,13 @@ func TestDeployment(t *testing.T) { svc, err := cli.InspectService(ctx, client.CaddyServiceName) require.NoError(t, err) assert.Len(t, svc.Containers, 1) - ctr0 := svc.Containers[0] + assertServiceMatchesSpec(t, svc, deployment.Spec) - machine0, err := cli.InspectMachine(ctx, ctr0.MachineID) - require.NoError(t, err) - assert.Equal(t, c.Machines[0].Name, machine0.Machine.Name) + assert.Equal(t, c.Machines[0].ID, svc.Containers[0].MachineID) + initialContainerID := svc.Containers[0].Container.ID - // Deploy to machines #0 and #2 - filter = func(m *pb.MachineInfo) bool { - return m.Name == c.Machines[0].Name || m.Name == c.Machines[2].Name - } - deployment, err = cli.NewCaddyDeployment(image, filter) + // Deploy to all machines without a placement constraint. + deployment, err = cli.NewCaddyDeployment(image, api.Placement{}) require.NoError(t, err) _, err = deployment.Run(ctx) @@ -379,26 +343,20 @@ func TestDeployment(t *testing.T) { svc, err = cli.InspectService(ctx, client.CaddyServiceName) require.NoError(t, err) - assert.Len(t, svc.Containers, 2) + assert.Len(t, svc.Containers, 3) + assertServiceMatchesSpec(t, svc, deployment.Spec) - // Existing container ctr0 on machine #0 should be left unchanged. - var ctr2 api.MachineServiceContainer - if ctr0.Container.ID == svc.Containers[0].Container.ID { - ctr2 = svc.Containers[1] - } else { - assert.Equal(t, ctr0.Container.ID, svc.Containers[1].Container.ID) - ctr2 = svc.Containers[0] - } - - machine2, err := cli.InspectMachine(ctx, ctr2.MachineID) - require.NoError(t, err) - assert.Equal(t, c.Machines[2].Name, machine2.Machine.Name) + // Initial container on machine #0 should be left unchanged. + machines := serviceMachines(svc) + assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine") + containers := serviceContainerIDs(svc) + assert.True(t, containers.Contains(initialContainerID), "Expected initial container to remain") }) t.Run("replicated", func(t *testing.T) { t.Parallel() - name := "replicated-deployment" + name := "test-replicated-deployment" t.Cleanup(func() { err := cli.RemoveService(ctx, name) if !errors.Is(err, api.ErrNotFound) { @@ -436,9 +394,9 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, spec) // Verify containers are on different machines for balanced distribution. - initialMachines := serviceMachines(t, svc) + initialMachines := serviceMachines(svc) assert.Len(t, initialMachines.ToSlice(), 2, "Expected 2 containers on 2 different machines") - initialContainers := serviceContainerIDs(t, svc) + initialContainers := serviceContainerIDs(svc) // 2. Update the service with a new configuration. init := true @@ -458,10 +416,10 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, updatedSpec) // Verify containers are on the same machines as before but the initial containers were replaced. - machines := serviceMachines(t, svc) + machines := serviceMachines(svc) assert.ElementsMatch(t, initialMachines.ToSlice(), machines.ToSlice(), "Expected containers on the same machines") - containers := serviceContainerIDs(t, svc) + containers := serviceContainerIDs(svc) assert.Empty(t, initialContainers.Intersect(containers).ToSlice(), "All existing containers should be replaced") @@ -485,9 +443,9 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, threeReplicaSpec) // Verify existing containers remain and a new one was added on a different machine. - machines = serviceMachines(t, svc) + machines = serviceMachines(svc) assert.Len(t, machines.ToSlice(), 3, "Expected 3 containers on 3 different machines") - containers = serviceContainerIDs(t, svc) + containers = serviceContainerIDs(svc) assert.Len(t, containers.Intersect(initialContainers).ToSlice(), 2, "Expected 2 initial containers to remain") // 4. Update to 5 replicas with a different configuration. @@ -510,13 +468,13 @@ func TestDeployment(t *testing.T) { assertServiceMatchesSpec(t, svc, fourReplicaSpec) // Verify all existing containers were replaced and new ones are evenly distributed. - machines = serviceMachines(t, svc) + machines = serviceMachines(svc) assert.Len(t, machines.ToSlice(), 3, "Expected containers on 3 different machines") - containers = serviceContainerIDs(t, svc) + containers = serviceContainerIDs(svc) assert.Empty(t, containers.Intersect(initialContainers).ToSlice(), "All existing containers should be replaced") - machineContainers := serviceContainersByMachine(t, svc) + machineContainers := serviceContainersByMachine(svc) for _, ctrs := range machineContainers { assert.LessOrEqual(t, len(ctrs), 2, "Expected at most 2 containers on each machine") } @@ -536,14 +494,14 @@ func TestDeployment(t *testing.T) { svc, err = cli.InspectService(ctx, name) require.NoError(t, err) - containers = serviceContainerIDs(t, svc) + containers = serviceContainerIDs(svc) assert.ElementsMatch(t, initialContainers.ToSlice(), containers.ToSlice()) }) - t.Run("replicated with machine filter", func(t *testing.T) { + t.Run("replicated with machine placement", func(t *testing.T) { t.Parallel() - name := "replicated-deployment-filtered" + name := "test-replicated-deployment-machine-placement" t.Cleanup(func() { err := cli.RemoveService(ctx, name) if !errors.Is(err, api.ErrNotFound) { @@ -551,21 +509,20 @@ func TestDeployment(t *testing.T) { } }) - // Create a replicated service with 2 replicas but limit to machines 0 and 1 + // Create a replicated service with 2 replicas but limit to machines 0 and 1. spec := api.ServiceSpec{ Name: name, Mode: api.ServiceModeReplicated, Container: api.ContainerSpec{ Image: "portainer/pause:latest", }, + Placement: api.Placement{ + Machines: []string{c.Machines[0].Name, c.Machines[1].Name}, + }, Replicas: 2, } - machine01Filter := func(m *pb.MachineInfo) bool { - return m.Name == c.Machines[0].Name || m.Name == c.Machines[1].Name - } - strategy := &deploy.RollingStrategy{MachineFilter: machine01Filter} - deployment := cli.NewDeployment(spec, strategy) + deployment := cli.NewDeployment(spec, nil) _, err = deployment.Run(ctx) require.NoError(t, err) @@ -573,27 +530,17 @@ func TestDeployment(t *testing.T) { // Verify service has 2 containers on machines 0 and 1. svc, err := cli.InspectService(ctx, name) require.NoError(t, err) + assertServiceMatchesSpec(t, svc, spec) assert.Len(t, svc.Containers, 2) - assert.NotEqual(t, svc.Containers[0].MachineID, svc.Containers[1].MachineID, - "Expected containers on different machines") - - machineNames := make(map[string]string) - for _, ctr := range svc.Containers { - machine, err := cli.InspectMachine(ctx, ctr.MachineID) - require.NoError(t, err) - machineNames[ctr.MachineID] = machine.Machine.Name - - // Should only be on machines 0 or 1. - assert.Contains(t, []string{c.Machines[0].Name, c.Machines[1].Name}, machine.Machine.Name) - } + machines := serviceMachines(svc) + assert.ElementsMatch(t, machines.ToSlice(), []string{c.Machines[0].ID, c.Machines[1].ID}) // Now update the filter to only allow machine 2 - machine2Filter := func(m *pb.MachineInfo) bool { - return m.Name == c.Machines[2].Name + spec.Placement = api.Placement{ + Machines: []string{c.Machines[2].Name}, } - strategy = &deploy.RollingStrategy{MachineFilter: machine2Filter} - deployment = cli.NewDeployment(spec, strategy) + deployment = cli.NewDeployment(spec, nil) _, err = deployment.Run(ctx) require.NoError(t, err) @@ -601,14 +548,11 @@ func TestDeployment(t *testing.T) { // Verify service now has containers only on machine 2. svc, err = cli.InspectService(ctx, name) require.NoError(t, err) + assertServiceMatchesSpec(t, svc, spec) assert.Len(t, svc.Containers, 2) // Still 2 replicas. - assert.Equal(t, svc.Containers[0].MachineID, svc.Containers[1].MachineID, - "Expected containers on the same machine") - - machine, err := cli.InspectMachine(ctx, svc.Containers[0].MachineID) - require.NoError(t, err) - assert.Equal(t, c.Machines[2].Name, machine.Machine.Name, "Containers should only be on machine #2") + machines = serviceMachines(svc) + assert.Equal(t, machines.ToSlice(), []string{c.Machines[2].ID}, "Expected containers on machine 2 only") }) // Deployments with volumes. @@ -880,7 +824,7 @@ func TestServiceLifecycle(t *testing.T) { Container: api.ContainerSpec{ Image: "portainer/pause:latest", }, - }, nil) + }) require.NoError(t, err) assert.NotEmpty(t, resp.ID) @@ -949,7 +893,7 @@ func TestServiceLifecycle(t *testing.T) { }, }, } - resp, err := cli.RunService(ctx, spec, nil) + resp, err := cli.RunService(ctx, spec) require.NoError(t, err) svc, err := cli.InspectService(ctx, resp.ID) @@ -979,7 +923,7 @@ func TestServiceLifecycle(t *testing.T) { Container: api.ContainerSpec{ Image: "portainer/pause:latest", }, - }, nil) + }) require.NoError(t, err) assert.NotEmpty(t, resp.ID)