diff --git a/internal/api/container.go b/internal/api/container.go index 16edc876..d1ef4aa3 100644 --- a/internal/api/container.go +++ b/internal/api/container.go @@ -69,6 +69,12 @@ func (c *Container) ServicePorts() ([]PortSpec, error) { return ports, nil } +func (c *Container) ServiceSpec() ServiceSpec { + // TODO: migrate api.Container type to use ContainerJSON to make it possible to construct + // a ServiceSpec from a Container. + return ServiceSpec{} +} + // runningStatusRegex matches the status string of a running container. // - "Up 3 minutes (healthy)" -> groups: ["Up 3 minutes (healthy)", "healthy"] // - "Up 5 seconds" -> groups: ["Up 5 seconds", ""] diff --git a/internal/api/service.go b/internal/api/service.go index 8db0d496..6d36f849 100644 --- a/internal/api/service.go +++ b/internal/api/service.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "github.com/distribution/reference" + "reflect" "uncloud/internal/machine/api/pb" ) @@ -37,6 +38,10 @@ func (s *ServiceSpec) Validate() error { return nil } +func (s *ServiceSpec) Equals(spec ServiceSpec) bool { + return reflect.DeepEqual(*s, spec) +} + type ContainerSpec struct { Command []string Image string diff --git a/internal/cli/client/container.go b/internal/cli/client/container.go index 0b386251..0d8317eb 100644 --- a/internal/cli/client/container.go +++ b/internal/cli/client/container.go @@ -23,6 +23,10 @@ func (cli *Client) CreateContainer( ) (container.CreateResponse, error) { var resp container.CreateResponse + if serviceID == "" { + return resp, errors.New("service ID is required") + } + machine, err := cli.InspectMachine(ctx, machineID) if err != nil { return resp, fmt.Errorf("inspect machine '%s': %w", machineID, err) diff --git a/internal/cli/client/deploy.go b/internal/cli/client/deploy.go index 073bbf8b..31aac9f7 100644 --- a/internal/cli/client/deploy.go +++ b/internal/cli/client/deploy.go @@ -7,20 +7,6 @@ import ( "uncloud/internal/api" ) -// Strategy defines how a service should be deployed or updated. Different implementations can provide various -// deployment patterns such as rolling updates, blue/green deployments, etc. -type Strategy interface { - // 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 *Client, svc *api.Service, spec api.ServiceSpec) (Operation, error) -} - -// Operation represents a single atomic operation in a deployment process. -// Operations can be composed to form complex deployment strategies. -type Operation interface { - Execute(ctx context.Context, cli *Client) error -} - // Deployment manages the process of creating or updating a service to match a desired state. // It coordinates the validation, planning, and execution of deployment operations. type Deployment struct { @@ -85,7 +71,6 @@ func (d *Deployment) Validate(ctx context.Context) error { return nil } - fmt.Printf("Service: %v\n", d.Service) if d.Service.Name != d.Spec.Name { return errors.New("service name cannot be changed") } @@ -99,79 +84,10 @@ func (d *Deployment) Validate(ctx context.Context) error { // Run executes the deployment plan. It will create a new plan if one hasn't been created yet. // The deployment will either create a new service or update an existing one to match the desired specification. func (d *Deployment) Run(ctx context.Context) error { - // TODO: create or get the plan, and run it. - return nil -} - -// RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time -// to minimize service disruption. -type RollingStrategy struct{} - -func (s *RollingStrategy) Plan( - ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, -) (Operation, error) { - switch spec.Mode { - case "", api.ServiceModeReplicated: - return s.planReplicated(ctx, cli, svc, spec) - case api.ServiceModeGlobal: - return s.planGlobal(ctx, cli, svc, spec) - default: - return nil, fmt.Errorf("unsupported service mode: %s", spec.Mode) + plan, err := d.Plan(ctx) + if err != nil { + return fmt.Errorf("plan: %w", err) } -} -// planReplicated creates a plan for a replicated service deployment. -func (s *RollingStrategy) planReplicated( - ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, -) (Operation, error) { - return nil, errors.New("not implemented") -} - -// planGlobal creates a plan for a global service deployment. -func (s *RollingStrategy) planGlobal( - ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, -) (Operation, error) { - // TODO - // - Prepare a map of machineID to the current container - // - Fetch a list of existing machines - // - For each machine, check if there is a container on it - // - If doesn't exist, add a RunContainerOperation - // - If exists, check if there are host port bindings that conflict with the new spec - // - If there are conflicts, add a RemoveContainerOperation and a RunContainerOperation - // - If there are no conflicts, add a RunContainerOperation and a RemoveContainerOperation - return nil, errors.New("not implemented") -} - -// RunContainerOperation creates and starts a new container on a specific machine. -type RunContainerOperation struct { - Spec api.ServiceSpec - MachineID string -} - -func (o *RunContainerOperation) Execute(ctx context.Context, cli *Client) error { - return nil -} - -// RemoveContainerOperation stops and removes a container from a specific machine. -type RemoveContainerOperation struct { - ContainerID string - MachineID string -} - -func (o *RemoveContainerOperation) Execute(ctx context.Context, cli *Client) error { - return nil -} - -// SequenceOperation is a composite operation that executes a sequence of operations in order. -type SequenceOperation struct { - Operations []Operation -} - -func (o *SequenceOperation) Execute(ctx context.Context, cli *Client) error { - for _, op := range o.Operations { - if err := op.Execute(ctx, cli); err != nil { - return err - } - } - return nil + return plan.Execute(ctx, d.cli) } diff --git a/internal/cli/client/operation.go b/internal/cli/client/operation.go new file mode 100644 index 00000000..4833b36d --- /dev/null +++ b/internal/cli/client/operation.go @@ -0,0 +1,104 @@ +package client + +import ( + "context" + "fmt" + "github.com/docker/docker/api/types/container" + "strings" + "uncloud/internal/api" +) + +// Operation represents a single atomic operation in a deployment process. +// Operations can be composed to form complex deployment strategies. +type Operation interface { + Execute(ctx context.Context, cli *Client) error + String() string +} + +// RunContainerOperation creates and starts a new container on a specific machine. +type RunContainerOperation struct { + ServiceID string + Spec api.ServiceSpec + MachineID string +} + +func (o *RunContainerOperation) Execute(ctx context.Context, cli *Client) error { + resp, err := cli.CreateContainer(ctx, o.ServiceID, o.Spec, o.MachineID) + if err != nil { + return fmt.Errorf("create container: %w", err) + } + if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil { + return fmt.Errorf("start container: %w", err) + } + + // TODO: wait for the container to become healthy + + return nil +} + +func (o *RunContainerOperation) String() string { + return fmt.Sprintf("RunContainerOperation[%s, %s, %s]", o.ServiceID, o.Spec.Name, o.MachineID) +} + +// StopContainerOperation stops a container on a specific machine. +type StopContainerOperation struct { + ServiceID string + ContainerID string + MachineID string +} + +func (o *StopContainerOperation) Execute(ctx context.Context, cli *Client) error { + if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, container.StopOptions{}); err != nil { + return fmt.Errorf("stop container: %w", err) + } + return nil +} + +func (o *StopContainerOperation) String() string { + return fmt.Sprintf("StopContainerOperation[%s, %s, %s]", o.ServiceID, o.ContainerID, o.MachineID) +} + +// RemoveContainerOperation stops and removes a container from a specific machine. +type RemoveContainerOperation struct { + ServiceID string + ContainerID string + MachineID string +} + +func (o *RemoveContainerOperation) Execute(ctx context.Context, cli *Client) error { + if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, container.StopOptions{}); err != nil { + return fmt.Errorf("stop container: %w", err) + } + if err := cli.RemoveContainer(ctx, o.ServiceID, o.ContainerID, container.RemoveOptions{}); err != nil { + return fmt.Errorf("remove container: %w", err) + } + + return nil +} + +func (o *RemoveContainerOperation) String() string { + return fmt.Sprintf("RemoveContainerOperation[%s, %s, %s]", o.ServiceID, o.ContainerID, o.MachineID) +} + +// SequenceOperation is a composite operation that executes a sequence of operations in order. +type SequenceOperation struct { + Operations []Operation +} + +func (o *SequenceOperation) Execute(ctx context.Context, cli *Client) error { + for _, op := range o.Operations { + if err := op.Execute(ctx, cli); err != nil { + return err + } + } + return nil +} + +func (o *SequenceOperation) String() string { + ops := make([]string, len(o.Operations)) + for i, op := range o.Operations { + ops[i] = op.String() + } + + return fmt.Sprintf("SequenceOperation[%s]", strings.Join(ops, ", ")) +} diff --git a/internal/cli/client/strategy.go b/internal/cli/client/strategy.go new file mode 100644 index 00000000..72063e9b --- /dev/null +++ b/internal/cli/client/strategy.go @@ -0,0 +1,177 @@ +package client + +import ( + "context" + "errors" + "fmt" + "uncloud/internal/api" + "uncloud/internal/machine/api/pb" + "uncloud/internal/secret" +) + +// Strategy defines how a service should be deployed or updated. Different implementations can provide various +// deployment patterns such as rolling updates, blue/green deployments, etc. +type Strategy interface { + // 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 *Client, svc *api.Service, spec api.ServiceSpec) (Operation, error) +} + +// RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time +// to minimize service disruption. +type RollingStrategy struct{} + +func (s *RollingStrategy) Plan( + ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, +) (Operation, error) { + switch spec.Mode { + case "", api.ServiceModeReplicated: + return s.planReplicated(ctx, cli, svc, spec) + case api.ServiceModeGlobal: + return s.planGlobal(ctx, cli, svc, spec) + default: + return nil, fmt.Errorf("unsupported service mode: %s", spec.Mode) + } +} + +// planReplicated creates a plan for a replicated service deployment. +func (s *RollingStrategy) planReplicated( + ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, +) (Operation, error) { + return nil, errors.New("not implemented") +} + +// planGlobal creates a plan for a global service deployment, ensuring one container runs on each available machine. +// For machines with an existing container, it attempts to start a new container before removing the old one if +// possible. If the new container would have port conflicts with the existing one, the old container is removed first. +// 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 *Client, svc *api.Service, spec api.ServiceSpec, +) (Operation, error) { + serviceID := "" + // Map machineID to service containers on that machine. For the global mode, there should be at most one + // container per machine but we use a slice to handle multiple containers that may exist due to a bug + // or interruption in the previous deployment. + containersOnMachine := make(map[string][]api.MachineContainer) + if svc != nil { + serviceID = svc.ID + for _, c := range svc.Containers { + containersOnMachine[c.MachineID] = append(containersOnMachine[c.MachineID], c) + } + } else { + // Generate a new service ID for the first service deployment. + var err error + serviceID, err = secret.NewID() + if err != nil { + return nil, fmt.Errorf("generate service ID: %w", err) + } + } + + machines, err := cli.ListMachines(ctx) + if err != nil { + return nil, fmt.Errorf("list machines: %w", err) + } + + plan := &SequenceOperation{} + 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) + continue + } + + containers := containersOnMachine[m.Machine.Id] + ops, err := reconcileGlobalContainer(containers, spec, serviceID, m.Machine.Id) + if err != nil { + return nil, err + } + plan.Operations = append(plan.Operations, ops...) + } + + return plan, nil +} + +func reconcileGlobalContainer( + containers []api.MachineContainer, spec api.ServiceSpec, serviceID, machineID string, +) ([]Operation, error) { + var ops []Operation + + if len(containers) == 0 { + // No containers on this machine, create a new one. + ops = append(ops, &RunContainerOperation{ + ServiceID: serviceID, + Spec: spec, + MachineID: machineID, + }) + return ops, nil + } + + // Check if there is a container with the same spec already running. If so, remove the rest. + upToDate := false + for i, c := range containers { + if c.Container.State != api.StateRunning && c.Container.State != api.StateRestarting { + // Skip containers that are not running. + continue + } + + svcSpec := c.Container.ServiceSpec() + if svcSpec.Equals(spec) { + // The container is already running with the same spec. + upToDate = true + for j, old := range containers { + if i == j { + continue + } + ops = append(ops, &RemoveContainerOperation{ + ServiceID: serviceID, + ContainerID: old.Container.ID, + MachineID: old.MachineID, + }) + } + break + } + } + if upToDate { + return ops, nil + } + + // The machine has containers but none of them match the new spec. + // Stop the old non-stopped containers that have conflicting ports with the new spec before running a new one. + for _, c := range containers { + if !c.Container.Stopped() { + conflictingPorts, err := c.Container.ConflictingServicePorts(spec.Ports) + if err != nil { + return nil, fmt.Errorf("check conflicting ports: %w", err) + } + + if len(conflictingPorts) > 0 { + // Stop the running container with conflicting ports. + ops = append(ops, &StopContainerOperation{ + ServiceID: serviceID, + ContainerID: c.Container.ID, + MachineID: c.MachineID, + }) + } + } + } + + // Run a new container. + ops = append(ops, &RunContainerOperation{ + ServiceID: serviceID, + Spec: spec, + MachineID: machineID, + }) + + // Remove the old containers. + for _, c := range containers { + ops = append(ops, &RemoveContainerOperation{ + ServiceID: serviceID, + ContainerID: c.Container.ID, + MachineID: c.MachineID, + }) + } + + return ops, nil +} diff --git a/internal/machine/cluster/cluster.go b/internal/machine/cluster/cluster.go index 75bfe459..4c9fdb0e 100644 --- a/internal/machine/cluster/cluster.go +++ b/internal/machine/cluster/cluster.go @@ -210,9 +210,13 @@ func (c *Cluster) ListMachines(ctx context.Context, _ *emptypb.Empty) (*pb.ListM state := pb.MachineMember_DOWN addr, _ := m.Network.ManagementIp.ToAddr() for _, s := range states { - if s.Addr.Addr().Compare(addr) == 0 && - (s.State == corrosion.MembershipStateAlive || s.State == corrosion.MembershipStateSuspect) { - state = pb.MachineMember_UP + if s.Addr.Addr().Compare(addr) == 0 { + switch s.State { + case corrosion.MembershipStateAlive: + state = pb.MachineMember_UP + case corrosion.MembershipStateSuspect: + state = pb.MachineMember_SUSPECT + } break } } diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index d72fc529..315007a7 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -3,6 +3,7 @@ package e2e import ( "context" "errors" + "fmt" "github.com/docker/docker/api/types/container" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -12,6 +13,147 @@ import ( "uncloud/internal/ucind" ) +func TestDeployment(t *testing.T) { + t.Parallel() + + clusterName := "ucind-test.deployment" + ctx := context.Background() + c, _ := createTestCluster(t, clusterName, ucind.CreateClusterOptions{Machines: 3}, true) + + cli, err := c.Machines[0].Connect(ctx) + require.NoError(t, err) + + t.Run("global", func(t *testing.T) { + t.Parallel() + + name := "global-deployment" + t.Cleanup(func() { + err := cli.RemoveService(ctx, name) + if errors.Is(err, client.ErrNotFound) { + require.NoError(t, err) + } + + _, err = cli.InspectService(ctx, name) + require.ErrorIs(t, err, client.ErrNotFound) + }) + + deploy, err := cli.NewDeployment(api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeGlobal, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + }, + }, nil) + require.NoError(t, err) + + err = deploy.Validate(ctx) + require.NoError(t, err) + + plan, err := deploy.Plan(ctx) + require.NoError(t, err) + assert.IsType(t, &client.SequenceOperation{}, plan) + assert.Len(t, plan.(*client.SequenceOperation).Operations, 3) // 3 run + fmt.Println("# First plan:", plan) + + err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err := cli.InspectService(ctx, name) + require.NoError(t, err) + assert.Equal(t, name, svc.Name) + assert.Equal(t, api.ServiceModeGlobal, svc.Mode) + assert.Len(t, svc.Containers, 3) + + // Deploy a published port. + deploy, err = cli.NewDeployment(api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeGlobal, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + }, + Ports: []api.PortSpec{ + { + PublishedPort: 8000, + ContainerPort: 8000, + Protocol: api.ProtocolTCP, + Mode: api.PortModeHost, + }, + }, + }, nil) + require.NoError(t, err) + + plan, err = deploy.Plan(ctx) + require.NoError(t, err) + assert.IsType(t, &client.SequenceOperation{}, plan) + assert.Len(t, plan.(*client.SequenceOperation).Operations, 6) // 3 run + 3 remove + fmt.Println("# Second plan:", plan) + + err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err = cli.InspectService(ctx, name) + require.NoError(t, err) + assert.Equal(t, name, svc.Name) + assert.Equal(t, api.ServiceModeGlobal, svc.Mode) + assert.Len(t, svc.Containers, 3) + + // Deploy the same conflicting port but with container spec changes + init := true + spec := api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeGlobal, + Container: api.ContainerSpec{ + Image: "portainer/pause:latest", + Init: &init, + }, + Ports: []api.PortSpec{ + { + PublishedPort: 8000, + ContainerPort: 8000, + Protocol: api.ProtocolTCP, + Mode: api.PortModeHost, + }, + }, + } + deploy, err = cli.NewDeployment(spec, nil) + require.NoError(t, err) + + plan, err = deploy.Plan(ctx) + require.NoError(t, err) + assert.IsType(t, &client.SequenceOperation{}, plan) + assert.Len(t, plan.(*client.SequenceOperation).Operations, 9) // 3 stop + 3 run + 3 remove + fmt.Println("# Third plan:", plan) + + err = deploy.Run(ctx) + require.NoError(t, err) + + svc, err = cli.InspectService(ctx, name) + require.NoError(t, err) + assert.Equal(t, name, svc.Name) + assert.Equal(t, api.ServiceModeGlobal, svc.Mode) + assert.Len(t, svc.Containers, 3) + + // Deploying the same spec should be a no-op. + //deploy, err = cli.NewDeployment(spec, nil) + //require.NoError(t, err) + // + //plan, err = deploy.Plan(ctx) + //require.NoError(t, err) + //assert.IsType(t, &client.SequenceOperation{}, plan) + //assert.Len(t, plan.(*client.SequenceOperation).Operations, 0) // no-op + //fmt.Println("# Forth plan:", plan) + // + //err = deploy.Run(ctx) + //require.NoError(t, err) + // + //svc, err = cli.InspectService(ctx, name) + //require.NoError(t, err) + //assert.Equal(t, name, svc.Name) + //assert.Equal(t, api.ServiceModeGlobal, svc.Mode) + //assert.Len(t, svc.Containers, 3) + }) +} + func TestRunService(t *testing.T) { t.Parallel()