diff --git a/internal/ucind/cluster.go b/internal/ucind/cluster.go index efc5e99a..986f4d52 100644 --- a/internal/ucind/cluster.go +++ b/internal/ucind/cluster.go @@ -119,17 +119,16 @@ func (p *Provisioner) CreateCluster(ctx context.Context, name string, opts Creat func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error { // Init a new cluster on the first machine. initMachine := machines[0] - - if err := WaitMachineReady(ctx, initMachine, 30*time.Second); err != nil { - return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err) - } - initClient, err := initMachine.Connect(ctx) if err != nil { return fmt.Errorf("create machine client over TCP '%s': %w", initMachine.APIAddress, err) } defer initClient.Close() + if err := initClient.WaitMachineReady(ctx, 30*time.Second); err != nil { + return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err) + } + req := &pb.InitClusterRequest{ MachineName: initMachine.Name, Network: pb.NewIPPrefix(cluster.DefaultNetwork), @@ -138,14 +137,22 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error if err != nil { return fmt.Errorf("init cluster: %w", err) } + fmt.Printf("Cluster %q initialised with machine %q\n", initMachine.ClusterName, initResp.Machine.Name) + fmt.Printf("Waiting for cluster to be ready...") + if err = initClient.WaitClusterReady(ctx, 30*time.Second); err != nil { + return fmt.Errorf("wait for cluster to be ready: %w", err) + } + fmt.Println(" done.") + + // Get the current store DB version from the init machine to pass to the join requests. + inspectResp, err := initClient.MachineClient.InspectMachine(ctx, &emptypb.Empty{}) + if err != nil { + return fmt.Errorf("inspect init machine: %w", err) + } // Join the rest of the machines to the cluster. for _, m := range machines[1:] { - if err = WaitMachineReady(ctx, m, 30*time.Second); err != nil { - return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err) - } - cli, err := m.Connect(ctx) if err != nil { return fmt.Errorf("create machine client over TCP '%s': %w", m.APIAddress, err) @@ -153,6 +160,10 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error //goland:noinspection GoDeferInLoop defer cli.Close() + if err := cli.WaitMachineReady(ctx, 30*time.Second); err != nil { + return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err) + } + tokenResp, err := cli.Token(ctx, &emptypb.Empty{}) if err != nil { return fmt.Errorf("get machine token: %w", err) @@ -181,8 +192,9 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error // Configure the machine to join the cluster. joinReq := &pb.JoinClusterRequest{ - Machine: addResp.Machine, - OtherMachines: []*pb.MachineInfo{initResp.Machine}, + Machine: addResp.Machine, + OtherMachines: []*pb.MachineInfo{initResp.Machine}, + MinStoreDbVersion: inspectResp.Machines[0].StoreDbVersion, } if _, err = cli.JoinCluster(ctx, joinReq); err != nil { return fmt.Errorf("join cluster: %w", err) @@ -256,16 +268,16 @@ func (p *Provisioner) InspectCluster(ctx context.Context, name string) (Cluster, // 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 := 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() + if err = cli.WaitClusterReady(ctx, timeout); err != nil { + return fmt.Errorf("wait for cluster to be ready: %w", err) + } + boff := backoff.WithContext(backoff.NewExponentialBackOff( backoff.WithInitialInterval(100*time.Millisecond), backoff.WithMaxInterval(1*time.Second), @@ -289,7 +301,25 @@ func (p *Provisioner) WaitClusterReady(ctx context.Context, c Cluster, timeout t } return nil } - return backoff.Retry(checkMachinesUp, boff) + if err = backoff.Retry(checkMachinesUp, boff); err != nil { + return err + } + + // Wait for each machine to sync the cluster store and be ready to serve cluster requests. + for _, m := range c.Machines[1:] { + mcli, err := m.Connect(ctx) + if err != nil { + return fmt.Errorf("connect to machine over TCP '%s': %w", m.APIAddress, err) + } + //goland:noinspection GoDeferInLoop + defer mcli.Close() + + if err = mcli.WaitClusterReady(ctx, timeout); err != nil { + return fmt.Errorf("wait for cluster to be ready on machine '%s': %w", m.Name, err) + } + } + + return nil } func (p *Provisioner) RemoveCluster(ctx context.Context, name string) error { diff --git a/internal/ucind/machine.go b/internal/ucind/machine.go index 8daef9d9..711b9dcb 100644 --- a/internal/ucind/machine.go +++ b/internal/ucind/machine.go @@ -9,7 +9,6 @@ import ( "net/netip" "time" - "github.com/cenkalti/backoff/v4" "github.com/containerd/errdefs" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/image" @@ -17,7 +16,6 @@ import ( "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/pkg/client" "github.com/psviderski/uncloud/pkg/client/connector" - "google.golang.org/protobuf/types/known/emptypb" ) const ( @@ -194,26 +192,3 @@ func randomMachineName() (string, error) { } return "machine-" + suffix, nil } - -// WaitMachineReady waits for the machine API to respond. -func WaitMachineReady(ctx context.Context, m Machine, timeout time.Duration) error { - cli, err := m.Connect(ctx) - if err != nil { - 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(10*time.Second), - backoff.WithMaxElapsedTime(timeout), - ), 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) -} diff --git a/pkg/client/machine.go b/pkg/client/machine.go index 151c9edc..1f022af8 100644 --- a/pkg/client/machine.go +++ b/pkg/client/machine.go @@ -4,7 +4,9 @@ import ( "context" "fmt" "strings" + "time" + "github.com/cenkalti/backoff/v4" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" "google.golang.org/grpc/codes" @@ -100,3 +102,46 @@ func (cli *Client) RenameMachine(ctx context.Context, nameOrID, newName string) return cli.UpdateMachine(ctx, req) } + +// WaitMachineReady waits for the machine API on the connected machine to respond. +func (cli *Client) WaitMachineReady(ctx context.Context, timeout time.Duration) error { + boff := backoff.WithContext(backoff.NewExponentialBackOff( + backoff.WithInitialInterval(100*time.Millisecond), + backoff.WithMaxInterval(1*time.Second), + backoff.WithMaxElapsedTime(timeout), + ), 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) +} + +// WaitClusterReady waits for the connected machine to be ready to server cluster requests. +func (cli *Client) WaitClusterReady(ctx context.Context, timeout time.Duration) error { + // Backoff is not really needed here as the default service config for the gRPC client is already + // doing retries with backoff for Unavailable errors. However, it's still convenient to use backoff + // to control the overall timeout for the operation. + boff := backoff.WithContext(backoff.NewExponentialBackOff( + backoff.WithInitialInterval(100*time.Millisecond), + backoff.WithMaxInterval(1*time.Second), + backoff.WithMaxElapsedTime(timeout), + ), ctx) + + listMachines := func() error { + _, err := cli.ListMachines(ctx, nil) + if err != nil { + if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable { + // Machine is not ready yet, retry. + return err + } + // Other non-Unavailable errors should not be retried. + return backoff.Permanent(err) + } + return nil + } + return backoff.Retry(listMachines, boff) +} diff --git a/test/e2e/cluster_test.go b/test/e2e/cluster_test.go index b667b90b..35498a6a 100644 --- a/test/e2e/cluster_test.go +++ b/test/e2e/cluster_test.go @@ -55,7 +55,7 @@ func createTestCluster( }) if waitReady { - require.NoError(t, p.WaitClusterReady(ctx, c, 15*time.Second)) + require.NoError(t, p.WaitClusterReady(ctx, c, 30*time.Second)) } return c, p @@ -73,7 +73,6 @@ func TestClusterLifecycle(t *testing.T) { // 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, ucind.WaitMachineReady(ctx, m, 5*time.Second)) clients[i], err = m.Connect(ctx) require.NoError(t, err) //goland:noinspection GoDeferInLoop @@ -86,11 +85,10 @@ func TestClusterLifecycle(t *testing.T) { require.Eventually(t, func() bool { machines, err := cli.ListMachines(ctx, nil) if err != nil { - // FailedPrecondition "cluster is not initialised" is expected until the store is reconciled. - if s, ok := status.FromError(err); ok { - if s.Code() == codes.FailedPrecondition { - return false - } + // Unavailable "machine is not ready to serve cluster requests" is expected until + // the store is reconciled. + if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable { + return false } require.NoError(t, err) }