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

refactor: deploy global service using Deployment with RollingStrategy

Browse Source
This commit is contained in:
Pavel Sviderski
2025-02-14 20:45:23 +10:00
parent a53fa2c1d8
commit c4398f9f06
7 changed files with 111 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cli/client/container.go
+4 -4
View File
@@ -233,7 +233,7 @@ func (cli *Client) InspectContainer(ctx context.Context, serviceID, containerID
}
for _, c := range svc.Containers {
if c.Container.ID == containerID || c.Container.Name == containerID {
if c.Container.ID == containerID || c.Container.NameWithoutSlash() == containerID {
ctr = c
}
}
@@ -258,7 +258,7 @@ func (cli *Client) StartContainer(ctx context.Context, serviceID, containerID st
ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.Name, machine.Machine.Name)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.NameWithoutSlash(), machine.Machine.Name)
pw.Event(progress.StartingEvent(eventID))
if err = cli.Docker.StartContainer(ctx, ctr.Container.ID, container.StartOptions{}); err != nil {
@@ -285,7 +285,7 @@ func (cli *Client) StopContainer(
ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.Name, machine.Machine.Name)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.NameWithoutSlash(), machine.Machine.Name)
pw.Event(progress.StoppingEvent(eventID))
if err = cli.Docker.StopContainer(ctx, ctr.Container.ID, opts); err != nil {
@@ -312,7 +312,7 @@ func (cli *Client) RemoveContainer(
ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.Name, machine.Machine.Name)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.NameWithoutSlash(), machine.Machine.Name)
pw.Event(progress.RemovingEvent(eventID))
if err = cli.Docker.RemoveContainer(ctx, ctr.Container.ID, opts); err != nil {
internal/cli/client/deploy.go
+49 -13
View File
@@ -4,7 +4,10 @@ import (
"context"
"errors"
"fmt"
"github.com/distribution/reference"
"strings"
"uncloud/internal/api"
"uncloud/internal/secret"
)
// Deployment manages the process of creating or updating a service to match a desired state.
@@ -14,12 +17,40 @@ type Deployment struct {
Spec api.ServiceSpec
Strategy Strategy
cli *Client
plan Operation
plan *Plan
}
type Plan struct {
ServiceID string
Operation
}
// 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) {
if err := spec.Validate(); err != nil {
return nil, fmt.Errorf("invalid service spec: %w", err)
}
if spec.Name == "" {
// Generate a random service name from the image when not provided.
img, err := reference.ParseDockerRef(spec.Container.Image)
if err != nil {
return nil, fmt.Errorf("invalid image: %w", err)
}
// Get the image name without the repository and tag/digest parts.
imageName := reference.FamiliarName(img)
// Get the last part of the image name (path), e.g. "nginx" from "bitnami/nginx".
if i := strings.LastIndex(imageName, "/"); i != -1 {
imageName = imageName[i+1:]
}
// Append a random suffix to the image name to generate an optimistically unique service name.
suffix, err := secret.RandomAlphaNumeric(4)
if err != nil {
return nil, fmt.Errorf("generate random suffix: %w", err)
}
spec.Name = fmt.Sprintf("%s-%s", imageName, suffix)
}
if strategy == nil {
strategy = &RollingStrategy{}
}
@@ -33,21 +64,21 @@ func (cli *Client) NewDeployment(spec api.ServiceSpec, strategy Strategy) (*Depl
// Plan returns a plan of operations to reconcile the service to the desired state.
// If a plan has already been created, the same plan will be returned.
func (d *Deployment) Plan(ctx context.Context) (Operation, error) {
func (d *Deployment) Plan(ctx context.Context) (Plan, error) {
if d.plan != nil {
return d.plan, nil
return *d.plan, nil
}
// Validate the new spec before planning.
if err := d.Validate(ctx); err != nil {
return nil, fmt.Errorf("invalid deployment: %w", err)
return Plan{}, fmt.Errorf("invalid deployment: %w", err)
}
plan, err := d.Strategy.Plan(ctx, d.cli, d.Service, d.Spec)
if err != nil {
return nil, fmt.Errorf("create plan using %T: %w", d.Strategy, err)
return Plan{}, fmt.Errorf("create plan using %T: %w", d.Strategy, err)
}
d.plan = plan
d.plan = &plan
return plan, nil
}
@@ -57,8 +88,11 @@ func (d *Deployment) Validate(ctx context.Context) error {
if err := d.Spec.Validate(); err != nil {
return fmt.Errorf("invalid service spec: %w", err)
}
if d.Spec.Name == "" {
return errors.New("service name is required")
}
if d.Service == nil && d.Spec.Name != "" {
if d.Service == nil {
svc, err := d.cli.InspectService(ctx, d.Spec.Name)
if err == nil {
d.Service = &svc
@@ -66,7 +100,7 @@ func (d *Deployment) Validate(ctx context.Context) error {
return fmt.Errorf("inspect service: %w", err)
}
}
// d.Service will be nil if the service doesn't exist yet (first deployment).
// d.Service is nil if the service doesn't exist yet (first deployment).
if d.Service == nil {
return nil
}
@@ -81,13 +115,15 @@ func (d *Deployment) Validate(ctx context.Context) error {
return nil
}
// 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 {
// Run executes the deployment plan and returns the ID of the created or updated service.
// It will create a new plan if one hasn't been created yet. The deployment will either create a new service or update
// the existing one to match the desired specification.
// TODO: forbid to run the same deployment more than once.
func (d *Deployment) Run(ctx context.Context) (string, error) {
plan, err := d.Plan(ctx)
if err != nil {
return fmt.Errorf("plan: %w", err)
return "", fmt.Errorf("plan: %w", err)
}
return plan.Execute(ctx, d.cli)
return plan.ServiceID, plan.Execute(ctx, d.cli)
}
internal/cli/client/service.go
+18 -71
View File
@@ -20,14 +20,8 @@ import (
)
type RunServiceResponse struct {
ID string
Name string
Containers []MachineContainerID
}
type MachineContainerID struct {
MachineID string
ContainerID string
ID string
Name string
}
func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunServiceResponse, error) {
@@ -77,7 +71,21 @@ func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunSer
case "", api.ServiceModeReplicated:
resp, err = cli.runReplicatedService(ctx, serviceID, spec)
case api.ServiceModeGlobal:
resp, err = cli.runGlobalService(ctx, serviceID, spec)
deploy, err := cli.NewDeployment(spec, &RollingStrategy{})
if err != nil {
return fmt.Errorf("create deployment: %w", err)
}
serviceID, err = deploy.Run(ctx)
if err != nil {
return err
}
resp.ID = serviceID
// TODO: get the service name from the plan when it's available.
resp.Name = spec.Name
return nil
default:
return fmt.Errorf("invalid mode: %q", spec.Mode)
}
@@ -120,16 +128,10 @@ func (cli *Client) runReplicatedService(ctx context.Context, id string, spec api
return resp, errors.New("no available machine to run the service")
}
runResp, err := cli.runContainer(ctx, id, spec, m.Machine)
if err != nil {
if _, err = cli.runContainer(ctx, id, spec, m.Machine); err != nil {
return resp, fmt.Errorf("run container: %w", err)
}
resp.Containers = append(resp.Containers, MachineContainerID{
MachineID: m.Machine.Id,
ContainerID: runResp.ID,
})
return resp, nil
}
@@ -150,61 +152,6 @@ func firstAvailableMachine(machines []*pb.MachineMember) *pb.MachineMember {
return nil
}
func (cli *Client) runGlobalService(ctx context.Context, id string, spec api.ServiceSpec) (RunServiceResponse, error) {
resp := RunServiceResponse{
ID: id,
Name: spec.Name,
}
machines, err := cli.ListMachines(ctx)
if err != nil {
return resp, fmt.Errorf("list machines: %w", err)
}
wg := sync.WaitGroup{}
errCh := make(chan error)
mu := sync.Mutex{}
// Run a service container on each available machine.
for _, m := range machines {
if m.State != pb.MachineMember_UP && m.State != pb.MachineMember_SUSPECT {
// TODO: return failed machines in the response.
fmt.Printf("WARNING: failed to run a service container on machine '%s' which is Down.\n", m.Machine.Name)
continue
}
wg.Add(1)
go func() {
defer wg.Done()
runResp, err := cli.runContainer(ctx, id, spec, m.Machine)
if err != nil {
errCh <- fmt.Errorf("run container on machine '%s': %w", m.Machine.Name, err)
return
}
mu.Lock()
resp.Containers = append(resp.Containers, MachineContainerID{
MachineID: m.Machine.Id,
ContainerID: runResp.ID,
})
mu.Unlock()
}()
}
go func() {
wg.Wait()
close(errCh)
}()
err = nil
for e := range errCh {
err = errors.Join(err, e)
}
return resp, err
}
func (cli *Client) runContainer(
ctx context.Context, serviceID string, spec api.ServiceSpec, machine *pb.MachineInfo,
) (container.CreateResponse, error) {
internal/cli/client/strategy.go
+18 -16
View File
@@ -14,7 +14,7 @@ import (
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)
Plan(ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec) (Plan, error)
}
// RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time
@@ -23,22 +23,22 @@ type RollingStrategy struct{}
func (s *RollingStrategy) Plan(
ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec,
) (Operation, error) {
) (Plan, 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)
return Plan{}, 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")
) (Plan, error) {
return Plan{}, errors.New("not implemented")
}
// planGlobal creates a plan for a global service deployment, ensuring one container runs on each available machine.
@@ -48,33 +48,34 @@ func (s *RollingStrategy) planReplicated(
// that are down.
func (s *RollingStrategy) planGlobal(
ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec,
) (Operation, error) {
serviceID := ""
) (Plan, error) {
var plan Plan
// 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
plan.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()
plan.ServiceID, err = secret.NewID()
if err != nil {
return nil, fmt.Errorf("generate service ID: %w", err)
return plan, fmt.Errorf("generate service ID: %w", err)
}
}
machines, err := cli.ListMachines(ctx)
if err != nil {
return nil, fmt.Errorf("list machines: %w", err)
return plan, fmt.Errorf("list machines: %w", err)
}
plan := &SequenceOperation{}
// TODO: figure out how to return a warning if there are machines down.
seqOp := &SequenceOperation{}
// 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.
var machinesDown []*pb.MachineInfo
for _, m := range machines {
// Skip machines that are down but collect them to report a warning later.
@@ -84,12 +85,13 @@ func (s *RollingStrategy) planGlobal(
}
containers := containersOnMachine[m.Machine.Id]
ops, err := reconcileGlobalContainer(containers, spec, serviceID, m.Machine.Id)
ops, err := reconcileGlobalContainer(containers, spec, plan.ServiceID, m.Machine.Id)
if err != nil {
return nil, err
return plan, err
}
plan.Operations = append(plan.Operations, ops...)
seqOp.Operations = append(seqOp.Operations, ops...)
}
plan.Operation = seqOp
return plan, nil
}