ucind: create machines in containers with published API ports

cmd/ucind/cluster/create.go

@@ -6,11 +6,8 @@ import (
 	"uncloud/internal/ucind"
 )
 
-type createOptions struct {
-}
-
 func NewCreateCommand() *cobra.Command {
-	//opts := createOptions{}
+	opts := ucind.CreateClusterOptions{}
 	cmd := &cobra.Command{
 		Use:   "create [NAME]",
 		Short: "Create a new cluster.",
@@ -23,12 +20,15 @@ func NewCreateCommand() *cobra.Command {
 				name = args[0]
 			}
 
-			if err := p.CreateCluster(cmd.Context(), name, ucind.CreateClusterOptions{}); err != nil {
+			if _, err := p.CreateCluster(cmd.Context(), name, opts); err != nil {
 				return fmt.Errorf("create cluster '%s': %w", name, err)
 			}
 			fmt.Printf("Cluster '%s' created.\n", name)
 			return nil
 		},
 	}
+
+	cmd.Flags().IntVarP(&opts.Machines, "machines", "m", 1, "Number of machines to create.")
+
 	return cmd
 }

internal/ucind/cluster.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/api/types/network"
 	"github.com/docker/docker/client"
 )
@@ -14,20 +16,23 @@ const (
 )
 
 type Cluster struct {
-	Name string
+	Name     string
+	Machines []Machine
 }
 
 type CreateClusterOptions struct {
 	Machines int
 }
 
-func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts CreateClusterOptions) error {
+func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts CreateClusterOptions) (Cluster, error) {
+	var c Cluster
+
 	_, err := p.InspectCluster(ctx, name)
 	if err == nil {
-		return fmt.Errorf("cluster with name '%s' already exists", name)
+		return c, fmt.Errorf("cluster with name '%s' already exists", name)
 	}
 	if !errors.Is(err, ErrNotFound) {
-		return fmt.Errorf("inspect cluster '%s': %w", name, err)
+		return c, fmt.Errorf("inspect cluster '%s': %w", name, err)
 	}
 
 	netOpts := network.CreateOptions{
@@ -36,36 +41,49 @@ func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts Creat
 			ManagedLabel:     "",
 		},
 	}
-	// Docker network name is the same as the cluster name.
+	// Create a Docker network with the same as the cluster name.
 	if _, err = p.client.NetworkCreate(ctx, name, netOpts); err != nil {
-		return fmt.Errorf("create Docker network '%s': %w", name, err)
+		return c, fmt.Errorf("create Docker network '%s': %w", name, err)
+	}
+	c.Name = name
+
+	// Create machines (containers) in the created cluster network.
+	for i := 1; i < opts.Machines+1; i++ {
+		mopts := CreateMachineOptions{
+			Name: fmt.Sprintf("machine-%d", i),
+		}
+		m, err := p.CreateMachine(ctx, name, mopts)
+		if err != nil {
+			return c, fmt.Errorf("create machine '%s': %w", mopts.Name, err)
+		}
+
+		c.Machines = append(c.Machines, m)
 	}
 
-	// TODO: create machines (containers) with the cluster name label.
+	return c, nil
-
-	return nil
 }
 
