add in docker service runner for corrosion

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
+}

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
+}

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 {
+		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
 		},
 	)

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())
+	manager := docker.NewManager(nc.dockerCli, nc.state.ID, nc.store)
-	if err != nil {
+	if err := manager.WaitDaemonReady(ctx); 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 {
 		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
 		}