diff --git a/internal/machine/docker/manager.go b/internal/machine/docker/manager.go index b5adbacb..1d15231a 100644 --- a/internal/machine/docker/manager.go +++ b/internal/machine/docker/manager.go @@ -2,7 +2,10 @@ package docker import ( "context" + "errors" "fmt" + "github.com/docker/docker/api/types/events" + "github.com/docker/docker/api/types/filters" "github.com/docker/docker/client" "log/slog" "time" @@ -12,18 +15,19 @@ import ( const ( NetworkName = "uncloud" UserChain = "DOCKER-USER" + // EventsDebounceInterval defines how long to wait before processing the next Docker event. Multiple events + // occurring within this window will be processed together as a single event to prevent system overload. + EventsDebounceInterval = 100 * time.Millisecond + // SyncInterval defines a regular interval to sync containers to the cluster store. + SyncInterval = 30 * time.Second ) type Manager struct { client *client.Client - store *store.Store } -func NewManager(client *client.Client, store *store.Store) *Manager { - return &Manager{ - client: client, - store: store, - } +func NewManager(client *client.Client) *Manager { + return &Manager{client: client} } // WaitDaemonReady waits for the Docker daemon to start and be ready to serve requests. @@ -53,3 +57,82 @@ func (d *Manager) WaitDaemonReady(ctx context.Context) error { } return nil } + +func (d *Manager) WatchAndSyncContainers(ctx context.Context, store *store.Store) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + // Filter only local container events. + opts := events.ListOptions{ + Filters: filters.NewArgs( + filters.Arg("scope", "local"), + filters.Arg("type", string(events.ContainerEventType)), + ), + } + + // Subscribe to Docker events before running the initial sync to avoid missing any events. + eventCh, errCh := d.client.Events(ctx, opts) + slog.Debug("Syncing containers to cluster store before processing Docker events.") + if err := d.syncContainersToStore(ctx, store); err != nil { + // The deferred cancel will stop the event subscription. + return fmt.Errorf("sync containers to cluster store: %w", err) + } + + var ( + // debouncer is used to debounce multiple Docker events into a single event sent to the debouncerCh + // to prevent system overload. + debouncer *time.Timer + debouncerCh = make(chan events.Message) + // ticker is used to trigger a regular sync of containers to the cluster store as a fallback. + ticker = time.NewTicker(SyncInterval) + ) + defer ticker.Stop() + + for { + select { + case e := <-eventCh: + switch e.Action { + case events.ActionStart, + events.ActionStop, + events.ActionPause, + events.ActionUnPause, + events.ActionKill, + events.ActionDie, + events.ActionOOM, + events.ActionHealthStatusHealthy, + events.ActionHealthStatusUnhealthy: + + if debouncer == nil { + debouncer = time.AfterFunc(EventsDebounceInterval, func() { + debouncerCh <- e + }) + } + } + case e := <-debouncerCh: + debouncer = nil + slog.Debug("Syncing containers to cluster store triggered by a Docker container event.", + "container_id", e.Actor.ID, + "container_name", e.Actor.Attributes["name"], + "action", e.Action) + + if err := d.syncContainersToStore(ctx, store); err != nil { + return fmt.Errorf("sync containers to cluster store: %w", err) + } + case <-ticker.C: + slog.Debug("Syncing containers to cluster store triggered by a regular interval.", + "interval", SyncInterval) + if err := d.syncContainersToStore(ctx, store); err != nil { + return fmt.Errorf("sync containers to cluster store: %w", err) + } + case err := <-errCh: + if errors.Is(err, context.Canceled) { + return nil + } + return fmt.Errorf("receive Docker event: %w", err) + } + } +} + +func (d *Manager) syncContainersToStore(ctx context.Context, store *store.Store) error { + // TODO: implement + return nil +} diff --git a/internal/machine/network.go b/internal/machine/network.go index 75487b66..b6d3fa98 100644 --- a/internal/machine/network.go +++ b/internal/machine/network.go @@ -98,29 +98,10 @@ func (nc *networkController) Run(ctx context.Context) error { }, ) - // Setup Docker network and iptables rules in a goroutine because it may block until the Docker daemon is ready. + // Setup Docker network and synchronise containers to the cluster store. errGroup.Go( func() 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.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 { - return err - } - slog.Info("Docker network configured.") - - //if err := d.WatchAndSyncContainers(ctx); err != nil { - // return fmt.Errorf("watch and sync containers to store: %w", err) - //} - return nil + return nc.prepareAndWatchDocker(ctx) }, ) @@ -192,11 +173,55 @@ func (nc *networkController) Run(ctx context.Context) error { return errGroup.Wait() } +// 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) + 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 { + return err + } + slog.Info("Docker network configured.") + + slog.Info("Watching Docker containers and syncing them to the cluster store.") + // Retry to watch and sync containers until the context is done. + boff := backoff.WithContext(backoff.NewExponentialBackOff( + backoff.WithInitialInterval(100*time.Millisecond), + backoff.WithMaxInterval(5*time.Second), + // Retry indefinitely. + backoff.WithMaxElapsedTime(0), + ), ctx) + watchAndSync := func() error { + if wErr := manager.WatchAndSyncContainers(ctx, nc.store); wErr != nil { + slog.Error("Failed to watch and sync containers to cluster store, retrying.", "err", wErr) + return wErr + } + return nil + } + if err = backoff.Retry(watchAndSync, boff); err != nil { + if errors.Is(err, context.Canceled) { + return nil + } + return fmt.Errorf("watch and sync containers to cluster store: %w", err) + } + + return nil +} + // handleMachineChanges subscribes to machine changes in the cluster and reconfigures the network peers accordingly. func (nc *networkController) handleMachineChanges(ctx context.Context) error { for { // Retry to subscribe to machine changes indefinitely until the context is done. - b := backoff.WithContext(backoff.NewExponentialBackOff( + boff := backoff.WithContext(backoff.NewExponentialBackOff( backoff.WithInitialInterval(1*time.Second), backoff.WithMaxInterval(60*time.Second), backoff.WithMaxElapsedTime(0), @@ -213,7 +238,7 @@ func (nc *networkController) handleMachineChanges(ctx context.Context) error { } return err } - if err = backoff.Retry(subscribe, b); err != nil { + if err = backoff.Retry(subscribe, boff); err != nil { if errors.Is(err, context.Canceled) { return nil } @@ -229,6 +254,9 @@ func (nc *networkController) handleMachineChanges(ctx context.Context) error { // For simplicity, reconfigure all peers on any change. for { select { + // TODO: test when Corrosion fails and the subscription fails to resubscribe (after 1 minute). It seems + // the changes channel will be closed and this will become a busy loop. Perhaps, the outer for loop should + // be reworked as well. case <-changes: slog.Info("Cluster machines changed, reconfiguring network peers.") if machines, err = nc.store.ListMachines(ctx); err != nil {