refactor: compose deployment with pluggable spec resolver, basic test

cmd/uncloud/deploy.go

@@ -67,7 +67,12 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error {
 	}
 	defer clusterClient.Close()
 
-	plan, err := client.PlanComposeDeployment(ctx, project, clusterClient)
+	composeDeploy, err := clusterClient.NewComposeDeployment(ctx, project)
+	if err != nil {
+		return fmt.Errorf("create compose deployment: %w", err)
+	}
+
+	plan, err := composeDeploy.Plan(ctx)
 	if err != nil {
 		return fmt.Errorf("plan deployment: %w", err)
 	}

internal/cli/client/compose.go

@@ -2,26 +2,58 @@ package client
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"github.com/compose-spec/compose-go/v2/graph"
 	"github.com/compose-spec/compose-go/v2/types"
+	"uncloud/internal/api"
 	"uncloud/internal/compose"
 )
 
-func PlanComposeDeployment(ctx context.Context, project *types.Project, cli *Client) (SequenceOperation, error) {
+func (cli *Client) NewComposeDeployment(ctx context.Context, project *types.Project) (*ComposeDeployment, error) {
+	domain, err := cli.GetDomain(ctx)
+	if err != nil && !errors.Is(err, ErrNotFound) {
+		return nil, fmt.Errorf("get cluster domain: %w", err)
+	}
+
+	resolver := &ServiceSpecResolver{
+		// If the domain is not found (not reserved), an empty domain is used for the resolver.
+		ClusterDomain: domain,
+		// TODO: provide an image resolver.
+	}
+
+	return &ComposeDeployment{
+		Client:       cli,
+		Project:      project,
+		SpecResolver: resolver,
+	}, nil
+}
+
+type ComposeDeployment struct {
+	Client       *Client
+	Project      *types.Project
+	SpecResolver *ServiceSpecResolver
+	plan         *SequenceOperation
+}
+
+func (d *ComposeDeployment) Plan(ctx context.Context) (SequenceOperation, error) {
+	if d.plan != nil {
+		return *d.plan, nil
+	}
+
 	plan := SequenceOperation{}
-	err := graph.InDependencyOrder(ctx, project,
+	err := graph.InDependencyOrder(ctx, d.Project,
 		func(ctx context.Context, name string, service types.ServiceConfig) error {
 			spec, err := compose.ServiceSpecFromCompose(name, service)
 			if err != nil {
 				return fmt.Errorf("convert compose service '%s' to service spec: %w", name, err)
 			}
-			if spec, err = cli.PrepareDeploymentSpec(ctx, spec); err != nil {
+			if spec, err = d.Client.PrepareDeploymentSpec(ctx, spec); err != nil {
 				return fmt.Errorf("prepare service '%s' spec ready for deployment: %w", name, err)
 			}
 
 			// TODO: properly handle dependency conditions in the service deployment plan as the first operation.
-			deploy, err := cli.NewDeployment(spec, nil)
+			deploy, err := d.Client.NewDeployment(spec, nil)
 			if err != nil {
 				return fmt.Errorf("create deployment for service '%s': %w", name, err)
 			}
@@ -38,6 +70,40 @@ func PlanComposeDeployment(ctx context.Context, project *types.Project, cli *Cli
 
 			return nil
 		})
+	if err != nil {
+		d.plan = &plan
+	}
 
 	return plan, err
 }
+
+// ServiceSpec returns the service specification for the given compose service that is ready for deployment.
+func (d *ComposeDeployment) ServiceSpec(name string) (api.ServiceSpec, error) {
+	service, err := d.Project.GetService(name)
+	if err != nil {
+		return api.ServiceSpec{}, fmt.Errorf("get config for compose service '%s': %w", name, err)
+	}
+
+	spec, err := compose.ServiceSpecFromCompose(name, service)
+	if err != nil {
+		return spec, fmt.Errorf("convert compose service '%s' to service spec: %w", name, err)
+	}
+
+	// TODO: resolve the image to a digest and supported platforms using an image resolver that broadcasts requests
+	//  to all machines in the cluster.
+	// TODO: configure placement filter based on the supported platforms of the image.
+	if err = d.SpecResolver.Resolve(&spec); err != nil {
+		return spec, fmt.Errorf("resolve service spec '%s': %w", name, err)
+	}
+
+	return spec, nil
+}
+
+func (d *ComposeDeployment) Run(ctx context.Context) error {
+	plan, err := d.Plan(ctx)
+	if err != nil {
+		return fmt.Errorf("create plan: %w", err)
+	}
+
+	return plan.Execute(ctx, d.Client)
+}

internal/cli/client/operation.go

@@ -11,6 +11,9 @@ import (
 // Operation represents a single atomic operation in a deployment process.
 // Operations can be composed to form complex deployment strategies.
 type Operation interface {
+	// Execute performs the operation using the provided client.
+	// TODO: Encapsulate the client in the operation as otherwise it gives an impression that different clients
+	//  can be provided. But in reality, the operation is tightly coupled with the client that was used to create it.
 	Execute(ctx context.Context, cli *Client) error
 	// Format returns a human-readable representation of the operation.
 	Format(resolver NameResolver) string

internal/cli/client/resolver.go

@@ -8,13 +8,14 @@ import (
 	"uncloud/internal/secret"
 )
 
+type ImageDigestResolver interface {
+	Resolve(image string) (string, error)
+}
+
 // ServiceSpecResolver transforms user-provided service specs into deployment-ready form.
 type ServiceSpecResolver struct {
 	ClusterDomain string
-}
-
-func NewServiceSpecResolver(clusterDomain string) *ServiceSpecResolver {
-	return &ServiceSpecResolver{ClusterDomain: clusterDomain}
+	ImageResolver ImageDigestResolver
 }
 
 // Resolve transforms a service spec into its fully resolved form ready for deployment.
@@ -26,6 +27,7 @@ func (r *ServiceSpecResolver) Resolve(spec *api.ServiceSpec) error {
 	steps := []func(*api.ServiceSpec) error{
 		r.applyDefaults,
 		r.resolveServiceName,
+		r.resolveImageDigest,
 		r.expandIngressPorts,
 	}
 
@@ -76,6 +78,21 @@ func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error {
 	return nil
 }
 
+func (r *ServiceSpecResolver) resolveImageDigest(spec *api.ServiceSpec) error {
+	if r.ImageResolver == nil {
+		// Skip digest resolution when no resolver is provided.
+		return nil
+	}
+
+	imageDigest, err := r.ImageResolver.Resolve(spec.Container.Image)
+	if err != nil {
+		return fmt.Errorf("resolve image digest: %w", err)
+	}
+	spec.Container.Image = imageDigest
+
+	return nil
+}
+
 // expandIngressPorts processes HTTP(S) ingress ports in a service spec by:
 // 1. Setting a default hostname (service-name.cluster-domain) for ports without a hostname.
 // 2. Duplicating a port with a cluster domain hostname for ports with external domains.

internal/cli/client/service.go

@@ -19,11 +19,14 @@ import (
 func (cli *Client) PrepareDeploymentSpec(ctx context.Context, spec api.ServiceSpec) (api.ServiceSpec, error) {
 	domain, err := cli.GetDomain(ctx)
 	if err != nil && !errors.Is(err, ErrNotFound) {
-		return spec, fmt.Errorf("get domain: %w", err)
+		return spec, fmt.Errorf("get cluster domain: %w", err)
 	}
 
+	resolver := ServiceSpecResolver{
 		// If the domain is not found (not reserved), an empty domain is used for the resolver.
-	resolver := NewServiceSpecResolver(domain)
+		ClusterDomain: domain,
+		// TODO: provide an image resolver.
+	}
 
 	if err = resolver.Resolve(&spec); err != nil {
 		return spec, err

internal/cli/client/strategy.go

@@ -311,6 +311,7 @@ func reconcileGlobalContainer(
 			}
 			break
 		}
+		// TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported.
 	}
 	if upToDate {
 		return ops, nil

internal/compose/project.go

@@ -1,5 +1,7 @@
 package compose
 
+// TODO: make compose, cli, and api packages public.
+
 import (
 	"context"
 	"fmt"

internal/compose/service.go

@@ -7,9 +7,6 @@ import (
 )
 
 func ServiceSpecFromCompose(name string, service types.ServiceConfig) (api.ServiceSpec, error) {
-	// TODO: resolve the image to a digest and supported platforms using an image resolver that broadcasts requests
-	//  to all machines in the cluster.
-	// TODO: configure placement filter based on the supported platforms of the image.
 	spec := api.ServiceSpec{
 		Container: api.ContainerSpec{
 			Command: service.Command,

test/e2e/compose_test.go

@@ -0,0 +1,71 @@
+package e2e
+
+import (
+	"context"
+	"errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"testing"
+	"uncloud/internal/api"
+	"uncloud/internal/cli/client"
+	"uncloud/internal/compose"
+	"uncloud/internal/ucind"
+)
+
+func TestComposeDeployment(t *testing.T) {
+	t.Parallel()
+
+	clusterName := "ucind-test.compose"
+	ctx := context.Background()
+	c, _ := createTestCluster(t, clusterName, ucind.CreateClusterOptions{Machines: 3}, true)
+
+	cli, err := c.Machines[0].Connect(ctx)
+	require.NoError(t, err)
+
+	t.Run("basic", func(t *testing.T) {
+		t.Parallel()
+
+		name := "basic"
+		t.Cleanup(func() {
+			err := cli.RemoveService(ctx, name)
+			if !errors.Is(err, client.ErrNotFound) {
+				require.NoError(t, err)
+			}
+		})
+
+		project, err := compose.LoadProject(ctx, []string{"fixtures/basic-compose.yaml"})
+		require.NoError(t, err)
+
+		deploy, err := cli.NewComposeDeployment(ctx, project)
+		require.NoError(t, err)
+
+		plan, err := deploy.Plan(ctx)
+		require.NoError(t, err)
+		assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy")
+
+		err = deploy.Run(ctx)
+		require.NoError(t, err)
+
+		svc, err := cli.InspectService(ctx, name)
+		require.NoError(t, err)
+
+		expectedSpec := api.ServiceSpec{
+			Name: name,
+			Mode: api.ServiceModeReplicated,
+			Container: api.ContainerSpec{
+				// TODO: resolve image digest and substitute the image with the image@digest.
+				Image: "portainer/pause:3.9",
+			},
+			Ports: []api.PortSpec{
+				{
+					Hostname:      "basic.example.com",
+					ContainerPort: 80,
+					Protocol:      api.ProtocolHTTPS,
+					Mode:          api.PortModeIngress,
+				},
+			},
+			Replicas: 1,
+		}
+		assertServiceMatchesSpec(t, svc, expectedSpec)
+	})
+}

test/e2e/fixtures/basic-compose.yaml

@@ -0,0 +1,5 @@
+services:
+  basic:
+    image: portainer/pause:3.9
+    x-ports:
+      - basic.example.com:80/https