From cd60d5178f29886c955bc64341694b0ec0a00476 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Thu, 28 Nov 2024 16:15:57 +1000 Subject: [PATCH] ucind: create machines in containers with published API ports --- cmd/ucind/cluster/create.go | 10 +- internal/ucind/cluster.go | 70 ++++++++++---- internal/ucind/machine.go | 180 ++++++++++++++++++++++++++++++++++++ 3 files changed, 237 insertions(+), 23 deletions(-) create mode 100644 internal/ucind/machine.go diff --git a/cmd/ucind/cluster/create.go b/cmd/ucind/cluster/create.go index fe31a931..f91de85a 100644 --- a/cmd/ucind/cluster/create.go +++ b/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 } diff --git a/internal/ucind/cluster.go b/internal/ucind/cluster.go index 373e60e2..00dbb598 100644 --- a/internal/ucind/cluster.go +++ b/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 nil + return c, 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{ - Name: name, - }, nil + return c, 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 diff --git a/internal/ucind/machine.go b/internal/ucind/machine.go new file mode 100644 index 00000000..78ca347f --- /dev/null +++ b/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 +}