From 4166474ee822ae9d548990d407023ae6bc68a913 Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Fri, 25 Jul 2025 16:28:32 +1000 Subject: [PATCH] fix: race on cluster init by waiting for corrosion service to become ready with schema applied --- internal/machine/cluster.go | 2 + internal/machine/corroservice/docker.go | 38 +++++++++-------- internal/machine/corroservice/service.go | 54 +++++++++++++++++++++++- internal/machine/corroservice/systemd.go | 18 +++----- internal/machine/machine.go | 1 + 5 files changed, 83 insertions(+), 30 deletions(-) diff --git a/internal/machine/cluster.go b/internal/machine/cluster.go index b6a67224..196fab55 100644 --- a/internal/machine/cluster.go +++ b/internal/machine/cluster.go @@ -108,11 +108,13 @@ func (cc *clusterController) Run(ctx context.Context) error { if err := cc.corroService.Restart(ctx); err != nil { return fmt.Errorf("restart corrosion service: %w", err) } + slog.Info("Corrosion service restarted.") } else { slog.Info("Starting corrosion service.") if err := cc.corroService.Start(ctx); err != nil { return fmt.Errorf("start corrosion service: %w", err) } + slog.Info("Corrosion service started.") } errGroup, ctx := errgroup.WithContext(ctx) diff --git a/internal/machine/corroservice/docker.go b/internal/machine/corroservice/docker.go index f5b70481..220c2184 100644 --- a/internal/machine/corroservice/docker.go +++ b/internal/machine/corroservice/docker.go @@ -27,29 +27,33 @@ type DockerService struct { User string } -func NewDockerService(cli *client.Client, image, name, dataDir string) *DockerService { - return &DockerService{ - Client: cli, - Image: image, - Name: name, - DataDir: dataDir, - } -} - func (s *DockerService) Start(ctx context.Context) error { _, err := s.Client.ContainerInspect(ctx, s.Name) if err != nil { - if client.IsErrNotFound(err) { - return s.startNewContainer(ctx) + if !client.IsErrNotFound(err) { + return fmt.Errorf("inspect container %q: %w", s.Name, err) + } + if err = s.startNewContainer(ctx); err != nil { + return err + } + } else { + // Container already exists. + // TODO: recreate only if the container configuration has to be changed. + if err = s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{Force: true}); err != nil { + return fmt.Errorf("remove container %q: %w", s.Name, err) + } + if err = s.startNewContainer(ctx); err != nil { + return err } - return fmt.Errorf("inspect container %q: %w", s.Name, err) - } - // TODO: recreate only if the container configuration has to be changed. - if err = s.Client.ContainerRemove(ctx, s.Name, container.RemoveOptions{Force: true}); err != nil { - return fmt.Errorf("remove container %q: %w", s.Name, err) } - return s.startNewContainer(ctx) + slog.Debug("Waiting for corrosion service to be ready.") + if err = WaitReady(ctx, s.DataDir); err != nil { + return err + } + slog.Debug("Corrosion service is ready.") + + return nil } func (s *DockerService) Stop(ctx context.Context) error { diff --git a/internal/machine/corroservice/service.go b/internal/machine/corroservice/service.go index af97fcf7..d8ffed72 100644 --- a/internal/machine/corroservice/service.go +++ b/internal/machine/corroservice/service.go @@ -1,6 +1,16 @@ package corroservice -import "context" +import ( + "context" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/BurntSushi/toml" + "github.com/cenkalti/backoff/v4" + "github.com/psviderski/uncloud/internal/corrosion" +) type Service interface { Start(ctx context.Context) error @@ -8,3 +18,45 @@ type Service interface { Restart(ctx context.Context) error Running() bool } + +// WaitReady waits for the Corrosion service to be ready with the uncloud schema applied. +func WaitReady(ctx context.Context, dataDir string) error { + // Read the config file to get the API address. + configPath := filepath.Join(dataDir, "config.toml") + configData, err := os.ReadFile(configPath) + if err != nil { + return fmt.Errorf("read config file: %w", err) + } + + var config Config + if err = toml.Unmarshal(configData, &config); err != nil { + return fmt.Errorf("unmarshal config: %w", err) + } + + corro, err := corrosion.NewAPIClient(config.API.Addr) + if err != nil { + return fmt.Errorf("create corrosion API client: %w", err) + } + + // Corrosion starts serving the API before applying the schema. Query the cluster table with exponential backoff + // to check if the uncloud schema has been applied. + checkReady := func() error { + rows, err := corro.QueryContext(ctx, "SELECT 1 FROM cluster LIMIT 1") + if err != nil { + return fmt.Errorf("query cluster table: %w", err) + } + defer rows.Close() + + return nil + } + + b := backoff.NewExponentialBackOff( + backoff.WithInitialInterval(50*time.Millisecond), + backoff.WithMaxInterval(1*time.Second), + backoff.WithMaxElapsedTime(15*time.Second), + ) + if err = backoff.Retry(checkReady, backoff.WithContext(b, ctx)); err != nil { + return fmt.Errorf("corrosion service did not become ready: %w", err) + } + return nil +} diff --git a/internal/machine/corroservice/systemd.go b/internal/machine/corroservice/systemd.go index 5efe573d..e2dfc5ae 100644 --- a/internal/machine/corroservice/systemd.go +++ b/internal/machine/corroservice/systemd.go @@ -5,7 +5,6 @@ import ( "fmt" "log/slog" "os/exec" - "time" ) const DefaultSystemdUnit = "uncloud-corrosion.service" @@ -44,20 +43,15 @@ func (s *SystemdService) startOrRestart(ctx context.Context, cmd string) error { if _, err := exec.Command("systemctl", cmd, s.Unit).Output(); err != nil { return fmt.Errorf("systemctl %s %s: %w", cmd, s.Unit, err) } - slog.Info(fmt.Sprintf("Corrosion systemd service %sed.", cmd), "unit", s.Unit) + slog.Debug(fmt.Sprintf("Corrosion systemd service %sed.", cmd), "unit", s.Unit) - // Optimistically wait for the corrosion service to start and initialise the database schema before proceeding. - timer := time.NewTimer(2 * time.Second) - defer timer.Stop() - - select { - case <-timer.C: - case <-ctx.Done(): - return nil + slog.Debug("Waiting for corrosion service to be ready.") + if err := WaitReady(ctx, s.DataDir); err != nil { + return err } - - // TODO: run a goroutine to check the status of the service and log any errors in the uncloud log. + slog.Debug("Corrosion service is ready.") s.running = true + return nil } diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 8756540d..b1276e42 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -321,6 +321,7 @@ func (m *Machine) Run(ctx context.Context) error { if err := m.config.CorrosionService.Start(ctx); err != nil { return fmt.Errorf("start corrosion service: %w", err) } + slog.Info("Corrosion service started.") } // Use an errgroup to coordinate error handling and graceful shutdown of multiple machine components.