sync new and updated containers to cluster store, no deletion yet

Makefile

@@ -1,11 +1,18 @@
 .PHONY: build
-uncloud-dev:
+update-dev:
 	GOOS=linux GOARCH=amd64 go build -o uncloudd-linux-amd64 ./cmd/uncloudd && \
 		scp uncloudd-linux-amd64 spy@192.168.40.243:~/ && \
 		ssh spy@192.168.40.243 sudo install ./uncloudd-linux-amd64 /usr/local/bin/uncloudd
 		scp uncloudd-linux-amd64 spy@192.168.40.176:~/ && \
 		ssh spy@192.168.40.176 sudo install ./uncloudd-linux-amd64 /usr/local/bin/uncloudd
 
+update-restart-dev:
+	GOOS=linux GOARCH=amd64 go build -o uncloudd-linux-amd64 ./cmd/uncloudd && \
+		scp uncloudd-linux-amd64 spy@192.168.40.243:~/ && \
+		ssh spy@192.168.40.243 "sudo install ./uncloudd-linux-amd64 /usr/local/bin/uncloudd && sudo systemctl restart uncloud" && \
+		scp uncloudd-linux-amd64 spy@192.168.40.176:~/ && \
+		ssh spy@192.168.40.176 "sudo install ./uncloudd-linux-amd64 /usr/local/bin/uncloudd && sudo systemctl restart uncloud"
+
 reset-dev:
 	ssh spy@192.168.40.243 "sudo systemctl stop uncloud && sudo rm -rf /var/lib/uncloud"
 	ssh spy@192.168.40.176 "sudo systemctl stop uncloud && sudo rm -rf /var/lib/uncloud"

internal/machine/docker/container/container.go

@@ -0,0 +1,9 @@
+package container
+
+import "github.com/docker/docker/api/types"
+
+type Container struct {
+	types.Container
+}
+
+// TODO: implement health related methods.

internal/machine/docker/manager.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	dockercontainer "github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/events"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
 	"log/slog"
 	"time"
+	"uncloud/internal/machine/docker/container"
 	"uncloud/internal/machine/store"
 )
 
@@ -20,26 +22,25 @@ const (
 	EventsDebounceInterval = 100 * time.Millisecond
 	// SyncInterval defines a regular interval to sync containers to the cluster store.
 	SyncInterval = 30 * time.Second
-
-	// SyncStatusSynced indicates that a container record is synchronised with the Docker daemon. The record may
-	// become outdated even when the status is "synced" if the machine crashes or a network partition occurs.
-	// The cluster membership state of the machine should also be checked to determine if the record can be trusted.
-	SyncStatusSynced = "synced"
-	// SyncStatusOutdated indicates that a container record may be outdated, for example, due to being unable
-	// to retrieve the container's state from the Docker daemon or when the machine is being stopped or restarted.
-	SyncStatusOutdated = "outdated"
 )
 
 type Manager struct {
 	client *client.Client
+	// machineID is the ID of the machine where the managed Docker daemon is running.
+	machineID string
+	store     *store.Store
 }
 
