From 9642863d7b097850918e2835d01aaa1aa34ef785 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Fri, 28 Feb 2025 17:53:58 +1000 Subject: [PATCH] feat(replicas): support --replicas and --machines for run command --- cmd/uncloud/caddy/deploy.go | 2 +- cmd/uncloud/service/run.go | 63 ++++++++++++----- internal/api/container.go | 1 + internal/api/service.go | 3 + internal/cli/client/deploy.go | 29 +------- internal/cli/client/operation.go | 1 + internal/cli/client/resolver.go | 17 ++++- internal/cli/client/service.go | 118 +++++-------------------------- 8 files changed, 87 insertions(+), 147 deletions(-) diff --git a/cmd/uncloud/caddy/deploy.go b/cmd/uncloud/caddy/deploy.go index 14a4ec6a..357684e3 100644 --- a/cmd/uncloud/caddy/deploy.go +++ b/cmd/uncloud/caddy/deploy.go @@ -211,7 +211,7 @@ func deploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error { fmt.Println() fmt.Println("DNS records updated to use only the internet-reachable machines running caddy service:") for _, r := range records { - fmt.Printf(" %s %s -> %s\n", r.Name, r.Type, strings.Join(r.Values, ", ")) + fmt.Printf(" %s %s → %s\n", r.Name, r.Type, strings.Join(r.Values, ", ")) } return nil diff --git a/cmd/uncloud/service/run.go b/cmd/uncloud/service/run.go index 6928ccbc..bd9feb6f 100644 --- a/cmd/uncloud/service/run.go +++ b/cmd/uncloud/service/run.go @@ -4,18 +4,23 @@ import ( "context" "fmt" "github.com/spf13/cobra" + "slices" + "strings" "uncloud/internal/api" "uncloud/internal/cli" + "uncloud/internal/cli/client" + "uncloud/internal/machine/api/pb" ) type runOptions struct { - command []string - image string - machine string - mode string - name string - publish []string - volumes []string + command []string + image string + machines []string + mode string + name string + publish []string + replicas uint + volumes []string cluster string } @@ -39,15 +44,13 @@ func NewRunCommand() *cobra.Command { }, } - // TODO: implement placement constraints and translate --machine to a constraint. - //cmd.Flags().StringVarP( - // &opts.machine, "machine", "m", "", - // "Name or ID of the machine to run the service on. (default is first available)", - //) cmd.Flags().StringVar(&opts.mode, "mode", api.ServiceModeReplicated, fmt.Sprintf("Replication mode of the service: either %q (a specified number of containers across "+ "the machines) or %q (one container on every machine).", api.ServiceModeReplicated, api.ServiceModeGlobal)) + cmd.Flags().StringSliceVarP(&opts.machines, "machine", "m", nil, + "Placement constraint by machine name, limiting which machines the service can run on. Can be specified "+ + "multiple times or as a comma-separated list of machine names. (default is any suitable machine)") cmd.Flags().StringVarP(&opts.name, "name", "n", "", "Assign a name to the service. A random name is generated if not specified.") cmd.Flags().StringSliceVarP(&opts.publish, "publish", "p", nil, @@ -60,6 +63,8 @@ func NewRunCommand() *cobra.Command { " -p app.example.com:8080/https Publish port 8080 as HTTPS via load balancer with custom hostname\n"+ " -p 9000:8080 Publish port 8080 as TCP port 9000 via load balancer\n"+ " -p 53:5353/udp@host Bind UDP port 5353 to host port 53") + cmd.Flags().UintVar(&opts.replicas, "replicas", 1, + "Number of containers to run for the service. Only valid for a replicated service.") cmd.Flags().StringSliceVarP(&opts.volumes, "volume", "v", nil, "Bind mount a host file or directory into a service container using the format "+ "/host/path:/container/path[:ro]. Can be specified multiple times.") @@ -74,11 +79,34 @@ func NewRunCommand() *cobra.Command { func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { switch opts.mode { - case "", api.ServiceModeReplicated, api.ServiceModeGlobal: + case api.ServiceModeReplicated, api.ServiceModeGlobal: default: return fmt.Errorf("invalid replication mode: %q", opts.mode) } + var machineFilter client.MachineFilter + if len(opts.machines) > 0 { + var machines []string + for _, value := range opts.machines { + if value == "" { + continue + } + + mlist := strings.Split(value, ",") + for _, m := range mlist { + if m = strings.TrimSpace(m); m != "" { + machines = append(machines, m) + } + } + } + + if len(machines) > 0 { + machineFilter = func(m *pb.MachineInfo) bool { + return slices.Contains(machines, m.Name) + } + } + } + ports := make([]api.PortSpec, len(opts.publish)) for i, publishPort := range opts.publish { port, err := api.ParsePortSpec(publishPort) @@ -95,9 +123,10 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { Image: opts.image, Volumes: opts.volumes, }, - Mode: opts.mode, - Name: opts.name, - Ports: ports, + Mode: opts.mode, + Name: opts.name, + Ports: ports, + Replicas: opts.replicas, } if err := spec.Validate(); err != nil { return fmt.Errorf("invalid service configuration: %w", err) @@ -109,7 +138,7 @@ func run(ctx context.Context, uncli *cli.CLI, opts runOptions) error { } defer client.Close() - resp, err := client.RunService(ctx, spec) + resp, err := client.RunService(ctx, spec, machineFilter) if err != nil { return fmt.Errorf("run service: %w", err) } diff --git a/internal/api/container.go b/internal/api/container.go index 0781399a..186811b8 100644 --- a/internal/api/container.go +++ b/internal/api/container.go @@ -21,6 +21,7 @@ type Container struct { } // NameWithoutSlash returns the container name without the leading slash. +// TODO: modify Name in original ContainerJSON structure when inspecting a Docker container and get rid of this method. func (c *Container) NameWithoutSlash() string { return c.Name[1:] } diff --git a/internal/api/service.go b/internal/api/service.go index 0d81aecf..987b17bd 100644 --- a/internal/api/service.go +++ b/internal/api/service.go @@ -23,6 +23,8 @@ type ServiceSpec struct { Name string // Ports defines what service ports to publish to make the service accessible outside the cluster. Ports []PortSpec + // Replicas is the number of containers to run for the service. Only valid for a replicated service. + Replicas uint } func (s *ServiceSpec) Validate() error { @@ -37,6 +39,7 @@ func (s *ServiceSpec) Validate() error { } // TODO: validate there is no conflict between ports. + // TODO: return error if there are non-HTTP/HTTPS ingress ports that we don't support yet. return nil } diff --git a/internal/cli/client/deploy.go b/internal/cli/client/deploy.go index 793e8ef2..14992bc6 100644 --- a/internal/cli/client/deploy.go +++ b/internal/cli/client/deploy.go @@ -4,11 +4,8 @@ import ( "context" "errors" "fmt" - "github.com/distribution/reference" - "strings" "uncloud/internal/api" "uncloud/internal/machine/api/pb" - "uncloud/internal/secret" ) // Deployment manages the process of creating or updating a service to match a desired state. @@ -35,29 +32,6 @@ 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 (cli *Client) NewDeployment(spec api.ServiceSpec, strategy Strategy) (*Deployment, error) { - if err := spec.Validate(); err != nil { - return nil, fmt.Errorf("invalid service spec: %w", err) - } - if spec.Name == "" { - // Generate a random service name from the image when not provided. - img, err := reference.ParseDockerRef(spec.Container.Image) - if err != nil { - return nil, 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 nil, fmt.Errorf("generate random suffix: %w", err) - } - spec.Name = fmt.Sprintf("%s-%s", imageName, suffix) - } - if strategy == nil { strategy = &RollingStrategy{} } @@ -118,6 +92,9 @@ func (d *Deployment) Validate(ctx context.Context) error { 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 } diff --git a/internal/cli/client/operation.go b/internal/cli/client/operation.go index 5ac28676..bc3cf595 100644 --- a/internal/cli/client/operation.go +++ b/internal/cli/client/operation.go @@ -17,6 +17,7 @@ type Operation interface { String() string } +// NameResolver resolves machine and container IDs to their names. type NameResolver interface { MachineName(machineID string) string ContainerName(containerID string) string diff --git a/internal/cli/client/resolver.go b/internal/cli/client/resolver.go index 26449803..67f6a4b0 100644 --- a/internal/cli/client/resolver.go +++ b/internal/cli/client/resolver.go @@ -24,6 +24,7 @@ func (r *ServiceSpecResolver) Resolve(spec *api.ServiceSpec) error { } steps := []func(*api.ServiceSpec) error{ + r.applyDefaults, r.resolveServiceName, r.expandIngressPorts, } @@ -37,6 +38,18 @@ func (r *ServiceSpecResolver) Resolve(spec *api.ServiceSpec) error { 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 @@ -63,14 +76,14 @@ func (r *ServiceSpecResolver) resolveServiceName(spec *api.ServiceSpec) error { return nil } -// expandIngressPorts processes ingress ports in a service spec by: +// 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.Mode != api.PortModeIngress { + if port.Protocol != api.ProtocolHTTP && port.Protocol != api.ProtocolHTTPS { continue } diff --git a/internal/cli/client/service.go b/internal/cli/client/service.go index 3420a372..17be70dc 100644 --- a/internal/cli/client/service.go +++ b/internal/cli/client/service.go @@ -14,7 +14,6 @@ import ( "sync" "uncloud/internal/api" "uncloud/internal/machine/api/pb" - "uncloud/internal/secret" ) func (cli *Client) PrepareDeploymentSpec(ctx context.Context, spec api.ServiceSpec) (api.ServiceSpec, error) { @@ -38,7 +37,9 @@ type RunServiceResponse struct { Name string } -func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunServiceResponse, error) { +func (cli *Client) RunService( + ctx context.Context, spec api.ServiceSpec, filter MachineFilter, +) (RunServiceResponse, error) { var resp RunServiceResponse if err := spec.Validate(); err != nil { @@ -61,112 +62,27 @@ func (cli *Client) RunService(ctx context.Context, spec api.ServiceSpec) (RunSer return resp, fmt.Errorf("prepare service spec ready for deployment: %w", err) } - serviceID, err := secret.NewID() - if err != nil { - return resp, fmt.Errorf("generate service ID: %w", err) - } - err = progress.RunWithTitle(ctx, func(ctx context.Context) error { - switch spec.Mode { - case "", api.ServiceModeReplicated: - resp, err = cli.runReplicatedService(ctx, serviceID, spec) - case api.ServiceModeGlobal: - deploy, err := cli.NewDeployment(spec, &RollingStrategy{}) - if err != nil { - return fmt.Errorf("create deployment: %w", err) - } - - serviceID, err = deploy.Run(ctx) - if err != nil { - return err - } - - resp.ID = serviceID - // TODO: get the service name from the plan when it's available. - resp.Name = spec.Name - - return nil - default: - return fmt.Errorf("invalid mode: %q", spec.Mode) + deploy, err := cli.NewDeployment(spec, &RollingStrategy{MachineFilter: filter}) + if err != nil { + return fmt.Errorf("create deployment: %w", err) } - return err - }, cli.progressOut(), "Running service "+spec.Name) + serviceID, err := deploy.Run(ctx) + if err != nil { + return err + } + + resp.ID = serviceID + // TODO: get the service name from the plan when it's available. + resp.Name = spec.Name + + return nil + }, cli.progressOut(), fmt.Sprintf("Running service %s (%s mode)", spec.Name, spec.Mode)) return resp, err } -func (cli *Client) runReplicatedService(ctx context.Context, id string, spec api.ServiceSpec) (RunServiceResponse, error) { - resp := RunServiceResponse{ - ID: id, - Name: spec.Name, - } - - // Find a machine to run a service replica on. - machines, err := cli.ListMachines(ctx) - if err != nil { - return resp, fmt.Errorf("list machines: %w", err) - } - - // TODO: support selecting a particular machine by ID or name through the user options. - //var machine *pb.MachineMember - //if opts.Machine != "" { - // // Check if the machine ID or name exists if it's explicitly specified. - // for _, m := range machines { - // if m.Machine.Name == opts.Machine || m.Machine.Id == opts.Machine { - // machine = m - // break - // } - // } - // if machine == nil { - // return resp, fmt.Errorf("machine %q not found", opts.Machine) - // } - //} - - m := firstAvailableMachine(machines) - if m == nil { - return resp, errors.New("no available machine to run the service") - } - - if _, err = cli.runContainer(ctx, id, spec, m.Machine); err != nil { - return resp, fmt.Errorf("run container: %w", err) - } - - return resp, nil -} - -func firstAvailableMachine(machines []*pb.MachineMember) *pb.MachineMember { - // Find the first UP machine. - for _, m := range machines { - if m.State == pb.MachineMember_UP { - return m - } - } - // There is no UP machine, try to find the first SUSPECT machine. - for _, m := range machines { - if m.State == pb.MachineMember_SUSPECT { - return m - } - } - - return nil -} - -func (cli *Client) runContainer( - ctx context.Context, serviceID string, spec api.ServiceSpec, machine *pb.MachineInfo, -) (container.CreateResponse, error) { - resp, err := cli.CreateContainer(ctx, serviceID, spec, machine.Name) - if err != nil { - return resp, fmt.Errorf("create container: %w", err) - } - - if err = cli.StartContainer(ctx, serviceID, resp.ID); err != nil { - return resp, fmt.Errorf("start container: %w", err) - } - - return resp, nil -} - // InspectService returns detailed information about a service and its containers. // The id parameter can be either a service ID or name. func (cli *Client) InspectService(ctx context.Context, id string) (api.Service, error) {