diff --git a/internal/machine/docker/server.go b/internal/machine/docker/server.go index a0b55f5b..a9a64dac 100644 --- a/internal/machine/docker/server.go +++ b/internal/machine/docker/server.go @@ -50,15 +50,42 @@ type Server struct { // internalDNSIP is a function that returns the IP address of the internal DNS server. It may return an empty // address if the address is unknown (e.g. when the machine is not initialised yet). internalDNSIP func() netip.Addr + // networkReady is a function that returns true if the Docker network is ready for containers. + networkReady func() bool + // waitForNetworkReady is a function that waits for the Docker network to be ready for containers. + waitForNetworkReady func(ctx context.Context) error +} + +// ServerOption configures the Docker server. +type ServerOption func(*Server) + +// WithNetworkReady sets the network readiness check function. +func WithNetworkReady(networkReady func() bool) ServerOption { + return func(s *Server) { + s.networkReady = networkReady + } +} + +// WithWaitForNetworkReady sets the network readiness wait function. +func WithWaitForNetworkReady(waitForNetworkReady func(ctx context.Context) error) ServerOption { + return func(s *Server) { + s.waitForNetworkReady = waitForNetworkReady + } } // NewServer creates a new Docker gRPC server with the provided Docker client. -func NewServer(cli *client.Client, db *sqlx.DB, internalDNSIP func() netip.Addr) *Server { - return &Server{ +func NewServer(cli *client.Client, db *sqlx.DB, internalDNSIP func() netip.Addr, opts ...ServerOption) *Server { + s := &Server{ client: cli, db: db, internalDNSIP: internalDNSIP, } + + for _, opt := range opts { + opt(s) + } + + return s } // CreateContainer creates a new container based on the given configuration. @@ -118,6 +145,15 @@ func (s *Server) InspectContainer(ctx context.Context, req *pb.InspectContainerR // StartContainer starts a container with the given ID and options. func (s *Server) StartContainer(ctx context.Context, req *pb.StartContainerRequest) (*emptypb.Empty, error) { + // Wait for Docker network to be ready before starting the container + if s.waitForNetworkReady != nil { + if err := s.waitForNetworkReady(ctx); err != nil { + return nil, status.Errorf(codes.Unavailable, "Docker network not ready: %v", err) + } + } else if s.networkReady != nil && !s.networkReady() { + return nil, status.Errorf(codes.Unavailable, "Docker network not ready") + } + var opts container.StartOptions if len(req.Options) > 0 { if err := json.Unmarshal(req.Options, &opts); err != nil { diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 27b6c643..313cde6c 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -13,6 +13,7 @@ import ( "path/filepath" "slices" "strconv" + "sync" "github.com/docker/docker/client" "github.com/docker/go-connections/sockets" @@ -149,6 +150,10 @@ type Machine struct { started chan struct{} // initialised is signalled when the machine is configured as a member of a cluster. initialised chan struct{} + // networkReady is signalled when the Docker network is configured and ready for containers. + networkReady chan struct{} + // networkReadyMu protects networkReady channel operations + networkReadyMu sync.RWMutex // store is the cluster store backed by a distributed Corrosion database. store *store.Store @@ -235,6 +240,7 @@ func NewMachine(config *Config) (*Machine, error) { state: state, started: make(chan struct{}), initialised: make(chan struct{}, 1), + networkReady: make(chan struct{}), store: corroStore, cluster: c, localProxyServer: localProxyServer, @@ -245,11 +251,17 @@ func NewMachine(config *Config) (*Machine, error) { internalDNSIP := func() netip.Addr { return m.IP() } - m.docker = machinedocker.NewServer(dockerCli, db, internalDNSIP) + m.docker = machinedocker.NewServer(dockerCli, db, internalDNSIP, + machinedocker.WithNetworkReady(m.IsNetworkReady), + machinedocker.WithWaitForNetworkReady(m.WaitForNetworkReady)) m.localMachineServer = newGRPCServer(m, c, m.docker) if m.Initialised() { m.initialised <- struct{}{} + } else { + // For non-initialized machines, signal network is ready immediately + // since there's no cluster network to set up + close(m.networkReady) } return m, nil @@ -361,6 +373,11 @@ func (m *Machine) Run(ctx context.Context) error { // It can be reset when leaving the cluster and then re-initialised again with a new configuration. case <-m.initialised: var err error + + // Reset networkReady channel for the new cluster configuration + m.networkReadyMu.Lock() + m.networkReady = make(chan struct{}) + m.networkReadyMu.Unlock() m.cluster.UpdateMachineID(m.state.ID) @@ -404,6 +421,7 @@ func (m *Machine) Run(ctx context.Context) error { caddyfileCtrl, dnsServer, dnsResolver, + m.networkReady, ) if err != nil { return fmt.Errorf("initialise network controller: %w", err) @@ -776,6 +794,47 @@ func (m *Machine) Inspect(_ context.Context, _ *emptypb.Empty) (*pb.MachineInfo, }, nil } +// IsNetworkReady returns true if the Docker network is ready for containers. +func (m *Machine) IsNetworkReady() bool { + if !m.Initialised() { + // If machine is not initialized, there's no network to check + return true + } + + // Check if network is ready by checking if the networkReady channel has been closed + m.networkReadyMu.RLock() + defer m.networkReadyMu.RUnlock() + + select { + case <-m.networkReady: + return true + default: + return false + } +} + +// WaitForNetworkReady waits for the Docker network to be ready for containers. +// It returns nil when the network is ready or an error if the context is cancelled. +func (m *Machine) WaitForNetworkReady(ctx context.Context) error { + if !m.Initialised() { + // If machine is not initialized, there's no network to wait for + return nil + } + + // Get a copy of the channel to wait on + m.networkReadyMu.RLock() + networkReady := m.networkReady + m.networkReadyMu.RUnlock() + + // Wait for network to be ready or context to be cancelled + select { + case <-networkReady: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + // Reset restores the machine to a clean state, removing all cluster-related сonfiguration and data and scheduling // a graceful shutdown. The uncloud daemon will restart the machine if managed by systemd. func (m *Machine) Reset(ctx context.Context, _ *pb.ResetRequest) (*emptypb.Empty, error) { diff --git a/internal/machine/network.go b/internal/machine/network.go index e3fa70b9..3184d679 100644 --- a/internal/machine/network.go +++ b/internal/machine/network.go @@ -44,6 +44,9 @@ type networkController struct { // dnsServer is the embedded internal DNS server for the cluster listening on the machine IP. dnsServer *dns.Server dnsResolver *dns.ClusterResolver + + // networkReady is signalled when the Docker network is configured and ready for containers. + networkReady chan<- struct{} } func newNetworkController( @@ -55,6 +58,7 @@ func newNetworkController( caddyfileCtrl *caddyconfig.Controller, dnsServer *dns.Server, dnsResolver *dns.ClusterResolver, + networkReady chan<- struct{}, ) ( *networkController, error, ) { @@ -76,6 +80,7 @@ func newNetworkController( caddyfileCtrl: caddyfileCtrl, dnsServer: dnsServer, dnsResolver: dnsResolver, + networkReady: networkReady, }, nil } @@ -225,6 +230,9 @@ func (nc *networkController) prepareAndWatchDocker(ctx context.Context) error { } slog.Info("Docker network configured.") + // Signal that the Docker network is ready for containers + close(nc.networkReady) + slog.Info("Watching Docker containers and syncing them to cluster store.") // Retry to watch and sync containers until the context is done. boff := backoff.WithContext(backoff.NewExponentialBackOff(