From e13408d79fbd802c09fb3dbb6f7860b7362a5929 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Mon, 17 Feb 2025 20:26:40 +1000 Subject: [PATCH] feat: allow to filter machines for global service deployments --- cmd/uncloud/caddy/deploy.go | 25 +++- internal/cli/client/caddy.go | 4 +- internal/cli/client/caddy_test.go | 4 +- internal/cli/client/deploy.go | 5 + internal/cli/client/strategy.go | 12 +- test/e2e/service_test.go | 187 +++++++++++++++++++++++++++++- 6 files changed, 227 insertions(+), 10 deletions(-) diff --git a/cmd/uncloud/caddy/deploy.go b/cmd/uncloud/caddy/deploy.go index b4928335..1e799b80 100644 --- a/cmd/uncloud/caddy/deploy.go +++ b/cmd/uncloud/caddy/deploy.go @@ -12,10 +12,12 @@ import ( "strings" "uncloud/internal/cli" "uncloud/internal/cli/client" + "uncloud/internal/machine/api/pb" ) type deployOptions struct { image string + machine string cluster string } @@ -35,6 +37,8 @@ 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().StringVarP( &opts.cluster, "cluster", "c", "", "Name of the cluster to deploy to. (default is the current cluster)", @@ -80,7 +84,17 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { fmt.Println() fmt.Println("Preparing a deployment plan...") - d, err := clusterClient.NewCaddyDeployment(opts.image) + + var filter client.MachineFilter + if opts.machine != "" { + machines := strings.Split(opts.machine, ",") + for i, m := range machines { + machines[i] = strings.TrimSpace(m) + } + filter = machineFilter(machines) + } + + d, err := clusterClient.NewCaddyDeployment(opts.image, filter) if err != nil { return fmt.Errorf("create caddy deployment: %w", err) } @@ -161,3 +175,12 @@ func confirm() (bool, error) { return confirmed, nil } + +func machineFilter(machines []string) client.MachineFilter { + if len(machines) == 0 { + return nil + } + return func(m *pb.MachineInfo) bool { + return slices.Contains(machines, m.Name) + } +} diff --git a/internal/cli/client/caddy.go b/internal/cli/client/caddy.go index ed8f7c72..f08e5c40 100644 --- a/internal/cli/client/caddy.go +++ b/internal/cli/client/caddy.go @@ -22,7 +22,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) (*Deployment, error) { +func (cli *Client) NewCaddyDeployment(image string, filter MachineFilter) (*Deployment, error) { latest, err := latestCaddyImage() if err != nil { return nil, fmt.Errorf("look up latest Caddy image: %w", err) @@ -56,7 +56,7 @@ func (cli *Client) NewCaddyDeployment(image string) (*Deployment, error) { }, } - return cli.NewDeployment(spec, &RollingStrategy{}) + return cli.NewDeployment(spec, &RollingStrategy{MachineFilter: filter}) } // latestCaddyImage returns the latest image of the official Caddy Docker image on Docker Hub. diff --git a/internal/cli/client/caddy_test.go b/internal/cli/client/caddy_test.go index 449f20e6..83e7998f 100644 --- a/internal/cli/client/caddy_test.go +++ b/internal/cli/client/caddy_test.go @@ -15,7 +15,7 @@ func TestClient_NewCaddyDeployment(t *testing.T) { t.Run("latest image from Docker Hub", func(t *testing.T) { t.Parallel() - deploy, err := cli.NewCaddyDeployment("") + deploy, err := cli.NewCaddyDeployment("", nil) require.NoError(t, err) assert.Equal(t, "caddy", deploy.Spec.Name) @@ -42,7 +42,7 @@ func TestClient_NewCaddyDeployment(t *testing.T) { t.Parallel() image := "my-caddy:1.2.3" - deploy, err := cli.NewCaddyDeployment(image) + deploy, err := cli.NewCaddyDeployment(image, nil) require.NoError(t, err) assert.Equal(t, "caddy", deploy.Spec.Name) diff --git a/internal/cli/client/deploy.go b/internal/cli/client/deploy.go index 1dc23341..652191b6 100644 --- a/internal/cli/client/deploy.go +++ b/internal/cli/client/deploy.go @@ -7,6 +7,7 @@ import ( "github.com/distribution/reference" "strings" "uncloud/internal/api" + "uncloud/internal/machine/api/pb" "uncloud/internal/secret" ) @@ -25,6 +26,10 @@ type Plan struct { SequenceOperation } +// MachineFilter determines which machines participate in a deployment operation by returning true for +// machines that should be included. +type MachineFilter func(m *pb.MachineInfo) bool + // NewDeployment creates a new deployment for the given service specification. // If strategy is nil, a default RollingStrategy will be used. func (cli *Client) NewDeployment(spec api.ServiceSpec, strategy Strategy) (*Deployment, error) { diff --git a/internal/cli/client/strategy.go b/internal/cli/client/strategy.go index c1542480..9b577c7c 100644 --- a/internal/cli/client/strategy.go +++ b/internal/cli/client/strategy.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "uncloud/internal/api" "uncloud/internal/machine/api/pb" "uncloud/internal/secret" @@ -19,7 +20,10 @@ type Strategy interface { // RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time // to minimize service disruption. -type RollingStrategy struct{} +type RollingStrategy struct { + // MachineFilter optionally restricts which machines participate in this deployment. + MachineFilter MachineFilter +} func (s *RollingStrategy) Plan( ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, @@ -72,6 +76,12 @@ func (s *RollingStrategy) planGlobal( if err != nil { return plan, fmt.Errorf("list machines: %w", err) } + // Filter machines if a machine filter is provided. + if s.MachineFilter != nil { + machines = slices.DeleteFunc(machines, func(m *pb.MachineMember) bool { + return !s.MachineFilter(m.Machine) + }) + } // TODO: figure out how to return a warning if there are machines down. Embed the machinesDown in the plan? // WARNING: failed to run a service container on machine '%s' which is Down. diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index fa880e0e..62f8c26c 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -9,6 +9,7 @@ import ( "testing" "uncloud/internal/api" "uncloud/internal/cli/client" + "uncloud/internal/machine/api/pb" "uncloud/internal/ucind" ) @@ -67,6 +68,12 @@ func TestDeployment(t *testing.T) { require.NoError(t, err) assert.True(t, svcSpec.Equals(spec)) + machines := make(map[string]struct{}) + for _, ctr := range svc.Containers { + machines[ctr.MachineID] = struct{}{} + } + assert.Len(t, machines, 3, "expected 1 container on each machine") + // Deploy a published port. specWithPort := api.ServiceSpec{ Name: name, @@ -162,10 +169,10 @@ func TestDeployment(t *testing.T) { assert.Len(t, svc.Containers, 3) }) - t.Run("caddy", func(t *testing.T) { + t.Run("global with machine filter", func(t *testing.T) { t.Parallel() - name := "caddy" + name := "global-deployment-filtered" t.Cleanup(func() { err := cli.RemoveService(ctx, name) if errors.Is(err, client.ErrNotFound) { @@ -173,7 +180,16 @@ func TestDeployment(t *testing.T) { } }) - deploy, err := cli.NewCaddyDeployment("") + // First deploy globally without filter to get containers on all machines. + spec := api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeGlobal, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + }, + } + + deploy, err := cli.NewDeployment(spec, nil) require.NoError(t, err) _, err = deploy.Run(ctx) @@ -181,7 +197,113 @@ func TestDeployment(t *testing.T) { svc, err := cli.InspectService(ctx, name) require.NoError(t, err) - assert.Equal(t, name, svc.Name) + assert.Len(t, svc.Containers, 3, "expected 1 container on each machine") + + // Store initial container IDs by machine. + initialContainers := make(map[string]string) // machineID -> containerID + for _, ctr := range svc.Containers { + initialContainers[ctr.MachineID] = ctr.Container.ID + } + + // Update spec with Init=true, but only deploy to machines #0 and #2. + init := true + specWithInit := spec + specWithInit.Container.Init = &init + + filter := func(m *pb.MachineInfo) bool { + return m.Name == c.Machines[0].Name || m.Name == c.Machines[2].Name + } + strategy := &client.RollingStrategy{MachineFilter: filter} + + deploy, err = cli.NewDeployment(specWithInit, strategy) + require.NoError(t, err) + + _, err = deploy.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") + + // 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) + + 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) + + svcSpec, err := 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 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, + }, + } + + deploy, err = cli.NewDeployment(specWithPort, nil) + require.NoError(t, err) + + _, err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err = cli.InspectService(ctx, name) + require.NoError(t, err) + assert.Len(t, svc.Containers, 3) + + // Verify all containers are updated with a published port. + for _, ctr := range svc.Containers { + svcSpec, err := ctr.Container.ServiceSpec() + require.NoError(t, err) + assert.Nil(t, svcSpec.Container.Init, + "Container on machine %s should have init disabled", ctr.MachineID) + + 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) + } + }) + + t.Run("caddy", func(t *testing.T) { + t.Cleanup(func() { + err := cli.RemoveService(ctx, client.CaddyServiceName) + if errors.Is(err, client.ErrNotFound) { + require.NoError(t, err) + } + }) + + deploy, err := cli.NewCaddyDeployment("", nil) + require.NoError(t, err) + + _, err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err := cli.InspectService(ctx, client.CaddyServiceName) + require.NoError(t, err) + assert.Equal(t, client.CaddyServiceName, svc.Name) assert.Equal(t, api.ServiceModeGlobal, svc.Mode) assert.Len(t, svc.Containers, 3) @@ -206,6 +328,63 @@ func TestDeployment(t *testing.T) { } assert.Equal(t, expectedPorts, ports) }) + + t.Run("caddy with machine filter", func(t *testing.T) { + t.Cleanup(func() { + err := cli.RemoveService(ctx, client.CaddyServiceName) + if errors.Is(err, client.ErrNotFound) { + require.NoError(t, err) + } + }) + + // Deploy to machine #0 + filter := func(m *pb.MachineInfo) bool { + return m.Name == c.Machines[0].Name + } + + deploy, err := cli.NewCaddyDeployment("", filter) + require.NoError(t, err) + + _, err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err := cli.InspectService(ctx, client.CaddyServiceName) + require.NoError(t, err) + assert.Len(t, svc.Containers, 1) + ctr0 := svc.Containers[0] + + machine0, err := cli.InspectMachine(ctx, ctr0.MachineID) + require.NoError(t, err) + assert.Equal(t, c.Machines[0].Name, machine0.Machine.Name) + + // 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 + } + + deploy, err = cli.NewCaddyDeployment("", filter) + require.NoError(t, err) + + _, err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err = cli.InspectService(ctx, client.CaddyServiceName) + require.NoError(t, err) + assert.Len(t, svc.Containers, 2) + + // Existing container ctr0 on machine #0 should be left unchanged. + var ctr2 api.MachineContainer + 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) + }) } func TestRunService(t *testing.T) {