diff --git a/internal/cli/progress/event.go b/internal/cli/progress/event.go new file mode 100644 index 00000000..7b3ed8ba --- /dev/null +++ b/internal/cli/progress/event.go @@ -0,0 +1,12 @@ +package progress + +import ( + "fmt" + + "github.com/psviderski/uncloud/internal/cli/tui" +) + +// PreDeployHookEventID returns a progress event identifier for pre-deploy hook operations. +func PreDeployHookEventID(serviceName, machineName string) string { + return fmt.Sprintf("Pre-deploy hook %s on %s", tui.NameStyle.Render(serviceName), tui.Bold.Render(machineName)) +} diff --git a/pkg/client/container.go b/pkg/client/container.go index 375cfd61..712bab6c 100644 --- a/pkg/client/container.go +++ b/pkg/client/container.go @@ -12,6 +12,7 @@ import ( "github.com/docker/compose/v2/pkg/progress" "github.com/docker/docker/api/types/container" "github.com/docker/docker/pkg/jsonmessage" + cliprogress "github.com/psviderski/uncloud/internal/cli/progress" "github.com/psviderski/uncloud/internal/docker" "github.com/psviderski/uncloud/internal/machine/api/pb" machinedocker "github.com/psviderski/uncloud/internal/machine/docker" @@ -66,16 +67,18 @@ func (cli *Client) createServiceContainerWithPull( if err != nil { return resp, fmt.Errorf("generate random suffix: %w", err) } + containerName := fmt.Sprintf("%s-%s", spec.Name, suffix) + eventID := fmt.Sprintf("Container %s on %s", containerName, machine.Machine.Name) if containerType == pb.CreateServiceContainerRequest_PRE_DEPLOY { containerName = fmt.Sprintf("%s-%s-%s", spec.Name, api.LabelHookPreDeploy, suffix) + eventID = cliprogress.PreDeployHookEventID(spec.Name, machine.Machine.Name) } // Proxy Docker gRPC requests to the selected machine. ctx = proxyToMachine(ctx, machine.Machine) pw := progress.ContextWriter(ctx) - eventID := fmt.Sprintf("Container %s on %s", containerName, machine.Machine.Name) pw.Event(progress.CreatingEvent(eventID)) if spec.Container.PullPolicy == api.PullPolicyAlways { diff --git a/pkg/client/deploy/operation/predeploy.go b/pkg/client/deploy/operation/predeploy.go new file mode 100644 index 00000000..7a78cc06 --- /dev/null +++ b/pkg/client/deploy/operation/predeploy.go @@ -0,0 +1,172 @@ +package operation + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/containerd/errdefs" + "github.com/docker/compose/v2/pkg/progress" + "github.com/docker/docker/api/types/container" + cliprogress "github.com/psviderski/uncloud/internal/cli/progress" + "github.com/psviderski/uncloud/internal/cli/tui" + "github.com/psviderski/uncloud/pkg/api" +) + +// DefaultPreDeployTimeout is the maximum duration to wait for a pre-deploy hook container to complete. +const DefaultPreDeployTimeout = 5 * time.Minute + +// StopPreDeployOperation stops a running pre-deploy hook container from a previous deployment. +type StopPreDeployOperation struct { + MachineID string + // MachineName is used for formatting the operation as part of the deployment plan. + MachineName string + Container api.ServiceContainer +} + +func (o *StopPreDeployOperation) Execute(ctx context.Context, cli Client) error { + if err := cli.StopContainer(ctx, o.Container.ServiceID(), o.Container.ID, container.StopOptions{}); err != nil { + if !errdefs.IsNotFound(err) { + return fmt.Errorf("stop pre-deploy hook container '%s': %w", o.Container.ID, err) + } + } + return nil +} + +func (o *StopPreDeployOperation) Format() string { + displayName := o.Container.ServiceSpec.Name + tui.Faint.Render("/") + o.Container.ShortID() + status, _ := o.Container.HumanState() + + return tui.BoldRed.Render("⏹") + " " + + tui.Faint.Render("stop pre-deploy hook") + " " + + displayName + " " + + tui.Faint.Render("("+status+")") + " " + + tui.Faint.Render("on") + " " + + o.MachineName +} + +func (o *StopPreDeployOperation) String() string { + return fmt.Sprintf("StopPreDeployOperation[machine_id=%s container_id=%s]", + o.MachineID, o.Container.ID) +} + +// RunPreDeployOperation runs a one-shot pre-deploy hook container before service deployment. +type RunPreDeployOperation struct { + ServiceID string + Spec api.ServiceSpec + MachineID string + // MachineName is used for formatting the operation as part of the deployment plan. + MachineName string + // OldContainerIDs are pre-deploy hook containers from previous deployments to remove. + OldContainerIDs []string +} + +func (o *RunPreDeployOperation) Execute(ctx context.Context, cli Client) error { + // Remove old pre-deploy containers. + for _, id := range o.OldContainerIDs { + _ = cli.StopContainer(ctx, o.ServiceID, id, container.StopOptions{}) + err := cli.RemoveContainer(ctx, o.ServiceID, id, container.RemoveOptions{RemoveVolumes: true}) + if err != nil && !errdefs.IsNotFound(err) { + return fmt.Errorf("remove old pre-deploy hook container '%s': %w", id, err) + } + } + + resp, err := cli.CreatePreDeployHookContainer(ctx, o.ServiceID, o.Spec, o.MachineID) + if err != nil { + return fmt.Errorf("create pre-deploy hook container: %w", err) + } + if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil { + return fmt.Errorf("start pre-deploy hook container: %w", err) + } + + timeout := DefaultPreDeployTimeout + if o.Spec.PreDeploy.Timeout != nil { + timeout = *o.Spec.PreDeploy.Timeout + } + + return o.waitForExit(ctx, cli, resp.ID, timeout) +} + +// waitForExit polls the container state until it exits or the timeout is reached. +func (o *RunPreDeployOperation) waitForExit( + ctx context.Context, cli Client, containerID string, timeout time.Duration, +) error { + pw := progress.ContextWriter(ctx) + eventID := cliprogress.PreDeployHookEventID(o.Spec.Name, o.MachineName) + pw.Event(progress.Waiting(eventID)) + + timeoutCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + + var ctr *api.ServiceContainer + for { + select { + case <-timeoutCtx.Done(): + // Stop the container on timeout or context cancellation. + pw.Event(progress.StoppingEvent(eventID)) + _ = cli.StopContainer(ctx, o.ServiceID, containerID, container.StopOptions{}) + + ctrID := containerID + if ctr != nil { + ctrID = fmt.Sprintf("%s/%s", o.Spec.Name, ctr.ShortID()) + } + if ctx.Err() != nil { + // The parent context has been cancelled before the timeout. + pw.Event(progress.NewEvent(eventID, progress.Error, "Cancelled")) + return fmt.Errorf("pre-deploy hook container '%s': %w", ctrID, ctx.Err()) + } + pw.Event(progress.NewEvent(eventID, progress.Error, fmt.Sprintf("Timeout (%s)", timeout))) + return fmt.Errorf("pre-deploy hook container '%s' timed out after %s. "+ + "It's stopped and available for inspection. Fetch logs with 'uc logs %s'", + ctrID, timeout, o.Spec.Name) + + case <-ticker.C: + mc, err := cli.InspectContainer(ctx, o.ServiceID, containerID) + if err != nil { + return fmt.Errorf("inspect pre-deploy hook container: %w", err) + } + ctr = &mc.Container + + if ctr.State.Running { + if state, err := ctr.Container.HumanState(); err == nil { + pw.Event(progress.NewEvent(eventID, progress.Working, fmt.Sprintf("Waiting (%s)", state))) + } + continue + } + + // Hook container has exited successfully. + if ctr.State.ExitCode == 0 { + pw.Event(progress.Event{ + ID: eventID, + Status: progress.Done, + }) + return nil + } + + pw.Event(progress.ErrorEvent(eventID)) + ctrID := fmt.Sprintf("%s/%s", o.Spec.Name, ctr.ShortID()) + return fmt.Errorf("pre-deploy hook container '%s' failed with exit code: %d. "+ + "It's stopped and available for inspection. Fetch logs with 'uc logs %s'", + ctrID, ctr.State.ExitCode, o.Spec.Name) + } + } +} + +func (o *RunPreDeployOperation) Format() string { + cmd := strings.Join(o.Spec.PreDeploy.Command, " ") + + return tui.BoldGreen.Render("▶") + " " + + tui.Faint.Render("run pre-deploy hook") + " " + + o.Spec.Name + " (" + cmd + ") " + + tui.Faint.Render("on") + " " + + o.MachineName +} + +func (o *RunPreDeployOperation) String() string { + return fmt.Sprintf("RunPreDeployOperation[machine_id=%s service_id=%s cmd=%v]", + o.MachineID, o.ServiceID, o.Spec.PreDeploy.Command) +} diff --git a/pkg/client/deploy/scheduler/state.go b/pkg/client/deploy/scheduler/state.go index 17eec0de..a4f6ca3d 100644 --- a/pkg/client/deploy/scheduler/state.go +++ b/pkg/client/deploy/scheduler/state.go @@ -67,3 +67,11 @@ func (s *ClusterState) Machine(nameOrID string) (*Machine, bool) { } return nil, false } + +// MachineName returns the machine name by ID from the cluster state. If the id is not found, ("", false) is returned. +func (s *ClusterState) MachineName(id string) (string, bool) { + if m, ok := s.Machine(id); ok { + return m.Info.Name, true + } + return "", false +} diff --git a/pkg/client/deploy/strategy.go b/pkg/client/deploy/strategy.go index 9a5617c6..6a6b6824 100644 --- a/pkg/client/deploy/strategy.go +++ b/pkg/client/deploy/strategy.go @@ -199,6 +199,10 @@ func (s *RollingStrategy) planReplicated(svc *api.Service, spec api.ServiceSpec) } } + if ops := s.preDeployOperations(svc, plan); len(ops) > 0 { + plan.Operations = append(ops, plan.Operations...) + } + return plan, nil } @@ -265,6 +269,10 @@ func (s *RollingStrategy) planGlobal(svc *api.Service, spec api.ServiceSpec) (Se } } + if ops := s.preDeployOperations(svc, plan); len(ops) > 0 { + plan.Operations = append(ops, plan.Operations...) + } + return plan, nil } @@ -324,6 +332,7 @@ func reconcileGlobalContainer( break } // TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported. + // Make sure to update preDeployOperation accordingly. } if upToDate { return ops, nil @@ -435,6 +444,64 @@ func determineUpdateOrder(oldContainer api.ServiceContainer, spec api.ServiceSpe return api.UpdateOrderStartFirst } +// preDeployOperations returns operations for the pre-deploy hook if the spec has one and the plan updates the service. +// It prepends StopPreDeployOperations for any running hook containers, followed by a RunPreDeployOperation on the same +// machine as the first run/replace operation. +func (s *RollingStrategy) preDeployOperations(svc *api.Service, plan ServicePlan) []operation.Operation { + if plan.Spec.PreDeploy == nil { + return nil + } + + // Find the first run or replace operation to determine the target machine. + var machineID, machineName string + for _, op := range plan.Operations { + switch o := op.(type) { + case *operation.RunContainerOperation: + machineID = o.MachineID + machineName = o.MachineName + case *operation.ReplaceContainerOperation: + machineID = o.MachineID + machineName = o.MachineName + default: + continue + } + break + } + + // Skip the hook as there are no run or replace operations. + if machineID == "" { + return nil + } + + var ops []operation.Operation + + // Collect old pre-deploy container IDs to clean up and stop any that are still running. + var oldContainerIDs []string + if svc != nil { + for _, c := range svc.HookContainers { + oldContainerIDs = append(oldContainerIDs, c.Container.ID) + if c.Container.State.Running { + hookMachineName, _ := s.state.MachineName(c.MachineID) + ops = append(ops, &operation.StopPreDeployOperation{ + MachineID: c.MachineID, + MachineName: hookMachineName, + Container: c.Container, + }) + } + } + } + + ops = append(ops, &operation.RunPreDeployOperation{ + ServiceID: plan.ServiceID, + Spec: plan.Spec, + MachineID: machineID, + MachineName: machineName, + OldContainerIDs: oldContainerIDs, + }) + + return ops +} + // newEmptyServicePlan creates a new empty plan for a service deployment with initialised service ID and name. func newEmptyServicePlan(svc *api.Service, spec api.ServiceSpec) (ServicePlan, error) { plan := ServicePlan{ diff --git a/pkg/client/deploy/strategy_test.go b/pkg/client/deploy/strategy_test.go index eca535f2..a065476d 100644 --- a/pkg/client/deploy/strategy_test.go +++ b/pkg/client/deploy/strategy_test.go @@ -9,6 +9,7 @@ import ( "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client/deploy/operation" + "github.com/psviderski/uncloud/pkg/client/deploy/scheduler" "github.com/stretchr/testify/assert" ) @@ -423,6 +424,246 @@ func TestReconcileGlobalContainer(t *testing.T) { } } +func TestPreDeployOperations(t *testing.T) { + hook := &api.PreDeployHook{ + Command: []string{"db", "migrate"}, + } + + strategy := &RollingStrategy{ + state: &scheduler.ClusterState{ + Machines: []*scheduler.Machine{ + {Info: &pb.MachineInfo{Id: "m-1", Name: "machine-1"}}, + {Info: &pb.MachineInfo{Id: "m-2", Name: "machine-2"}}, + {Info: &pb.MachineInfo{Id: "m-3", Name: "machine-3"}}, + }, + }, + } + + runningHook1 := newServiceContainer("running-hook-1", container.State{Running: true, Status: "running"}) + runningHook2 := newServiceContainer("running-hook-2", container.State{Running: true, Status: "running"}) + + tests := []struct { + name string + plan ServicePlan + svc *api.Service + expected []operation.Operation + }{ + { + name: "no pre-deploy hook in spec", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + }, + }, + }, + expected: nil, + }, + { + name: "plan has RunContainerOperation", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + }, + }, + }, + expected: []operation.Operation{ + &operation.RunPreDeployOperation{ + ServiceID: "svc-1", + MachineID: "m-1", + MachineName: "machine-1", + }, + }, + }, + { + name: "plan has ReplaceContainerOperation", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.ReplaceContainerOperation{MachineID: "m-2", MachineName: "machine-2"}, + }, + }, + }, + expected: []operation.Operation{ + &operation.RunPreDeployOperation{ + ServiceID: "svc-1", + MachineID: "m-2", + MachineName: "machine-2", + }, + }, + }, + { + name: "plan has only RemoveContainerOperation", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.RemoveContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + }, + }, + }, + expected: nil, + }, + { + name: "plan has only StopContainerOperation", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.StopContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + }, + }, + }, + expected: nil, + }, + { + name: "empty plan with no operations", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + }, + expected: nil, + }, + { + name: "stopped hook containers are collected for cleanup", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + }, + }, + }, + svc: &api.Service{ + HookContainers: []api.MachineServiceContainer{ + {MachineID: "m-1", Container: newServiceContainer("old-hook-1", container.State{Status: "exited"})}, + }, + }, + expected: []operation.Operation{ + &operation.RunPreDeployOperation{ + ServiceID: "svc-1", + MachineID: "m-1", + MachineName: "machine-1", + OldContainerIDs: []string{"old-hook-1"}, + }, + }, + }, + { + name: "running hook containers are stopped before run", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + }, + }, + }, + svc: &api.Service{ + HookContainers: []api.MachineServiceContainer{ + {MachineID: "m-2", Container: runningHook1}, + }, + }, + expected: []operation.Operation{ + &operation.StopPreDeployOperation{ + MachineID: "m-2", + MachineName: "machine-2", + Container: runningHook1, + }, + &operation.RunPreDeployOperation{ + ServiceID: "svc-1", + MachineID: "m-1", + MachineName: "machine-1", + OldContainerIDs: []string{"running-hook-1"}, + }, + }, + }, + { + name: "mixed stopped and running hook containers and mixed container operations", + plan: ServicePlan{ + ServiceID: "svc-1", + ServiceName: "app", + Spec: api.ServiceSpec{PreDeploy: hook}, + SequenceOperation: operation.SequenceOperation{ + Operations: []operation.Operation{ + &operation.StopContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + &operation.RemoveContainerOperation{MachineID: "m-1", MachineName: "machine-1"}, + &operation.ReplaceContainerOperation{MachineID: "m-2", MachineName: "machine-2"}, + &operation.RunContainerOperation{MachineID: "m-3", MachineName: "machine-3"}, + &operation.RemoveContainerOperation{MachineID: "m-3", MachineName: "machine-3"}, + }, + }, + }, + svc: &api.Service{ + HookContainers: []api.MachineServiceContainer{ + {MachineID: "m-1", Container: newServiceContainer("stopped-hook", container.State{Status: "exited"})}, + {MachineID: "m-1", Container: runningHook1}, + {MachineID: "m-3", Container: runningHook2}, + }, + }, + expected: []operation.Operation{ + &operation.StopPreDeployOperation{ + MachineID: "m-1", + MachineName: "machine-1", + Container: runningHook1, + }, + &operation.StopPreDeployOperation{ + MachineID: "m-3", + MachineName: "machine-3", + Container: runningHook2, + }, + &operation.RunPreDeployOperation{ + ServiceID: "svc-1", + MachineID: "m-2", + MachineName: "machine-2", + OldContainerIDs: []string{"stopped-hook", "running-hook-1", "running-hook-2"}, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := strategy.preDeployOperations(tt.svc, tt.plan) + if tt.expected == nil { + assert.Nil(t, result) + return + } + assertOperationsEqual(t, tt.expected, result) + }) + } +} + +// newServiceContainer creates an api.ServiceContainer with the given ID and state. +func newServiceContainer(id string, state container.State) api.ServiceContainer { + return api.ServiceContainer{Container: api.Container{ + InspectResponse: container.InspectResponse{ + ContainerJSONBase: &container.ContainerJSONBase{ + ID: id, + State: &state, + }, + }, + }} +} + // assertOperationsEqual compares expected and actual operations, ignoring the Spec field // which is passed separately to the function and not the focus of these tests. func assertOperationsEqual(t *testing.T, expected, actual []operation.Operation) { @@ -430,6 +671,7 @@ func assertOperationsEqual(t *testing.T, expected, actual []operation.Operation) opts := cmp.Options{ cmpopts.IgnoreFields(operation.RunContainerOperation{}, "Spec"), cmpopts.IgnoreFields(operation.ReplaceContainerOperation{}, "Spec"), + cmpopts.IgnoreFields(operation.RunPreDeployOperation{}, "Spec"), cmpopts.IgnoreUnexported(api.Container{}), } if diff := cmp.Diff(expected, actual, opts); diff != "" {