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

wait for test cluster to be ready (all machines up)

Browse Source
This commit is contained in:
Pavel Sviderski
2024-12-05 13:45:46 +10:00
parent a40fd613e8
commit 8bddbec7bd
5 changed files with 72 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cli/client/service.go
+1
View File
@@ -471,6 +471,7 @@ func (cli *Client) RemoveService(ctx context.Context, id string) error {
return
}
removeCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs("machines", machineIP))
// TODO: gracefully stop the container before removing it without force.
err := cli.RemoveContainer(removeCtx, mc.Container.ID, container.RemoveOptions{Force: true})
if err != nil {
if !dockerclient.IsErrNotFound(err) {
internal/ucind/cluster.go
+44 -11
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"github.com/cenkalti/backoff/v4"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/network"
@@ -11,8 +12,6 @@ import (
"google.golang.org/protobuf/types/known/emptypb"
"net/netip"
"time"
"uncloud/internal/cli/client"
"uncloud/internal/cli/client/connector"
"uncloud/internal/machine"
"uncloud/internal/machine/api/pb"
"uncloud/internal/machine/cluster"
@@ -90,13 +89,11 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
// 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 {
if err := p.WaitMachineReady(ctx, initMachine, 5*time.Second); 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))
initClient, err := initMachine.Connect(ctx)
if err != nil {
return fmt.Errorf("create machine client over TCP '%s': %w", initMachine.APIAddress, err)
}
@@ -114,14 +111,11 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
// Join the rest of the machines to the cluster.
for _, m := range machines[1:] {
waitCtx, cancel = context.WithTimeout(ctx, 5*time.Second)
//goland:noinspection GoDeferInLoop
defer cancel()
if err = p.WaitMachineReady(waitCtx, m); err != nil {
if err = p.WaitMachineReady(ctx, m, 5*time.Second); 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))
cli, err := m.Connect(ctx)
if err != nil {
return fmt.Errorf("create machine client over TCP '%s': %w", m.APIAddress, err)
}
@@ -224,6 +218,45 @@ func (p *Provisioner) InspectCluster(ctx context.Context, name string) (Cluster,
return c, nil
}
// WaitClusterReady waits for all machines in the cluster to be ready and UP.
func (p *Provisioner) WaitClusterReady(ctx context.Context, c Cluster, timeout time.Duration) error {
firstMachine := c.Machines[0]
if err := p.WaitMachineReady(ctx, firstMachine, timeout); err != nil {
return fmt.Errorf("wait for machine '%s' to be ready: %w", firstMachine.Name, err)
}
cli, err := firstMachine.Connect(ctx)
if err != nil {
return fmt.Errorf("connect to machine over TCP '%s': %w", firstMachine.APIAddress, err)
}
defer cli.Close()
boff := backoff.WithContext(backoff.NewExponentialBackOff(
backoff.WithInitialInterval(100*time.Millisecond),
backoff.WithMaxInterval(1*time.Second),
backoff.WithMaxElapsedTime(timeout),
), ctx)
checkMachinesUp := func() error {
machines, err := cli.ListMachines(ctx)
if err != nil {
return fmt.Errorf("list machines: %w", err)
}
if len(machines) != len(c.Machines) {
return fmt.Errorf("expected %d machines, got %d", len(c.Machines), len(machines))
}
for _, m := range machines {
if m.State != pb.MachineMember_UP {
return fmt.Errorf("machine '%s' state is not UP: %s", m.Machine.Name, m.State)
}
}
return nil
}
return backoff.Retry(checkMachinesUp, boff)
}
func (p *Provisioner) RemoveCluster(ctx context.Context, name string) error {
if _, err := p.InspectCluster(ctx, name); err != nil {
if errors.Is(err, ErrNotFound) {
internal/ucind/machine.go
+4 -4
View File
@@ -185,17 +185,17 @@ func randomMachineName() (string, error) {
}
// 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))
func (p *Provisioner) WaitMachineReady(ctx context.Context, m Machine, timeout time.Duration) error {
cli, err := m.Connect(ctx)
if err != nil {
return fmt.Errorf("create machine client over TCP '%s': %w", m.APIAddress, err)
return fmt.Errorf("connect to machine 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),
backoff.WithMaxElapsedTime(timeout),
), ctx)
inspect := func() error {
test/e2e/cluster_test.go
+12 -5
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
dockerclient "github.com/docker/docker/client"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -15,7 +16,9 @@ import (
"uncloud/internal/ucind"
)
func createTestCluster(t *testing.T, name string, opts ucind.CreateClusterOptions) (ucind.Cluster, *ucind.Provisioner) {
func createTestCluster(
t *testing.T, name string, opts ucind.CreateClusterOptions, waitReady bool,
) (ucind.Cluster, *ucind.Provisioner) {
dockerCli, err := dockerclient.NewClientWithOpts(dockerclient.FromEnv, dockerclient.WithAPIVersionNegotiation())
require.NoError(t, err)
@@ -39,13 +42,17 @@ func createTestCluster(t *testing.T, name string, opts ucind.CreateClusterOption
c, err := p.CreateCluster(ctx, name, opts)
require.NoError(t, err)
require.Equal(t, name, c.Name)
require.Len(t, c.Machines, opts.Machines)
assert.Equal(t, name, c.Name)
assert.Len(t, c.Machines, opts.Machines)
t.Cleanup(func() {
require.NoError(t, p.RemoveCluster(ctx, name))
})
if waitReady {
require.NoError(t, p.WaitClusterReady(ctx, c, 15*time.Second))
}
return c, p
}
@@ -54,14 +61,14 @@ func TestClusterLifecycle(t *testing.T) {
name := "ucind-test.cluster-lifecycle"
ctx := context.Background()
c, p := createTestCluster(t, name, ucind.CreateClusterOptions{Machines: 3})
c, p := createTestCluster(t, name, ucind.CreateClusterOptions{Machines: 3}, false)
t.Run("each machine reconciled cluster store", func(t *testing.T) {
var err error
// Create a client for each machine and wait for it to be ready.
clients := make([]*client.Client, len(c.Machines))
for i, m := range c.Machines {
require.NoError(t, p.WaitMachineReady(ctx, m))
require.NoError(t, p.WaitMachineReady(ctx, m, 5*time.Second))
clients[i], err = m.Connect(ctx)
require.NoError(t, err)
//goland:noinspection GoDeferInLoop
test/e2e/service_test.go
+11 -8
View File
@@ -14,26 +14,29 @@ import (
func TestRunService(t *testing.T) {
t.Parallel()
name := "ucind-test.run-service"
clusterName := "ucind-test.run-service"
ctx := context.Background()
c, _ := createTestCluster(t, name, ucind.CreateClusterOptions{Machines: 3})
c, _ := createTestCluster(t, clusterName, ucind.CreateClusterOptions{Machines: 3}, true)
cli, err := c.Machines[0].Connect(ctx)
require.NoError(t, err)
t.Run("global mode", func(t *testing.T) {
t.Parallel()
name := "busybox-global"
t.Cleanup(func() {
err := cli.RemoveService(ctx, "busybox-global")
err := cli.RemoveService(ctx, name)
if !dockerclient.IsErrNotFound(err) {
require.NoError(t, err)
}
_, err = cli.InspectService(ctx, "busybox-global")
_, err = cli.InspectService(ctx, name)
require.ErrorIs(t, err, client.ErrNotFound)
})
resp, err := cli.RunService(ctx, api.ServiceSpec{
Name: "busybox-global",
Name: name,
Mode: api.ServiceModeGlobal,
Container: api.ContainerSpec{
Command: []string{"sleep", "infinity"},
@@ -43,14 +46,14 @@ func TestRunService(t *testing.T) {
require.NoError(t, err)
assert.NotEmpty(t, resp.ID)
assert.Equal(t, "busybox-global", resp.Name)
assert.Equal(t, name, resp.Name)
assert.Len(t, resp.Containers, 3, "expected 1 container on each machine")
svc, err := cli.InspectService(ctx, "busybox-global")
svc, err := cli.InspectService(ctx, name)
require.NoError(t, err)
assert.Equal(t, resp.ID, svc.ID)
assert.Equal(t, "busybox-global", svc.Name)
assert.Equal(t, name, svc.Name)
assert.Equal(t, api.ServiceModeGlobal, svc.Mode)
assert.Len(t, svc.Containers, 3, "expected 1 container on each machine")
})