subscribe to Docker container events with backoff

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 {
+func NewManager(client *client.Client) *Manager {
-	return &Manager{
+	return &Manager{client: client}
-		client: client,
-		store:  store,
-	}
 }
 
 // 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
+}

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())
+			return nc.prepareAndWatchDocker(ctx)
-			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
 		},
 	)
 
@@ -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 {