feat(replicas): support --replicas and --machines for run command

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

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
+	command  []string
-	image   string
+	image    string
-	machine string
+	machines []string
-	mode    string
+	mode     string
-	name    string
+	name     string
-	publish []string
+	publish  []string
-	volumes []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,
+		Mode:     opts.mode,
-		Name:  opts.name,
+		Name:     opts.name,
-		Ports: ports,
+		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)
 	}

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:]
 }

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
 }

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
 }

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

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
 		}
 

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 {
+		deploy, err := cli.NewDeployment(spec, &RollingStrategy{MachineFilter: filter})
-		case "", api.ServiceModeReplicated:
+		if err != nil {
-			resp, err = cli.runReplicatedService(ctx, serviceID, spec)
+			return fmt.Errorf("create deployment: %w", err)
-		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)
 		}
 
-		return err
+		serviceID, err := deploy.Run(ctx)
-	}, cli.progressOut(), "Running service "+spec.Name)
+		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) {