philipp/uncloud
1
0
Fork 0
mirror of https://github.com/psviderski/uncloud.git synced 2026-08-26 11:03:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: store ServiceContainer (includes service spec) instead of Container in Corrosion store

Browse Source
This commit is contained in:
Pasha Sviderski
2025-08-14 14:53:33 +10:00
parent 9186d31d12
commit 4cc1e556dd
10 changed files with 174 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/machine/docker/manager.go → internal/machine/docker/controller.go
+24 -39
View File
@@ -7,12 +7,11 @@ import (
"log/slog"
"time"
dockercontainer "github.com/docker/docker/api/types/container"
"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"
"github.com/psviderski/uncloud/internal/machine/store"
"github.com/psviderski/uncloud/pkg/api"
)
const (
@@ -24,23 +23,26 @@ const (
SyncInterval = 30 * time.Second
)
type Manager struct {
client *client.Client
// Controller monitors Docker events and synchronises service containers with the cluster store.
type Controller struct {
// machineID is the ID of the machine where the managed Docker daemon is running.
machineID string
client *client.Client
service *Service
store *store.Store
}
func NewManager(client *client.Client, machineID string, store *store.Store) *Manager {
return &Manager{
client: client,
func NewController(machineID string, service *Service, store *store.Store) *Controller {
return &Controller{
machineID: machineID,
client: service.Client,
service: service,
store: store,
}
}
// WaitDaemonReady waits for the Docker daemon to start and be ready to serve requests.
func (m *Manager) WaitDaemonReady(ctx context.Context) error {
func (c *Controller) WaitDaemonReady(ctx context.Context) error {
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
@@ -50,7 +52,7 @@ func (m *Manager) WaitDaemonReady(ctx context.Context) error {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
_, err := m.client.Ping(ctx)
_, err := c.client.Ping(ctx)
if err == nil {
ready = true
break
@@ -67,7 +69,7 @@ func (m *Manager) WaitDaemonReady(ctx context.Context) error {
return nil
}
func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
func (c *Controller) WatchAndSyncContainers(ctx context.Context) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// Filter only local container events.
@@ -79,9 +81,9 @@ func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
}
// Subscribe to Docker events before running the initial sync to avoid missing any events.
eventCh, errCh := m.client.Events(ctx, opts)
eventCh, errCh := c.service.Client.Events(ctx, opts)
slog.Debug("Syncing containers to cluster store before processing Docker events.")
if err := m.syncContainersToStore(ctx); err != nil {
if err := c.syncContainersToStore(ctx); err != nil {
// The deferred cancel will stop the event subscription.
return fmt.Errorf("sync containers to cluster store: %w", err)
}
@@ -126,13 +128,13 @@ func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
"container_name", e.Actor.Attributes["name"],
"action", e.Action)
if err := m.syncContainersToStore(ctx); err != nil {
if err := c.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 := m.syncContainersToStore(ctx); err != nil {
if err := c.syncContainersToStore(ctx); err != nil {
return fmt.Errorf("sync containers to cluster store: %w", err)
}
case err := <-errCh:
@@ -144,33 +146,16 @@ func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
}
}
func (m *Manager) syncContainersToStore(ctx context.Context) error {
storeContainers, err := m.store.ListContainers(ctx, store.ListOptions{MachineIDs: []string{m.machineID}})
func (c *Controller) syncContainersToStore(ctx context.Context) error {
storeContainers, err := c.store.ListContainers(ctx, store.ListOptions{MachineIDs: []string{c.machineID}})
if err != nil {
return fmt.Errorf("list containers from store: %w", err)
}
// List only Uncloud service containers identified by their labels.
containerSummaries, err := m.client.ContainerList(ctx, dockercontainer.ListOptions{
Filters: filters.NewArgs(
filters.Arg("label", api.LabelServiceID),
filters.Arg("label", api.LabelServiceName),
filters.Arg("label", api.LabelManaged),
),
})
containers, err := c.service.ListServiceContainers(ctx, "", container.ListOptions{})
if err != nil {
// TODO: mark all containers as outdated in the store.
return fmt.Errorf("list Docker containers: %w", err)
}
// Inspect each container to get the full container details.
containers := make([]api.Container, len(containerSummaries))
for i, cs := range containerSummaries {
ctr, err := m.client.ContainerInspect(ctx, cs.ID)
if err != nil {
return fmt.Errorf("inspect container '%s': %w", cs.ID, err)
}
containers[i] = api.Container{ContainerJSON: ctr}
return fmt.Errorf("list service containers: %w", err)
}
// Delete containers from the store that are no longer present in the Docker daemon.
@@ -190,15 +175,15 @@ func (m *Manager) syncContainersToStore(ctx context.Context) error {
var storeErr error
if len(deleteIDs) > 0 {
if err = m.store.DeleteContainers(ctx, store.DeleteOptions{IDs: deleteIDs}); err != nil {
if err = c.store.DeleteContainers(ctx, store.DeleteOptions{IDs: deleteIDs}); err != nil {
storeErr = fmt.Errorf("delete containers from store: %w", err)
}
}
// Create or update the current Docker containers in the store.
for _, c := range containers {
if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil {
storeErr = errors.Join(storeErr, fmt.Errorf("create or update container %q: %w", c.ID, err))
for _, ctr := range containers {
if err = c.store.CreateOrUpdateContainer(ctx, ctr, c.machineID); err != nil {
storeErr = errors.Join(storeErr, fmt.Errorf("create or update container '%s': %w", ctr.ID, err))
}
}
return storeErr
internal/machine/docker/manager_darwin.go → internal/machine/docker/controller_darwin.go
+2 -2
View File
@@ -9,11 +9,11 @@ import (
)
// EnsureUncloudNetwork is a stub for Darwin.
func (m *Manager) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix, dnsServer netip.Addr) error {
func (c *Controller) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix, dnsServer netip.Addr) error {
return fmt.Errorf("not supported on Darwin")
}
// Cleanup is a stub for Darwin.
func (m *Manager) Cleanup() error {
func (c *Controller) Cleanup() error {
return fmt.Errorf("not supported on Darwin")
}
internal/machine/docker/manager_linux.go → internal/machine/docker/controller_linux.go
+11 -11
View File
@@ -22,10 +22,10 @@ import (
// EnsureUncloudNetwork creates the Docker bridge network NetworkName with the provided machine subnet
// if it doesn't exist. If the network exists but has a different subnet, it removes and recreates the network.
// It also configures iptables to allow container access from the WireGuard network.
func (m *Manager) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix, dnsServer netip.Addr) error {
func (c *Controller) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix, dnsServer netip.Addr) error {
// Ensure the Docker network 'uncloud' is created with the correct subnet.
needsCreation := false
nw, err := m.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{})
nw, err := c.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{})
if err != nil {
if !client.IsErrNotFound(err) {
return fmt.Errorf("inspect Docker network '%s': %w", NetworkName, err)
@@ -37,7 +37,7 @@ func (m *Manager) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix,
slog.Info(
"Removing Docker network with old subnet.", "name", NetworkName, "subnet", nw.IPAM.Config[0].Subnet,
)
if err = m.client.NetworkRemove(ctx, NetworkName); err != nil {
if err = c.client.NetworkRemove(ctx, NetworkName); err != nil {
// It can still fail if the network is in use by a container. Leave it to the user to resolve the issue.
return fmt.Errorf("remove Docker network '%s': %w", NetworkName, err)
}
@@ -45,7 +45,7 @@ func (m *Manager) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix,
}
if needsCreation {
if _, err = m.client.NetworkCreate(
if _, err = c.client.NetworkCreate(
ctx, NetworkName, dnetwork.CreateOptions{
Driver: "bridge",
Scope: "local",
@@ -70,7 +70,7 @@ func (m *Manager) EnsureUncloudNetwork(ctx context.Context, subnet netip.Prefix,
}
slog.Info("Docker network created.", "name", NetworkName, "subnet", subnet.String())
if nw, err = m.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{}); err != nil {
if nw, err = c.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{}); err != nil {
return fmt.Errorf("inspect Docker network '%s': %w", NetworkName, err)
}
}
@@ -168,12 +168,12 @@ func cleanupIptables(bridgeName string, subnet netip.Prefix) error {
}
// Cleanup removes all uncloud-managed containers and the uncloud Docker network.
func (m *Manager) Cleanup() error {
func (c *Controller) Cleanup() error {
ctx := context.Background()
var errs []error
// Remove uncloud-managed Docker containers.
containers, err := m.client.ContainerList(ctx, dockercontainer.ListOptions{
containers, err := c.client.ContainerList(ctx, dockercontainer.ListOptions{
All: true, // Include stopped containers.
Filters: filters.NewArgs(
filters.Arg("label", api.LabelManaged),
@@ -186,12 +186,12 @@ func (m *Manager) Cleanup() error {
removed := 0
for _, ctr := range containers {
err = m.client.ContainerStop(ctx, ctr.ID, dockercontainer.StopOptions{})
err = c.client.ContainerStop(ctx, ctr.ID, dockercontainer.StopOptions{})
if err != nil && !client.IsErrNotFound(err) {
errs = append(errs, fmt.Errorf("stop container '%s': %w", ctr.ID, err))
}
err = m.client.ContainerRemove(ctx, ctr.ID, dockercontainer.RemoveOptions{
err = c.client.ContainerRemove(ctx, ctr.ID, dockercontainer.RemoveOptions{
// Remove anonymous volumes created by the container.
RemoveVolumes: true,
})
@@ -205,7 +205,7 @@ func (m *Manager) Cleanup() error {
}
// Remove the uncloud Docker network and related iptables rules.
nw, err := m.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{})
nw, err := c.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{})
if err == nil {
bridgeName := "br-" + nw.ID[:12]
var subnet netip.Prefix
@@ -221,7 +221,7 @@ func (m *Manager) Cleanup() error {
}
}
if err = m.client.NetworkRemove(ctx, NetworkName); err == nil {
if err = c.client.NetworkRemove(ctx, NetworkName); err == nil {
slog.Info("Docker network removed.", "name", NetworkName)
} else if !client.IsErrNotFound(err) {
errs = append(errs, fmt.Errorf("remove Docker network '%s': %w", NetworkName, err))
internal/machine/docker/server.go
+4 -3
View File
@@ -73,10 +73,11 @@ func WithWaitForNetworkReady(waitForNetworkReady func(ctx context.Context) error
}
}
// NewServer creates a new Docker gRPC server with the provided Docker client.
func NewServer(cli *client.Client, db *sqlx.DB, internalDNSIP func() netip.Addr, opts ...ServerOption) *Server {
// NewServer creates a new Docker gRPC server with the provided Docker service.
// TODO: refactor to use methods from the Docker service instead of querying the database directly.
func NewServer(service *Service, db *sqlx.DB, internalDNSIP func() netip.Addr, opts ...ServerOption) *Server {
s := &Server{
client: cli,
client: service.Client,
db: db,
internalDNSIP: internalDNSIP,
}
internal/machine/docker/service.go
+103
View File
@@ -0,0 +1,103 @@
package docker
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/client"
"github.com/jmoiron/sqlx"
"github.com/psviderski/uncloud/pkg/api"
)
// Service provides higher-level Docker operations that extends Docker API with Uncloud-specific data
// from the machine database.
type Service struct {
Client *client.Client
db *sqlx.DB
}
// NewService creates a new Docker service instance.
func NewService(client *client.Client, db *sqlx.DB) *Service {
return &Service{
Client: client,
db: db,
}
}
// InspectServiceContainer inspects a Docker container and retrieves its associated ServiceSpec
// from the machine database, returning a complete ServiceContainer.
func (s *Service) InspectServiceContainer(ctx context.Context, nameOrID string) (api.ServiceContainer, error) {
var serviceCtr api.ServiceContainer
ctr, err := s.Client.ContainerInspect(ctx, nameOrID)
if err != nil {
return serviceCtr, err
}
if _, ok := ctr.Config.Labels[api.LabelManaged]; !ok {
return serviceCtr, fmt.Errorf("container '%s' is not managed by Uncloud", nameOrID)
}
serviceCtr.Container = api.Container{ContainerJSON: ctr}
// Retrieve ServiceSpec from the machine database.
var specBytes []byte
err = s.db.QueryRowContext(ctx, `SELECT service_spec FROM containers WHERE id = $1`, ctr.ID).Scan(&specBytes)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
// If this happens, there is a bug in the code, or someone manually removed the container from the DB,
// or created a managed container out of band or by previous uncloud installation.
return serviceCtr, fmt.Errorf("service spec not found for container '%s' in machine DB", ctr.ID)
}
return serviceCtr, fmt.Errorf("get service spec for container '%s' from machine DB: %w", ctr.ID, err)
}
if err = json.Unmarshal(specBytes, &serviceCtr.ServiceSpec); err != nil {
return serviceCtr, fmt.Errorf("unmarshal service spec for container '%s': %w", ctr.ID, err)
}
return serviceCtr, nil
}
// ListServiceContainers lists Docker containers that belong to the service with the given name or ID.
// If serviceIDOrName is empty, all service containers are returned. The opts parameter allows additional filtering.
func (s *Service) ListServiceContainers(
ctx context.Context, serviceNameOrID string, opts container.ListOptions,
) ([]api.ServiceContainer, error) {
if opts.Filters.Len() == 0 {
opts.Filters = filters.NewArgs()
}
// Add labels to existing filters to list only Uncloud-managed service containers.
opts.Filters.Add("label", api.LabelServiceID)
opts.Filters.Add("label", api.LabelManaged)
containerSummaries, err := s.Client.ContainerList(ctx, opts)
if err != nil {
return nil, err
}
var containers []api.ServiceContainer
for _, cs := range containerSummaries {
// Filter by service name or ID if provided.
if serviceNameOrID != "" &&
cs.Labels[api.LabelServiceID] != serviceNameOrID &&
cs.Labels[api.LabelServiceName] != serviceNameOrID {
continue
}
ctr, err := s.InspectServiceContainer(ctx, cs.ID)
if err != nil {
// Log error but continue with other containers.
slog.Error("Failed to inspect service container.", "service", serviceNameOrID, "id", cs.ID, "err", err)
continue
}
containers = append(containers, ctr)
}
return containers, nil
}