philipp/uncloud
1
0
Fork 0
mirror of https://github.com/psviderski/uncloud.git synced 2026-08-26 19:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: allow to filter machines for global service deployments

Browse Source
This commit is contained in:
Pavel Sviderski
2025-02-17 20:26:40 +10:00
parent 8610322682
commit e13408d79f
6 changed files with 227 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/uncloud/caddy/deploy.go
+24 -1
View File
@@ -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)
}
}
internal/cli/client/caddy.go
+2 -2
View File
@@ -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.
internal/cli/client/caddy_test.go
+2 -2
View File
@@ -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)
internal/cli/client/deploy.go
+5
View File
@@ -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) {
internal/cli/client/strategy.go
+11 -1
View File
@@ -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.
test/e2e/service_test.go
+183 -4
View File
@@ -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) {