From 6e0df53f13c9d60cbe77ef5f848e46a5deca037e Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Wed, 27 Nov 2024 10:34:42 +1000 Subject: [PATCH] add in docker service runner for corrosion --- internal/docker/client.go | 49 +++++++++++++ internal/machine/corroservice/docker.go | 91 +++++++++++++++++++++++++ internal/machine/machine.go | 61 ++++++++++++++--- internal/machine/network.go | 18 ++--- 4 files changed, 200 insertions(+), 19 deletions(-) create mode 100644 internal/docker/client.go create mode 100644 internal/machine/corroservice/docker.go diff --git a/internal/docker/client.go b/internal/docker/client.go new file mode 100644 index 00000000..87d26a20 --- /dev/null +++ b/internal/docker/client.go @@ -0,0 +1,49 @@ +package docker + +import ( + "context" + "errors" + "fmt" + "github.com/cenkalti/backoff/v4" + "github.com/docker/docker/client" + "log/slog" + "time" +) + +// WaitDaemonReady waits for the Docker daemon to start and be ready to serve requests. +func WaitDaemonReady(ctx context.Context, cli *client.Client) error { + // Retry to ping the Docker daemon until it's ready or the context is canceled. + boff := backoff.WithContext(backoff.NewExponentialBackOff( + backoff.WithInitialInterval(100*time.Millisecond), + backoff.WithMaxInterval(1*time.Second), + backoff.WithMaxElapsedTime(0), + ), ctx) + + waitingLogged := false + ping := func() error { + _, err := cli.Ping(ctx) + if err == nil { + if waitingLogged { + slog.Info("Docker daemon is ready.") + } + return nil + } + if !client.IsErrConnectionFailed(err) { + return backoff.Permanent(fmt.Errorf("connect to Docker daemon: %w", err)) + } + + if !waitingLogged { + slog.Info("Waiting for Docker daemon to start and be ready.") + waitingLogged = true + } + return err + } + + if err := backoff.Retry(ping, boff); err != nil { + if errors.Is(err, context.Canceled) { + return nil + } + return fmt.Errorf("ping Docker: %w", err) + } + return nil +} diff --git a/internal/machine/corroservice/docker.go b/internal/machine/corroservice/docker.go new file mode 100644 index 00000000..3e865c34 --- /dev/null +++ b/internal/machine/corroservice/docker.go @@ -0,0 +1,91 @@ +package corroservice + +import ( + "context" + "fmt" + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/network" + "github.com/docker/docker/client" + "path/filepath" +) + +const ( + LatestImage = "corrosion:latest" +) + +type DockerService struct { + Client *client.Client + Image string + Name string + DataDir string + // TODO: uid/guid +} + +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) + } + return fmt.Errorf("inspect container %q: %w", s.Name, err) + } + // Container already exists, recreate it if its configuration has to be changed. + // TODO: check config equal to the new one + 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) +} + +func (s *DockerService) Restart(ctx context.Context) error { + if err := s.Client.ContainerRestart(ctx, s.Name, container.StopOptions{}); err != nil { + return fmt.Errorf("restart container %q: %w", s.Name, err) + } + return nil +} + +func (s *DockerService) Running() bool { + c, err := s.Client.ContainerInspect(context.Background(), s.Name) + if err != nil { + return false + } + + return c.State.Running +} + +func (s *DockerService) containerConfig() *container.Config { + return &container.Config{ + Image: s.Image, + Env: []string{ + fmt.Sprintf("CONFIG_PATH=%s", filepath.Join(s.DataDir, "config.toml")), + }, + } +} + +func (s *DockerService) hostConfig() *container.HostConfig { + return &container.HostConfig{ + NetworkMode: network.NetworkHost, + } +} + +func (s *DockerService) startNewContainer(ctx context.Context) error { + _, err := s.Client.ContainerCreate(ctx, s.containerConfig(), s.hostConfig(), nil, nil, s.Name) + if err != nil { + return fmt.Errorf("create container: %w", err) + } + if err = s.Client.ContainerStart(ctx, s.Name, container.StartOptions{}); err != nil { + return fmt.Errorf("start container: %w", err) + } + + return nil +} diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 8f4a7fd4..8e642298 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -20,11 +20,12 @@ import ( "path/filepath" "strconv" "uncloud/internal/corrosion" + "uncloud/internal/docker" "uncloud/internal/machine/api/pb" apiproxy "uncloud/internal/machine/api/proxy" "uncloud/internal/machine/cluster" "uncloud/internal/machine/corroservice" - "uncloud/internal/machine/docker" + machinedocker "uncloud/internal/machine/docker" "uncloud/internal/machine/network" "uncloud/internal/machine/store" ) @@ -46,10 +47,13 @@ type Config struct { CorrosionAPIAddr netip.AddrPort CorrosionAdminSockPath string CorrosionService corroservice.Service + + // DockerClient manages system and user containers using the local Docker daemon. + DockerClient *client.Client } // SetDefaults returns a new Config with default values set where not provided. -func (c *Config) SetDefaults() *Config { +func (c *Config) SetDefaults() (*Config, error) { // Copy c into a new Config to avoid modifying the original. cfg := *c @@ -62,6 +66,15 @@ func (c *Config) SetDefaults() *Config { if cfg.UncloudSockPath == "" { cfg.UncloudSockPath = DefaultUncloudSockPath } + + if cfg.DockerClient == nil { + cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) + if err != nil { + return nil, fmt.Errorf("create Docker client: %w", err) + } + cfg.DockerClient = cli + } + if cfg.CorrosionDir == "" { cfg.CorrosionDir = filepath.Join(cfg.DataDir, "corrosion") } @@ -77,9 +90,25 @@ func (c *Config) SetDefaults() *Config { cfg.CorrosionAdminSockPath = filepath.Join(cfg.CorrosionDir, "admin.sock") } if cfg.CorrosionService == nil { - cfg.CorrosionService = corroservice.DefaultSystemdService(cfg.CorrosionDir) + if isRunningInDocker() { + // Run corrosion in a nested Docker container if the machine is running in a container. + cfg.CorrosionService = corroservice.NewDockerService( + cfg.DockerClient, + corroservice.LatestImage, + "uncloud-corrosion", + cfg.CorrosionDir, + ) + } else { + cfg.CorrosionService = corroservice.DefaultSystemdService(cfg.CorrosionDir) + } } - return &cfg + return &cfg, nil +} + +// isRunningInDocker returns true if the current process is running in a Docker container. +func isRunningInDocker() bool { + _, err := os.Stat("/.dockerenv") + return err == nil } type Machine struct { @@ -95,7 +124,7 @@ type Machine struct { // store is the cluster store backed by a distributed Corrosion database. store *store.Store cluster *cluster.Cluster - docker *docker.Server + docker *machinedocker.Server // localMachineServer is the gRPC server for the machine API listening on the local Unix socket. localMachineServer *grpc.Server @@ -108,7 +137,10 @@ type Machine struct { } func NewMachine(config *Config) (*Machine, error) { - config = config.SetDefaults() + config, err := config.SetDefaults() + if err != nil { + return nil, fmt.Errorf("set default config values: %w", err) + } // Load the existing machine state or create a new one. statePath := StatePath(config.DataDir) @@ -153,7 +185,7 @@ func NewMachine(config *Config) (*Machine, error) { if err != nil { return nil, fmt.Errorf("create Docker client: %w", err) } - dockerServer := docker.NewServer(dockerCli) + dockerServer := machinedocker.NewServer(dockerCli) // Init a local gRPC proxy server that proxies requests to the local or remote machine API servers. proxyDirector := apiproxy.NewDirector(config.MachineSockPath, APIPort) @@ -207,6 +239,11 @@ func (m *Machine) Initialised() bool { } func (m *Machine) Run(ctx context.Context) error { + // Docker dependency is essential for the machine to function. Block until it's ready. + if err := docker.WaitDaemonReady(ctx, m.config.DockerClient); err != nil { + return fmt.Errorf("wait for Docker daemon: %w", err) + } + // Configure and start the corrosion service on the loopback if the machine is not initialised as a cluster // member. This provides the store required for the machine to initialise a new cluster on it. Once the machine // is initialised, the corrosion service is managed by the networkController. @@ -297,7 +334,13 @@ func (m *Machine) Run(ctx context.Context) error { ), ) - ctrl, err = newNetworkController(m.state, m.store, proxyServer, m.config.CorrosionService) + ctrl, err = newNetworkController( + m.state, + m.store, + proxyServer, + m.config.CorrosionService, + m.config.DockerClient, + ) if err != nil { return fmt.Errorf("initialise network controller: %w", err) } @@ -343,6 +386,8 @@ func (m *Machine) Run(ctx context.Context) error { // Close the proxy director to close all backend connections. m.proxyDirector.Close() slog.Info("Local API proxy server stopped.") + + m.config.DockerClient.Close() return nil }, ) diff --git a/internal/machine/network.go b/internal/machine/network.go index df74f249..a368af4e 100644 --- a/internal/machine/network.go +++ b/internal/machine/network.go @@ -34,13 +34,14 @@ type networkController struct { server *grpc.Server corroService corroservice.Service + dockerCli *client.Client // TODO: DNS server/resolver listening on the machine IP, e.g. 10.210.0.1:53. It can't listen on 127.0.X.X // like resolved does because it needs to be reachable from both the host and the containers. } func newNetworkController( - state *State, store *store.Store, server *grpc.Server, corroService corroservice.Service, + state *State, store *store.Store, server *grpc.Server, corroService corroservice.Service, dockerCli *client.Client, ) ( *networkController, error, ) { @@ -58,6 +59,7 @@ func newNetworkController( endpointChanges: endpointChanges, server: server, corroService: corroService, + dockerCli: dockerCli, }, nil } @@ -176,18 +178,12 @@ func (nc *networkController) Run(ctx context.Context) error { // prepareAndWatchDocker configures the Docker network and watches local Docker containers to sync them // to the cluster store. func (nc *networkController) prepareAndWatchDocker(ctx context.Context) error { - cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) - if err != nil { - return fmt.Errorf("init Docker client: %w", err) - } - defer cli.Close() - - manager := docker.NewManager(cli, nc.state.ID, nc.store) - if err = manager.WaitDaemonReady(ctx); err != nil { + manager := docker.NewManager(nc.dockerCli, nc.state.ID, nc.store) + if err := manager.WaitDaemonReady(ctx); err != nil { return fmt.Errorf("wait for Docker daemon: %w", err) } - if err = manager.EnsureUncloudNetwork(ctx, nc.state.Network.Subnet); err != nil { + if err := manager.EnsureUncloudNetwork(ctx, nc.state.Network.Subnet); err != nil { return fmt.Errorf("ensure Docker network: %w", err) } slog.Info("Docker network configured.") @@ -206,7 +202,7 @@ func (nc *networkController) prepareAndWatchDocker(ctx context.Context) error { } return nil } - if err = backoff.Retry(watchAndSync, boff); err != nil { + if err := backoff.Retry(watchAndSync, boff); err != nil { if errors.Is(err, context.Canceled) { return nil }