From b4c2ec3a2a0fd656288c12ba08bca6fab9326686 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Tue, 19 Nov 2024 16:06:16 +1000 Subject: [PATCH] sync new and updated containers to cluster store, no deletion yet --- Makefile | 9 ++- .../machine/docker/container/container.go | 9 +++ internal/machine/docker/manager.go | 52 +++++++++++------- internal/machine/network.go | 4 +- internal/machine/store/container.go | 23 ++++++++ internal/machine/store/db | Bin 0 -> 45056 bytes internal/machine/store/schema.sql | 1 + internal/machine/store/store.go | 31 +++++++++++ 8 files changed, 107 insertions(+), 22 deletions(-) create mode 100644 internal/machine/docker/container/container.go create mode 100644 internal/machine/store/container.go create mode 100644 internal/machine/store/db 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 0000000000000000000000000000000000000000..3c34a76c581ef20de578b6321c0bdec1e704bf16 GIT binary patch literal 45056 zcmeI)&u-&H9KdlqNt4DUn5u^$N~P(tQfg_FZQ5NSdV$aeD4VuRlcMb@a$^_F=Fc{E zcDqzPEFf{_X^@b325!6oToxe?JOJY)j@_oI5T|YUT1_$;&&<#BnO_u5UA6jI#qn%$ z-08MVPpm6%D60C}2SO-{l9rD-`3R1bEF^+A@>PwLCl%95q5V@v|3lHVHx>Or|M~T| z^xJd4WPY1_HkZ(Tkd+(=AbA z-L<<DTV~@+IqP~fWcs?0VTa2h!!yR`sbu!E^2(ub9DWF52>O~^w^g-5OYs3vyz z*V8eZ&eygjwmOZ+b~kKQsPxpQR$#k6S=eVoY?%Ny!k+^eC2l9^`|m`wKD2y7=tSs za>&)1JXCJbG|0NJX{kCK+&=ZXX2T1Izaw(D3l;OoZn}kAy>_G7=~;#000qaol?&nr zgZhgxn96?~)E^)mw)ey`J=?0wEt(YC-YHjWrJaZ2%8Zzb@4dU4UoFZXvAVj^|H)~_ zdNP^)cv)5Y+2)=!<&Ljz_Pjy%ML*cZ5vZH`CKXR*mzLCT&3-Bci59#jE~HUV8%?5% zD>;@t*AO}F<4&L4Z!@x$+2)_PI?;9pAJc8W!IN&s>m2otMay~3T561(6wtQ@jEwx5rs z?U 0 { + slog.Debug("Container record updated in store DB.", "id", c.ID, "machine_id", machineID) + } + + return nil +}