-func (p *Provisioner) InspectCluster(ctx context.Context, name string) (*Cluster, error) {
+func (p *Provisioner) InspectCluster(ctx context.Context, name string) (Cluster, error) {
+	var c Cluster
+
 	// Docker network name is the same as the cluster name.
 	net, err := p.client.NetworkInspect(ctx, name, network.InspectOptions{})
 	if err != nil {
 		if client.IsErrNotFound(err) {
-			return nil, ErrNotFound
+			return c, ErrNotFound
 		}
-		return nil, fmt.Errorf("inspect Docker network '%s': %w", name, err)
+		return c, fmt.Errorf("inspect Docker network '%s': %w", name, err)
 	}
 
 	if _, ok := net.Labels[ManagedLabel]; !ok {
 		// The network with the cluster name exists, but it's not managed by ucind.
-		return nil, ErrNotFound
+		return c, ErrNotFound
 	}
 
+	c.Name = name
 	// TODO: list containers (machines) with the cluster name label and include them in the cluster struct.
 
-	return &Cluster{
+	return c, nil
-		Name: name,
-	}, nil
 }
 
 func (p *Provisioner) RemoveCluster(ctx context.Context, name string) error {
@@ -73,9 +91,25 @@ func (p *Provisioner) RemoveCluster(ctx context.Context, name string) error {
 		return err
 	}
 
-	// TODO: remove machines (containers) with the cluster name label.
+	// Remove all containers (machines) with the cluster name label.
+	opts := container.ListOptions{
+		All: true,
+		Filters: filters.NewArgs(
+			filters.Arg("label", ClusterNameLabel+"="+name),
+			filters.Arg("label", ManagedLabel),
+		),
+	}
+	containers, err := p.client.ContainerList(ctx, opts)
+	if err != nil {
+		return fmt.Errorf("list Docker containers with cluster name '%s': %w", name, err)
+	}
+	for _, c := range containers {
+		if err = p.client.ContainerRemove(ctx, c.ID, container.RemoveOptions{Force: true}); err != nil {
+			return fmt.Errorf("remove Docker container '%s': %w", c.ID, err)
+		}
+	}
 
-	if err := p.client.NetworkRemove(ctx, name); err != nil {
+	if err = p.client.NetworkRemove(ctx, name); err != nil {
 		return fmt.Errorf("remove Docker network '%s': %w", name, err)
 	}
 	return nil

internal/ucind/machine.go

@@ -0,0 +1,180 @@
+package ucind
+
+import (
+	"context"
+	"crypto/rand"
+	"errors"
+	"fmt"
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/client"
+	"github.com/docker/go-connections/nat"
+	"io"
+	"math/big"
+	"net"
+	"net/netip"
+	"time"
+)
+
+const (
+	DefaultImage   = "ghcr.io/psviderski/ucind:latest"
+	UncloudAPIPort = 51000
+
+	MachineNameLabel = "ucind.machine.name"
+)
+
+type Machine struct {
+	ClusterName   string
+	ContainerName string
+	Name          string
+	APIAddress    netip.AddrPort
+}
+
+type CreateMachineOptions struct {
+	Name  string
+	Image string
+}
+
+func (p *Provisioner) CreateMachine(ctx context.Context, clusterName string, opts CreateMachineOptions) (Machine, error) {
+	var m Machine
+
+	machineName := opts.Name
+	if machineName == "" {
+		var err error
+		if machineName, err = p.generateMachineName(ctx, clusterName); err != nil {
+			return m, fmt.Errorf("generate machine name: %w", err)
+		}
+	}
+	containerName := clusterName + "-" + machineName
+
+	img := DefaultImage
+	if opts.Image != "" {
+		img = opts.Image
+	}
+
+	apiPort := nat.Port(fmt.Sprintf("%d/tcp", UncloudAPIPort))
+	config := &container.Config{
+		Image: img,
+		Labels: map[string]string{
+			ClusterNameLabel: clusterName,
+			MachineNameLabel: machineName,
+			ManagedLabel:     "",
+		},
+		ExposedPorts: nat.PortSet{
+			apiPort: struct{}{},
+		},
+	}
+	hostConfig := &container.HostConfig{
+		NetworkMode: container.NetworkMode(clusterName),
+		PortBindings: nat.PortMap{
+			apiPort: []nat.PortBinding{
+				{
+					HostIP: "127.0.0.1",
+					// Host port is a random available port.
+				},
+			},
+		},
+		Privileged: true,
+		RestartPolicy: container.RestartPolicy{
+			Name: container.RestartPolicyAlways,
+		},
+	}
+
+	if _, err := p.createContainerWithImagePull(ctx, containerName, config, hostConfig); err != nil {
+		return m, err
+	}
+	if err := p.client.ContainerStart(ctx, containerName, container.StartOptions{}); err != nil {
+		return m, fmt.Errorf("start Docker container: %w", err)
+	}
+
+	apiPortBindings, err := p.waitPortPublished(ctx, containerName, apiPort)
+	if err != nil {
+		return m, fmt.Errorf("wait for machine API port '%s' to be published: %w", apiPort, err)
+	}
+	apiAddr, err := netip.ParseAddrPort(net.JoinHostPort(apiPortBindings[0].HostIP, apiPortBindings[0].HostPort))
+	if err != nil {
+		return m, fmt.Errorf("parse machine API port binding: %w", err)
+	}
+
+	m = Machine{
+		ClusterName:   clusterName,
+		ContainerName: containerName,
+		Name:          machineName,
+		APIAddress:    apiAddr,
+	}
+	return m, nil
+}
+
+// createContainerWithImagePull creates a Docker container. If the image is missing, it pulls the image first.
+func (p *Provisioner) createContainerWithImagePull(
+	ctx context.Context, name string, config *container.Config, hostConfig *container.HostConfig,
+) (container.CreateResponse, error) {
+	var resp container.CreateResponse
+
+	_, err := p.client.ContainerCreate(ctx, config, hostConfig, nil, nil, name)
+	if err == nil {
+		return resp, nil
+	}
+
+	if !client.IsErrNotFound(err) {
+		return resp, fmt.Errorf("create Docker container: %w", err)
+	}
+
+	respBody, err := p.client.ImagePull(ctx, config.Image, image.PullOptions{})
+	if err != nil {
+		return resp, fmt.Errorf("pull Docker image: %w", err)
+	}
+	defer respBody.Close()
+
+	// Wait for pull to complete.
+	if _, err = io.Copy(io.Discard, respBody); err != nil {
+		return resp, fmt.Errorf("read Docker pull response: %w", err)
+	}
+
+	// Create container again after image pull.
+	if resp, err = p.client.ContainerCreate(ctx, config, hostConfig, nil, nil, name); err != nil {
+		return resp, fmt.Errorf("create Docker container: %w", err)
+	}
+
+	return resp, nil
+}
+
+// waitPortPublished waits for a Docker container port to be published on the host which happens asynchronously.
+func (p *Provisioner) waitPortPublished(ctx context.Context, containerID string, port nat.Port) ([]nat.PortBinding, error) {
+	ctx, cancel := context.WithDeadline(ctx, time.Now().Add(5*time.Second))
+	defer cancel()
+
+	for {
+		c, err := p.client.ContainerInspect(ctx, containerID)
+		if err != nil {
+			return nil, fmt.Errorf("inspect container: %w", err)
+		}
+
+		binding, ok := c.NetworkSettings.Ports[port]
+		if !ok {
+			return nil, fmt.Errorf("port '%s' not published", port)
+		}
+		if len(binding) > 0 {
+			return binding, nil
+		}
+
+		select {
+		case <-time.After(10 * time.Millisecond):
+		case <-ctx.Done():
+			if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+				return nil, errors.New("timeout")
+			}
+			return nil, ctx.Err()
+		}
+	}
+}
+
+func (p *Provisioner) generateMachineName(ctx context.Context, clusterName string) (string, error) {
+	// TODO: list existing containers and extract the last number X from machine-X names.
+	r, err := rand.Int(rand.Reader, big.NewInt(1000))
+	if err != nil {
+		return "", fmt.Errorf("generate machine name: %w", err)
+	}
+	i := r.Int64() + 10
+	return fmt.Sprintf("machine-%d", i), nil
+}