From 6b80ef9db8473ee21d9a221853534b4eef447c19 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Wed, 11 Dec 2024 15:14:32 +1000 Subject: [PATCH] subscribe to container changes in Caddyfile controller --- internal/machine/caddyfile/controller.go | 60 ++++++++++++++++++++- internal/machine/docker/manager.go | 1 + internal/machine/store/container.go | 66 +++++++++++++++++++++++- internal/machine/store/schema.sql | 2 +- internal/machine/store/store.go | 9 +++- 5 files changed, 134 insertions(+), 4 deletions(-) diff --git a/internal/machine/caddyfile/controller.go b/internal/machine/caddyfile/controller.go index d7208755..c7ba28c6 100644 --- a/internal/machine/caddyfile/controller.go +++ b/internal/machine/caddyfile/controller.go @@ -3,9 +3,11 @@ package caddyfile import ( "context" "fmt" + "log/slog" "os" "path/filepath" "sync" + "uncloud/internal/api" "uncloud/internal/machine/store" ) @@ -30,6 +32,62 @@ func NewController(store *store.Store, path string) (*Controller, error) { }, nil } -func (cc *Controller) Run(ctx context.Context) error { +func (c *Controller) Run(ctx context.Context) error { + containerRecords, changes, err := c.store.SubscribeContainers(ctx) + if err != nil { + return fmt.Errorf("subscribe to container changes: %w", err) + } + slog.Info("Subscribed to container changes in the cluster to generate Caddy configuration.") + + containers, err := c.filterAvailableContainers(containerRecords) + if err != nil { + return fmt.Errorf("filter available containers: %w", err) + } + if err = c.generateConfig(containers); err != nil { + return fmt.Errorf("generate Caddy configuration: %w", err) + } + + for { + select { + case _, ok := <-changes: + if !ok { + return fmt.Errorf("containers subscription failed") + } + slog.Debug("Cluster containers changed, updating Caddy configuration.") + + containerRecords, err = c.store.ListContainers(ctx, store.ListOptions{}) + if err != nil { + slog.Error("Failed to list containers.", "err", err) + continue + } + containers, err = c.filterAvailableContainers(containerRecords) + if err != nil { + slog.Error("Failed to filter available containers.", "err", err) + continue + } + if err = c.generateConfig(containers); err != nil { + slog.Error("Failed to generate Caddy configuration.", "err", err) + } + + slog.Debug("Updated Caddy configuration.", "path", c.path) + case <-ctx.Done(): + return nil + } + } +} + +// filterAvailableContainers filters out containers that are likely unavailable from this machine. The availability +// is determined by the cluster membership state of the machine that the container is running on. +// TODO: implement machine membership check using Corrossion Admin client. +func (c *Controller) filterAvailableContainers(containerRecords []*store.ContainerRecord) ([]*api.Container, error) { + containers := make([]*api.Container, len(containerRecords)) + for i, cr := range containerRecords { + containers[i] = cr.Container + } + return containers, nil +} + +func (c *Controller) generateConfig(containers []*api.Container) error { + // TODO return nil } diff --git a/internal/machine/docker/manager.go b/internal/machine/docker/manager.go index 8f725b7b..24a4e5ca 100644 --- a/internal/machine/docker/manager.go +++ b/internal/machine/docker/manager.go @@ -184,6 +184,7 @@ func (m *Manager) syncContainersToStore(ctx context.Context) error { } // Create or update the current Docker containers in the store. + // TODO: update only the changed containers to reduce unnecessary gossip traffic and not trigger other controllers. for _, dc := range containers { c := &api.Container{Container: dc} if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil { diff --git a/internal/machine/store/container.go b/internal/machine/store/container.go index 40200c30..9a1b7da3 100644 --- a/internal/machine/store/container.go +++ b/internal/machine/store/container.go @@ -76,7 +76,8 @@ func (s *Store) CreateOrUpdateContainer(ctx context.Context, c *api.Container, m // ListContainers returns a list of container records from the store database that match the given options. func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]*ContainerRecord, error) { - q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers") + q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers"). + Where(sq.Eq{"sync_status": SyncStatusSynced}) if len(opts.MachineIDs) > 0 { q = q.Where(sq.Eq{"machine_id": opts.MachineIDs}) @@ -154,3 +155,66 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error return nil } + +// SubscribeContainers returns a list of containers and a channel that signals changes to the list. The channel doesn't +// receive any values, it just signals when a container(s) has been added, updated, or deleted in the database. +func (s *Store) SubscribeContainers(ctx context.Context) ([]*ContainerRecord, <-chan struct{}, error) { + // TODO: figure out whether we need sync_status at all. + q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers"). + Where(sq.Eq{"sync_status": SyncStatusSynced}) + query, args, err := q.ToSql() + if err != nil { + return nil, nil, fmt.Errorf("build query: %w", err) + } + + sub, err := s.corro.SubscribeContext(ctx, query, args, false) + if err != nil { + return nil, nil, err + } + + var containers []*ContainerRecord + var cJSON, updatedAtStr string + + rows := sub.Rows() + for rows.Next() { + var cr ContainerRecord + if err = rows.Scan(&cJSON, &cr.MachineID, &cr.SyncStatus, &updatedAtStr); err != nil { + return nil, nil, err + } + + if err = json.Unmarshal([]byte(cJSON), &cr.Container); err != nil { + return nil, nil, fmt.Errorf("unmarshal container: %w", err) + } + if cr.UpdatedAt, err = time.Parse(time.DateTime, updatedAtStr); err != nil { + return nil, nil, fmt.Errorf("parse updated_at: %w", err) + } + containers = append(containers, &cr) + } + events, err := sub.Changes() + if err != nil { + return nil, nil, fmt.Errorf("get subscription changes: %w", err) + } + + changes := make(chan struct{}) + go func() { + defer close(changes) + for { + select { + case <-ctx.Done(): + return + case _, ok := <-events: + if !ok { + // events channel has been closed. + if sub.Err() != nil { + slog.Error("Containers subscription failed.", "id", sub.ID(), "err", sub.Err()) + } + return + } + // Just signal that there is a change in the containers list. + changes <- struct{}{} + } + } + }() + + return containers, changes, nil +} diff --git a/internal/machine/store/schema.sql b/internal/machine/store/schema.sql index 788baab8..5ab1da63 100644 --- a/internal/machine/store/schema.sql +++ b/internal/machine/store/schema.sql @@ -19,7 +19,7 @@ CREATE TABLE machines CREATE TABLE containers ( id TEXT NOT NULL PRIMARY KEY, - -- container is a JSON-serialized Docker container.Summary struct. + -- container is a JSON-serialized api.Container struct. container TEXT NOT NULL DEFAULT '{}' CHECK (json_valid(container)), machine_id TEXT NOT NULL DEFAULT '', service_id TEXT AS (json_extract(container, '$.Labels."uncloud.service.id"')), diff --git a/internal/machine/store/store.go b/internal/machine/store/store.go index 6b746829..1cc5ef67 100644 --- a/internal/machine/store/store.go +++ b/internal/machine/store/store.go @@ -120,7 +120,14 @@ func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-cha select { case <-ctx.Done(): return - case <-events: + case _, ok := <-events: + if !ok { + // events channel has been closed. + if sub.Err() != nil { + slog.Error("Machines subscription failed.", "id", sub.ID(), "err", sub.Err()) + } + return + } // Just signal that there is a change in the machines list. changes <- struct{}{} }