philipp/uncloud
1
0
Fork 0
mirror of https://github.com/psviderski/uncloud.git synced 2026-08-26 19:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: decouple deploy and compose packages from client dependency

Browse Source
This commit is contained in:
Pavel Sviderski
2025-03-24 14:38:16 +10:00
parent 1471a94162
commit 042dd594e0
27 changed files with 509 additions and 459 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/uncloud/caddy/deploy.go
+9 -7
View File
@@ -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
}
cmd/uncloud/deploy.go
+7 -6
View File
@@ -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)
cmd/uncloud/dns/reserve.go
+2 -1
View File
@@ -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
cmd/uncloud/dns/show.go
+2 -2
View File
@@ -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
cmd/uncloud/machine/add.go
+2 -1
View File
@@ -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 {
cmd/uncloud/service/run.go
+2 -2
View File
@@ -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 {
cmd/uncloud/service/scale.go
+3 -7
View File
@@ -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
internal/cli/cli.go
+2 -1
View File
@@ -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()
pkg/api/client.go
+37
View File
@@ -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)
}
pkg/api/error.go
+5
View File
@@ -0,0 +1,5 @@
package api
import "errors"
var ErrNotFound = errors.New("not found")
pkg/client/caddy.go
+3 -2
View File
@@ -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.
pkg/client/client.go
+3 -2
View File
@@ -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)
pkg/client/cluster.go
+2 -1
View File
@@ -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) {
pkg/client/container.go
+1 -24
View File
@@ -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
}
pkg/client/deploy.go
+4 -107
View File
@@ -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)
}
pkg/client/dns.go
+2 -1
View File
@@ -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
}
pkg/client/resolver.go
+30 -110
View File
@@ -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
}
pkg/client/service.go
+11 -13
View File
@@ -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)
pkg/compose/deploy.go
+21 -19
View File
@@ -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)
}
pkg/deploy/container.go
+29
View File
@@ -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
}
pkg/deploy/deploy.go
+120
View File
@@ -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)
}
pkg/client/operation.go → pkg/deploy/operation.go
+6 -52
View File
@@ -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
}
pkg/deploy/resolver.go
+133
View File
@@ -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
}
pkg/client/strategy.go → pkg/deploy/strategy.go
+5 -5
View File
@@ -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 {
test/e2e/assert.go
+3 -3
View File
@@ -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.
test/e2e/compose_test.go
+1 -2
View File
@@ -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)
}
})
test/e2e/service_test.go
+64 -91
View File
@@ -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)
}
})