update run command to run a stub service container on the connected node

cmd/uncloud/run.go

@@ -1,19 +1,16 @@
 package main
 
 import (
-	"fmt"
 	"github.com/spf13/cobra"
 	"uncloud/internal/cli"
 )
 
-type runOptions struct {
-	name    string
-	publish []string
-	cluster string
-}
-
 func NewRunCommand() *cobra.Command {
-	opts := runOptions{}
+	var (
+		cluster string
+		opts    cli.ServiceOptions
+	)
+
 	cmd := &cobra.Command{
 		Use:   "run IMAGE",
 		Short: "Run a service in a cluster.",
@@ -21,21 +18,20 @@ func NewRunCommand() *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			uncli := cmd.Context().Value("cli").(*cli.CLI)
 
-			image := args[0]
+			opts.Image = args[0]
-
+			return uncli.RunService(cmd.Context(), cluster, &opts)
-			//return uncli.RunService(cmd.Context(), ...)
-			return fmt.Errorf("not implemented: run image %q %s", image, uncli)
 		},
 	}
-	cmd.Flags().StringVarP(&opts.name, "name", "n", "",
+
+	cmd.Flags().StringVarP(&opts.Name, "name", "n", "",
 		"Assign a name to the service. A random name is generated if not specified.")
-	cmd.Flags().StringVarP(
+	cmd.Flags().StringSliceVarP(&opts.Publish, "publish", "p", nil,
-		&opts.cluster, "cluster", "c", "",
-		"Name of the cluster to run the service in. (default is the current cluster)",
-	)
-	cmd.Flags().StringSliceVarP(&opts.publish, "publish", "p", nil,
 		"Publish a service port to make it accessible outside the cluster. Can be specified multiple times. "+
 			"Format: [load_balancer_port:]container_port[/protocol]")
+	cmd.Flags().StringVarP(
+		&cluster, "cluster", "c", "",
+		"Name of the cluster to run the service in. (default is the current cluster)",
+	)
 
 	return cmd
 }

internal/cli/client/client.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"google.golang.org/grpc"
 	"uncloud/internal/machine/api/pb"
+	"uncloud/internal/machine/docker"
 )
 
 // Client is a client for the machine API.
@@ -15,8 +16,12 @@ type Client struct {
 
 	pb.MachineClient
 	pb.ClusterClient
+	*DockerClient
 }
 
+// DockerClient is a type alias for the Docker client to embed it in Client with a more specific name.
+type DockerClient = docker.Client
+
 // Connector is an interface for establishing a connection to the machine API.
 type Connector interface {
 	Connect(ctx context.Context) (*grpc.ClientConn, error)
@@ -37,10 +42,10 @@ func New(ctx context.Context, connector Connector) (*Client, error) {
 
 	c.MachineClient = pb.NewMachineClient(c.conn)
 	c.ClusterClient = pb.NewClusterClient(c.conn)
+	c.DockerClient = docker.NewClient(c.conn)
 	return c, nil
 }
 
 func (c *Client) Close() error {
-	err := c.conn.Close()
+	return errors.Join(c.conn.Close(), c.connector.Close())
-	return errors.Join(err, c.connector.Close())
 }

internal/cli/service.go

@@ -0,0 +1,41 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"github.com/docker/docker/api/types/container"
+)
+
+// ServiceOptions contains all the options for creating a service.
+type ServiceOptions struct {
+	Image   string
+	Name    string
+	Publish []string
+}
+
+func (cli *CLI) RunService(ctx context.Context, clusterName string, opts *ServiceOptions) error {
+	c, err := cli.ConnectCluster(ctx, clusterName)
+	if err != nil {
+		return fmt.Errorf("connect to cluster: %w", err)
+	}
+	defer func() {
+		_ = c.Close()
+	}()
+
+	config := &container.Config{
+		Image: opts.Image,
+	}
+	// TODO: generate a container name from the service name.
+	// TODO: set service labels on the container.
+	resp, err := c.CreateContainer(ctx, config, nil, nil, nil, opts.Name)
+	if err != nil {
+		return fmt.Errorf("create container: %w", err)
+	}
+
+	if err = c.StartContainer(ctx, resp.ID, container.StartOptions{}); err != nil {
+		return fmt.Errorf("start container: %w", err)
+	}
+
+	fmt.Printf("Service %q started with container ID %q\n", opts.Name, resp.ID)
+	return nil
+}

internal/machine/machine.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/docker/docker/client"
 	"github.com/docker/go-connections/sockets"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
@@ -21,6 +22,7 @@ import (
 	"uncloud/internal/machine/api/pb"
 	"uncloud/internal/machine/cluster"
 	"uncloud/internal/machine/corroservice"
+	"uncloud/internal/machine/docker"
 	"uncloud/internal/machine/network"
 	"uncloud/internal/machine/store"
 )
@@ -82,6 +84,7 @@ type Machine struct {
 	// store is the cluster store backed by a distributed Corrosion database.
 	store       *store.Store
 	cluster     *cluster.Cluster
+	docker      *docker.Server
 	localServer *grpc.Server
 }
 
@@ -122,6 +125,13 @@ func NewMachine(config *Config) (*Machine, error) {
 	corroStore := store.New(corro)
 	c := cluster.NewCluster(corroStore)
 
+	// Init a gRPC Docker server that proxies requests to the local Docker daemon.
+	dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+	if err != nil {
+		return nil, fmt.Errorf("create Docker client: %w", err)
+	}
+	dockerServer := docker.NewServer(dockerCli)
+
 	m := &Machine{
 		config:      *config,
 		state:       state,
@@ -129,8 +139,9 @@ func NewMachine(config *Config) (*Machine, error) {
 		initialised: make(chan struct{}, 1),
 		store:       corroStore,
 		cluster:     c,
+		docker:      dockerServer,
 	}
-	m.localServer = newGRPCServer(m, c)
+	m.localServer = newGRPCServer(m, c, dockerServer)
 
 	if m.Initialised() {
 		m.initialised <- struct{}{}
@@ -139,10 +150,11 @@ func NewMachine(config *Config) (*Machine, error) {
 	return m, nil
 }
 
-func newGRPCServer(m pb.MachineServer, c pb.ClusterServer) *grpc.Server {
+func newGRPCServer(m pb.MachineServer, c pb.ClusterServer, d pb.DockerServer) *grpc.Server {
 	s := grpc.NewServer()
 	pb.RegisterMachineServer(s, m)
 	pb.RegisterClusterServer(s, c)
+	pb.RegisterDockerServer(s, d)
 	return s
 }
 
@@ -223,7 +235,7 @@ func (m *Machine) Run(ctx context.Context) error {
 					slog.Info("Configured corrosion service.", "dir", m.config.CorrosionDir)
 
 					slog.Info("Starting network controller.")
-					networkServer := newGRPCServer(m, m.cluster)
+					networkServer := newGRPCServer(m, m.cluster, m.docker)
 					ctrl, err = newNetworkController(m.state, m.store, networkServer, m.config.CorrosionService)
 					if err != nil {
 						return fmt.Errorf("initialise network controller: %w", err)