feat(pre-deploy): add support for pre-deploy hook containers in service management, sync *all* containers to store

pkg/api/client.go

@@ -21,6 +21,9 @@ type ContainerClient interface {
 	CreateContainer(
 		ctx context.Context, serviceID string, spec ServiceSpec, machineID string,
 	) (container.CreateResponse, error)
+	CreatePreDeployHookContainer(
+		ctx context.Context, serviceID string, spec ServiceSpec, machineID string,
+	) (container.CreateResponse, error)
 	ExecContainer(ctx context.Context, serviceNameOrID, containerNameOrID string, config ExecOptions) (int, error)
 	InspectContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) (MachineServiceContainer, error)
 	StartContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) error

pkg/api/container.go

@@ -18,11 +18,16 @@ const (
 	// DockerNetworkName is the name of the Docker network used by uncloud. Keep the value in sync with NetworkName
 	// in internal/machine/docker/manager.go.
 	DockerNetworkName = "uncloud"
+
 	LabelManaged      = "uncloud.managed"
 	LabelServiceID    = "uncloud.service.id"
 	LabelServiceName  = "uncloud.service.name"
 	LabelServiceMode  = "uncloud.service.mode"
 	LabelServicePorts = "uncloud.service.ports"
+	// LabelHook marks a container as a deployment hook. The value indicates the hook type (e.g. LabelHookPreDeploy).
+	LabelHook = "uncloud.service.hook"
+	// LabelHookPreDeploy indicates that the container is a pre-deploy hook that runs before deploying the service.
+	LabelHookPreDeploy = "pre-deploy"
 )
 
 type Container struct {
@@ -181,10 +186,17 @@ func (c *ServiceContainer) ServiceName() string {
 
 // ServiceMode returns the replication mode of the service this container belongs to.
 func (c *ServiceContainer) ServiceMode() string {
-	return c.Config.Labels[LabelServiceMode]
+	return c.ServiceSpec.Mode
+}
+
+// IsHook returns true if the container is a deployment hook (e.g. pre-deploy).
+func (c *ServiceContainer) IsHook() bool {
+	_, ok := c.Config.Labels[LabelHook]
+	return ok
 }
 
 // ServicePorts returns the ports this container publishes as part of its service.
+// TODO: return ports from ServiceSpec to allow updating ingress ports without recreating containers.
 func (c *ServiceContainer) ServicePorts() ([]PortSpec, error) {
 	encoded, ok := c.Config.Labels[LabelServicePorts]
 	if !ok {

pkg/api/service.go

@@ -454,8 +454,8 @@ type PreDeployHook struct {
 	Command []string
 	// Env defines additional environment variables for the container, merged with the service's environment variables.
 	Env EnvVars `json:",omitempty"`
-	// Privileged gives extended privileges to the container.
-	Privileged bool `json:",omitempty"`
+	// Privileged overrides the container's privileged mode. nil means inherit from the service.
+	Privileged *bool `json:",omitempty"`
 	// Timeout is the maximum duration to wait for the command to complete. On timeout, the container is stopped
 	// and the deployment fails. If nil, a default timeout is used.
 	Timeout *time.Duration `json:",omitempty"`
@@ -505,10 +505,13 @@ type RunServiceResponse struct {
 }
 
 type Service struct {
-	ID         string
-	Name       string
-	Mode       string
+	ID   string
+	Name string
+	Mode string
+	// Containers is the regular long-running service containers.
 	Containers []MachineServiceContainer
+	// HookContainers are one-shot containers for deployment hooks (e.g. pre-deploy).
+	HookContainers []MachineServiceContainer
 }
 
 type MachineServiceContainer struct {

pkg/client/container.go

@@ -7,14 +7,17 @@ import (
 	"strings"
 	"time"
 
-	"github.com/containerd/errdefs"
+	"encoding/json"
+
 	"github.com/docker/compose/v2/pkg/progress"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/psviderski/uncloud/internal/docker"
+	"github.com/psviderski/uncloud/internal/machine/api/pb"
 	machinedocker "github.com/psviderski/uncloud/internal/machine/docker"
 	"github.com/psviderski/uncloud/internal/secret"
 	"github.com/psviderski/uncloud/pkg/api"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
@@ -24,6 +27,27 @@ import (
 // CreateContainer creates a new container for the given service on the specified machine.
 func (cli *Client) CreateContainer(
 	ctx context.Context, serviceID string, spec api.ServiceSpec, machineID string,
+) (container.CreateResponse, error) {
+	return cli.createServiceContainerWithPull(ctx, serviceID, spec, machineID, pb.CreateServiceContainerRequest_SERVICE)
+}
+
+// CreatePreDeployHookContainer creates a one-shot container for a pre-deploy hook for the given service
+// on the specified machine.
+func (cli *Client) CreatePreDeployHookContainer(
+	ctx context.Context, serviceID string, spec api.ServiceSpec, machineID string,
+) (container.CreateResponse, error) {
+	return cli.createServiceContainerWithPull(
+		ctx, serviceID, spec, machineID, pb.CreateServiceContainerRequest_PRE_DEPLOY)
+}
+
+// createServiceContainerWithPull creates a regular or deployment hook container for the service
+// on the specified machine, pulling the image if needed.
+func (cli *Client) createServiceContainerWithPull(
+	ctx context.Context,
+	serviceID string,
+	spec api.ServiceSpec,
+	machineID string,
+	containerType pb.CreateServiceContainerRequest_ContainerType,
 ) (container.CreateResponse, error) {
 	var resp container.CreateResponse
 
@@ -43,6 +67,9 @@ func (cli *Client) CreateContainer(
 		return resp, fmt.Errorf("generate random suffix: %w", err)
 	}
 	containerName := fmt.Sprintf("%s-%s", spec.Name, suffix)
+	if containerType == pb.CreateServiceContainerRequest_PRE_DEPLOY {
+		containerName = fmt.Sprintf("%s-%s-%s", spec.Name, api.LabelHookPreDeploy, suffix)
+	}
 
 	// Proxy Docker gRPC requests to the selected machine.
 	ctx = proxyToMachine(ctx, machine.Machine)
@@ -57,7 +84,18 @@ func (cli *Client) CreateContainer(
 		}
 	}
 
-	resp, err = cli.Docker.CreateServiceContainer(ctx, serviceID, spec, containerName)
+	specBytes, err := json.Marshal(spec)
+	if err != nil {
+		return resp, fmt.Errorf("marshal service spec: %w", err)
+	}
+	req := &pb.CreateServiceContainerRequest{
+		ServiceId:     serviceID,
+		ServiceSpec:   specBytes,
+		ContainerName: containerName,
+		ContainerType: containerType,
+	}
+
+	grpcResp, err := cli.Docker.GRPCClient.CreateServiceContainer(ctx, req)
 	if err != nil {
 		switch spec.Container.PullPolicy {
 		case api.PullPolicyAlways, api.PullPolicyNever:
@@ -68,7 +106,7 @@ func (cli *Client) CreateContainer(
 		}
 
 		// NotFound (No such image) error is expected if the image is missing.
-		if !errdefs.IsNotFound(err) || !strings.Contains(err.Error(), "No such image") {
+		if status.Code(err) != codes.NotFound || !strings.Contains(err.Error(), "No such image") {
 			return resp, err
 		}
 
@@ -76,10 +114,14 @@ func (cli *Client) CreateContainer(
 		if err = cli.pullImageWithProgress(ctx, spec.Container.Image, machine.Machine.Name, eventID); err != nil {
 			return resp, err
 		}
-		if resp, err = cli.Docker.CreateServiceContainer(ctx, serviceID, spec, containerName); err != nil {
+		if grpcResp, err = cli.Docker.GRPCClient.CreateServiceContainer(ctx, req); err != nil {
 			return resp, err
 		}
 	}
+
+	if err = json.Unmarshal(grpcResp.Response, &resp); err != nil {
+		return resp, fmt.Errorf("unmarshal gRPC response: %w", err)
+	}
 	pw.Event(progress.CreatedEvent(eventID))
 
 	return resp, nil

pkg/client/service.go

@@ -109,7 +109,7 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
 	}
 	listCtx := metadata.NewOutgoingContext(ctx, md)
 
-	// List all service containers including stopped ones.
+	// List all service containers including stopped ones and deployment hooks.
 	opts := container.ListOptions{All: true}
 	machineContainers, err := cli.Docker.ListServiceContainers(listCtx, nameOrID, opts)
 	if err != nil {
@@ -146,16 +146,15 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
 			}
 		}
 
-		for _, ctr := range mc.Containers {
-			if ctr.ServiceID() == nameOrID || ctr.ServiceName() == nameOrID {
-				containers = append(containers, api.MachineServiceContainer{
-					MachineID: machineID,
-					Container: ctr,
-				})
+		// Collect both regular and hook containers for the service.
+		for _, ctr := range append(mc.Containers, mc.HookContainers...) {
+			containers = append(containers, api.MachineServiceContainer{
+				MachineID: machineID,
+				Container: ctr,
+			})
 
-				if ctr.ServiceID() == nameOrID {
-					foundByID = true
-				}
+			if ctr.ServiceID() == nameOrID {
+				foundByID = true
 			}
 		}
 	}
@@ -181,14 +180,22 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
 		}
 	}
 
-	svc = api.Service{
-		ID:         containers[0].Container.ServiceID(),
-		Name:       containers[0].Container.ServiceName(),
-		Mode:       containers[0].Container.ServiceMode(),
-		Containers: containers,
+	// Partition containers into regular service containers and hook containers.
+	var serviceContainers, hookContainers []api.MachineServiceContainer
+	for _, mc := range containers {
+		if mc.Container.IsHook() {
+			hookContainers = append(hookContainers, mc)
+		} else {
+			serviceContainers = append(serviceContainers, mc)
+		}
 	}
-	if svc.Mode == "" {
-		svc.Mode = api.ServiceModeReplicated
+
+	svc = api.Service{
+		ID:             containers[0].Container.ServiceID(),
+		Name:           containers[0].Container.ServiceName(),
+		Mode:           containers[0].Container.ServiceMode(),
+		Containers:     serviceContainers,
+		HookContainers: hookContainers,
 	}
 
 	return svc, nil