From c4398f9f06e78a9a3446d5b72ccb16d931849fff Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Fri, 14 Feb 2025 20:45:23 +1000 Subject: [PATCH] refactor: deploy global service using Deployment with RollingStrategy --- internal/api/container.go | 5 ++ internal/api/service.go | 3 +- internal/cli/client/container.go | 8 +-- internal/cli/client/deploy.go | 62 +++++++++++++++++----- internal/cli/client/service.go | 89 +++++++------------------------- internal/cli/client/strategy.go | 34 ++++++------ test/e2e/service_test.go | 30 ++++++----- 7 files changed, 111 insertions(+), 120 deletions(-) diff --git a/internal/api/container.go b/internal/api/container.go index 200dc057..0781399a 100644 --- a/internal/api/container.go +++ b/internal/api/container.go @@ -20,6 +20,11 @@ type Container struct { types.ContainerJSON } +// NameWithoutSlash returns the container name without the leading slash. +func (c *Container) NameWithoutSlash() string { + return c.Name[1:] +} + // ServiceID returns the ID of the service this container belongs to. func (c *Container) ServiceID() string { return c.Config.Labels[LabelServiceID] diff --git a/internal/api/service.go b/internal/api/service.go index a195dbfa..ebda592d 100644 --- a/internal/api/service.go +++ b/internal/api/service.go @@ -53,8 +53,7 @@ type ContainerSpec struct { } func (s *ContainerSpec) Validate() error { - _, err := reference.ParseDockerRef(s.Image) - if err != nil { + if _, err := reference.ParseDockerRef(s.Image); err != nil { return fmt.Errorf("invalid image: %w", err) } diff --git a/internal/cli/client/container.go b/internal/cli/client/container.go index 18e719dd..1b016ea1 100644 --- a/internal/cli/client/container.go +++ b/internal/cli/client/container.go @@ -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 { diff --git a/internal/cli/client/deploy.go b/internal/cli/client/deploy.go index 31aac9f7..5668295c 100644 --- a/internal/cli/client/deploy.go +++ b/internal/cli/client/deploy.go @@ -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) } diff --git a/internal/cli/client/service.go b/internal/cli/client/service.go index a0cf57f3..101fbda3 100644 --- a/internal/cli/client/service.go +++ b/internal/cli/client/service.go @@ -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) { diff --git a/internal/cli/client/strategy.go b/internal/cli/client/strategy.go index db3326d5..b7c740b7 100644 --- a/internal/cli/client/strategy.go +++ b/internal/cli/client/strategy.go @@ -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 } diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 13df2335..e85b1a33 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -51,11 +51,12 @@ func TestDeployment(t *testing.T) { 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 + assert.IsType(t, &client.SequenceOperation{}, plan.Operation) + assert.Len(t, plan.Operation.(*client.SequenceOperation).Operations, 3) // 3 run - err = deploy.Run(ctx) + svcID, err := deploy.Run(ctx) require.NoError(t, err) + assert.NotEmpty(t, svcID) svc, err := cli.InspectService(ctx, name) require.NoError(t, err) @@ -88,11 +89,12 @@ func TestDeployment(t *testing.T) { 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 + assert.IsType(t, &client.SequenceOperation{}, plan.Operation) + assert.Len(t, plan.Operation.(*client.SequenceOperation).Operations, 6) // 3 run + 3 remove - err = deploy.Run(ctx) + svcID, err = deploy.Run(ctx) require.NoError(t, err) + assert.NotEmpty(t, svcID) svc, err = cli.InspectService(ctx, name) require.NoError(t, err) @@ -127,11 +129,12 @@ func TestDeployment(t *testing.T) { 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 + assert.IsType(t, &client.SequenceOperation{}, plan.Operation) + assert.Len(t, plan.Operation.(*client.SequenceOperation).Operations, 9) // 3 stop + 3 run + 3 remove - err = deploy.Run(ctx) + svcID, err = deploy.Run(ctx) require.NoError(t, err) + assert.NotEmpty(t, svcID) svc, err = cli.InspectService(ctx, name) require.NoError(t, err) @@ -149,11 +152,12 @@ func TestDeployment(t *testing.T) { 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 + assert.IsType(t, &client.SequenceOperation{}, plan.Operation) + assert.Len(t, plan.Operation.(*client.SequenceOperation).Operations, 0) // no-op - err = deploy.Run(ctx) + svcID, err = deploy.Run(ctx) require.NoError(t, err) + assert.NotEmpty(t, svcID) svc, err = cli.InspectService(ctx, name) require.NoError(t, err) @@ -235,7 +239,6 @@ func TestRunService(t *testing.T) { assert.NotEmpty(t, resp.ID) assert.Equal(t, name, resp.Name) - assert.Len(t, resp.Containers, 1) svc, err := cli.InspectService(ctx, name) require.NoError(t, err) @@ -340,7 +343,6 @@ func TestRunService(t *testing.T) { assert.NotEmpty(t, resp.ID) assert.Equal(t, name, resp.Name) - assert.Len(t, resp.Containers, 3, "expected 1 container on each machine") svc, err := cli.InspectService(ctx, name) require.NoError(t, err)