-func NewManager(client *client.Client) *Manager {
+func NewManager(client *client.Client, machineID string, store *store.Store) *Manager {
-	return &Manager{client: client}
+	return &Manager{
+		client:    client,
+		machineID: machineID,
+		store:     store,
+	}
 }
 
 // WaitDaemonReady waits for the Docker daemon to start and be ready to serve requests.
-func (d *Manager) WaitDaemonReady(ctx context.Context) error {
+func (m *Manager) WaitDaemonReady(ctx context.Context) error {
 	ticker := time.NewTicker(1 * time.Second)
 	defer ticker.Stop()
 
@@ -49,7 +50,7 @@ func (d *Manager) WaitDaemonReady(ctx context.Context) error {
 		case <-ctx.Done():
 			return ctx.Err()
 		case <-ticker.C:
-			_, err := d.client.Ping(ctx)
+			_, err := m.client.Ping(ctx)
 			if err == nil {
 				ready = true
 				break
@@ -66,7 +67,7 @@ func (d *Manager) WaitDaemonReady(ctx context.Context) error {
 	return nil
 }
 
-func (d *Manager) WatchAndSyncContainers(ctx context.Context, store *store.Store) error {
+func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	// Filter only local container events.
@@ -78,9 +79,9 @@ func (d *Manager) WatchAndSyncContainers(ctx context.Context, store *store.Store
 	}
 
 	// Subscribe to Docker events before running the initial sync to avoid missing any events.
-	eventCh, errCh := d.client.Events(ctx, opts)
+	eventCh, errCh := m.client.Events(ctx, opts)
 	slog.Debug("Syncing containers to cluster store before processing Docker events.")
-	if err := d.syncContainersToStore(ctx, store); err != nil {
+	if err := m.syncContainersToStore(ctx); err != nil {
 		// The deferred cancel will stop the event subscription.
 		return fmt.Errorf("sync containers to cluster store: %w", err)
 	}
@@ -123,13 +124,13 @@ func (d *Manager) WatchAndSyncContainers(ctx context.Context, store *store.Store
 				"container_name", e.Actor.Attributes["name"],
 				"action", e.Action)
 
-			if err := d.syncContainersToStore(ctx, store); err != nil {
+			if err := m.syncContainersToStore(ctx); 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 {
+			if err := m.syncContainersToStore(ctx); err != nil {
 				return fmt.Errorf("sync containers to cluster store: %w", err)
 			}
 		case err := <-errCh:
@@ -141,7 +142,20 @@ func (d *Manager) WatchAndSyncContainers(ctx context.Context, store *store.Store
 	}
 }
 
-func (d *Manager) syncContainersToStore(ctx context.Context, store *store.Store) error {
+func (m *Manager) syncContainersToStore(ctx context.Context) error {
-	// TODO: implement
+	containers, err := m.client.ContainerList(ctx, dockercontainer.ListOptions{})
+	if err != nil {
+		// TODO: mark all containers as outdated in the store.
+		return fmt.Errorf("list containers: %w", err)
+	}
+	for _, dc := range containers {
+		c := &container.Container{
+			Container: dc,
+		}
+		if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil {
+			return fmt.Errorf("create or update container: %w", err)
+		}
+	}
+
 	return nil
 }

internal/machine/network.go

@@ -182,7 +182,7 @@ func (nc *networkController) prepareAndWatchDocker(ctx context.Context) error {
 	}
 	defer cli.Close()
 
-	manager := docker.NewManager(cli)
+	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)
 	}
@@ -200,7 +200,7 @@ func (nc *networkController) prepareAndWatchDocker(ctx context.Context) error {
 		backoff.WithMaxElapsedTime(0),
 	), ctx)
 	watchAndSync := func() error {
-		if wErr := manager.WatchAndSyncContainers(ctx, nc.store); wErr != nil {
+		if wErr := manager.WatchAndSyncContainers(ctx); wErr != nil {
 			slog.Error("Failed to watch and sync containers to cluster store, retrying.", "err", wErr)
 			return wErr
 		}

internal/machine/store/container.go

@@ -0,0 +1,23 @@
+package store
+
+import (
+	"time"
+	"uncloud/internal/machine/docker/container"
+)
+
+const (
+	// SyncStatusSynced indicates that a container record is synchronised with the Docker daemon. The record may
+	// become outdated even when the status is "synced" if the machine crashes or a network partition occurs.
+	// The cluster membership state of the machine should also be checked to determine if the record can be trusted.
+	SyncStatusSynced = "synced"
+	// SyncStatusOutdated indicates that a container record may be outdated, for example, due to being unable
+	// to retrieve the container's state from the Docker daemon or when the machine is being stopped or restarted.
+	SyncStatusOutdated = "outdated"
+)
+
+type ContainerRecord struct {
+	Container  *container.Container
+	MachineID  string
+	SyncStatus string
+	UpdatedAt  time.Time
+}

internal/machine/store/schema.sql

@@ -12,6 +12,7 @@ CREATE TABLE machines
     name TEXT AS (json_extract(info, '$.name')),
     -- info is a JSON-serialized MachineInfo protobuf message.
     info TEXT NOT NULL DEFAULT '{}' CHECK (json_valid(info))
+    -- TODO: add created_at and updated_at fields to track machine age and last update time.
 );
 
 -- containers table stores the Uncloud-managed Docker containers created in the cluster.

internal/machine/store/store.go

@@ -3,12 +3,14 @@ package store
 import (
 	"context"
 	_ "embed"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"google.golang.org/protobuf/encoding/protojson"
 	"log/slog"
 	"uncloud/internal/corrosion"
 	"uncloud/internal/machine/api/pb"
+	"uncloud/internal/machine/docker/container"
 )
 
 var (
@@ -129,3 +131,32 @@ func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-cha
 
 	return machines, changes, nil
 }
+
+// CreateOrUpdateContainer creates a new container record or updates an existing one in the store database.
+// The container is associated with the given machine ID that indicates which machine the container is running on.
+func (s *Store) CreateOrUpdateContainer(ctx context.Context, c *container.Container, machineID string) error {
+	cJSON, err := json.Marshal(c)
+	if err != nil {
+		return fmt.Errorf("marshal container: %w", err)
+	}
+
+	// Insert or update the container record if the container or machine ID has changed.
+	res, err := s.corro.ExecContext(ctx, `
+		INSERT INTO containers (id, container, machine_id, sync_status, updated_at)
+		VALUES (?, ?, ?, ?, datetime('now'))
+		ON CONFLICT (id) DO UPDATE SET container   = excluded.container,
+									   machine_id  = excluded.machine_id,
+									   sync_status = excluded.sync_status,
+									   updated_at  = excluded.updated_at
+		WHERE containers.container != excluded.container
+		  OR containers.machine_id != excluded.machine_id`,
+		c.ID, string(cJSON), machineID, SyncStatusSynced)
+	if err != nil {
+		return fmt.Errorf("upsert query: %w", err)
+	}
+	if res.RowsAffected > 0 {
+		slog.Debug("Container record updated in store DB.", "id", c.ID, "machine_id", machineID)
+	}
+
+	return nil
+}