From 7a9eac593f9be0f02b42b1bef5b82f311e6af243 Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Fri, 27 Feb 2026 21:36:40 +1000 Subject: [PATCH] feat: implement container health monitoring and rollback during rolling deployment (closes #24) --- pkg/api/service.go | 9 +- pkg/client/compose/project.go | 21 +++ pkg/client/compose/service.go | 6 +- pkg/client/compose/service_test.go | 109 ++++++++++-- pkg/client/deploy/operation/container.go | 47 ++++- test/e2e/main_test.go | 15 ++ test/e2e/service_test.go | 214 +++++++++++++++++++++++ 7 files changed, 396 insertions(+), 25 deletions(-) create mode 100644 test/e2e/main_test.go diff --git a/pkg/api/service.go b/pkg/api/service.go index 4697e4a3..873905fe 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -438,9 +438,14 @@ type LogDriver struct { // UpdateConfig configures how a service is updated during a deployment. type UpdateConfig struct { // Order specifies the order of operations during an update. - // Valid values are "start-first" (default for stateless services) and "stop-first" (default for services with volumes). - // Empty value means the strategy will determine the order based on service characteristics. + // Valid values are "start-first" (default for stateless services) and "stop-first" (default for services with + // volumes). Empty value means the strategy will determine the order based on service characteristics. Order string `json:",omitempty"` + // MonitorPeriod is how long to wait after starting a container before checking that it's still running + // and not restarting. Containers with a health check that become healthy before the period ends succeed early. + // nil means use the default api.DefaultHealthMonitorPeriod. + // Zero skips the monitoring and checks the container's health immediately after starting. + MonitorPeriod *time.Duration `json:",omitempty"` } type RunServiceResponse struct { diff --git a/pkg/client/compose/project.go b/pkg/client/compose/project.go index fc36b35d..0cc4de36 100644 --- a/pkg/client/compose/project.go +++ b/pkg/client/compose/project.go @@ -6,13 +6,23 @@ import ( "os" "path/filepath" "strings" + "sync" composecli "github.com/compose-spec/compose-go/v2/cli" + "github.com/compose-spec/compose-go/v2/transform" + "github.com/compose-spec/compose-go/v2/tree" "github.com/compose-spec/compose-go/v2/types" + "github.com/psviderski/uncloud/pkg/api" ) +var registerComposeOverrides sync.Once + // LoadProject loads a Compose project from the default locations or the given paths. func LoadProject(ctx context.Context, paths []string, opts ...composecli.ProjectOptionsFn) (*types.Project, error) { + registerComposeOverrides.Do(func() { + transform.RegisterDefaultValue("services.*.deploy.update_config", setUpdateConfigDefaults) + }) + defaultOpts := []composecli.ProjectOptionsFn{ // First apply os.Environment, always wins. composecli.WithOsEnv, @@ -92,3 +102,14 @@ func removeProjectPrefixFromNames(project *types.Project) { project.Volumes[name] = vol } } + +// setUpdateConfigDefaults sets default values for deploy.update_config attributes when not specified in the compose file. +func setUpdateConfigDefaults(data any, _ tree.Path, _ bool) (any, error) { + switch v := data.(type) { + case map[string]any: + if _, ok := v["monitor"]; !ok { + v["monitor"] = api.DefaultHealthMonitorPeriod.String() + } + } + return data, nil +} diff --git a/pkg/client/compose/service.go b/pkg/client/compose/service.go index 83cbce65..1f69a461 100644 --- a/pkg/client/compose/service.go +++ b/pkg/client/compose/service.go @@ -101,7 +101,6 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser return spec, fmt.Errorf("unsupported deploy mode: '%s'", service.Deploy.Mode) } - // Parse update_config.order if cfg := service.Deploy.UpdateConfig; cfg != nil { switch cfg.Order { case "": @@ -111,8 +110,11 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser case "stop-first": spec.UpdateConfig.Order = api.UpdateOrderStopFirst default: - return spec, fmt.Errorf("unsupported update_config.order: '%s'", cfg.Order) + return spec, fmt.Errorf("unsupported deploy.update_config.order: '%s'", cfg.Order) } + + d := time.Duration(cfg.Monitor) + spec.UpdateConfig.MonitorPeriod = &d } } diff --git a/pkg/client/compose/service_test.go b/pkg/client/compose/service_test.go index 5fbf2e85..07a07b0b 100644 --- a/pkg/client/compose/service_test.go +++ b/pkg/client/compose/service_test.go @@ -207,7 +207,8 @@ func TestServiceSpecFromCompose(t *testing.T) { }, Replicas: 3, UpdateConfig: api.UpdateConfig{ - Order: api.UpdateOrderStopFirst, + Order: api.UpdateOrderStopFirst, + MonitorPeriod: &api.DefaultHealthMonitorPeriod, }, Volumes: []api.VolumeSpec{ { @@ -312,6 +313,8 @@ func TestServiceSpecFromCompose(t *testing.T) { } func TestServiceSpecFromCompose_Caddy(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string @@ -460,6 +463,8 @@ services: } func TestServiceSpecFromCompose_GPUs(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string @@ -647,6 +652,8 @@ services: } func TestServiceSpecFromCompose_VolumeDriverOpts(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string @@ -755,6 +762,8 @@ volumes: } func TestServiceSpecFromCompose_Ulimits(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string @@ -843,6 +852,8 @@ services: } func TestServiceSpecFromCompose_UpdateConfig(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string @@ -858,6 +869,33 @@ services: `, expected: api.UpdateConfig{}, }, + { + name: "empty update_config", + composeYAML: ` +services: + test: + image: nginx + deploy: + update_config: {} +`, + expected: api.UpdateConfig{ + MonitorPeriod: &api.DefaultHealthMonitorPeriod, + }, + }, + { + name: "update_config with unsupported attributes ignored", + composeYAML: ` +services: + test: + image: nginx + deploy: + update_config: + parallelism: 1 +`, + expected: api.UpdateConfig{ + MonitorPeriod: &api.DefaultHealthMonitorPeriod, + }, + }, { name: "update_config with stop-first order", composeYAML: ` @@ -869,7 +907,8 @@ services: order: stop-first `, expected: api.UpdateConfig{ - Order: api.UpdateOrderStopFirst, + Order: api.UpdateOrderStopFirst, + MonitorPeriod: &api.DefaultHealthMonitorPeriod, }, }, { @@ -883,7 +922,8 @@ services: order: start-first `, expected: api.UpdateConfig{ - Order: api.UpdateOrderStartFirst, + Order: api.UpdateOrderStartFirst, + MonitorPeriod: &api.DefaultHealthMonitorPeriod, }, }, { @@ -898,18 +938,6 @@ services: `, expectError: true, }, - { - name: "update_config with empty order", - composeYAML: ` -services: - test: - image: nginx - deploy: - update_config: - parallelism: 1 -`, - expected: api.UpdateConfig{}, - }, { name: "update_config with replicas and order", composeYAML: ` @@ -922,7 +950,52 @@ services: order: stop-first `, expected: api.UpdateConfig{ - Order: api.UpdateOrderStopFirst, + Order: api.UpdateOrderStopFirst, + MonitorPeriod: &api.DefaultHealthMonitorPeriod, + }, + }, + { + name: "update_config with custom monitor", + composeYAML: ` +services: + test: + image: nginx + deploy: + update_config: + monitor: 10s +`, + expected: api.UpdateConfig{ + MonitorPeriod: api.AsPtr(10 * time.Second), + }, + }, + { + name: "update_config with monitor and order", + composeYAML: ` +services: + test: + image: nginx + deploy: + update_config: + order: start-first + monitor: 30s +`, + expected: api.UpdateConfig{ + Order: api.UpdateOrderStartFirst, + MonitorPeriod: api.AsPtr(30 * time.Second), + }, + }, + { + name: "update_config with zero monitor skips monitoring", + composeYAML: ` +services: + test: + image: nginx + deploy: + update_config: + monitor: 0s +`, + expected: api.UpdateConfig{ + MonitorPeriod: api.AsPtr(time.Duration(0)), }, }, } @@ -949,6 +1022,8 @@ services: } func TestServiceSpecFromCompose_XMachinesPlacement(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string @@ -1107,6 +1182,8 @@ services: } func TestServiceSpecFromCompose_Devices(t *testing.T) { + t.Parallel() + tests := []struct { name string composeYAML string diff --git a/pkg/client/deploy/operation/container.go b/pkg/client/deploy/operation/container.go index 90be5ccd..19da345a 100644 --- a/pkg/client/deploy/operation/container.go +++ b/pkg/client/deploy/operation/container.go @@ -4,7 +4,9 @@ import ( "context" "fmt" + "github.com/docker/compose/v2/pkg/progress" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/pkg/stringid" "github.com/psviderski/uncloud/pkg/api" ) @@ -24,7 +26,11 @@ func (o *RunContainerOperation) Execute(ctx context.Context, cli Client) error { return fmt.Errorf("start container: %w", err) } - // TODO: wait for the container to become healthy + opts := api.WaitContainerHealthyOptions{MonitorPeriod: o.Spec.UpdateConfig.MonitorPeriod} + if err = cli.WaitContainerHealthy(ctx, o.ServiceID, resp.ID, opts); err != nil { + return fmt.Errorf("container '%s/%s' failed to become healthy: %w", + o.Spec.Name, stringid.TruncateID(resp.ID), err) + } return nil } @@ -111,14 +117,12 @@ func (o *ReplaceContainerOperation) Execute(ctx context.Context, cli Client) err stopFirst := o.Order == api.UpdateOrderStopFirst if stopFirst { + // TODO: inspect and remember the current status of the old container. if err := cli.StopContainer(ctx, o.ServiceID, o.OldContainer.ID, container.StopOptions{}); err != nil { return fmt.Errorf("stop old container: %w", err) } } - // TODO: Rollback support - if new container fails to start, stop new, collect logs, and restart old container (#24) - // TODO: When parallelism is added, rollback becomes more complex - need to track which containers - // were stopped and restore them all on failure resp, err := cli.CreateContainer(ctx, o.ServiceID, o.Spec, o.MachineID) if err != nil { return fmt.Errorf("create container: %w", err) @@ -127,11 +131,44 @@ func (o *ReplaceContainerOperation) Execute(ctx context.Context, cli Client) err return fmt.Errorf("start container: %w", err) } - // TODO: wait for the container to become healthy. If unhealthy, stop new container, collect logs, and start old. + opts := api.WaitContainerHealthyOptions{MonitorPeriod: o.Spec.UpdateConfig.MonitorPeriod} + if err = cli.WaitContainerHealthy(ctx, o.ServiceID, resp.ID, opts); err != nil { + // New container failed to become healthy. Stop it and roll back to the previous container. + // Don't remove the new stopped container to allow users to inspect logs and state. + // TODO: collect logs from the new container and include in the error message to speed up debugging. + + // Use context without progress to not overwrite the container Unhealthy status with Stopped. + ctxWithoutProgress := progress.WithContextWriter(ctx, nil) + _ = cli.StopContainer(ctxWithoutProgress, o.ServiceID, resp.ID, container.StopOptions{}) + + newCtr := fmt.Sprintf("%s/%s", o.Spec.Name, stringid.TruncateID(resp.ID)) + oldCtr := fmt.Sprintf("%s/%s", o.OldContainer.ServiceSpec.Name, o.OldContainer.ShortID()) + + if stopFirst { + // Restart the old container since we stopped it earlier. + // TODO: restart only if the old container was running before we stopped it to avoid starting the failed + // or intentionally stopped container. + if rollbackErr := cli.StartContainer(ctx, o.ServiceID, o.OldContainer.ID); rollbackErr != nil { + return fmt.Errorf( + "new container '%s' failed to become healthy: %w; "+ + "rolled back to previous container '%s' but failed to restart it: %w", + newCtr, rollbackErr, oldCtr, err, + ) + } + } + + return fmt.Errorf("new container '%s' failed to become healthy: %w. Rolled back to previous container '%s'. "+ + "New container has been stopped and is available for inspection. Fetch logs with 'uc logs %s'", + newCtr, err, oldCtr, o.Spec.Name) + } // For start-first, we need to stop before removing. // For stop-first, the container is already stopped. if !stopFirst { + // TODO: the new container is propagated to Caddy upstreams through the cluster store asynchronously. + // There still might be a brief downtime (for a 1 replica service) when Caddy doesn't know about + // the new container but we're stopping the old container. We should somehow ensure Caddy is updated + // with the new container before we stop the old one to avoid this downtime. if err := cli.StopContainer(ctx, o.ServiceID, o.OldContainer.ID, container.StopOptions{}); err != nil { return fmt.Errorf("stop old container: %w", err) } diff --git a/test/e2e/main_test.go b/test/e2e/main_test.go new file mode 100644 index 00000000..0ee1039e --- /dev/null +++ b/test/e2e/main_test.go @@ -0,0 +1,15 @@ +package e2e + +import ( + "os" + "testing" + + "github.com/psviderski/uncloud/pkg/api" +) + +func TestMain(m *testing.M) { + // Disable the default health monitor period to speed up tests. Tests that need a non-zero monitor period + // should set it explicitly via UpdateConfig.MonitorPeriod in their service spec. + api.DefaultHealthMonitorPeriod = 0 + os.Exit(m.Run()) +} diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 43659a81..3723371e 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -56,11 +56,16 @@ func TestDeployment(t *testing.T) { require.ErrorIs(t, err, api.ErrNotFound) }) + // Explicit short period as the default 5s is disabled in tests for faster test execution. + monitorPeriod := 1 * time.Second spec := api.ServiceSpec{ Mode: api.ServiceModeGlobal, Container: api.ContainerSpec{ Image: "portainer/pause:latest", }, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, } deployment := cli.NewDeployment(spec, nil) @@ -73,7 +78,9 @@ func TestDeployment(t *testing.T) { assert.NotEmpty(t, plan.ServiceName) assert.Len(t, plan.SequenceOperation.Operations, 3) // 3 run + start := time.Now() runPlan, err := deployment.Run(ctx) + duration := time.Since(start) require.NoError(t, err) assert.Equal(t, plan, runPlan) @@ -88,6 +95,13 @@ func TestDeployment(t *testing.T) { machines := serviceMachines(svc) assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine") + for _, ctr := range svc.Containers { + assert.True(t, ctr.Container.Healthy(), "Expected deployed containers to be healthy") + } + assert.True(t, duration >= 3*monitorPeriod, + "Expected deployment to wait for at least the health monitor period for each container "+ + "before checking health") + // Deploy a published port. initialContainers := serviceContainerIDs(svc) @@ -105,6 +119,9 @@ func TestDeployment(t *testing.T) { Mode: api.PortModeHost, }, }, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, } deployment = cli.NewDeployment(specWithPort, nil) @@ -112,7 +129,9 @@ func TestDeployment(t *testing.T) { require.NoError(t, err) assert.Len(t, plan.SequenceOperation.Operations, 3) // 3 replace + start = time.Now() _, err = deployment.Run(ctx) + duration = time.Since(start) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -126,6 +145,13 @@ func TestDeployment(t *testing.T) { assert.Empty(t, initialContainers.Intersect(containers).ToSlice(), "All existing containers should be replaced") + for _, ctr := range svc.Containers { + assert.True(t, ctr.Container.Healthy(), "Expected redeployed containers to be healthy") + } + assert.True(t, duration >= 3*monitorPeriod, + "Expected deployment to wait for at least the health monitor period for each container "+ + "before checking health") + // Deploy the same conflicting port but with container spec changes initialContainers = containers @@ -500,6 +526,8 @@ myapp.example.com { }) // 1. Create a basic replicated service with 2 replicas. + // Explicit short period as the default 5s is disabled in tests for faster test execution. + monitorPeriod := 1 * time.Second spec := api.ServiceSpec{ Name: name, Mode: api.ServiceModeReplicated, @@ -507,6 +535,9 @@ myapp.example.com { Image: "portainer/pause:latest", }, Replicas: 2, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, } deployment := cli.NewDeployment(spec, nil) @@ -519,7 +550,9 @@ myapp.example.com { assert.Equal(t, name, plan.ServiceName) assert.Len(t, plan.SequenceOperation.Operations, 2) // 2 run operations for 2 replicas + start := time.Now() runPlan, err := deployment.Run(ctx) + duration := time.Since(start) require.NoError(t, err) assert.Equal(t, plan, runPlan) @@ -533,6 +566,13 @@ myapp.example.com { assert.Len(t, initialMachines.ToSlice(), 2, "Expected 2 containers on 2 different machines") initialContainers := serviceContainerIDs(svc) + for _, ctr := range svc.Containers { + assert.True(t, ctr.Container.Healthy(), "Expected deployed containers to be healthy") + } + assert.True(t, duration >= 2*monitorPeriod, + "Expected deployment to wait for at least the health monitor period for each container "+ + "before checking health") + // 2. Update the service with a new configuration. init := true updatedSpec := spec @@ -543,7 +583,9 @@ myapp.example.com { require.NoError(t, err) assert.Len(t, plan.Operations, 2, "Expected 2 replace operations") + start = time.Now() _, err = deployment.Run(ctx) + duration = time.Since(start) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -558,6 +600,13 @@ myapp.example.com { assert.Empty(t, initialContainers.Intersect(containers).ToSlice(), "All existing containers should be replaced") + for _, ctr := range svc.Containers { + assert.True(t, ctr.Container.Healthy(), "Expected deployed containers to be healthy") + } + assert.True(t, duration >= 2*monitorPeriod, + "Expected deployment to wait for at least the health monitor period for each container "+ + "before checking health") + // 3. Scale to 3 replicas. initialMachines = machines initialContainers = containers // Reset container tracking. @@ -1213,6 +1262,9 @@ myapp.example.com { Container: api.ContainerSpec{ Image: uniqueImage, }, + Placement: api.Placement{ + Machines: []string{c.Machines[0].Name, c.Machines[1].Name}, + }, Replicas: 2, } @@ -1264,6 +1316,168 @@ myapp.example.com { } }) + t.Run("healthcheck becomes healthy", func(t *testing.T) { + t.Parallel() + + name := "test-health-ok" + t.Cleanup(func() { + err := cli.RemoveService(ctx, name) + if !errors.Is(err, api.ErrNotFound) { + require.NoError(t, err) + } + }) + + monitorPeriod := 60 * time.Second + spec := api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeReplicated, + Container: api.ContainerSpec{ + Image: "busybox:1.37.0-musl", + Command: []string{"sh", "-c", "sleep 3600"}, + Healthcheck: &api.HealthcheckSpec{ + Test: []string{"CMD-SHELL", "exit 0"}, + Interval: 1 * time.Second, + Retries: 2, + }, + }, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, + } + deployment := cli.NewDeployment(spec, nil) + + start := time.Now() + _, err := deployment.Run(ctx) + duration := time.Since(start) + require.NoError(t, err) + + svc, err := cli.InspectService(ctx, name) + require.NoError(t, err) + assertServiceMatchesSpec(t, svc, spec) + + assert.True(t, svc.Containers[0].Container.Healthy()) + assert.True(t, duration >= 1*time.Second, + "Deployment should wait for at least one health check interval before marking container as healthy") + assert.True(t, duration < monitorPeriod, + "Deployment should mark container as healthy after first successful health check "+ + "and not wait for the entire monitor period") + }) + + t.Run("healthcheck becomes unhealthy", func(t *testing.T) { + t.Parallel() + + name := "test-health-fail" + t.Cleanup(func() { + err := cli.RemoveService(ctx, name) + if !errors.Is(err, api.ErrNotFound) { + require.NoError(t, err) + } + }) + + monitorPeriod := 1 * time.Second + spec := api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeReplicated, + Container: api.ContainerSpec{ + Image: "busybox:1.37.0-musl", + Command: []string{"sh", "-c", "sleep 3600"}, + Healthcheck: &api.HealthcheckSpec{ + Test: []string{"CMD-SHELL", "exit 1"}, + Interval: 1 * time.Second, + Retries: 2, + }, + }, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, + } + deployment := cli.NewDeployment(spec, nil) + + start := time.Now() + _, err := deployment.Run(ctx) + duration := time.Since(start) + require.Error(t, err) + + assert.ErrorContains(t, err, "unhealthy") + assert.True(t, duration >= monitorPeriod, + "Deployment should wait for at least the monitor period before checking health status") + }) + + t.Run("container crashes on startup with healthcheck", func(t *testing.T) { + t.Parallel() + + name := "test-crash-startup-healthcheck" + t.Cleanup(func() { + err := cli.RemoveService(ctx, name) + if !errors.Is(err, api.ErrNotFound) { + require.NoError(t, err) + } + }) + + monitorPeriod := 1 * time.Second + spec := api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeReplicated, + Container: api.ContainerSpec{ + Image: "busybox:1.37.0-musl", + Command: []string{"false"}, + Healthcheck: &api.HealthcheckSpec{ + Test: []string{"CMD-SHELL", "exit 0"}, + Interval: 1 * time.Second, + Retries: 2, + }, + }, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, + } + deployment := cli.NewDeployment(spec, nil) + + start := time.Now() + _, err := deployment.Run(ctx) + duration := time.Since(start) + require.Error(t, err) + + assert.ErrorContains(t, err, "restarting") + assert.True(t, duration >= monitorPeriod, + "Deployment should wait for at least the monitor period before marking container as unhealthy") + }) + + t.Run("container crashes on startup without healthcheck", func(t *testing.T) { + t.Parallel() + + name := "test-crash-startup-no-healthcheck" + t.Cleanup(func() { + err := cli.RemoveService(ctx, name) + if !errors.Is(err, api.ErrNotFound) { + require.NoError(t, err) + } + }) + + monitorPeriod := 1 * time.Second + spec := api.ServiceSpec{ + Name: name, + Mode: api.ServiceModeReplicated, + Container: api.ContainerSpec{ + Image: "busybox:1.37.0-musl", + Command: []string{"false"}, + }, + UpdateConfig: api.UpdateConfig{ + MonitorPeriod: &monitorPeriod, + }, + } + deployment := cli.NewDeployment(spec, nil) + + start := time.Now() + _, err := deployment.Run(ctx) + duration := time.Since(start) + require.Error(t, err) + + assert.ErrorContains(t, err, "restarting") + assert.True(t, duration >= monitorPeriod, + "Deployment should wait for at least the monitor period before marking container as unhealthy") + }) + // TODO: test deployments with unreachable machines. See https://github.com/psviderski/uncloud/issues/29. }