diff --git a/Makefile b/Makefile index bec44bae..cf502249 100644 --- a/Makefile +++ b/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" diff --git a/internal/machine/docker/container/container.go b/internal/machine/docker/container/container.go new file mode 100644 index 00000000..ee596062 --- /dev/null +++ b/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. diff --git a/internal/machine/docker/manager.go b/internal/machine/docker/manager.go index 1bb87ed9..3a31b0c8 100644 --- a/internal/machine/docker/manager.go +++ b/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 { - return &Manager{client: client} +func NewManager(client *client.Client, machineID string, store *store.Store) *Manager { + 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 { - // TODO: implement +func (m *Manager) syncContainersToStore(ctx context.Context) error { + 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 } diff --git a/internal/machine/network.go b/internal/machine/network.go index c0f766c6..df74f249 100644 --- a/internal/machine/network.go +++ b/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 } diff --git a/internal/machine/store/container.go b/internal/machine/store/container.go new file mode 100644 index 00000000..c9de5852 --- /dev/null +++ b/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 +} diff --git a/internal/machine/store/db b/internal/machine/store/db new file mode 100644 index 00000000..3c34a76c Binary files /dev/null and b/internal/machine/store/db differ diff --git a/internal/machine/store/schema.sql b/internal/machine/store/schema.sql index 8ce87966..788baab8 100644 --- a/internal/machine/store/schema.sql +++ b/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. diff --git a/internal/machine/store/store.go b/internal/machine/store/store.go index 8bb7ff30..7166b736 100644 --- a/internal/machine/store/store.go +++ b/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 +}