From 042dd594e0e9c087599c0abaea427ff040f3e76a Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Mon, 24 Mar 2025 14:38:16 +1000 Subject: [PATCH] refactor: decouple deploy and compose packages from client dependency --- cmd/uncloud/caddy/deploy.go | 16 +-- cmd/uncloud/deploy.go | 13 +-- cmd/uncloud/dns/reserve.go | 3 +- cmd/uncloud/dns/show.go | 4 +- cmd/uncloud/machine/add.go | 3 +- cmd/uncloud/service/run.go | 4 +- cmd/uncloud/service/scale.go | 10 +- internal/cli/cli.go | 3 +- pkg/api/client.go | 37 +++++++ pkg/api/error.go | 5 + pkg/client/caddy.go | 5 +- pkg/client/client.go | 5 +- pkg/client/cluster.go | 3 +- pkg/client/container.go | 25 +---- pkg/client/deploy.go | 111 +------------------- pkg/client/dns.go | 3 +- pkg/client/resolver.go | 140 ++++++------------------- pkg/client/service.go | 24 ++--- pkg/compose/deploy.go | 40 +++---- pkg/deploy/container.go | 29 ++++++ pkg/deploy/deploy.go | 120 +++++++++++++++++++++ pkg/{client => deploy}/operation.go | 58 ++--------- pkg/deploy/resolver.go | 133 ++++++++++++++++++++++++ pkg/{client => deploy}/strategy.go | 10 +- test/e2e/assert.go | 6 +- test/e2e/compose_test.go | 3 +- test/e2e/service_test.go | 155 ++++++++++++---------------- 27 files changed, 509 insertions(+), 459 deletions(-) create mode 100644 pkg/api/client.go create mode 100644 pkg/api/error.go create mode 100644 pkg/deploy/container.go create mode 100644 pkg/deploy/deploy.go rename pkg/{client => deploy}/operation.go (73%) create mode 100644 pkg/deploy/resolver.go rename pkg/{client => deploy}/strategy.go (97%) diff --git a/cmd/uncloud/caddy/deploy.go b/cmd/uncloud/caddy/deploy.go index a836726c..17dfa7f9 100644 --- a/cmd/uncloud/caddy/deploy.go +++ b/cmd/uncloud/caddy/deploy.go @@ -8,7 +8,9 @@ import ( "github.com/docker/compose/v2/pkg/progress" "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/deploy" "github.com/spf13/cobra" "maps" "slices" @@ -31,7 +33,7 @@ func NewDeployCommand() *cobra.Command { "A rolling update is performed when updating existing containers to minimise disruption.", RunE: func(cmd *cobra.Command, args []string) error { uncli := cmd.Context().Value("cli").(*cli.CLI) - return deploy(cmd.Context(), uncli, opts) + return runDeploy(cmd.Context(), uncli, opts) }, } @@ -47,7 +49,7 @@ func NewDeployCommand() *cobra.Command { return cmd } -func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { +func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { clusterClient, err := uncli.ConnectCluster(ctx, opts.cluster) if err != nil { return fmt.Errorf("connect to cluster: %w", err) @@ -56,7 +58,7 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { svc, err := clusterClient.InspectService(ctx, client.CaddyServiceName) if err != nil { - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { return fmt.Errorf("inspect caddy service: %w", err) } fmt.Printf("Service: %s (not running)\n", client.CaddyServiceName) @@ -85,7 +87,7 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { fmt.Println() fmt.Println("Preparing a deployment plan...") - var filter client.MachineFilter + var filter deploy.MachineFilter if opts.machine != "" { machines := strings.Split(opts.machine, ",") for i, m := range machines { @@ -105,7 +107,7 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { plan, err := d.Plan(ctx) if err != nil { - if errors.Is(err, client.ErrNoMatchingMachines) { + if errors.Is(err, deploy.ErrNoMatchingMachines) { return fmt.Errorf("no machines found matching: %s", opts.machine) } return fmt.Errorf("plan caddy deployment: %w", err) @@ -169,7 +171,7 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { func UpdateDomainRecords(ctx context.Context, clusterClient *client.Client, progressOut *streams.Out) error { if _, err := clusterClient.GetDomain(ctx); err != nil { - if errors.Is(err, client.ErrNotFound) { + if errors.Is(err, api.ErrNotFound) { fmt.Println("Skipping DNS records update as no cluster domain is reserved (see 'uc dns').") return nil } @@ -213,7 +215,7 @@ func UpdateDomainRecords(ctx context.Context, clusterClient *client.Client, prog return nil } -func machineFilter(machines []string) client.MachineFilter { +func machineFilter(machines []string) deploy.MachineFilter { if len(machines) == 0 { return nil } diff --git a/cmd/uncloud/deploy.go b/cmd/uncloud/deploy.go index 1531f117..8393da01 100644 --- a/cmd/uncloud/deploy.go +++ b/cmd/uncloud/deploy.go @@ -7,8 +7,9 @@ import ( "github.com/compose-spec/compose-go/v2/types" "github.com/docker/compose/v2/pkg/progress" "github.com/psviderski/uncloud/internal/cli" - "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/compose" + "github.com/psviderski/uncloud/pkg/deploy" "github.com/spf13/cobra" "strings" ) @@ -34,7 +35,7 @@ func NewDeployCommand() *cobra.Command { opts.services = args } - return deploy(cmd.Context(), uncli, opts) + return runDeploy(cmd.Context(), uncli, opts) }, } @@ -46,8 +47,8 @@ func NewDeployCommand() *cobra.Command { return cmd } -// deploy parses the Compose file(s) and deploys the services. -func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { +// runDeploy parses the Compose file(s) and deploys the services. +func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { project, err := compose.LoadProject(ctx, opts.files) if err != nil { return fmt.Errorf("load compose file(s): %w", err) @@ -85,14 +86,14 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { fmt.Println("Deployment plan:") for _, op := range plan.Operations { - svcPlan, ok := op.(*client.Plan) + svcPlan, ok := op.(*deploy.Plan) if !ok { return fmt.Errorf("expected service Plan, got: %T", op) } svc, err := clusterClient.InspectService(ctx, svcPlan.ServiceID) if err != nil { - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { return fmt.Errorf("inspect service: %w", err) } fmt.Printf("- Run service [name=%s]\n", svcPlan.ServiceName) diff --git a/cmd/uncloud/dns/reserve.go b/cmd/uncloud/dns/reserve.go index 66a6d9f9..5375c90d 100644 --- a/cmd/uncloud/dns/reserve.go +++ b/cmd/uncloud/dns/reserve.go @@ -7,6 +7,7 @@ import ( "github.com/psviderski/uncloud/cmd/uncloud/caddy" "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" "github.com/spf13/cobra" "google.golang.org/grpc/codes" @@ -61,7 +62,7 @@ func reserve(ctx context.Context, uncli *cli.CLI, opts reserveOptions) error { // Update cluster domain records in Uncloud DNS to point to machines running caddy service if it has been deployed. if _, err = clusterClient.InspectService(ctx, client.CaddyServiceName); err != nil { - if errors.Is(err, client.ErrNotFound) { + if errors.Is(err, api.ErrNotFound) { fmt.Println("Deploy the Caddy reverse proxy service ('uc caddy deploy') to enable internet access " + "to your services via the reserved or your custom domain.") return nil diff --git a/cmd/uncloud/dns/show.go b/cmd/uncloud/dns/show.go index 4edccb85..0c23039d 100644 --- a/cmd/uncloud/dns/show.go +++ b/cmd/uncloud/dns/show.go @@ -5,7 +5,7 @@ import ( "errors" "fmt" "github.com/psviderski/uncloud/internal/cli" - "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/api" "github.com/spf13/cobra" ) @@ -42,7 +42,7 @@ func show(ctx context.Context, uncli *cli.CLI, opts showOptions) error { domain, err := clusterClient.GetDomain(ctx) if err != nil { - if errors.Is(err, client.ErrNotFound) { + if errors.Is(err, api.ErrNotFound) { return errors.New("no domain reserved") } return err diff --git a/cmd/uncloud/machine/add.go b/cmd/uncloud/machine/add.go index 7e74636a..4941269c 100644 --- a/cmd/uncloud/machine/add.go +++ b/cmd/uncloud/machine/add.go @@ -10,6 +10,7 @@ import ( "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli/config" "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" "github.com/spf13/cobra" "google.golang.org/grpc/codes" @@ -116,7 +117,7 @@ func add(ctx context.Context, uncli *cli.CLI, remoteMachine cli.RemoteMachine, o caddyImage := "" caddySvc, err := machineClient.InspectService(ctx, client.CaddyServiceName) if err != nil { - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { return fmt.Errorf("inspect caddy service: %w", err) } } else { diff --git a/cmd/uncloud/service/run.go b/cmd/uncloud/service/run.go index 19f14242..9f3f9f08 100644 --- a/cmd/uncloud/service/run.go +++ b/cmd/uncloud/service/run.go @@ -6,7 +6,7 @@ import ( "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" - "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/deploy" "github.com/spf13/cobra" "slices" "strings" @@ -89,7 +89,7 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { return fmt.Errorf("invalid replication mode: %q", opts.mode) } - var machineFilter client.MachineFilter + var machineFilter deploy.MachineFilter if len(opts.machines) > 0 { var machines []string for _, value := range opts.machines { diff --git a/cmd/uncloud/service/scale.go b/cmd/uncloud/service/scale.go index c06e7ebe..b04020c9 100644 --- a/cmd/uncloud/service/scale.go +++ b/cmd/uncloud/service/scale.go @@ -89,12 +89,8 @@ func scale(ctx context.Context, uncli *cli.CLI, opts scaleOptions) error { spec.Replicas = opts.replicas - deploy, err := clusterClient.NewDeployment(spec, nil) - if err != nil { - return fmt.Errorf("create deployment: %w", err) - } - - plan, err := deploy.Plan(ctx) + deployment := clusterClient.NewDeployment(spec, nil) + plan, err := deployment.Plan(ctx) if err != nil { return fmt.Errorf("plan deployment: %w", err) } @@ -128,7 +124,7 @@ func scale(ctx context.Context, uncli *cli.CLI, opts scaleOptions) error { title := fmt.Sprintf("Scaling service %s (%d → %d replicas)", svc.Name, currentReplicas, opts.replicas) err = progress.RunWithTitle(ctx, func(ctx context.Context) error { - if _, err = deploy.Run(ctx); err != nil { + if _, err = deployment.Run(ctx); err != nil { return fmt.Errorf("deploy service: %w", err) } return nil diff --git a/internal/cli/cli.go b/internal/cli/cli.go index bab8ba53..ee73c384 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -10,6 +10,7 @@ import ( "github.com/psviderski/uncloud/internal/machine" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/sshexec" + "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" "github.com/psviderski/uncloud/pkg/client/connector" "net/netip" @@ -56,7 +57,7 @@ func (cli *CLI) CreateCluster(name string) error { func (cli *CLI) SetCurrentCluster(name string) error { if _, ok := cli.config.Clusters[name]; !ok { - return client.ErrNotFound + return api.ErrNotFound } cli.config.CurrentCluster = name return cli.config.Save() diff --git a/pkg/api/client.go b/pkg/api/client.go new file mode 100644 index 00000000..52629d05 --- /dev/null +++ b/pkg/api/client.go @@ -0,0 +1,37 @@ +package api + +import ( + "context" + "github.com/docker/docker/api/types/container" + "github.com/psviderski/uncloud/internal/machine/api/pb" +) + +type Client interface { + ContainerClient + DNSClient + MachineClient + ServiceClient +} + +type ContainerClient interface { + CreateContainer( + ctx context.Context, serviceID string, spec ServiceSpec, machineID string, + ) (container.CreateResponse, error) + InspectContainer(ctx context.Context, serviceID, containerID string) (MachineContainer, error) + RemoveContainer(ctx context.Context, serviceID, containerID string, opts container.RemoveOptions) error + StartContainer(ctx context.Context, serviceID, containerID string) error + StopContainer(ctx context.Context, serviceID, containerID string, opts container.StopOptions) error +} + +type DNSClient interface { + GetDomain(ctx context.Context) (string, error) +} + +type MachineClient interface { + InspectMachine(ctx context.Context, id string) (*pb.MachineMember, error) + ListMachines(ctx context.Context) ([]*pb.MachineMember, error) +} + +type ServiceClient interface { + InspectService(ctx context.Context, id string) (Service, error) +} diff --git a/pkg/api/error.go b/pkg/api/error.go new file mode 100644 index 00000000..618fb746 --- /dev/null +++ b/pkg/api/error.go @@ -0,0 +1,5 @@ +package api + +import "errors" + +var ErrNotFound = errors.New("not found") diff --git a/pkg/client/caddy.go b/pkg/client/caddy.go index 50490f91..a144bb98 100644 --- a/pkg/client/caddy.go +++ b/pkg/client/caddy.go @@ -7,6 +7,7 @@ import ( "github.com/google/go-containerregistry/pkg/name" "github.com/google/go-containerregistry/pkg/v1/remote" "github.com/psviderski/uncloud/pkg/api" + "github.com/psviderski/uncloud/pkg/deploy" "regexp" ) @@ -21,7 +22,7 @@ var caddyImageTagRegex = regexp.MustCompile(`^2\.\d+\.\d+$`) // NewCaddyDeployment creates a new deployment for a Caddy reverse proxy service. // The service is deployed in global mode to all machines in the cluster. If the image is not provided, the latest // version of the official Caddy Docker image is used. -func (cli *Client) NewCaddyDeployment(image string, filter MachineFilter) (*Deployment, error) { +func (cli *Client) NewCaddyDeployment(image string, filter deploy.MachineFilter) (*deploy.Deployment, error) { latest, err := latestCaddyImage() if err != nil { return nil, fmt.Errorf("look up latest Caddy image: %w", err) @@ -55,7 +56,7 @@ func (cli *Client) NewCaddyDeployment(image string, filter MachineFilter) (*Depl }, } - return cli.NewDeployment(spec, &RollingStrategy{MachineFilter: filter}) + return cli.NewDeployment(spec, &deploy.RollingStrategy{MachineFilter: filter}), nil } // latestCaddyImage returns the latest image of the official Caddy Docker image on Docker Hub. diff --git a/pkg/client/client.go b/pkg/client/client.go index 31c2f0a2..1577dea9 100644 --- a/pkg/client/client.go +++ b/pkg/client/client.go @@ -7,13 +7,12 @@ import ( "github.com/docker/cli/cli/streams" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/docker" + "github.com/psviderski/uncloud/pkg/api" "google.golang.org/grpc" "google.golang.org/grpc/metadata" "os" ) -var ErrNotFound = errors.New("not found") - // Client is a client for the machine API. type Client struct { connector Connector @@ -26,6 +25,8 @@ type Client struct { Docker *docker.Client } +var _ api.Client = (*Client)(nil) + // Connector is an interface for establishing a connection to the machine API. type Connector interface { Connect(ctx context.Context) (*grpc.ClientConn, error) diff --git a/pkg/client/cluster.go b/pkg/client/cluster.go index bc424af7..17c0f8cf 100644 --- a/pkg/client/cluster.go +++ b/pkg/client/cluster.go @@ -3,6 +3,7 @@ package client import ( "context" "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" "google.golang.org/protobuf/types/known/emptypb" ) @@ -18,7 +19,7 @@ func (cli *Client) InspectMachine(ctx context.Context, id string) (*pb.MachineMe } } - return nil, ErrNotFound + return nil, api.ErrNotFound } func (cli *Client) ListMachines(ctx context.Context) ([]*pb.MachineMember, error) { diff --git a/pkg/client/container.go b/pkg/client/container.go index 74762b40..dc0a19a0 100644 --- a/pkg/client/container.go +++ b/pkg/client/container.go @@ -256,7 +256,7 @@ func (cli *Client) InspectContainer(ctx context.Context, serviceID, containerID } } if ctr.MachineID == "" { - return ctr, ErrNotFound + return ctr, api.ErrNotFound } return ctr, nil @@ -340,26 +340,3 @@ func (cli *Client) RemoveContainer( return nil } - -type ContainerSpecStatus string - -const ContainerUpToDate ContainerSpecStatus = "up-to-date" -const ContainerNeedsUpdate ContainerSpecStatus = "needs-update" -const ContainerNeedsRecreate ContainerSpecStatus = "needs-recreate" - -func CompareContainerToSpec(ctr api.Container, spec api.ServiceSpec) (ContainerSpecStatus, error) { - specHash, err := spec.ImmutableHash() - if err != nil { - return "", fmt.Errorf("calculate immutable hash for service spec: %w", err) - } - - // Is the hash label is unset, there is no easy way to compare its configuration with the spec, - // so let's recreate as well. - if ctr.Config.Labels[api.LabelServiceSpecHash] != specHash { - return ContainerNeedsRecreate, nil - } - - // TODO: compare mutable properties such as memory or CPU limits when they are implemented. - - return ContainerUpToDate, nil -} diff --git a/pkg/client/deploy.go b/pkg/client/deploy.go index bbff09e3..ebbe37f1 100644 --- a/pkg/client/deploy.go +++ b/pkg/client/deploy.go @@ -1,115 +1,12 @@ package client import ( - "context" - "errors" - "fmt" - "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" + "github.com/psviderski/uncloud/pkg/deploy" ) -// Deployment manages the process of creating or updating a service to match a desired state. -// It coordinates the validation, planning, and execution of deployment operations. -type Deployment struct { - Service *api.Service - Spec api.ServiceSpec - Strategy Strategy - cli *Client - plan *Plan -} - -type Plan struct { - ServiceID string - ServiceName string - SequenceOperation -} - -// MachineFilter determines which machines participate in a deployment operation by returning true for -// machines that should be included. -type MachineFilter func(m *pb.MachineInfo) bool - -var ErrNoMatchingMachines = errors.New("no machines match the filter") - // NewDeployment creates a new deployment for the given service specification. -// If strategy is nil, a default RollingStrategy will be used. -// TODO(refactor): do not return error -func (cli *Client) NewDeployment(spec api.ServiceSpec, strategy Strategy) (*Deployment, error) { - if strategy == nil { - strategy = &RollingStrategy{} - } - - return &Deployment{ - Spec: spec, - Strategy: strategy, - cli: cli, - }, nil -} - -// 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) (Plan, error) { - if d.plan != nil { - return *d.plan, nil - } - - // Validate the new spec before planning. - if err := d.Validate(ctx); err != nil { - return Plan{}, fmt.Errorf("invalid deployment: %w", err) - } - - plan, err := d.Strategy.Plan(ctx, d.cli, d.Service, d.Spec) - if err != nil { - return Plan{}, fmt.Errorf("create plan using %s strategy: %w", d.Strategy.Type(), err) - } - d.plan = &plan - - return plan, nil -} - -// Validate checks if the deployment specification is valid. -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 { - svc, err := d.cli.InspectService(ctx, d.Spec.Name) - if err == nil { - d.Service = &svc - } else if !errors.Is(err, ErrNotFound) { - return fmt.Errorf("inspect service: %w", err) - } - } - // d.Service is nil if the service doesn't exist yet (first deployment). - if d.Service == nil { - return nil - } - - if d.Service.Name != d.Spec.Name { - return errors.New("service name cannot be changed") - } - if d.Service.Mode != d.Spec.Mode { - return errors.New("service mode cannot be changed") - } - if d.Spec.Mode == api.ServiceModeReplicated && d.Spec.Replicas < 1 { - return errors.New("number of replicas must be at least 1") - } - - return nil -} - -// 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) (Plan, error) { - plan, err := d.Plan(ctx) - if err != nil { - return plan, fmt.Errorf("create plan: %w", err) - } - - return plan, plan.Execute(ctx, d.cli) +// If strategy is nil, a default deploy.RollingStrategy will be used. +func (cli *Client) NewDeployment(spec api.ServiceSpec, strategy deploy.Strategy) *deploy.Deployment { + return deploy.NewDeployment(cli, spec, strategy) } diff --git a/pkg/client/dns.go b/pkg/client/dns.go index 6818f047..5e986a4d 100644 --- a/pkg/client/dns.go +++ b/pkg/client/dns.go @@ -8,6 +8,7 @@ import ( "github.com/docker/compose/v2/pkg/progress" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/caddyfile" + "github.com/psviderski/uncloud/pkg/api" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "io" @@ -21,7 +22,7 @@ func (cli *Client) GetDomain(ctx context.Context) (string, error) { domain, err := cli.ClusterClient.GetDomain(ctx, nil) if err != nil { if status.Convert(err).Code() == codes.NotFound { - return "", ErrNotFound + return "", api.ErrNotFound } return "", err } diff --git a/pkg/client/resolver.go b/pkg/client/resolver.go index 7f8073af..464c8172 100644 --- a/pkg/client/resolver.go +++ b/pkg/client/resolver.go @@ -1,133 +1,53 @@ package client import ( + "context" "fmt" - "github.com/distribution/reference" - "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/pkg/api" - "strings" ) -type ImageDigestResolver interface { - Resolve(image string) (string, error) +// MapNameResolver resolves machine and container IDs to their names using a static map. +type MapNameResolver struct { + machines map[string]string + containers map[string]string } -// ServiceSpecResolver transforms user-provided service specs into deployment-ready form. -type ServiceSpecResolver struct { - ClusterDomain string - ImageResolver ImageDigestResolver +func NewNameResolver(machines, containers map[string]string) *MapNameResolver { + return &MapNameResolver{ + machines: machines, + containers: containers, + } } -// Resolve transforms a service spec into its fully resolved form ready for deployment. -func (r *ServiceSpecResolver) Resolve(spec *api.ServiceSpec) error { - if err := spec.Validate(); err != nil { - return fmt.Errorf("invalid service spec: %w", err) +func (r *MapNameResolver) MachineName(machineID string) string { + if name, ok := r.machines[machineID]; ok { + return name } - - steps := []func(*api.ServiceSpec) error{ - r.applyDefaults, - r.resolveServiceName, - r.resolveImageDigest, - r.expandIngressPorts, - } - - for _, step := range steps { - if err := step(spec); err != nil { - return err - } - } - - return nil + return machineID } -func (r *ServiceSpecResolver) applyDefaults(spec *api.ServiceSpec) error { - if spec.Mode == "" { - spec.Mode = api.ServiceModeReplicated +func (r *MapNameResolver) ContainerName(containerID string) string { + if name, ok := r.containers[containerID]; ok { + return name } - // Ensure the replicated service has at least one replica. - if spec.Mode == api.ServiceModeReplicated && spec.Replicas == 0 { - spec.Replicas = 1 - } - - return nil + return containerID } -func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error { - if spec.Name != "" { - return nil - } - - // Generate a random service name from the image when not provided. - img, err := reference.ParseDockerRef(spec.Container.Image) +// ServiceOperationNameResolver returns a machine and container name resolver for a service that can be used to format +// deployment operations. +func (cli *Client) ServiceOperationNameResolver(ctx context.Context, svc api.Service) (*MapNameResolver, error) { + machines, err := cli.ListMachines(ctx) if err != nil { - return fmt.Errorf("invalid image: %w", err) + return nil, fmt.Errorf("list machines: %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:] + machineNames := make(map[string]string, len(machines)) + for _, m := range machines { + machineNames[m.Machine.Id] = m.Machine.Name } - // Append a random suffix to the image name to generate an optimistically unique service name. - suffix, err := secret.RandomAlphaNumeric(4) - if err != nil { - return fmt.Errorf("generate random suffix: %w", err) + containerNames := make(map[string]string, len(svc.Containers)) + for _, c := range svc.Containers { + containerNames[c.Container.ID] = c.Container.NameWithoutSlash() } - spec.Name = fmt.Sprintf("%s-%s", imageName, suffix) - 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. -// This ensures every ingress port is accessible via the cluster domain, while preserving any custom domains specified -// by the user. -func (r *ServiceSpecResolver) expandIngressPorts(spec *api.ServiceSpec) error { - for i, port := range spec.Ports { - if port.Protocol != api.ProtocolHTTP && port.Protocol != api.ProtocolHTTPS { - continue - } - - if port.Hostname == "" { - if r.ClusterDomain == "" { - return fmt.Errorf("cluster domain must be reserved to generate hostname for ingress port: %d/%s", - port.ContainerPort, port.Protocol) - } - // Assign the default hostname (service-name.cluster-domain). - spec.Ports[i].Hostname = fmt.Sprintf("%s.%s", spec.Name, r.ClusterDomain) - } else { - if r.ClusterDomain == "" { - // When no cluster domain is reserved, use only the provided hostname. - continue - } - - if strings.HasSuffix(port.Hostname, "."+r.ClusterDomain) { - // If the hostname is already a cluster subdomain, use as is. - continue - } - // For external domains, duplicate the port with a service-name.cluster-domain hostname so the service - // can be accessed via both hostnames. - newPort := port - newPort.Hostname = fmt.Sprintf("%s.%s", spec.Name, r.ClusterDomain) - spec.Ports = append(spec.Ports, newPort) - } - } - - return nil + return NewNameResolver(machineNames, containerNames), nil } diff --git a/pkg/client/service.go b/pkg/client/service.go index d8ced27a..a06e0bc1 100644 --- a/pkg/client/service.go +++ b/pkg/client/service.go @@ -9,6 +9,7 @@ import ( "github.com/docker/docker/api/types/filters" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" + "github.com/psviderski/uncloud/pkg/deploy" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" @@ -18,11 +19,11 @@ 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) { + if err != nil && !errors.Is(err, api.ErrNotFound) { return spec, fmt.Errorf("get cluster domain: %w", err) } - resolver := ServiceSpecResolver{ + resolver := deploy.ServiceSpecResolver{ // If the domain is not found (not reserved), an empty domain is used for the resolver. ClusterDomain: domain, // TODO: provide an image resolver. @@ -41,7 +42,7 @@ type RunServiceResponse struct { } func (cli *Client) RunService( - ctx context.Context, spec api.ServiceSpec, filter MachineFilter, + ctx context.Context, spec api.ServiceSpec, filter deploy.MachineFilter, ) (RunServiceResponse, error) { var resp RunServiceResponse @@ -55,7 +56,7 @@ func (cli *Client) RunService( if err == nil { return resp, fmt.Errorf("service with name '%s' already exists", spec.Name) } - if !errors.Is(err, ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { return resp, fmt.Errorf("inspect service: %w", err) } } @@ -66,12 +67,9 @@ func (cli *Client) RunService( } err = progress.RunWithTitle(ctx, func(ctx context.Context) error { - deploy, err := cli.NewDeployment(spec, &RollingStrategy{MachineFilter: filter}) - if err != nil { - return fmt.Errorf("create deployment: %w", err) - } + deployment := cli.NewDeployment(spec, &deploy.RollingStrategy{MachineFilter: filter}) - plan, err := deploy.Run(ctx) + plan, err := deployment.Run(ctx) if err != nil { return err } @@ -168,7 +166,7 @@ func (cli *Client) InspectService(ctx context.Context, id string) (api.Service, } if len(containers) == 0 { - return svc, ErrNotFound + return svc, api.ErrNotFound } // Containers from different services may share the same service name (distributed and eventually consistent store @@ -211,7 +209,7 @@ func (cli *Client) InspectServiceFromStore(ctx context.Context, id string) (api. if err != nil { if s, ok := status.FromError(err); ok { if s.Code() == codes.NotFound { - return svc, ErrNotFound + return svc, api.ErrNotFound } } return svc, err @@ -259,7 +257,7 @@ func (cli *Client) RemoveService(ctx context.Context, id string) error { } err = cli.RemoveContainer(ctx, svc.ID, mc.Container.ID, container.RemoveOptions{}) - if err != nil && !errors.Is(err, ErrNotFound) { + if err != nil && !errors.Is(err, api.ErrNotFound) { errCh <- fmt.Errorf("remove container '%s': %w", mc.Container.ID, err) } }() @@ -327,7 +325,7 @@ func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) { svc, err := cli.InspectService(ctx, ctr.ServiceID()) if err != nil { - if errors.Is(err, ErrNotFound) { + if errors.Is(err, api.ErrNotFound) { continue } return nil, fmt.Errorf("inspect service: %w", err) diff --git a/pkg/compose/deploy.go b/pkg/compose/deploy.go index 52dcf8f3..5cb2401f 100644 --- a/pkg/compose/deploy.go +++ b/pkg/compose/deploy.go @@ -7,16 +7,28 @@ import ( "github.com/compose-spec/compose-go/v2/graph" "github.com/compose-spec/compose-go/v2/types" "github.com/psviderski/uncloud/pkg/api" - "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/deploy" ) -func NewDeployment(ctx context.Context, cli *client.Client, project *types.Project) (*Deployment, error) { +type Client interface { + api.DNSClient + deploy.Client +} + +type Deployment struct { + Client Client + Project *types.Project + SpecResolver *deploy.ServiceSpecResolver + plan *deploy.SequenceOperation +} + +func NewDeployment(ctx context.Context, cli Client, project *types.Project) (*Deployment, error) { domain, err := cli.GetDomain(ctx) - if err != nil && !errors.Is(err, client.ErrNotFound) { + if err != nil && !errors.Is(err, api.ErrNotFound) { return nil, fmt.Errorf("get cluster domain: %w", err) } - resolver := &client.ServiceSpecResolver{ + resolver := &deploy.ServiceSpecResolver{ // If the domain is not found (not reserved), an empty domain is used for the resolver. ClusterDomain: domain, // TODO: provide an image resolver. @@ -29,19 +41,12 @@ func NewDeployment(ctx context.Context, cli *client.Client, project *types.Proje }, nil } -type Deployment struct { - Client *client.Client - Project *types.Project - SpecResolver *client.ServiceSpecResolver - plan *client.SequenceOperation -} - -func (d *Deployment) Plan(ctx context.Context) (client.SequenceOperation, error) { +func (d *Deployment) Plan(ctx context.Context) (deploy.SequenceOperation, error) { if d.plan != nil { return *d.plan, nil } - plan := client.SequenceOperation{} + plan := deploy.SequenceOperation{} err := graph.InDependencyOrder(ctx, d.Project, func(ctx context.Context, name string, _ types.ServiceConfig) error { spec, err := d.ServiceSpec(name) @@ -49,13 +54,10 @@ func (d *Deployment) Plan(ctx context.Context) (client.SequenceOperation, error) return fmt.Errorf("convert compose service '%s' to service spec: %w", name, err) } - // TODO: properly handle dependency conditions in the service deployment plan as the first operation. - deploy, err := d.Client.NewDeployment(spec, nil) - if err != nil { - return fmt.Errorf("create deployment for service '%s': %w", name, err) - } + // TODO: properly handle depends_on conditions in the service deployment plan as the first operation. + deployment := deploy.NewDeployment(d.Client, spec, nil) - servicePlan, err := deploy.Plan(ctx) + servicePlan, err := deployment.Plan(ctx) if err != nil { return fmt.Errorf("create deployment plan for service '%s': %w", name, err) } diff --git a/pkg/deploy/container.go b/pkg/deploy/container.go new file mode 100644 index 00000000..d2ae708a --- /dev/null +++ b/pkg/deploy/container.go @@ -0,0 +1,29 @@ +package deploy + +import ( + "fmt" + "github.com/psviderski/uncloud/pkg/api" +) + +type ContainerSpecStatus string + +const ContainerUpToDate ContainerSpecStatus = "up-to-date" +const ContainerNeedsUpdate ContainerSpecStatus = "needs-update" +const ContainerNeedsRecreate ContainerSpecStatus = "needs-recreate" + +func CompareContainerToSpec(ctr api.Container, spec api.ServiceSpec) (ContainerSpecStatus, error) { + specHash, err := spec.ImmutableHash() + if err != nil { + return "", fmt.Errorf("calculate immutable hash for service spec: %w", err) + } + + // Is the hash label is unset, there is no easy way to compare its configuration with the spec, + // so let's recreate as well. + if ctr.Config.Labels[api.LabelServiceSpecHash] != specHash { + return ContainerNeedsRecreate, nil + } + + // TODO: compare mutable properties such as memory or CPU limits when they are implemented. + + return ContainerUpToDate, nil +} diff --git a/pkg/deploy/deploy.go b/pkg/deploy/deploy.go new file mode 100644 index 00000000..6c1700dc --- /dev/null +++ b/pkg/deploy/deploy.go @@ -0,0 +1,120 @@ +package deploy + +import ( + "context" + "errors" + "fmt" + "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/pkg/api" +) + +type Client interface { + api.ContainerClient + api.MachineClient + api.ServiceClient +} + +// Deployment manages the process of creating or updating a service to match a desired state. +// It coordinates the validation, planning, and execution of deployment operations. +type Deployment struct { + Service *api.Service + Spec api.ServiceSpec + Strategy Strategy + cli Client + plan *Plan +} + +type Plan struct { + ServiceID string + ServiceName string + SequenceOperation +} + +// MachineFilter determines which machines participate in a deployment operation by returning true for +// machines that should be included. +type MachineFilter func(m *pb.MachineInfo) bool + +var ErrNoMatchingMachines = errors.New("no machines match the filter") + +// NewDeployment creates a new deployment for the given service specification. +// If strategy is nil, a default RollingStrategy will be used. +func NewDeployment(cli Client, spec api.ServiceSpec, strategy Strategy) *Deployment { + if strategy == nil { + strategy = &RollingStrategy{} + } + + return &Deployment{ + Spec: spec, + Strategy: strategy, + cli: cli, + } +} + +// 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) (Plan, error) { + if d.plan != nil { + return *d.plan, nil + } + + // Validate the new spec before planning. + if err := d.Validate(ctx); err != nil { + return Plan{}, fmt.Errorf("invalid deployment: %w", err) + } + + plan, err := d.Strategy.Plan(ctx, d.cli, d.Service, d.Spec) + if err != nil { + return Plan{}, fmt.Errorf("create plan using %s strategy: %w", d.Strategy.Type(), err) + } + d.plan = &plan + + return plan, nil +} + +// Validate checks if the deployment specification is valid. +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 { + svc, err := d.cli.InspectService(ctx, d.Spec.Name) + if err == nil { + d.Service = &svc + } else if !errors.Is(err, api.ErrNotFound) { + return fmt.Errorf("inspect service: %w", err) + } + } + // d.Service is nil if the service doesn't exist yet (first deployment). + if d.Service == nil { + return nil + } + + if d.Service.Name != d.Spec.Name { + return errors.New("service name cannot be changed") + } + if d.Service.Mode != d.Spec.Mode { + return errors.New("service mode cannot be changed") + } + if d.Spec.Mode == api.ServiceModeReplicated && d.Spec.Replicas < 1 { + return errors.New("number of replicas must be at least 1") + } + + return nil +} + +// 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) (Plan, error) { + plan, err := d.Plan(ctx) + if err != nil { + return plan, fmt.Errorf("create plan: %w", err) + } + + return plan, plan.Execute(ctx, d.cli) +} diff --git a/pkg/client/operation.go b/pkg/deploy/operation.go similarity index 73% rename from pkg/client/operation.go rename to pkg/deploy/operation.go index 327ca535..cf2ae337 100644 --- a/pkg/client/operation.go +++ b/pkg/deploy/operation.go @@ -1,4 +1,4 @@ -package client +package deploy import ( "context" @@ -14,7 +14,7 @@ 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 + Execute(ctx context.Context, cli Client) error // Format returns a human-readable representation of the operation. Format(resolver NameResolver) string String() string @@ -33,7 +33,7 @@ type RunContainerOperation struct { MachineID string } -func (o *RunContainerOperation) Execute(ctx context.Context, cli *Client) error { +func (o *RunContainerOperation) Execute(ctx context.Context, cli Client) error { resp, err := cli.CreateContainer(ctx, o.ServiceID, o.Spec, o.MachineID) if err != nil { return fmt.Errorf("create container: %w", err) @@ -64,7 +64,7 @@ type StopContainerOperation struct { MachineID string } -func (o *StopContainerOperation) Execute(ctx context.Context, cli *Client) error { +func (o *StopContainerOperation) Execute(ctx context.Context, cli Client) error { if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, container.StopOptions{}); err != nil { return fmt.Errorf("stop container: %w", err) } @@ -88,7 +88,7 @@ type RemoveContainerOperation struct { MachineID string } -func (o *RemoveContainerOperation) Execute(ctx context.Context, cli *Client) error { +func (o *RemoveContainerOperation) Execute(ctx context.Context, cli Client) error { if err := cli.StopContainer(ctx, o.ServiceID, o.ContainerID, container.StopOptions{}); err != nil { return fmt.Errorf("stop container: %w", err) } @@ -114,7 +114,7 @@ type SequenceOperation struct { Operations []Operation } -func (o *SequenceOperation) Execute(ctx context.Context, cli *Client) error { +func (o *SequenceOperation) Execute(ctx context.Context, cli Client) error { for _, op := range o.Operations { if err := op.Execute(ctx, cli); err != nil { return err @@ -140,49 +140,3 @@ func (o *SequenceOperation) String() string { return fmt.Sprintf("SequenceOperation[%s]", strings.Join(ops, ", ")) } - -// MapNameResolver resolves machine and container IDs to their names using a static map. -type MapNameResolver struct { - machines map[string]string - containers map[string]string -} - -func NewNameResolver(machines, containers map[string]string) *MapNameResolver { - return &MapNameResolver{ - machines: machines, - containers: containers, - } -} - -func (r *MapNameResolver) MachineName(machineID string) string { - if name, ok := r.machines[machineID]; ok { - return name - } - return machineID -} - -func (r *MapNameResolver) ContainerName(containerID string) string { - if name, ok := r.containers[containerID]; ok { - return name - } - return containerID -} - -// ServiceOperationNameResolver returns a machine and container name resolver for a service that can be used to format -// deployment operations. -func (cli *Client) ServiceOperationNameResolver(ctx context.Context, svc api.Service) (*MapNameResolver, error) { - machines, err := cli.ListMachines(ctx) - if err != nil { - return nil, fmt.Errorf("list machines: %w", err) - } - machineNames := make(map[string]string, len(machines)) - for _, m := range machines { - machineNames[m.Machine.Id] = m.Machine.Name - } - containerNames := make(map[string]string, len(svc.Containers)) - for _, c := range svc.Containers { - containerNames[c.Container.ID] = c.Container.NameWithoutSlash() - } - - return NewNameResolver(machineNames, containerNames), nil -} diff --git a/pkg/deploy/resolver.go b/pkg/deploy/resolver.go new file mode 100644 index 00000000..f5cf612f --- /dev/null +++ b/pkg/deploy/resolver.go @@ -0,0 +1,133 @@ +package deploy + +import ( + "fmt" + "github.com/distribution/reference" + "github.com/psviderski/uncloud/internal/secret" + "github.com/psviderski/uncloud/pkg/api" + "strings" +) + +type ImageDigestResolver interface { + Resolve(image string) (string, error) +} + +// ServiceSpecResolver transforms user-provided service specs into deployment-ready form. +type ServiceSpecResolver struct { + ClusterDomain string + ImageResolver ImageDigestResolver +} + +// Resolve transforms a service spec into its fully resolved form ready for deployment. +func (r *ServiceSpecResolver) Resolve(spec *api.ServiceSpec) error { + if err := spec.Validate(); err != nil { + return fmt.Errorf("invalid service spec: %w", err) + } + + steps := []func(*api.ServiceSpec) error{ + r.applyDefaults, + r.resolveServiceName, + r.resolveImageDigest, + r.expandIngressPorts, + } + + for _, step := range steps { + if err := step(spec); err != nil { + return err + } + } + + return nil +} + +func (r *ServiceSpecResolver) applyDefaults(spec *api.ServiceSpec) error { + if spec.Mode == "" { + spec.Mode = api.ServiceModeReplicated + } + // Ensure the replicated service has at least one replica. + if spec.Mode == api.ServiceModeReplicated && spec.Replicas == 0 { + spec.Replicas = 1 + } + + return nil +} + +func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error { + if spec.Name != "" { + return nil + } + + // Generate a random service name from the image when not provided. + img, err := reference.ParseDockerRef(spec.Container.Image) + if err != nil { + return 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 fmt.Errorf("generate random suffix: %w", err) + } + spec.Name = fmt.Sprintf("%s-%s", imageName, suffix) + + 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. +// This ensures every ingress port is accessible via the cluster domain, while preserving any custom domains specified +// by the user. +func (r *ServiceSpecResolver) expandIngressPorts(spec *api.ServiceSpec) error { + for i, port := range spec.Ports { + if port.Protocol != api.ProtocolHTTP && port.Protocol != api.ProtocolHTTPS { + continue + } + + if port.Hostname == "" { + if r.ClusterDomain == "" { + return fmt.Errorf("cluster domain must be reserved to generate hostname for ingress port: %d/%s", + port.ContainerPort, port.Protocol) + } + // Assign the default hostname (service-name.cluster-domain). + spec.Ports[i].Hostname = fmt.Sprintf("%s.%s", spec.Name, r.ClusterDomain) + } else { + if r.ClusterDomain == "" { + // When no cluster domain is reserved, use only the provided hostname. + continue + } + + if strings.HasSuffix(port.Hostname, "."+r.ClusterDomain) { + // If the hostname is already a cluster subdomain, use as is. + continue + } + // For external domains, duplicate the port with a service-name.cluster-domain hostname so the service + // can be accessed via both hostnames. + newPort := port + newPort.Hostname = fmt.Sprintf("%s.%s", spec.Name, r.ClusterDomain) + spec.Ports = append(spec.Ports, newPort) + } + } + + return nil +} diff --git a/pkg/client/strategy.go b/pkg/deploy/strategy.go similarity index 97% rename from pkg/client/strategy.go rename to pkg/deploy/strategy.go index 29bc0d11..53a8a4db 100644 --- a/pkg/client/strategy.go +++ b/pkg/deploy/strategy.go @@ -1,4 +1,4 @@ -package client +package deploy import ( "context" @@ -17,7 +17,7 @@ type Strategy interface { Type() string // 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) (Plan, error) + Plan(ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec) (Plan, error) } // RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time @@ -32,7 +32,7 @@ func (s *RollingStrategy) Type() string { } func (s *RollingStrategy) Plan( - ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, + ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec, ) (Plan, error) { // We can assume that the spec is valid at this point because it has been validated by the deployment. switch spec.Mode { @@ -49,7 +49,7 @@ func (s *RollingStrategy) Plan( // For replicated services, we want to maintain a specific number of containers (replicas) across the available machines // in the cluster. func (s *RollingStrategy) planReplicated( - ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, + ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec, ) (Plan, error) { plan, err := newEmptyPlan(svc, spec) if err != nil { @@ -211,7 +211,7 @@ func (s *RollingStrategy) planReplicated( // It handles multiple containers per machine (though this should not occur in normal operation) and skips machines // that are down. func (s *RollingStrategy) planGlobal( - ctx context.Context, cli *Client, svc *api.Service, spec api.ServiceSpec, + ctx context.Context, cli api.MachineClient, svc *api.Service, spec api.ServiceSpec, ) (Plan, error) { plan, err := newEmptyPlan(svc, spec) if err != nil { diff --git a/test/e2e/assert.go b/test/e2e/assert.go index d6817c5c..30a8db5b 100644 --- a/test/e2e/assert.go +++ b/test/e2e/assert.go @@ -3,7 +3,7 @@ package e2e import ( mapset "github.com/deckarep/golang-set/v2" "github.com/psviderski/uncloud/pkg/api" - "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/deploy" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "testing" @@ -23,9 +23,9 @@ func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpe } func assertContainerMatchesSpec(t *testing.T, ctr api.Container, spec api.ServiceSpec) { - status, err := client.CompareContainerToSpec(ctr, spec) + status, err := deploy.CompareContainerToSpec(ctr, spec) require.NoError(t, err) - assert.Equal(t, client.ContainerUpToDate, status) + assert.Equal(t, deploy.ContainerUpToDate, status) } // serviceContainersByMachine returns a map of machine ID to service containers on that machine. diff --git a/test/e2e/compose_test.go b/test/e2e/compose_test.go index 59156556..b56001f1 100644 --- a/test/e2e/compose_test.go +++ b/test/e2e/compose_test.go @@ -5,7 +5,6 @@ import ( "errors" "github.com/psviderski/uncloud/internal/ucind" "github.com/psviderski/uncloud/pkg/api" - "github.com/psviderski/uncloud/pkg/client" "github.com/psviderski/uncloud/pkg/compose" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -28,7 +27,7 @@ func TestComposeDeployment(t *testing.T) { name := "basic" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 2a3f9ef4..4eebbca2 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -11,6 +11,7 @@ import ( "github.com/psviderski/uncloud/internal/ucind" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/deploy" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "net/netip" @@ -42,12 +43,12 @@ func TestDeployment(t *testing.T) { name := "global-deployment" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } _, err = cli.InspectService(ctx, name) - require.ErrorIs(t, err, client.ErrNotFound) + require.ErrorIs(t, err, api.ErrNotFound) }) spec := api.ServiceSpec{ @@ -57,19 +58,17 @@ func TestDeployment(t *testing.T) { Image: "portainer/pause:latest", }, } - deploy, err := cli.NewDeployment(spec, nil) + deployment := cli.NewDeployment(spec, nil) + err = deployment.Validate(ctx) require.NoError(t, err) - err = deploy.Validate(ctx) - require.NoError(t, err) - - plan, err := deploy.Plan(ctx) + plan, err := deployment.Plan(ctx) require.NoError(t, err) assert.NotEmpty(t, plan.ServiceID) assert.Equal(t, name, plan.ServiceName) assert.Len(t, plan.SequenceOperation.Operations, 3) // 3 run - runPlan, err := deploy.Run(ctx) + runPlan, err := deployment.Run(ctx) require.NoError(t, err) assert.Equal(t, plan, runPlan) @@ -99,14 +98,12 @@ func TestDeployment(t *testing.T) { }, }, } - deploy, err = cli.NewDeployment(specWithPort, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(specWithPort, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Len(t, plan.SequenceOperation.Operations, 6) // 3 run + 3 remove - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -140,14 +137,12 @@ func TestDeployment(t *testing.T) { }, }, } - deploy, err = cli.NewDeployment(specWithPortAndInit, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(specWithPortAndInit, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Len(t, plan.SequenceOperation.Operations, 9) // 3 stop + 3 run + 3 remove - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -164,14 +159,12 @@ func TestDeployment(t *testing.T) { // Deploying the same spec should be a no-op. initialContainers = containers - deploy, err = cli.NewDeployment(specWithPortAndInit, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(specWithPortAndInit, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Len(t, plan.SequenceOperation.Operations, 0) // no-op - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -187,7 +180,7 @@ func TestDeployment(t *testing.T) { name := "global-deployment-filtered" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -201,10 +194,8 @@ func TestDeployment(t *testing.T) { }, } - deploy, err := cli.NewDeployment(spec, nil) - require.NoError(t, err) - - _, err = deploy.Run(ctx) + deployment := cli.NewDeployment(spec, nil) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err := cli.InspectService(ctx, name) @@ -225,12 +216,10 @@ func TestDeployment(t *testing.T) { filter := func(m *pb.MachineInfo) bool { return m.Name == c.Machines[0].Name || m.Name == c.Machines[2].Name } - strategy := &client.RollingStrategy{MachineFilter: filter} + strategy := &deploy.RollingStrategy{MachineFilter: filter} - deploy, err = cli.NewDeployment(specWithInit, strategy) - require.NoError(t, err) - - _, err = deploy.Run(ctx) + deployment = cli.NewDeployment(specWithInit, strategy) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -275,10 +264,8 @@ func TestDeployment(t *testing.T) { }, } - deploy, err = cli.NewDeployment(specWithPort, nil) - require.NoError(t, err) - - _, err = deploy.Run(ctx) + deployment = cli.NewDeployment(specWithPort, nil) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -302,15 +289,15 @@ func TestDeployment(t *testing.T) { t.Run("caddy", func(t *testing.T) { t.Cleanup(func() { err := cli.RemoveService(ctx, client.CaddyServiceName) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) - deploy, err := cli.NewCaddyDeployment("", nil) + deployment, err := cli.NewCaddyDeployment("", nil) require.NoError(t, err) - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err := cli.InspectService(ctx, client.CaddyServiceName) @@ -349,7 +336,7 @@ func TestDeployment(t *testing.T) { t.Run("caddy with machine filter", func(t *testing.T) { t.Cleanup(func() { err := cli.RemoveService(ctx, client.CaddyServiceName) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -359,10 +346,10 @@ func TestDeployment(t *testing.T) { return m.Name == c.Machines[0].Name } - deploy, err := cli.NewCaddyDeployment("", filter) + deployment, err := cli.NewCaddyDeployment("", filter) require.NoError(t, err) - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err := cli.InspectService(ctx, client.CaddyServiceName) @@ -379,10 +366,10 @@ func TestDeployment(t *testing.T) { return m.Name == c.Machines[0].Name || m.Name == c.Machines[2].Name } - deploy, err = cli.NewCaddyDeployment("", filter) + deployment, err = cli.NewCaddyDeployment("", filter) require.NoError(t, err) - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, client.CaddyServiceName) @@ -409,7 +396,7 @@ func TestDeployment(t *testing.T) { name := "replicated-deployment" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -424,19 +411,17 @@ func TestDeployment(t *testing.T) { Replicas: 2, } - deploy, err := cli.NewDeployment(spec, nil) + deployment := cli.NewDeployment(spec, nil) + err = deployment.Validate(ctx) require.NoError(t, err) - err = deploy.Validate(ctx) - require.NoError(t, err) - - plan, err := deploy.Plan(ctx) + plan, err := deployment.Plan(ctx) require.NoError(t, err) assert.NotEmpty(t, plan.ServiceID) assert.Equal(t, name, plan.ServiceName) assert.Len(t, plan.SequenceOperation.Operations, 2) // 2 run operations for 2 replicas - runPlan, err := deploy.Run(ctx) + runPlan, err := deployment.Run(ctx) require.NoError(t, err) assert.Equal(t, plan, runPlan) @@ -455,14 +440,12 @@ func TestDeployment(t *testing.T) { updatedSpec := spec updatedSpec.Container.Init = &init - deploy, err = cli.NewDeployment(updatedSpec, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(updatedSpec, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Len(t, plan.Operations, 4, "Expected 2 run + 2 remove operations") - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -484,14 +467,12 @@ func TestDeployment(t *testing.T) { threeReplicaSpec := updatedSpec threeReplicaSpec.Replicas = 3 - deploy, err = cli.NewDeployment(threeReplicaSpec, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(threeReplicaSpec, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Len(t, plan.Operations, 1, "Expected 1 run operation") - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -511,14 +492,12 @@ func TestDeployment(t *testing.T) { fourReplicaSpec.Container.Command = []string{"updated"} fourReplicaSpec.Replicas = 5 - deploy, err = cli.NewDeployment(fourReplicaSpec, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(fourReplicaSpec, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Len(t, plan.Operations, 8, "Expected 5 run + 3 remove operations") - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -540,14 +519,12 @@ func TestDeployment(t *testing.T) { // 5. Redeploy the exact same spec and verify it's a noop. initialContainers = containers // Reset container tracking. - deploy, err = cli.NewDeployment(fourReplicaSpec, nil) - require.NoError(t, err) - - plan, err = deploy.Plan(ctx) + deployment = cli.NewDeployment(fourReplicaSpec, nil) + plan, err = deployment.Plan(ctx) require.NoError(t, err) assert.Empty(t, plan.Operations, "Redeploying the same spec should be a no-op") - _, err = deploy.Run(ctx) + _, err = deployment.Run(ctx) require.NoError(t, err) svc, err = cli.InspectService(ctx, name) @@ -563,7 +540,7 @@ func TestDeployment(t *testing.T) { name := "replicated-deployment-filtered" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -581,12 +558,10 @@ func TestDeployment(t *testing.T) { machine01Filter := func(m *pb.MachineInfo) bool { return m.Name == c.Machines[0].Name || m.Name == c.Machines[1].Name } - strategy := &client.RollingStrategy{MachineFilter: machine01Filter} + strategy := &deploy.RollingStrategy{MachineFilter: machine01Filter} - deploy, err := cli.NewDeployment(spec, strategy) - require.NoError(t, err) - - _, err = deploy.Run(ctx) + deployment := cli.NewDeployment(spec, strategy) + _, err = deployment.Run(ctx) require.NoError(t, err) // Verify service has 2 containers on machines 0 and 1. @@ -612,11 +587,9 @@ func TestDeployment(t *testing.T) { return m.Name == c.Machines[2].Name } - strategy = &client.RollingStrategy{MachineFilter: machine2Filter} - deploy, err = cli.NewDeployment(spec, strategy) - require.NoError(t, err) - - _, err = deploy.Run(ctx) + strategy = &deploy.RollingStrategy{MachineFilter: machine2Filter} + deployment = cli.NewDeployment(spec, strategy) + _, err = deployment.Run(ctx) require.NoError(t, err) // Verify service now has containers only on machine 2. @@ -663,7 +636,7 @@ func TestServiceLifecycle(t *testing.T) { t.Cleanup(func() { err := cli.RemoveContainer(ctx, serviceID, resp.ID, container.RemoveOptions{Force: true}) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -738,7 +711,7 @@ func TestServiceLifecycle(t *testing.T) { t.Cleanup(func() { err := cli.RemoveContainer(ctx, serviceID, resp.ID, container.RemoveOptions{Force: true}) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -803,7 +776,7 @@ func TestServiceLifecycle(t *testing.T) { t.Cleanup(func() { err := cli.RemoveContainer(ctx, serviceID, ctr.ID, container.RemoveOptions{Force: true}) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -819,7 +792,7 @@ func TestServiceLifecycle(t *testing.T) { require.NoError(t, err) err = cli.RemoveContainer(ctx, serviceID, ctr.ID, container.RemoveOptions{}) - require.ErrorIs(t, err, client.ErrNotFound) + require.ErrorIs(t, err, api.ErrNotFound) }) t.Run("1 replica", func(t *testing.T) { @@ -828,12 +801,12 @@ func TestServiceLifecycle(t *testing.T) { name := "1-replica" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } _, err = cli.InspectService(ctx, name) - require.ErrorIs(t, err, client.ErrNotFound) + require.ErrorIs(t, err, api.ErrNotFound) }) resp, err := cli.RunService(ctx, api.ServiceSpec{ @@ -878,7 +851,7 @@ func TestServiceLifecycle(t *testing.T) { name := "1-replica-ports" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } }) @@ -930,7 +903,7 @@ func TestServiceLifecycle(t *testing.T) { name := "global" t.Cleanup(func() { err := cli.RemoveService(ctx, name) - if !errors.Is(err, client.ErrNotFound) { + if !errors.Is(err, api.ErrNotFound) { require.NoError(t, err) } })