fix: prevent race condition with Docker network creation (#89)

This commit is contained in:
Evgenii Orlov
2025-07-10 17:51:08 +10:00
committed by GitHub
parent 31cd4c77e9
commit 10bbe9fbc5
3 changed files with 106 additions and 3 deletions
+38 -2
View File
@@ -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 {
+60 -1
View File
@@ -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
@@ -362,6 +374,11 @@ func (m *Machine) Run(ctx context.Context) error {
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)
// Ensure the corrosion config is up to date, including a new gossip address if the machine
@@ -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) {
+8
View File
@@ -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(