philipp/uncloud
1
0
Fork 0
mirror of https://github.com/psviderski/uncloud.git synced 2026-08-26 19:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

ucind: wait for machine to become ready through polling API

Browse Source
This commit is contained in:
Pavel Sviderski
2024-11-29 12:08:57 +10:00
parent bb967461d0
commit faf3378659
3 changed files with 67 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ucind/cluster.go
+23 -12
View File
@@ -49,7 +49,7 @@ func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts Creat
},
}
// Create a Docker network with the same as the cluster name.
if _, err = p.client.NetworkCreate(ctx, name, netOpts); err != nil {
if _, err = p.dockerCli.NetworkCreate(ctx, name, netOpts); err != nil {
return c, fmt.Errorf("create Docker network '%s': %w", name, err)
}
c.Name = name
@@ -79,11 +79,15 @@ func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts Creat
}
func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error {
// TODO: properly wait for the API to respond.
time.Sleep(2 * time.Second) // Wait for the machines to be up and running.
// Init a new cluster on the first machine.
initMachine := machines[0]
waitCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
if err := p.WaitMachineReady(waitCtx, initMachine); err != nil {
return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err)
}
initClient, err := client.New(ctx, connector.NewTCPConnector(initMachine.APIAddress))
if err != nil {
return fmt.Errorf("create machine client over TCP '%s': %w", initMachine.APIAddress, err)
@@ -102,14 +106,21 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
// Join the rest of the machines to the cluster.
for _, m := range machines[1:] {
mcli, err := client.New(ctx, connector.NewTCPConnector(m.APIAddress))
waitCtx, cancel = context.WithTimeout(ctx, 5*time.Second)
//goland:noinspection GoDeferInLoop
defer cancel()
if err = p.WaitMachineReady(waitCtx, m); err != nil {
return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err)
}
cli, err := client.New(ctx, connector.NewTCPConnector(m.APIAddress))
if err != nil {
return fmt.Errorf("create machine client over TCP '%s': %w", m.APIAddress, err)
}
//goland:noinspection GoDeferInLoop
defer mcli.Close()
defer cli.Close()
tokenResp, err := mcli.Token(ctx, &emptypb.Empty{})
tokenResp, err := cli.Token(ctx, &emptypb.Empty{})
if err != nil {
return fmt.Errorf("get machine token: %w", err)
}
@@ -140,7 +151,7 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
Machine: addResp.Machine,
OtherMachines: []*pb.MachineInfo{initResp.Machine},
}
if _, err = mcli.JoinCluster(ctx, joinReq); err != nil {
if _, err = cli.JoinCluster(ctx, joinReq); err != nil {
return fmt.Errorf("join cluster: %w", err)
}
@@ -154,7 +165,7 @@ func (p *Provisioner) InspectCluster(ctx context.Context, name string) (Cluster,
var c Cluster
// Docker network name is the same as the cluster name.
net, err := p.client.NetworkInspect(ctx, name, network.InspectOptions{})
net, err := p.dockerCli.NetworkInspect(ctx, name, network.InspectOptions{})
if err != nil {
if dockerclient.IsErrNotFound(err) {
return c, ErrNotFound
@@ -186,17 +197,17 @@ func (p *Provisioner) RemoveCluster(ctx context.Context, name string) error {
filters.Arg("label", ManagedLabel),
),
}
containers, err := p.client.ContainerList(ctx, opts)
containers, err := p.dockerCli.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 {
if err = p.dockerCli.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.dockerCli.NetworkRemove(ctx, name); err != nil {
return fmt.Errorf("remove Docker network '%s': %w", name, err)
}
return nil
internal/ucind/machine.go
+42 -18
View File
@@ -2,18 +2,21 @@ package ucind
import (
"context"
"crypto/rand"
"errors"
"fmt"
"github.com/cenkalti/backoff/v4"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/image"
"github.com/docker/docker/client"
dockerclient "github.com/docker/docker/client"
"github.com/docker/go-connections/nat"
"google.golang.org/protobuf/types/known/emptypb"
"io"
"math/big"
"net"
"net/netip"
"time"
"uncloud/internal/cli/client"
"uncloud/internal/cli/client/connector"
"uncloud/internal/secret"
)
const (
@@ -41,8 +44,8 @@ func (p *Provisioner) CreateMachine(ctx context.Context, clusterName string, opt
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)
if machineName, err = randomMachineName(); err != nil {
return m, fmt.Errorf("generate random machine name: %w", err)
}
}
containerName := clusterName + "-" + machineName
@@ -83,7 +86,7 @@ func (p *Provisioner) CreateMachine(ctx context.Context, clusterName string, opt
if _, err := p.createContainerWithImagePull(ctx, containerName, config, hostConfig); err != nil {
return m, err
}
if err := p.client.ContainerStart(ctx, containerName, container.StartOptions{}); err != nil {
if err := p.dockerCli.ContainerStart(ctx, containerName, container.StartOptions{}); err != nil {
return m, fmt.Errorf("start Docker container: %w", err)
}
@@ -111,16 +114,16 @@ func (p *Provisioner) createContainerWithImagePull(
) (container.CreateResponse, error) {
var resp container.CreateResponse
_, err := p.client.ContainerCreate(ctx, config, hostConfig, nil, nil, name)
_, err := p.dockerCli.ContainerCreate(ctx, config, hostConfig, nil, nil, name)
if err == nil {
return resp, nil
}
if !client.IsErrNotFound(err) {
if !dockerclient.IsErrNotFound(err) {
return resp, fmt.Errorf("create Docker container: %w", err)
}
respBody, err := p.client.ImagePull(ctx, config.Image, image.PullOptions{})
respBody, err := p.dockerCli.ImagePull(ctx, config.Image, image.PullOptions{})
if err != nil {
return resp, fmt.Errorf("pull Docker image: %w", err)
}
@@ -132,7 +135,7 @@ func (p *Provisioner) createContainerWithImagePull(
}
// Create container again after image pull.
if resp, err = p.client.ContainerCreate(ctx, config, hostConfig, nil, nil, name); err != nil {
if resp, err = p.dockerCli.ContainerCreate(ctx, config, hostConfig, nil, nil, name); err != nil {
return resp, fmt.Errorf("create Docker container: %w", err)
}
@@ -141,11 +144,11 @@ func (p *Provisioner) createContainerWithImagePull(
// 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))
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
for {
c, err := p.client.ContainerInspect(ctx, containerID)
c, err := p.dockerCli.ContainerInspect(ctx, containerID)
if err != nil {
return nil, fmt.Errorf("inspect container: %w", err)
}
@@ -169,12 +172,33 @@ func (p *Provisioner) waitPortPublished(ctx context.Context, containerID string,
}
}
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))
func randomMachineName() (string, error) {
suffix, err := secret.RandomAlphaNumeric(4)
if err != nil {
return "", fmt.Errorf("generate machine name: %w", err)
return "", fmt.Errorf("generate random suffix: %w", err)
}
i := r.Int64() + 10
return fmt.Sprintf("machine-%d", i), nil
return "machine-" + suffix, nil
}
// WaitMachineReady waits for the machine API to respond.
func (p *Provisioner) WaitMachineReady(ctx context.Context, m Machine) error {
cli, err := client.New(ctx, connector.NewTCPConnector(m.APIAddress))
if err != nil {
return fmt.Errorf("create machine client over TCP '%s': %w", m.APIAddress, err)
}
defer cli.Close()
boff := backoff.WithContext(backoff.NewExponentialBackOff(
backoff.WithInitialInterval(100*time.Millisecond),
backoff.WithMaxInterval(1*time.Second),
backoff.WithMaxElapsedTime(0),
), ctx)
inspect := func() error {
if _, err := cli.Inspect(ctx, &emptypb.Empty{}); err != nil {
return fmt.Errorf("inspect machine: %w", err)
}
return nil
}
return backoff.Retry(inspect, boff)
}
internal/ucind/provision.go
+2 -2
View File
@@ -8,11 +8,11 @@ import (
var ErrNotFound = errors.New("not found")
type Provisioner struct {
client *client.Client
dockerCli *client.Client
}
func NewProvisioner(cli *client.Client) *Provisioner {
return &Provisioner{
client: cli,
dockerCli: cli,
}
}