Compare commits

..
14 Commits
Author SHA1 Message Date
Pasha Sviderski 1ce3e62dbb fix: use local and remote Docker credentials to pull image from private registry 2025-08-15 14:09:40 +10:00
Pasha Sviderski 4be8339c51 chore: generate a minimal Caddyfile with verify handler alongside caddy.json 2025-08-14 19:43:40 +10:00
Pasha Sviderski 8dd69b46da chore: refactor docker gRPC server to use docker service for inspecting and listing containers 2025-08-14 15:34:59 +10:00
Pasha Sviderski 4cc1e556dd chore: store ServiceContainer (includes service spec) instead of Container in Corrosion store 2025-08-14 14:53:33 +10:00
Pasha Sviderski 9186d31d12 chore: go mod tidy 2025-08-13 19:31:13 +10:00
Pasha Sviderski dd7bc6c982 chore: trim spaces for x-caddy, diff Caddy configs when comparing service specs 2025-08-13 19:27:40 +10:00
Pasha Sviderski 12c07812a2 chore: add Caddy config to ServiceSpec, load x-caddy to it 2025-08-13 18:44:38 +10:00
Pasha Sviderski ec73f9ecd8 chore: handle x-caddy: path/to/Caddyfile to read Caddy config in compose from file 2025-08-13 13:45:39 +10:00
Pasha Sviderski 879c7c1876 test: x-caddy extension parsing 2025-08-13 13:45:39 +10:00
Anton Ovchinnikov c67127f83f feat: Add basic LLM instruction files 2025-08-11 23:57:13 +02:00
Pasha Sviderski 5d3f1fe225 chore: x-caddy extension type in compose 2025-08-11 21:11:40 +10:00
Pasha Sviderski 2e585d0183 feat: add --recreate flag for deploy command to force container recreation 2025-08-07 18:09:35 +10:00
Pasha Sviderski ae9f943404 chore: change default restart policy for service containers always -> unless-stopped 2025-08-07 18:09:35 +10:00
Anton Ovchinnikov 8805178a58 doc: Add Sentry to sponsors 2025-08-07 00:03:38 +02:00
39 changed files with 1506 additions and 241 deletions
+1
View File
@@ -0,0 +1 @@
../AI.md
+244
View File
@@ -0,0 +1,244 @@
# AI.md - Uncloud Project Guide
This document provides comprehensive information about the Uncloud project for AI assistants to understand the codebase, architecture, and development practices.
## Project Overview
**Uncloud** is a lightweight clustering and container orchestration tool that enables deployment and management of web applications across cloud VMs and bare metal servers. It creates a secure WireGuard mesh network between Docker hosts and provides automatic service discovery, load balancing, HTTPS ingress, and simple CLI commands for application management.
### Key Characteristics
- **Language**: Go
- **Architecture**: Decentralized, no control plane
- **Target**: Self-hosted infrastructure without Kubernetes complexity
- **License**: View LICENSE file for details
- **Status**: Active development, not yet ready for production
## Core Features
### 🏗️ Infrastructure
- **Multi-machine deployment**: Combine cloud VMs, dedicated servers, and bare metal
- **Zero-config networking**: Automatic WireGuard mesh with NAT traversal
- **Decentralized design**: No central control plane, all machines are equal
- **Service discovery**: Built-in DNS server resolves service names to container IPs
### 🚀 Application Management
- **Docker Compose compatibility**: Uses familiar Docker Compose format
- **Zero-downtime deployments**: Rolling updates without service interruption
- **Automatic HTTPS**: Caddy reverse proxy with Let's Encrypt integration
- **Managed DNS**: Free `*.cluster.uncloud.run` subdomains via Uncloud DNS service
- **Cross-machine scaling**: Run containers across multiple machines
### 🔧 Developer Experience
- **Docker-like CLI**: Familiar commands (`uc` binary)
- **Imperative operations**: Direct commands vs. declarative state reconciliation
- **Remote management**: Control entire infrastructure via SSH to any machine
- **Minimal overhead**: ~150MB RAM footprint per machine
## Architecture
### Core Components
1. **CLI (`uc`)** - Main user interface for cluster management
2. **Daemon (`uncloudd`)** - Machine daemon running on each node
3. **Corrosion** - Distributed SQLite database for cluster state (Fly.io project)
4. **Caddy** - Reverse proxy for HTTPS termination and routing
5. **WireGuard** - Secure mesh networking between machines
### Network Architecture
- Each machine gets unique subnet (e.g., `10.210.0.0/24`, `10.210.1.0/24`)
- Containers get cluster-unique IPs for direct communication
- Automatic peer discovery and key management
- NAT traversal for machines behind firewalls
### State Management
- **CRDT-based distributed storage** using Corrosion
- **Eventually consistent** state across all machines
- **Gossip protocol** (Serf) for state propagation
- **No quorum requirements** - partial network splits remain functional
## Project Structure
### Key Directories
- **`cmd/`**: Contains main applications
- `uncloud/`: CLI tool with subcommands for machine, service, volume management
- `uncloudd/`: Daemon that runs on each machine
- `ucind/`: Development cluster management for testing
- **`internal/`**: Internal implementation packages
- `cli/`: Command-line interface logic
- `machine/`: Machine lifecycle and state management
- `daemon/`: Daemon implementation and gRPC services
- `dns/`: Internal DNS server for service discovery
- **`pkg/`**: Public API packages for external use
- `api/`: Core API types and definitions
- `client/`: Client libraries for interacting with Uncloud
- **`experiment/`**: Experimental features and prototypes
- **`scripts/`**: Installation and utility scripts
- **`test/`**: Test suites and test infrastructure
- **`website/`**: Documentation website (Docusaurus)
- **`misc/`**: Design documents and guides
## Key Technologies
### Core Dependencies
```go
// Networking and orchestration
github.com/docker/docker // Docker API client
github.com/docker/compose/v2 // Docker Compose integration
golang.zx2c4.com/wireguard // WireGuard implementation
github.com/hashicorp/serf // Gossip protocol
// State management
github.com/ipfs/go-ds-crdt // CRDT distributed storage
github.com/dgraph-io/badger/v3 // Embedded database
// Web proxy
github.com/caddyserver/caddy/v2 // HTTP server and reverse proxy
// CLI and UX
github.com/spf13/cobra // CLI framework
github.com/charmbracelet/huh // Interactive forms
// gRPC and networking
google.golang.org/grpc // gRPC framework
github.com/siderolabs/grpc-proxy // gRPC proxy for forwarding
```
## Development Workflow
### Build and Development
```bash
# Build binaries
go build -o uncloud ./cmd/uncloud
go build -o uncloudd ./cmd/uncloudd
```
### Key Make Targets
- `proto`: Generate protobuf code
- `ucind-cluster`: Create development cluster
- `update-dev`: Deploy to development machines
- `demo-reset`: Reset demo environment
- `fmt`: Format code
- `test`: Run all tests
- `lint`: Lint the code using golangci-lint
- `lint-and-fix`: Lint the code and fix issues whenever possible
## CLI Commands Structure
The `uc` CLI provides these main command groups:
### Machine Management
```bash
uc machine init <user@host> # Initialize new cluster
uc machine add <user@host> # Add machine to cluster
uc machine ls # List machines
uc machine rm <name> # Remove machine
```
### Service Management
```bash
uc run <image> # Run container from image
uc deploy # Deploy from compose.yaml
uc scale <service> <count> # Scale service replicas
uc ls # List services
uc rm <service> # Remove service
```
### Context and Connectivity
```bash
uc context ls # List available contexts
uc context use <name> # Switch context
```
### Global Flags
- `--connect`: Connect to remote machine directly, without a config file
- `--uncloud-config`: Override config file path
## Development Guidelines
### Code Organization
- **Package naming**: Use clear, descriptive names
- **Error handling**: Wrap errors with context using `fmt.Errorf`
- **Logging**: Use structured logging with levels
- **gRPC**: Services defined in `internal/machine/api/pb/`
### Testing
- Unit tests alongside source files (`*_test.go`)
- Integration tests in `test/e2e/`
- Test fixtures in `test/fixtures/`
### Dependencies
- Prefer standard library when possible
- Pin versions in `go.mod`
- Document rationale for external dependencies
### Configuration
- Support environment variables for key settings
- Validate configuration early
- Provide sensible defaults
## Troubleshooting and Debugging
### Common Issues
- **Networking**: Check WireGuard status, iptables rules
- **DNS**: Verify service discovery resolution
- **Containers**: Use standard Docker debugging tools
- **State sync**: Check Corrosion logs for replication issues
### Debugging Tools
- Standard Linux networking tools (`ping`, `traceroute`, `wireshark`)
- Docker commands (`docker ps`, `docker logs`)
- SSH access to machines for direct inspection
- gRPC debugging tools
### Logs and Monitoring
- Systemd services (getting logs via `journalctl -u SERVICE_NAME`)
- `uncloud` -- Uncloud daemon
- `uncloud-corrosion` -- Corrosion process
- Machine daemon logs
- Container logs via Docker
## File Patterns and Conventions
### Important Files to Understand
- `cmd/uncloud/main.go`: CLI entry point and command structure
- `internal/cli/cli.go`: CLI implementation and configuration
- `internal/machine/machine.go`: Core machine management
- `pkg/api/`: Public API definitions
- `misc/design.md`: Architecture and design philosophy
- `README.md`: User-facing documentation
### Configuration Files
- `go.mod/go.sum`: Go dependency management
- `Makefile`: Build and development tasks
- `Dockerfile`: Container build instructions forUncloud-in-Docker (used for testing)
This document should help AI assistants understand the project structure, make informed suggestions, and contribute effectively to the Uncloud codebase.
Symlink
+1
View File
@@ -0,0 +1 @@
./AI.md
+9
View File
@@ -323,6 +323,15 @@ SQLite database used to share Uncloud's cluster state.
features, and be the first to know when it's ready for production use. features, and be the first to know when it's ready for production use.
* Watch this repository for releases. * Watch this repository for releases.
## 💖 Sponsors
These companies and projects are helping Uncloud with their generous sponsorship and/or services:
<!-- Sentry -->
<a href="https://sentry.io/welcome/">
<img height="100" alt="Sentry" src="https://github.com/user-attachments/assets/6c1439c0-d20d-40dc-a669-c9aa94651dfa" />
</a>
## ❤️ Contributors ## ❤️ Contributors
Thank you [@cedws](https://github.com/cedws) for being the first contributor to Uncloud! 🎉 Thank you [@cedws](https://github.com/cedws) for being the first contributor to Uncloud! 🎉
+8 -1
View File
@@ -21,6 +21,7 @@ type deployOptions struct {
profiles []string profiles []string
services []string services []string
noBuild bool noBuild bool
recreate bool
context string context string
} }
@@ -50,6 +51,8 @@ func NewDeployCommand() *cobra.Command {
"Name of the cluster context to deploy to (default is the current context)") "Name of the cluster context to deploy to (default is the current context)")
cmd.Flags().BoolVarP(&opts.noBuild, "no-build", "n", false, cmd.Flags().BoolVarP(&opts.noBuild, "no-build", "n", false,
"Do not build images before deploying services. (default false)") "Do not build images before deploying services. (default false)")
cmd.Flags().BoolVar(&opts.recreate, "recreate", false,
"Recreate containers even if their configuration and image haven't changed.")
// TODO: Consider adding a filter flag to specify which machines to deploy to but keep the rest running. // TODO: Consider adding a filter flag to specify which machines to deploy to but keep the rest running.
// Could be useful to test a new version on a subset of machines before rolling out to all. // Could be useful to test a new version on a subset of machines before rolling out to all.
@@ -108,7 +111,11 @@ func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error {
} }
defer clusterClient.Close() defer clusterClient.Close()
composeDeploy, err := compose.NewDeployment(ctx, clusterClient, project) var strategy deploy.Strategy
if opts.recreate {
strategy = &deploy.RollingStrategy{ForceRecreate: true}
}
composeDeploy, err := compose.NewDeploymentWithStrategy(ctx, clusterClient, project, strategy)
if err != nil { if err != nil {
return fmt.Errorf("create compose deployment: %w", err) return fmt.Errorf("create compose deployment: %w", err)
} }
+1 -1
View File
@@ -35,6 +35,7 @@ require (
github.com/jmoiron/sqlx v1.4.0 github.com/jmoiron/sqlx v1.4.0
github.com/lmittmann/tint v1.0.5 github.com/lmittmann/tint v1.0.5
github.com/miekg/dns v1.1.65 github.com/miekg/dns v1.1.65
github.com/mitchellh/mapstructure v1.5.0
github.com/moby/term v0.5.0 github.com/moby/term v0.5.0
github.com/opencontainers/go-digest v1.0.0 github.com/opencontainers/go-digest v1.0.0
github.com/opencontainers/image-spec v1.1.0 github.com/opencontainers/image-spec v1.1.0
@@ -204,7 +205,6 @@ require (
github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/mitchellh/go-homedir v1.1.0 // indirect
github.com/mitchellh/go-ps v1.0.0 // indirect github.com/mitchellh/go-ps v1.0.0 // indirect
github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect
github.com/moby/buildkit v0.17.2 // indirect github.com/moby/buildkit v0.17.2 // indirect
github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect
+17
View File
@@ -0,0 +1,17 @@
package caddyconfig
import (
"fmt"
"github.com/psviderski/uncloud/pkg/api"
)
func GenerateCaddyfile(containers []api.ServiceContainer, verifyResponse string) (string, error) {
return fmt.Sprintf(`http:// {
handle %s {
respond "%s" 200
}
log
}
`, VerifyPath, verifyResponse), nil
}
+51 -27
View File
@@ -23,23 +23,24 @@ const (
// network. // network.
type Controller struct { type Controller struct {
store *store.Store store *store.Store
path string configDir string
verifyResponse string verifyResponse string
log *slog.Logger
} }
func NewController(store *store.Store, path string, verifyResponse string) (*Controller, error) { func NewController(store *store.Store, configDir string, verifyResponse string) (*Controller, error) {
dir := filepath.Dir(path) if err := os.MkdirAll(configDir, 0o750); err != nil {
if err := os.MkdirAll(dir, 0o750); err != nil { return nil, fmt.Errorf("create directory for Caddy configuration '%s': %w", configDir, err)
return nil, fmt.Errorf("create parent directory for Caddy configuration '%s': %w", dir, err)
} }
if err := fs.Chown(dir, "", CaddyGroup); err != nil { if err := fs.Chown(configDir, "", CaddyGroup); err != nil {
return nil, fmt.Errorf("change owner of parent directory for Caddy configuration '%s': %w", dir, err) return nil, fmt.Errorf("change owner of directory for Caddy configuration '%s': %w", configDir, err)
} }
return &Controller{ return &Controller{
store: store, store: store,
path: path, configDir: configDir,
verifyResponse: verifyResponse, verifyResponse: verifyResponse,
log: slog.With("component", "caddy-controller"),
}, nil }, nil
} }
@@ -48,14 +49,18 @@ func (c *Controller) Run(ctx context.Context) error {
if err != nil { if err != nil {
return fmt.Errorf("subscribe to container changes: %w", err) return fmt.Errorf("subscribe to container changes: %w", err)
} }
slog.Info("Subscribed to container changes in the cluster to generate Caddy configuration.") c.log.Info("Subscribed to container changes in the cluster to generate Caddy configuration.")
containers, err := c.filterAvailableContainers(containerRecords) containers, err := c.filterAvailableContainers(containerRecords)
if err != nil { if err != nil {
return fmt.Errorf("filter available containers: %w", err) return fmt.Errorf("filter available containers: %w", err)
} }
if err = c.generateConfig(containers); err != nil {
return fmt.Errorf("generate Caddy configuration: %w", err) if err = c.generateCaddyfile(containers); err != nil {
return fmt.Errorf("generate Caddyfile configuration: %w", err)
}
if err = c.generateJSONConfig(containers); err != nil {
return fmt.Errorf("generate Caddy JSON configuration: %w", err)
} }
for { for {
@@ -64,23 +69,27 @@ func (c *Controller) Run(ctx context.Context) error {
if !ok { if !ok {
return fmt.Errorf("containers subscription failed") return fmt.Errorf("containers subscription failed")
} }
slog.Debug("Cluster containers changed, updating Caddy configuration.") c.log.Info("Cluster containers changed, updating Caddy configuration.")
containerRecords, err = c.store.ListContainers(ctx, store.ListOptions{}) containerRecords, err = c.store.ListContainers(ctx, store.ListOptions{})
if err != nil { if err != nil {
slog.Error("Failed to list containers.", "err", err) c.log.Info("Failed to list containers.", "err", err)
continue continue
} }
containers, err = c.filterAvailableContainers(containerRecords) containers, err = c.filterAvailableContainers(containerRecords)
if err != nil { if err != nil {
slog.Error("Failed to filter available containers.", "err", err) c.log.Info("Failed to filter available containers.", "err", err)
continue continue
} }
if err = c.generateConfig(containers); err != nil {
slog.Error("Failed to generate Caddy configuration.", "err", err) if err = c.generateCaddyfile(containers); err != nil {
c.log.Info("Failed to generate Caddyfile configuration.", "err", err)
}
if err = c.generateJSONConfig(containers); err != nil {
c.log.Info("Failed to generate Caddy JSON configuration.", "err", err)
} }
slog.Debug("Updated Caddy configuration.", "path", c.path) c.log.Info("Updated Caddy configuration.", "dir", c.configDir)
case <-ctx.Done(): case <-ctx.Done():
return nil return nil
} }
@@ -95,16 +104,30 @@ func (c *Controller) filterAvailableContainers(
) ([]api.ServiceContainer, error) { ) ([]api.ServiceContainer, error) {
containers := make([]api.ServiceContainer, len(containerRecords)) containers := make([]api.ServiceContainer, len(containerRecords))
for i, cr := range containerRecords { for i, cr := range containerRecords {
containers[i] = api.ServiceContainer{ containers[i] = cr.Container
Container: cr.Container,
// TODO: restore ServiceSpec from the container record once it's saved in the store.
}
} }
return containers, nil return containers, nil
} }
func (c *Controller) generateConfig(containers []api.ServiceContainer) error { func (c *Controller) generateCaddyfile(containers []api.ServiceContainer) error {
config, err := GenerateConfig(containers, c.verifyResponse) caddyfile, err := GenerateCaddyfile(containers, c.verifyResponse)
if err != nil {
return fmt.Errorf("generate Caddyfile: %w", err)
}
caddyfilePath := filepath.Join(c.configDir, "Caddyfile")
if err = os.WriteFile(caddyfilePath, []byte(caddyfile), 0o640); err != nil {
return fmt.Errorf("write Caddyfile to file '%s': %w", caddyfilePath, err)
}
if err = fs.Chown(caddyfilePath, "", CaddyGroup); err != nil {
return fmt.Errorf("change owner of Caddyfile '%s': %w", caddyfilePath, err)
}
return nil
}
func (c *Controller) generateJSONConfig(containers []api.ServiceContainer) error {
config, err := GenerateJSONConfig(containers, c.verifyResponse)
if err != nil { if err != nil {
return err return err
} }
@@ -113,12 +136,13 @@ func (c *Controller) generateConfig(containers []api.ServiceContainer) error {
if err != nil { if err != nil {
return fmt.Errorf("marshal Caddy configuration: %w", err) return fmt.Errorf("marshal Caddy configuration: %w", err)
} }
configPath := filepath.Join(c.configDir, "caddy.json")
if err = os.WriteFile(c.path, configBytes, 0o640); err != nil { if err = os.WriteFile(configPath, configBytes, 0o640); err != nil {
return fmt.Errorf("write Caddy configuration to file '%s': %w", c.path, err) return fmt.Errorf("write Caddy configuration to file '%s': %w", configPath, err)
} }
if err = fs.Chown(c.path, "", CaddyGroup); err != nil { if err = fs.Chown(configPath, "", CaddyGroup); err != nil {
return fmt.Errorf("change owner of Caddy configuration file '%s': %w", c.path, err) return fmt.Errorf("change owner of Caddy configuration file '%s': %w", configPath, err)
} }
return nil return nil
@@ -19,7 +19,7 @@ import (
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
) )
func GenerateConfig(containers []api.ServiceContainer, verifyResponse string) (*caddy.Config, error) { func GenerateJSONConfig(containers []api.ServiceContainer, verifyResponse string) (*caddy.Config, error) {
// Maps hostnames to lists of upstreams (container IP:port pairs). // Maps hostnames to lists of upstreams (container IP:port pairs).
httpHostUpstreams := make(map[string][]string) httpHostUpstreams := make(map[string][]string)
httpsHostUpstreams := make(map[string][]string) httpsHostUpstreams := make(map[string][]string)
@@ -378,7 +378,7 @@ func TestGenerateConfig(t *testing.T) {
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
config, err := GenerateConfig(tt.containers, "verification-response-body") config, err := GenerateJSONConfig(tt.containers, "verification-response-body")
if tt.wantErr { if tt.wantErr {
assert.Error(t, err) assert.Error(t, err)
+10 -12
View File
@@ -12,7 +12,6 @@ import (
"time" "time"
"github.com/cenkalti/backoff/v4" "github.com/cenkalti/backoff/v4"
"github.com/docker/docker/client"
"github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/api/pb"
"github.com/psviderski/uncloud/internal/machine/caddyconfig" "github.com/psviderski/uncloud/internal/machine/caddyconfig"
"github.com/psviderski/uncloud/internal/machine/constants" "github.com/psviderski/uncloud/internal/machine/constants"
@@ -36,10 +35,9 @@ type clusterController struct {
wgnet *network.WireGuardNetwork wgnet *network.WireGuardNetwork
endpointChanges <-chan network.EndpointChangeEvent endpointChanges <-chan network.EndpointChangeEvent
server *grpc.Server server *grpc.Server
corroService corroservice.Service corroService corroservice.Service
dockerCli *client.Client dockerCtrl *docker.Controller
dockerManager *docker.Manager
// dockerReady is signalled when Docker is configured and ready for containers. // dockerReady is signalled when Docker is configured and ready for containers.
dockerReady chan<- struct{} dockerReady chan<- struct{}
caddyconfigCtrl *caddyconfig.Controller caddyconfigCtrl *caddyconfig.Controller
@@ -57,7 +55,7 @@ func newClusterController(
store *store.Store, store *store.Store,
server *grpc.Server, server *grpc.Server,
corroService corroservice.Service, corroService corroservice.Service,
dockerCli *client.Client, dockerService *docker.Service,
dockerReady chan<- struct{}, dockerReady chan<- struct{},
caddyfileCtrl *caddyconfig.Controller, caddyfileCtrl *caddyconfig.Controller,
dnsServer *dns.Server, dnsServer *dns.Server,
@@ -77,8 +75,7 @@ func newClusterController(
endpointChanges: endpointChanges, endpointChanges: endpointChanges,
server: server, server: server,
corroService: corroService, corroService: corroService,
dockerCli: dockerCli, dockerCtrl: docker.NewController(state.ID, dockerService, store),
dockerManager: docker.NewManager(dockerCli, state.ID, store),
dockerReady: dockerReady, dockerReady: dockerReady,
caddyconfigCtrl: caddyfileCtrl, caddyconfigCtrl: caddyfileCtrl,
dnsServer: dnsServer, dnsServer: dnsServer,
@@ -238,11 +235,11 @@ func (cc *clusterController) Run(ctx context.Context) error {
// ensureDockerNetwork ensures that the Docker network is configured and ready for containers. // ensureDockerNetwork ensures that the Docker network is configured and ready for containers.
func (cc *clusterController) ensureDockerNetwork(ctx context.Context) error { func (cc *clusterController) ensureDockerNetwork(ctx context.Context) error {
if err := cc.dockerManager.WaitDaemonReady(ctx); err != nil { if err := cc.dockerCtrl.WaitDaemonReady(ctx); err != nil {
return fmt.Errorf("wait for Docker daemon: %w", err) return fmt.Errorf("wait for Docker daemon: %w", err)
} }
if err := cc.dockerManager.EnsureUncloudNetwork( if err := cc.dockerCtrl.EnsureUncloudNetwork(
ctx, ctx,
cc.state.Network.Subnet, cc.state.Network.Subnet,
cc.dnsServer.ListenAddr(), cc.dnsServer.ListenAddr(),
@@ -257,6 +254,7 @@ func (cc *clusterController) ensureDockerNetwork(ctx context.Context) error {
} }
// syncDockerContainers watches local Docker containers and syncs them to the cluster store. // syncDockerContainers watches local Docker containers and syncs them to the cluster store.
// TODO: move this to the Docker controller.
func (cc *clusterController) syncDockerContainers(ctx context.Context) error { func (cc *clusterController) syncDockerContainers(ctx context.Context) error {
// Retry to watch and sync containers until the context is done. // Retry to watch and sync containers until the context is done.
boff := backoff.WithContext(backoff.NewExponentialBackOff( boff := backoff.WithContext(backoff.NewExponentialBackOff(
@@ -265,7 +263,7 @@ func (cc *clusterController) syncDockerContainers(ctx context.Context) error {
backoff.WithMaxElapsedTime(0), backoff.WithMaxElapsedTime(0),
), ctx) ), ctx)
watchAndSync := func() error { watchAndSync := func() error {
if wErr := cc.dockerManager.WatchAndSyncContainers(ctx); wErr != nil { if wErr := cc.dockerCtrl.WatchAndSyncContainers(ctx); wErr != nil {
slog.Error("Failed to watch and sync containers to cluster store, retrying.", "err", wErr) slog.Error("Failed to watch and sync containers to cluster store, retrying.", "err", wErr)
return wErr return wErr
} }
@@ -413,7 +411,7 @@ func (cc *clusterController) Cleanup() error {
<-cc.stopped <-cc.stopped
var errs []error var errs []error
if err := cc.dockerManager.Cleanup(); err != nil { if err := cc.dockerCtrl.Cleanup(); err != nil {
errs = append(errs, fmt.Errorf("cleanup Docker resources: %w", err)) errs = append(errs, fmt.Errorf("cleanup Docker resources: %w", err))
} }
if err := cc.wgnet.Cleanup(); err != nil { if err := cc.wgnet.Cleanup(); err != nil {
+2 -8
View File
@@ -5,12 +5,10 @@ import (
"fmt" "fmt"
"log/slog" "log/slog"
"net/netip" "net/netip"
"strings"
"sync" "sync"
"time" "time"
"github.com/psviderski/uncloud/internal/machine/store" "github.com/psviderski/uncloud/internal/machine/store"
"github.com/psviderski/uncloud/pkg/api"
) )
// ClusterResolver implements Resolver by tracking containers in the cluster and resolving service names // ClusterResolver implements Resolver by tracking containers in the cluster and resolving service names
@@ -84,17 +82,13 @@ func (r *ClusterResolver) updateServiceIPs(containers []store.ContainerRecord) {
continue continue
} }
ctr := api.ServiceContainer{Container: record.Container} ctr := record.Container
if ctr.ServiceID() == "" || ctr.ServiceName() == "" { if ctr.ServiceID() == "" || ctr.ServiceName() == "" {
// Container is not part of a service, skip it. // Container is not part of a service, skip it.
continue continue
} }
// TODO: remove normalisation after implementing service name validation: newServiceIPs[ctr.ServiceName()] = append(newServiceIPs[ctr.ServiceName()], ip)
//.https://github.com/psviderski/uncloud/issues/53
serviceName := strings.ToLower(ctr.ServiceName())
newServiceIPs[serviceName] = append(newServiceIPs[serviceName], ip)
// Also add the service ID as a valid lookup. // Also add the service ID as a valid lookup.
newServiceIPs[ctr.ServiceID()] = append(newServiceIPs[ctr.ServiceID()], ip) newServiceIPs[ctr.ServiceID()] = append(newServiceIPs[ctr.ServiceID()], ip)
containersCount++ containersCount++
+16 -2
View File
@@ -196,13 +196,27 @@ func (c *Client) RemoveContainer(ctx context.Context, id string, opts container.
return err return err
} }
// PullOptions defines the options for pulling an image from a remote registry.
// This is a copy of image.PullOptions from the Docker API without the PrivilegeFunc field that is non-serialisable.
type PullOptions struct {
All bool
// RegistryAuth is the base64 encoded credentials for the registry.
RegistryAuth string
Platform string
}
type PullImageMessage struct { type PullImageMessage struct {
Message jsonmessage.JSONMessage Message jsonmessage.JSONMessage
Err error Err error
} }
func (c *Client) PullImage(ctx context.Context, image string) (<-chan PullImageMessage, error) { func (c *Client) PullImage(ctx context.Context, image string, opts PullOptions) (<-chan PullImageMessage, error) {
stream, err := c.grpcClient.PullImage(ctx, &pb.PullImageRequest{Image: image}) optsBytes, err := json.Marshal(opts)
if err != nil {
return nil, fmt.Errorf("marshal options: %w", err)
}
stream, err := c.grpcClient.PullImage(ctx, &pb.PullImageRequest{Image: image, Options: optsBytes})
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -7,12 +7,11 @@ import (
"log/slog" "log/slog"
"time" "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/events"
"github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/filters"
"github.com/docker/docker/client" "github.com/docker/docker/client"
"github.com/psviderski/uncloud/internal/machine/store" "github.com/psviderski/uncloud/internal/machine/store"
"github.com/psviderski/uncloud/pkg/api"
) )
const ( const (
@@ -24,23 +23,26 @@ const (
SyncInterval = 30 * time.Second SyncInterval = 30 * time.Second
) )
type Manager struct { // Controller monitors Docker events and synchronises service containers with the cluster store.
client *client.Client type Controller struct {
// machineID is the ID of the machine where the managed Docker daemon is running. // machineID is the ID of the machine where the managed Docker daemon is running.
machineID string machineID string
client *client.Client
service *Service
store *store.Store store *store.Store
} }
func NewManager(client *client.Client, machineID string, store *store.Store) *Manager { func NewController(machineID string, service *Service, store *store.Store) *Controller {
return &Manager{ return &Controller{
client: client,
machineID: machineID, machineID: machineID,
client: service.Client,
service: service,
store: store, store: store,
} }
} }
// WaitDaemonReady waits for the Docker daemon to start and be ready to serve requests. // 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) ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop() defer ticker.Stop()
@@ -50,7 +52,7 @@ func (m *Manager) WaitDaemonReady(ctx context.Context) error {
case <-ctx.Done(): case <-ctx.Done():
return ctx.Err() return ctx.Err()
case <-ticker.C: case <-ticker.C:
_, err := m.client.Ping(ctx) _, err := c.client.Ping(ctx)
if err == nil { if err == nil {
ready = true ready = true
break break
@@ -67,7 +69,7 @@ func (m *Manager) WaitDaemonReady(ctx context.Context) error {
return nil return nil
} }
func (m *Manager) WatchAndSyncContainers(ctx context.Context) error { func (c *Controller) WatchAndSyncContainers(ctx context.Context) error {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
defer cancel() defer cancel()
// Filter only local container events. // 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. // 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.") 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. // The deferred cancel will stop the event subscription.
return fmt.Errorf("sync containers to cluster store: %w", err) 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"], "container_name", e.Actor.Attributes["name"],
"action", e.Action) "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) return fmt.Errorf("sync containers to cluster store: %w", err)
} }
case <-ticker.C: case <-ticker.C:
slog.Debug("Syncing containers to cluster store triggered by a regular interval.", slog.Debug("Syncing containers to cluster store triggered by a regular interval.",
"interval", SyncInterval) "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) return fmt.Errorf("sync containers to cluster store: %w", err)
} }
case err := <-errCh: case err := <-errCh:
@@ -144,33 +146,16 @@ func (m *Manager) WatchAndSyncContainers(ctx context.Context) error {
} }
} }
func (m *Manager) syncContainersToStore(ctx context.Context) error { func (c *Controller) syncContainersToStore(ctx context.Context) error {
storeContainers, err := m.store.ListContainers(ctx, store.ListOptions{MachineIDs: []string{m.machineID}}) storeContainers, err := c.store.ListContainers(ctx, store.ListOptions{MachineIDs: []string{c.machineID}})
if err != nil { if err != nil {
return fmt.Errorf("list containers from store: %w", err) return fmt.Errorf("list containers from store: %w", err)
} }
// List only Uncloud service containers identified by their labels. containers, err := c.service.ListServiceContainers(ctx, "", container.ListOptions{})
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),
),
})
if err != nil { if err != nil {
// TODO: mark all containers as outdated in the store. // TODO: mark all containers as outdated in the store.
return fmt.Errorf("list Docker containers: %w", err) return fmt.Errorf("list service 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}
} }
// Delete containers from the store that are no longer present in the Docker daemon. // 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 var storeErr error
if len(deleteIDs) > 0 { 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) storeErr = fmt.Errorf("delete containers from store: %w", err)
} }
} }
// Create or update the current Docker containers in the store. // Create or update the current Docker containers in the store.
for _, c := range containers { for _, ctr := range containers {
if err = m.store.CreateOrUpdateContainer(ctx, c, m.machineID); err != nil { if err = c.store.CreateOrUpdateContainer(ctx, ctr, c.machineID); err != nil {
storeErr = errors.Join(storeErr, fmt.Errorf("create or update container %q: %w", c.ID, err)) storeErr = errors.Join(storeErr, fmt.Errorf("create or update container '%s': %w", ctr.ID, err))
} }
} }
return storeErr return storeErr
@@ -9,11 +9,11 @@ import (
) )
// EnsureUncloudNetwork is a stub for Darwin. // 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") return fmt.Errorf("not supported on Darwin")
} }
// Cleanup is a stub for Darwin. // Cleanup is a stub for Darwin.
func (m *Manager) Cleanup() error { func (c *Controller) Cleanup() error {
return fmt.Errorf("not supported on Darwin") return fmt.Errorf("not supported on Darwin")
} }
@@ -22,10 +22,10 @@ import (
// EnsureUncloudNetwork creates the Docker bridge network NetworkName with the provided machine subnet // 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. // 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. // 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. // Ensure the Docker network 'uncloud' is created with the correct subnet.
needsCreation := false needsCreation := false
nw, err := m.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{}) nw, err := c.client.NetworkInspect(ctx, NetworkName, dnetwork.InspectOptions{})
if err != nil { if err != nil {
if !client.IsErrNotFound(err) { if !client.IsErrNotFound(err) {
return fmt.Errorf("inspect Docker network '%s': %w", NetworkName, 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( slog.Info(
"Removing Docker network with old subnet.", "name", NetworkName, "subnet", nw.IPAM.Config[0].Subnet, "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. // 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) 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 needsCreation {
if _, err = m.client.NetworkCreate( if _, err = c.client.NetworkCreate(
ctx, NetworkName, dnetwork.CreateOptions{ ctx, NetworkName, dnetwork.CreateOptions{
Driver: "bridge", Driver: "bridge",
Scope: "local", 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()) 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) 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. // Cleanup removes all uncloud-managed containers and the uncloud Docker network.
func (m *Manager) Cleanup() error { func (c *Controller) Cleanup() error {
ctx := context.Background() ctx := context.Background()
var errs []error var errs []error
// Remove uncloud-managed Docker containers. // 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. All: true, // Include stopped containers.
Filters: filters.NewArgs( Filters: filters.NewArgs(
filters.Arg("label", api.LabelManaged), filters.Arg("label", api.LabelManaged),
@@ -186,12 +186,12 @@ func (m *Manager) Cleanup() error {
removed := 0 removed := 0
for _, ctr := range containers { 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) { if err != nil && !client.IsErrNotFound(err) {
errs = append(errs, fmt.Errorf("stop container '%s': %w", ctr.ID, 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. // Remove anonymous volumes created by the container.
RemoveVolumes: true, RemoveVolumes: true,
}) })
@@ -205,7 +205,7 @@ func (m *Manager) Cleanup() error {
} }
// Remove the uncloud Docker network and related iptables rules. // 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 { if err == nil {
bridgeName := "br-" + nw.ID[:12] bridgeName := "br-" + nw.ID[:12]
var subnet netip.Prefix 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) slog.Info("Docker network removed.", "name", NetworkName)
} else if !client.IsErrNotFound(err) { } else if !client.IsErrNotFound(err) {
errs = append(errs, fmt.Errorf("remove Docker network '%s': %w", NetworkName, err)) errs = append(errs, fmt.Errorf("remove Docker network '%s': %w", NetworkName, err))
+35 -54
View File
@@ -2,19 +2,21 @@ package docker
import ( import (
"context" "context"
"database/sql"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log/slog" "log/slog"
"net/netip" "net/netip"
"os"
"regexp" "regexp"
"slices" "slices"
"strconv" "strconv"
"strings" "strings"
"github.com/distribution/reference" "github.com/distribution/reference"
dockercommand "github.com/docker/cli/cli/command"
dockerconfig "github.com/docker/cli/cli/config"
"github.com/docker/docker/api/types" "github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/filters"
@@ -45,8 +47,9 @@ var fullDockerIDRegex = regexp.MustCompile(`^[a-f0-9]{64}$`)
// Server implements the gRPC Docker service that proxies requests to the Docker daemon. // Server implements the gRPC Docker service that proxies requests to the Docker daemon.
type Server struct { type Server struct {
pb.UnimplementedDockerServer pb.UnimplementedDockerServer
client *client.Client client *client.Client
db *sqlx.DB service *Service
db *sqlx.DB
// internalDNSIP is a function that returns the IP address of the internal DNS server. It may return an empty // internalDNSIP is a function that returns the IP address of the internal DNS server. It may return an empty
// address if the address is unknown (e.g. when the machine is not initialised yet). // address if the address is unknown (e.g. when the machine is not initialised yet).
internalDNSIP func() netip.Addr internalDNSIP func() netip.Addr
@@ -73,10 +76,11 @@ func WithWaitForNetworkReady(waitForNetworkReady func(ctx context.Context) error
} }
} }
// NewServer creates a new Docker gRPC server with the provided Docker client. // NewServer creates a new Docker gRPC server with the provided Docker service.
func NewServer(cli *client.Client, db *sqlx.DB, internalDNSIP func() netip.Addr, opts ...ServerOption) *Server { func NewServer(service *Service, db *sqlx.DB, internalDNSIP func() netip.Addr, opts ...ServerOption) *Server {
s := &Server{ s := &Server{
client: cli, client: service.Client,
service: service,
db: db, db: db,
internalDNSIP: internalDNSIP, internalDNSIP: internalDNSIP,
} }
@@ -265,7 +269,6 @@ func (s *Server) RemoveContainer(ctx context.Context, req *pb.RemoveContainerReq
func (s *Server) PullImage(req *pb.PullImageRequest, stream grpc.ServerStreamingServer[pb.JSONMessage]) error { func (s *Server) PullImage(req *pb.PullImageRequest, stream grpc.ServerStreamingServer[pb.JSONMessage]) error {
ctx := stream.Context() ctx := stream.Context()
// TODO: replace with another JSON serializable type (PullOptions.PrivilegeFunc is not serializable).
var opts image.PullOptions var opts image.PullOptions
if len(req.Options) > 0 { if len(req.Options) > 0 {
if err := json.Unmarshal(req.Options, &opts); err != nil { if err := json.Unmarshal(req.Options, &opts); err != nil {
@@ -273,6 +276,14 @@ func (s *Server) PullImage(req *pb.PullImageRequest, stream grpc.ServerStreaming
} }
} }
if opts.RegistryAuth == "" {
// Try to retrieve the authentication token for the image from the default local Docker config file.
dockerConfig := dockerconfig.LoadDefaultConfigFile(os.Stderr)
if encodedAuth, err := dockercommand.RetrieveAuthTokenFromImage(dockerConfig, req.Image); err == nil {
opts.RegistryAuth = encodedAuth
}
}
respBody, err := s.client.ImagePull(ctx, req.Image, opts) respBody, err := s.client.ImagePull(ctx, req.Image, opts)
if err != nil { if err != nil {
return status.Errorf(codes.Internal, err.Error()) return status.Errorf(codes.Internal, err.Error())
@@ -455,6 +466,7 @@ func (s *Server) RemoveVolume(ctx context.Context, req *pb.RemoveVolumeRequest)
} }
// CreateServiceContainer creates a new container for the service with the given specifications. // CreateServiceContainer creates a new container for the service with the given specifications.
// TODO: move the main logic to the Docker service and remove db dependency from the server.
func (s *Server) CreateServiceContainer( func (s *Server) CreateServiceContainer(
ctx context.Context, req *pb.CreateServiceContainerRequest, ctx context.Context, req *pb.CreateServiceContainerRequest,
) (*pb.CreateContainerResponse, error) { ) (*pb.CreateContainerResponse, error) {
@@ -546,10 +558,10 @@ func (s *Server) CreateServiceContainer(
Memory: spec.Container.Resources.Memory, Memory: spec.Container.Resources.Memory,
MemoryReservation: spec.Container.Resources.MemoryReservation, MemoryReservation: spec.Container.Resources.MemoryReservation,
}, },
// Always restart service containers if they exit or a machine restarts. // Restart service containers if they exit or a machine restarts unless they are explicitly stopped.
// For one-off containers and batch jobs we plan to use a different service type/mode. // For one-off containers and batch jobs we plan to use a different service type/mode.
RestartPolicy: container.RestartPolicy{ RestartPolicy: container.RestartPolicy{
Name: container.RestartPolicyAlways, Name: container.RestartPolicyUnlessStopped,
}, },
} }
@@ -709,7 +721,7 @@ func (s *Server) verifyDockerVolumesExist(ctx context.Context, mounts []mount.Mo
func (s *Server) InspectServiceContainer( func (s *Server) InspectServiceContainer(
ctx context.Context, req *pb.InspectContainerRequest, ctx context.Context, req *pb.InspectContainerRequest,
) (*pb.ServiceContainer, error) { ) (*pb.ServiceContainer, error) {
ctr, err := s.client.ContainerInspect(ctx, req.Id) serviceCtr, err := s.service.InspectServiceContainer(ctx, req.Id)
if err != nil { if err != nil {
if client.IsErrNotFound(err) { if client.IsErrNotFound(err) {
return nil, status.Errorf(codes.NotFound, err.Error()) return nil, status.Errorf(codes.NotFound, err.Error())
@@ -717,19 +729,14 @@ func (s *Server) InspectServiceContainer(
return nil, status.Errorf(codes.Internal, err.Error()) return nil, status.Errorf(codes.Internal, err.Error())
} }
ctrBytes, err := json.Marshal(ctr) ctrBytes, err := json.Marshal(serviceCtr.Container)
if err != nil { if err != nil {
return nil, status.Errorf(codes.Internal, "marshal response: %v", err) return nil, status.Errorf(codes.Internal, "marshal container: %v", err)
} }
var specBytes []byte specBytes, err := json.Marshal(serviceCtr.ServiceSpec)
err = s.db.QueryRowContext(ctx, `SELECT service_spec FROM containers WHERE id = $1`, ctr.ID).Scan(&specBytes)
if err != nil { if err != nil {
if errors.Is(err, sql.ErrNoRows) { return nil, status.Errorf(codes.Internal, "marshal service spec: %v", err)
return nil, status.Errorf(codes.NotFound, "service spec not found for container: '%s'", ctr.ID)
}
return nil, status.Errorf(codes.Internal, "get service spec for container '%s' from machine database: %v",
ctr.ID, err)
} }
return &pb.ServiceContainer{ return &pb.ServiceContainer{
@@ -761,54 +768,28 @@ func (s *Server) ListServiceContainers(
return nil, status.Errorf(codes.InvalidArgument, "unmarshal filters: %v", err) return nil, status.Errorf(codes.InvalidArgument, "unmarshal filters: %v", err)
} }
opts.Filters = args opts.Filters = args
} else {
opts.Filters = filters.NewArgs()
} }
} }
// Only uncloud-managed containers that belong to some service.
opts.Filters.Add("label", api.LabelServiceID)
opts.Filters.Add("label", api.LabelManaged)
containerSummaries, err := s.client.ContainerList(ctx, opts) containers, err := s.service.ListServiceContainers(ctx, req.ServiceId, opts)
if err != nil { if err != nil {
return nil, status.Error(codes.Internal, err.Error()) return nil, status.Error(codes.Internal, err.Error())
} }
containers := make([]*pb.ServiceContainer, 0, len(containerSummaries)) // Convert to protobuf format.
for _, cs := range containerSummaries { pbContainers := make([]*pb.ServiceContainer, 0, len(containers))
if req.ServiceId != "" && for _, ctr := range containers {
cs.Labels[api.LabelServiceID] != req.ServiceId && cs.Labels[api.LabelServiceName] != req.ServiceId { ctrBytes, err := json.Marshal(ctr.Container)
continue
}
ctr, err := s.client.ContainerInspect(ctx, cs.ID)
if err != nil {
if client.IsErrNotFound(err) {
// The listed container may have been removed while we were inspecting other containers.
continue
}
return nil, status.Errorf(codes.Internal, "inspect container %s: %v", cs.ID, err)
}
ctrBytes, err := json.Marshal(ctr)
if err != nil { if err != nil {
return nil, status.Errorf(codes.Internal, "marshal container: %v", err) return nil, status.Errorf(codes.Internal, "marshal container: %v", err)
} }
var specBytes []byte specBytes, err := json.Marshal(ctr.ServiceSpec)
err = s.db.QueryRowContext(ctx, `SELECT service_spec FROM containers WHERE id = $1`, ctr.ID).Scan(&specBytes)
if err != nil { if err != nil {
if errors.Is(err, sql.ErrNoRows) { return nil, status.Errorf(codes.Internal, "marshal service spec: %v", err)
// 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.
slog.Error("Service container not found in machine database.", "id", ctr.ID)
// Just ignore such a container to not fail the list operation as it's not easily recoverable.
continue
}
return nil, status.Errorf(codes.Internal, "get service spec for container '%s' from machine database: %v",
ctr.ID, err)
} }
containers = append(containers, &pb.ServiceContainer{ pbContainers = append(pbContainers, &pb.ServiceContainer{
Container: ctrBytes, Container: ctrBytes,
ServiceSpec: specBytes, ServiceSpec: specBytes,
}) })
@@ -817,7 +798,7 @@ func (s *Server) ListServiceContainers(
return &pb.ListServiceContainersResponse{ return &pb.ListServiceContainersResponse{
Messages: []*pb.MachineServiceContainers{ Messages: []*pb.MachineServiceContainers{
{ {
Containers: containers, Containers: pbContainers,
}, },
}, },
}, nil }, nil
+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
}
+18 -19
View File
@@ -30,7 +30,6 @@ import (
machinedocker "github.com/psviderski/uncloud/internal/machine/docker" machinedocker "github.com/psviderski/uncloud/internal/machine/docker"
"github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/machine/network"
"github.com/psviderski/uncloud/internal/machine/store" "github.com/psviderski/uncloud/internal/machine/store"
"github.com/psviderski/uncloud/pkg/api"
"github.com/siderolabs/grpc-proxy/proxy" "github.com/siderolabs/grpc-proxy/proxy"
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
"google.golang.org/grpc" "google.golang.org/grpc"
@@ -62,9 +61,9 @@ type Config struct {
// DockerClient manages system and user containers using the local Docker daemon. // DockerClient manages system and user containers using the local Docker daemon.
DockerClient *client.Client DockerClient *client.Client
// CaddyConfigPath specifies where the machine generates the Caddy reverse proxy configuration file for routing // CaddyConfigDir specifies the directory where the machine generates the Caddy reverse proxy configuration file
// external traffic to service containers across the internal network. Default is DataDir/caddy/caddy.json. // for routing external traffic to service containers across the internal network. Default is DataDir/caddy.
CaddyConfigPath string CaddyConfigDir string
// DNSUpstreams specifies the upstream DNS servers for the embedded internal DNS server. // DNSUpstreams specifies the upstream DNS servers for the embedded internal DNS server.
DNSUpstreams []netip.AddrPort DNSUpstreams []netip.AddrPort
} }
@@ -129,8 +128,8 @@ func (c *Config) SetDefaults() (*Config, error) {
} }
} }
if cfg.CaddyConfigPath == "" { if cfg.CaddyConfigDir == "" {
cfg.CaddyConfigPath = filepath.Join(cfg.DataDir, "caddy", "caddy.json") cfg.CaddyConfigDir = filepath.Join(cfg.DataDir, "caddy")
} }
return &cfg, nil return &cfg, nil
@@ -162,7 +161,9 @@ type Machine struct {
// store is the cluster store backed by a distributed Corrosion database. // store is the cluster store backed by a distributed Corrosion database.
store *store.Store store *store.Store
cluster *cluster.Cluster cluster *cluster.Cluster
docker *machinedocker.Server // dockerService provides high-level operations for managing Docker containers.
dockerService *machinedocker.Service
dockerServer *machinedocker.Server
// localMachineServer is the gRPC server for the machine API listening on the local Unix socket. // localMachineServer is the gRPC server for the machine API listening on the local Unix socket.
localMachineServer *grpc.Server localMachineServer *grpc.Server
@@ -222,17 +223,14 @@ func NewMachine(config *Config) (*Machine, error) {
c := cluster.NewCluster(corroStore, corroAdmin) c := cluster.NewCluster(corroStore, corroAdmin)
// Init dependencies for a gRPC Docker server that proxies requests to the local Docker daemon. // Init dependencies for a gRPC Docker server that proxies requests to the local Docker daemon.
dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
if err != nil {
return nil, fmt.Errorf("create Docker client: %w", err)
}
dbFilePath := filepath.Join(config.DataDir, DBFileName) dbFilePath := filepath.Join(config.DataDir, DBFileName)
db, err := NewDB(dbFilePath) db, err := NewDB(dbFilePath)
if err != nil { if err != nil {
return nil, fmt.Errorf("init machine database: %w", err) return nil, fmt.Errorf("init machine database: %w", err)
} }
dockerService := machinedocker.NewService(config.DockerClient, db)
// Init a local gRPC proxy server that proxies requests to the local or remote machine API servers. // Init a local gRPC proxy server that proxies requests to the local or remote machine API servers.
proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort) proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort)
localProxyServer := grpc.NewServer( localProxyServer := grpc.NewServer(
@@ -250,6 +248,7 @@ func NewMachine(config *Config) (*Machine, error) {
networkReady: make(chan struct{}), networkReady: make(chan struct{}),
store: corroStore, store: corroStore,
cluster: c, cluster: c,
dockerService: dockerService,
localProxyServer: localProxyServer, localProxyServer: localProxyServer,
proxyDirector: proxyDirector, proxyDirector: proxyDirector,
} }
@@ -258,10 +257,10 @@ func NewMachine(config *Config) (*Machine, error) {
internalDNSIP := func() netip.Addr { internalDNSIP := func() netip.Addr {
return m.IP() return m.IP()
} }
m.docker = machinedocker.NewServer(dockerCli, db, internalDNSIP, m.dockerServer = machinedocker.NewServer(dockerService, db, internalDNSIP,
machinedocker.WithNetworkReady(m.IsNetworkReady), machinedocker.WithNetworkReady(m.IsNetworkReady),
machinedocker.WithWaitForNetworkReady(m.WaitForNetworkReady)) machinedocker.WithWaitForNetworkReady(m.WaitForNetworkReady))
m.localMachineServer = newGRPCServer(m, c, m.docker) m.localMachineServer = newGRPCServer(m, c, m.dockerServer)
if m.Initialised() { if m.Initialised() {
m.initialised <- struct{}{} m.initialised <- struct{}{}
@@ -388,7 +387,7 @@ func (m *Machine) Run(ctx context.Context) error {
// Create a new caddyconfig controller for managing the Caddy reverse proxy configuration. // Create a new caddyconfig controller for managing the Caddy reverse proxy configuration.
// It will also serve the current machine ID at /.uncloud-verify to verify Caddy reachability. // It will also serve the current machine ID at /.uncloud-verify to verify Caddy reachability.
caddyconfigCtrl, err := caddyconfig.NewController(m.store, m.config.CaddyConfigPath, m.state.ID) caddyconfigCtrl, err := caddyconfig.NewController(m.store, m.config.CaddyConfigDir, m.state.ID)
if err != nil { if err != nil {
return fmt.Errorf("create caddyconfig controller: %w", err) return fmt.Errorf("create caddyconfig controller: %w", err)
} }
@@ -405,7 +404,7 @@ func (m *Machine) Run(ctx context.Context) error {
m.store, m.store,
proxyServer, proxyServer,
m.config.CorrosionService, m.config.CorrosionService,
m.config.DockerClient, m.dockerService,
m.networkReady, m.networkReady,
caddyconfigCtrl, caddyconfigCtrl,
dnsServer, dnsServer,
@@ -569,7 +568,7 @@ func (m *Machine) cleanup() error {
} }
// CheckPrerequisites verifies if the machine meets all necessary system requirements to participate in the cluster. // CheckPrerequisites verifies if the machine meets all necessary system requirements to participate in the cluster.
func (m *Machine) CheckPrerequisites(ctx context.Context, _ *emptypb.Empty) (*pb.CheckPrerequisitesResponse, error) { func (m *Machine) CheckPrerequisites(_ context.Context, _ *emptypb.Empty) (*pb.CheckPrerequisitesResponse, error) {
// Check DNS port (UDP) availability. // Check DNS port (UDP) availability.
if err := checkDNSPortAvailable(); err != nil { if err := checkDNSPortAvailable(); err != nil {
return &pb.CheckPrerequisitesResponse{ return &pb.CheckPrerequisitesResponse{
@@ -835,7 +834,7 @@ func (m *Machine) WaitForNetworkReady(ctx context.Context) error {
// Reset restores the machine to a clean state, scheduling a graceful shutdown and removing all cluster-related // Reset restores the machine to a clean state, scheduling a graceful shutdown and removing all cluster-related
// configuration and resource. The uncloud daemon will restart the machine if managed by systemd. // configuration and resource. The uncloud daemon will restart the machine if managed by systemd.
func (m *Machine) Reset(ctx context.Context, _ *pb.ResetRequest) (*emptypb.Empty, error) { func (m *Machine) Reset(_ context.Context, _ *pb.ResetRequest) (*emptypb.Empty, error) {
if !m.Initialised() { if !m.Initialised() {
return nil, nil return nil, nil
} }
@@ -890,7 +889,7 @@ func (m *Machine) InspectService(
} }
} }
ctr := api.ServiceContainer{Container: records[0].Container} ctr := records[0].Container
svc := &pb.Service{ svc := &pb.Service{
Id: ctr.ServiceID(), Id: ctr.ServiceID(),
Name: ctr.ServiceName(), Name: ctr.ServiceName(),
+5 -3
View File
@@ -23,7 +23,7 @@ const (
) )
type ContainerRecord struct { type ContainerRecord struct {
Container api.Container Container api.ServiceContainer
MachineID string MachineID string
SyncStatus string SyncStatus string
UpdatedAt time.Time UpdatedAt time.Time
@@ -48,10 +48,12 @@ type DeleteOptions struct {
// CreateOrUpdateContainer creates a new container record or updates an existing one in the store database. // 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. // 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, ctr api.Container, machineID string) error { func (s *Store) CreateOrUpdateContainer(ctx context.Context, ctr api.ServiceContainer, machineID string) error {
// Remove the environment variables from the container record before storing it in the database // Remove the environment variables from the container record before storing it in the database
// to avoid leaking secrets. // to avoid leaking secrets.
ctr.Config.Env = nil ctr.Config.Env = nil
ctr.ServiceSpec.Container.Env = nil
cJSON, err := json.Marshal(ctr) cJSON, err := json.Marshal(ctr)
if err != nil { if err != nil {
return fmt.Errorf("marshal container: %w", err) return fmt.Errorf("marshal container: %w", err)
@@ -118,7 +120,7 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain
return nil, fmt.Errorf("scan container record: %w", err) return nil, fmt.Errorf("scan container record: %w", err)
} }
var c api.Container var c api.ServiceContainer
if err = json.Unmarshal([]byte(cJSON), &c); err != nil { if err = json.Unmarshal([]byte(cJSON), &c); err != nil {
return nil, fmt.Errorf("unmarshal container: %w", err) return nil, fmt.Errorf("unmarshal container: %w", err)
} }
+21
View File
@@ -0,0 +1,21 @@
package api
import "strings"
// CaddySpec is the Caddy reverse proxy configuration for a service.
type CaddySpec struct {
// Config contains the Caddy config (Caddyfile) content. It must not conflict with the Caddy configs
// of other services.
Config string
}
func (c *CaddySpec) Equals(other *CaddySpec) bool {
if c == nil {
return other == nil || strings.TrimSpace(other.Config) == ""
}
if other == nil {
return strings.TrimSpace(c.Config) == ""
}
return strings.TrimSpace(c.Config) == strings.TrimSpace(other.Config)
}
+20 -2
View File
@@ -6,6 +6,7 @@ import (
"maps" "maps"
"regexp" "regexp"
"slices" "slices"
"strings"
"github.com/distribution/reference" "github.com/distribution/reference"
"github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp"
@@ -42,6 +43,10 @@ func ValidateServiceID(id string) bool {
// ServiceSpec defines the desired state of a service. // ServiceSpec defines the desired state of a service.
// ATTENTION: after changing this struct, verify if deploy.EvalContainerSpecChange needs to be updated. // ATTENTION: after changing this struct, verify if deploy.EvalContainerSpecChange needs to be updated.
type ServiceSpec struct { type ServiceSpec struct {
// Caddy is the optional Caddy reverse proxy configuration for the service.
// Caddy and Ports cannot be specified simultaneously.
Caddy *CaddySpec `json:",omitempty"`
// Container defines the desired state of each container in the service.
Container ContainerSpec Container ContainerSpec
// Mode is the replication mode of the service. Default is ServiceModeReplicated if empty. // Mode is the replication mode of the service. Default is ServiceModeReplicated if empty.
Mode string Mode string
@@ -49,6 +54,7 @@ type ServiceSpec struct {
// Placement defines the placement constraints for the service. // Placement defines the placement constraints for the service.
Placement Placement Placement Placement
// Ports defines what service ports to publish to make the service accessible outside the cluster. // Ports defines what service ports to publish to make the service accessible outside the cluster.
// Caddy and Ports cannot be specified simultaneously.
Ports []PortSpec Ports []PortSpec
// Replicas is the number of containers to run for the service. Only valid for a replicated service. // Replicas is the number of containers to run for the service. Only valid for a replicated service.
Replicas uint `json:",omitempty"` Replicas uint `json:",omitempty"`
@@ -112,10 +118,17 @@ func (s *ServiceSpec) Validate() error {
return fmt.Errorf("service name too long (max 63 characters): %q", s.Name) return fmt.Errorf("service name too long (max 63 characters): %q", s.Name)
} }
if !dnsLabelRegexp.MatchString(s.Name) { if !dnsLabelRegexp.MatchString(s.Name) {
return fmt.Errorf("invalid service name: %q. must be 1-63 characters, lowercase letters, numbers, and dashes only; must start and end with a letter or number", s.Name) return fmt.Errorf("invalid service name: %q. must be 1-63 characters, lowercase letters, numbers, "+
"and dashes only; must start and end with a letter or number", s.Name)
} }
} }
// Validate that Caddy and Ports are not used together.
if s.Caddy != nil && strings.TrimSpace(s.Caddy.Config) != "" && len(s.Ports) > 0 {
return fmt.Errorf("ports and Caddy configuration cannot be specified simultaneously: " +
"Caddy config is auto-generated from ports, use only one of them")
}
for _, p := range s.Ports { for _, p := range s.Ports {
if (p.Mode == "" || p.Mode == PortModeIngress) && if (p.Mode == "" || p.Mode == PortModeIngress) &&
p.Protocol != ProtocolHTTP && p.Protocol != ProtocolHTTPS { p.Protocol != ProtocolHTTP && p.Protocol != ProtocolHTTPS {
@@ -151,11 +164,16 @@ func (s *ServiceSpec) Validate() error {
func (s *ServiceSpec) Clone() ServiceSpec { func (s *ServiceSpec) Clone() ServiceSpec {
spec := *s spec := *s
if s.Caddy != nil {
caddyCopy := *s.Caddy
spec.Caddy = &caddyCopy
}
spec.Container = s.Container.Clone()
if s.Ports != nil { if s.Ports != nil {
spec.Ports = make([]PortSpec, len(s.Ports)) spec.Ports = make([]PortSpec, len(s.Ports))
copy(spec.Ports, s.Ports) copy(spec.Ports, s.Ports)
} }
spec.Container = s.Container.Clone()
if s.Volumes != nil { if s.Volumes != nil {
spec.Volumes = make([]VolumeSpec, len(s.Volumes)) spec.Volumes = make([]VolumeSpec, len(s.Volumes))
+104
View File
@@ -0,0 +1,104 @@
package api
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestServiceSpec_Validate_CaddyAndPorts(t *testing.T) {
tests := []struct {
name string
spec ServiceSpec
wantErr string
}{
{
name: "valid with neither Caddy nor Ports",
spec: ServiceSpec{
Name: "test",
Container: ContainerSpec{
Image: "nginx:latest",
},
},
wantErr: "",
},
{
name: "valid with Caddy only",
spec: ServiceSpec{
Name: "test",
Container: ContainerSpec{
Image: "nginx:latest",
},
Caddy: &CaddySpec{
Config: "example.com {\n reverse_proxy :8080\n}",
},
},
wantErr: "",
},
{
name: "valid with Ports only",
spec: ServiceSpec{
Name: "test",
Container: ContainerSpec{
Image: "nginx:latest",
},
Ports: []PortSpec{
{
ContainerPort: 80,
Protocol: ProtocolHTTP,
},
},
},
wantErr: "",
},
{
name: "valid with empty Caddy config and Ports",
spec: ServiceSpec{
Name: "test",
Container: ContainerSpec{
Image: "nginx:latest",
},
Caddy: &CaddySpec{
Config: "",
},
Ports: []PortSpec{
{
ContainerPort: 80,
Protocol: ProtocolHTTP,
},
},
},
wantErr: "",
},
{
name: "invalid with both Caddy and Ports",
spec: ServiceSpec{
Name: "test",
Container: ContainerSpec{
Image: "nginx:latest",
},
Caddy: &CaddySpec{
Config: "example.com {\n reverse_proxy :8080\n}",
},
Ports: []PortSpec{
{
ContainerPort: 80,
Protocol: ProtocolHTTP,
},
},
},
wantErr: "ports and Caddy configuration cannot be specified simultaneously",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := tt.spec.Validate()
if tt.wantErr == "" {
require.NoError(t, err)
} else {
require.ErrorContains(t, err, tt.wantErr)
}
})
}
}
+94
View File
@@ -0,0 +1,94 @@
package compose
import (
"fmt"
"os"
"path/filepath"
"strings"
"github.com/compose-spec/compose-go/v2/types"
"github.com/mitchellh/mapstructure"
)
const CaddyExtensionKey = "x-caddy"
type Caddy struct {
Config string `yaml:"config" json:"config"`
}
// DecodeMapstructure decodes x-caddy extension from either a string or an object.
// When x-caddy is a string, it's mapped directly to the Config field.
func (c *Caddy) DecodeMapstructure(value any) error {
switch v := value.(type) {
case *Caddy:
// Already decoded, happens when mapstructure is called after initial parsing.
*c = *v
return nil
case string:
// Handle x-caddy: "Caddyfile config"
*c = Caddy{Config: v}
case map[string]any:
// Use mapstructure to decode the map directly to the struct.
decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
Result: c,
ErrorUnused: true, // Error if there are extra keys not in the struct.
WeaklyTypedInput: false, // Enforce strict type matching.
})
if err != nil {
return fmt.Errorf("create decoder for x-caddy extension: %w", err)
}
if err := decoder.Decode(v); err != nil {
return fmt.Errorf("decode x-caddy extension: %w", err)
}
default:
return fmt.Errorf("invalid type %T for x-caddy extension: expected string or object", value)
}
return nil
}
// isCaddyfilePath determines if a string is likely a file path rather than inline Caddyfile config.
func isCaddyfilePath(s string) bool {
s = strings.TrimSpace(s)
if s == "" {
return false
}
// For simplicity, multi-line string is considered an inline Caddyfile content.
return !strings.Contains(s, "\n")
}
// transformServicesCaddyExtension processes Caddy extensions to load configs from files if needed.
func transformServicesCaddyExtension(project *types.Project) (*types.Project, error) {
return project.WithServicesTransform(func(name string, service types.ServiceConfig) (types.ServiceConfig, error) {
ext, ok := service.Extensions[CaddyExtensionKey]
if !ok {
return service, nil
}
caddy, ok := ext.(Caddy)
if !ok {
return service, nil
}
// Load the Caddyfile config from file if it's a path and replace the path with its content.
if isCaddyfilePath(caddy.Config) {
configPath := caddy.Config
if !filepath.IsAbs(configPath) {
configPath = filepath.Join(project.WorkingDir, configPath)
}
content, err := os.ReadFile(configPath)
if err != nil {
return service, fmt.Errorf("read Caddy config (Caddyfile) from file '%s' for service '%s': %w",
caddy.Config, name, err)
}
caddy.Config = string(content)
}
caddy.Config = strings.TrimSpace(caddy.Config)
service.Extensions[CaddyExtensionKey] = caddy
return service, nil
})
}
+217
View File
@@ -0,0 +1,217 @@
package compose
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestCaddyExtension(t *testing.T) {
tests := []struct {
name string
composeYAML string
wantConfig string
wantErr string
}{
{
name: "x-caddy as string",
composeYAML: `
services:
web:
image: nginx
x-caddy: |
example.com {
reverse_proxy web:80
}
`,
wantConfig: `example.com {
reverse_proxy web:80
}`,
},
{
name: "x-caddy as string with extra spaces",
composeYAML: `
services:
web:
image: nginx
x-caddy: |+
example.com {
reverse_proxy web:80
}
`,
wantConfig: `example.com {
reverse_proxy web:80
}`,
},
{
name: "x-caddy as object with config field",
composeYAML: `
services:
web:
image: nginx
x-caddy:
config: |
example.com {
reverse_proxy web:80
}
`,
wantConfig: `example.com {
reverse_proxy web:80
}`,
},
{
name: "x-caddy as object with config field and extra spaces",
composeYAML: `
services:
web:
image: nginx
x-caddy:
config: |+
example.com {
reverse_proxy web:80
}
`,
wantConfig: `example.com {
reverse_proxy web:80
}`,
},
{
name: "x-caddy with path to Caddyfile",
composeYAML: `
services:
web:
image: nginx
x-caddy: testdata/Caddyfile
`,
wantConfig: `test.example.com {
reverse_proxy test:8000
}`,
},
{
name: "x-caddy with empty object",
composeYAML: `
services:
web:
image: nginx
x-caddy: {}
`,
wantConfig: "",
},
{
name: "x-caddy with empty string",
composeYAML: `
services:
web:
image: nginx
x-caddy: ""
`,
wantConfig: "",
},
{
name: "x-caddy with extra unknown field should fail",
composeYAML: `
services:
web:
image: nginx
x-caddy:
config: |
example.com {
reverse_proxy web:80
}
unknown_field: "should cause error"
`,
wantErr: "invalid keys: unknown_field",
},
{
name: "x-caddy with non-string config field should fail",
composeYAML: `
services:
web:
image: nginx
x-caddy:
config: 123
`,
wantErr: "expected type 'string'",
},
{
name: "x-caddy with x-ports conflict",
composeYAML: `
services:
web:
image: nginx
x-caddy: |
example.com {
reverse_proxy web:80
}
x-ports:
- example.com:80/http
`,
wantErr: "cannot specify both 'x-caddy' and 'x-ports'",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
project, err := loadProjectFromContent(t, tt.composeYAML)
if tt.wantErr != "" {
require.ErrorContains(t, err, tt.wantErr)
return
}
require.NoError(t, err)
service, err := project.GetService("web")
require.NoError(t, err)
// Verify the x-caddy extension was parsed correctly.
caddyExt, ok := service.Extensions[CaddyExtensionKey]
require.True(t, ok, "x-caddy extension not found")
caddy, ok := caddyExt.(Caddy)
require.True(t, ok, "x-caddy extension is not Caddy type")
assert.Equal(t, tt.wantConfig, caddy.Config)
})
}
}
func TestIsCaddyfilePath(t *testing.T) {
tests := []struct {
name string
input string
want bool
}{
// Should be detected as file paths.
{"relative path with slash", "./Caddyfile", true},
{"relative path parent", "../Caddyfile", true},
{"relative path", "relative/path/to/file", true},
{"absolute path", "/etc/caddy/Caddyfile", true},
{"just Caddyfile", "Caddyfile", true},
{"Caddyfile with suffix", "Caddyfile.app", true},
{"caddyfile lowercase", "caddyfile", true},
{"with .caddyfile extension", "my.caddyfile", true},
{"with .Caddyfile extension", "my.Caddyfile", true},
{"with .caddy extension", "config.caddy", true},
{"with .conf extension", "caddy.conf", true},
{"simple filename", "config", true},
// Should NOT be detected as file paths.
{"multiline config", "example.com {\n reverse_proxy :8080\n}", false},
{"empty string", "", false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result := isCaddyfilePath(tt.input)
assert.Equal(t, tt.want, result, "isCaddyfilePath(%q) should be %v", tt.input, tt.want)
})
}
}
+11 -2
View File
@@ -25,11 +25,16 @@ type Deployment struct {
Client Client Client Client
Project *types.Project Project *types.Project
SpecResolver *deploy.ServiceSpecResolver SpecResolver *deploy.ServiceSpecResolver
Strategy deploy.Strategy
state *scheduler.ClusterState state *scheduler.ClusterState
plan *deploy.SequenceOperation plan *deploy.SequenceOperation
} }
func NewDeployment(ctx context.Context, cli Client, project *types.Project) (*Deployment, error) { func NewDeployment(ctx context.Context, cli Client, project *types.Project) (*Deployment, error) {
return NewDeploymentWithStrategy(ctx, cli, project, nil)
}
func NewDeploymentWithStrategy(ctx context.Context, cli Client, project *types.Project, strategy deploy.Strategy) (*Deployment, error) {
state, err := scheduler.InspectClusterState(ctx, cli) state, err := scheduler.InspectClusterState(ctx, cli)
if err != nil { if err != nil {
return nil, fmt.Errorf("inspect cluster state: %w", err) return nil, fmt.Errorf("inspect cluster state: %w", err)
@@ -39,16 +44,20 @@ func NewDeployment(ctx context.Context, cli Client, project *types.Project) (*De
if err != nil && !errors.Is(err, api.ErrNotFound) { if err != nil && !errors.Is(err, api.ErrNotFound) {
return nil, fmt.Errorf("get cluster domain: %w", err) return nil, fmt.Errorf("get cluster domain: %w", err)
} }
resolver := &deploy.ServiceSpecResolver{ resolver := &deploy.ServiceSpecResolver{
// If the domain is not found (not reserved), an empty domain is used for the resolver. // If the domain is not found (not reserved), an empty domain is used for the resolver.
ClusterDomain: domain, ClusterDomain: domain,
} }
if strategy == nil {
strategy = &deploy.RollingStrategy{State: state}
}
return &Deployment{ return &Deployment{
Client: cli, Client: cli,
Project: project, Project: project,
SpecResolver: resolver, SpecResolver: resolver,
Strategy: strategy,
state: state, state: state,
}, nil }, nil
} }
@@ -90,7 +99,7 @@ func (d *Deployment) Plan(ctx context.Context) (deploy.SequenceOperation, error)
for _, spec := range serviceSpecs { for _, spec := range serviceSpecs {
// TODO: properly handle depends_on conditions in the service deployment plan as the first operation. // TODO: properly handle depends_on conditions in the service deployment plan as the first operation.
// Pass the update cluster state with scheduled volumes to the deployment. // Pass the update cluster state with scheduled volumes to the deployment.
deployment := deploy.NewDeployment(d.Client, spec, &deploy.RollingStrategy{State: d.state}) deployment := deploy.NewDeployment(d.Client, spec, d.Strategy)
servicePlan, err := deployment.Plan(ctx) servicePlan, err := deployment.Plan(ctx)
if err != nil { if err != nil {
return plan, fmt.Errorf("create deployment plan for service '%s': %w", spec.Name, err) return plan, fmt.Errorf("create deployment plan for service '%s': %w", spec.Name, err)
+10 -1
View File
@@ -24,8 +24,9 @@ func LoadProject(ctx context.Context, paths []string, opts ...composecli.Project
composecli.WithConfigFileEnv, composecli.WithConfigFileEnv,
// If none was selected, get default Compose file names from current or parent folders. // If none was selected, get default Compose file names from current or parent folders.
composecli.WithDefaultConfigPath, composecli.WithDefaultConfigPath,
composecli.WithExtension(PortsExtensionKey, PortsSource{}), composecli.WithExtension(CaddyExtensionKey, Caddy{}),
composecli.WithExtension(MachinesExtensionKey, MachinesSource{}), composecli.WithExtension(MachinesExtensionKey, MachinesSource{}),
composecli.WithExtension(PortsExtensionKey, PortsSource{}),
} }
options, err := composecli.NewProjectOptions( options, err := composecli.NewProjectOptions(
@@ -41,9 +42,17 @@ func LoadProject(ctx context.Context, paths []string, opts ...composecli.Project
return nil, err return nil, err
} }
if project, err = transformServicesCaddyExtension(project); err != nil {
return nil, err
}
if project, err = transformServicesPortsExtension(project); err != nil { if project, err = transformServicesPortsExtension(project); err != nil {
return nil, err return nil, err
} }
// Validate extension combinations after all transformations.
if err = validateServicesExtensions(project); err != nil {
return nil, err
}
return project, nil return project, nil
} }
+29
View File
@@ -57,6 +57,12 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser
Mode: api.ServiceModeReplicated, Mode: api.ServiceModeReplicated,
} }
// Map x-caddy extension to spec.Caddy if specified.
if caddy, ok := service.Extensions[CaddyExtensionKey].(Caddy); ok && caddy.Config != "" {
spec.Caddy = &api.CaddySpec{
Config: caddy.Config,
}
}
if ports, ok := service.Extensions[PortsExtensionKey].([]api.PortSpec); ok { if ports, ok := service.Extensions[PortsExtensionKey].([]api.PortSpec); ok {
spec.Ports = ports spec.Ports = ports
} }
@@ -242,3 +248,26 @@ func tmpfsVolumeSpecFromCompose(serviceVolume types.ServiceVolumeConfig) api.Vol
return spec return spec
} }
// validateServicesExtensions validates extension combinations across all services in the project.
func validateServicesExtensions(project *types.Project) error {
for _, service := range project.Services {
// Check for x-caddy and x-ports conflict.
hasCaddy := false
if caddy, ok := service.Extensions[CaddyExtensionKey].(Caddy); ok && caddy.Config != "" {
hasCaddy = true
}
hasPorts := false
if ports, ok := service.Extensions[PortsExtensionKey].([]api.PortSpec); ok && len(ports) > 0 {
hasPorts = true
}
if hasCaddy && hasPorts {
return fmt.Errorf("service '%s' cannot specify both 'x-caddy' and 'x-ports': "+
"Caddy config is auto-generated from ports, use only one of them", service.Name)
}
}
return nil
}
+185 -3
View File
@@ -2,6 +2,7 @@ package compose
import ( import (
"context" "context"
"net/netip"
"path/filepath" "path/filepath"
"slices" "slices"
"strings" "strings"
@@ -38,6 +39,7 @@ func loadProjectFromContent(t *testing.T, content string) (*types.Project, error
if o.KnownExtensions == nil { if o.KnownExtensions == nil {
o.KnownExtensions = map[string]any{} o.KnownExtensions = map[string]any{}
} }
o.KnownExtensions[CaddyExtensionKey] = Caddy{}
o.KnownExtensions[PortsExtensionKey] = PortsSource{} o.KnownExtensions[PortsExtensionKey] = PortsSource{}
o.KnownExtensions[MachinesExtensionKey] = MachinesSource{} o.KnownExtensions[MachinesExtensionKey] = MachinesSource{}
}) })
@@ -45,11 +47,19 @@ func loadProjectFromContent(t *testing.T, content string) (*types.Project, error
return nil, err return nil, err
} }
// Apply ports extension transformation since we're not using LoadProject // Apply extension transformations since we're not using LoadProject.
if project, err = transformServicesCaddyExtension(project); err != nil {
return nil, err
}
if project, err = transformServicesPortsExtension(project); err != nil { if project, err = transformServicesPortsExtension(project); err != nil {
return nil, err return nil, err
} }
// Validate extension combinations after all transformations.
if err = validateServicesExtensions(project); err != nil {
return nil, err
}
return project, nil return project, nil
} }
@@ -182,6 +192,25 @@ func TestServiceSpecFromCompose(t *testing.T) {
}, },
}, },
}, },
Ports: []api.PortSpec{
{
Hostname: "test.example.com",
ContainerPort: 80,
Protocol: api.ProtocolHTTPS,
Mode: api.PortModeIngress,
},
{
ContainerPort: 8000,
Protocol: api.ProtocolHTTP,
Mode: api.PortModeIngress,
},
{
ContainerPort: 3000,
PublishedPort: 5000,
Protocol: "tcp",
Mode: api.PortModeHost,
},
},
Replicas: 3, Replicas: 3,
Volumes: []api.VolumeSpec{ Volumes: []api.VolumeSpec{
{ {
@@ -225,6 +254,19 @@ func TestServiceSpecFromCompose(t *testing.T) {
}, },
}, },
}, },
"test-caddy-config": {
Name: "test-caddy-config",
Mode: api.ServiceModeReplicated,
Container: api.ContainerSpec{
Image: "myapp:1.2.3",
PullPolicy: api.PullPolicyMissing,
},
Caddy: &api.CaddySpec{
Config: `test-caddy-config.example.com {
reverse_proxy {{ upstreams 80 }}
}`,
},
},
}, },
}, },
} }
@@ -245,13 +287,153 @@ func TestServiceSpecFromCompose(t *testing.T) {
return strings.Compare(a.Name, b.Name) return strings.Compare(a.Name, b.Name)
}) })
assert.True(t, cmp.Equal(spec, expectedSpec, cmpopts.EquateEmpty()), cmpOpts := cmp.Options{cmpopts.EquateEmpty(), cmpopts.EquateComparable(netip.Addr{})}
cmp.Diff(spec, expectedSpec, cmpopts.EquateEmpty())) assert.True(t, cmp.Equal(spec, expectedSpec, cmpOpts...), cmp.Diff(spec, expectedSpec, cmpOpts...))
} }
}) })
} }
} }
func TestServiceSpecFromCompose_Caddy(t *testing.T) {
tests := []struct {
name string
composeYAML string
want *api.CaddySpec
}{
{
name: "x-caddy as string",
composeYAML: `
services:
web:
image: nginx
x-caddy: |
example.com {
reverse_proxy web:80
}
`,
want: &api.CaddySpec{
Config: `example.com {
reverse_proxy web:80
}`,
},
},
{
name: "x-caddy as string with extra spaces",
composeYAML: `
services:
web:
image: nginx
x-caddy: |+
example.com {
reverse_proxy web:80
}
`,
want: &api.CaddySpec{
Config: `example.com {
reverse_proxy web:80
}`,
},
},
{
name: "x-caddy as object with config field",
composeYAML: `
services:
web:
image: nginx
x-caddy:
config: |
example.com {
reverse_proxy web:80
}
`,
want: &api.CaddySpec{
Config: `example.com {
reverse_proxy web:80
}`,
},
},
{
name: "x-caddy as object with config field and extra spaces",
composeYAML: `
services:
web:
image: nginx
x-caddy:
config: |+
example.com {
reverse_proxy web:80
}
`,
want: &api.CaddySpec{
Config: `example.com {
reverse_proxy web:80
}`,
},
},
{
name: "x-caddy with path to Caddyfile",
composeYAML: `
services:
web:
image: nginx
x-caddy: testdata/Caddyfile
`,
want: &api.CaddySpec{
Config: `test.example.com {
reverse_proxy test:8000
}`,
},
},
{
name: "x-caddy with empty string",
composeYAML: `
services:
web:
image: nginx
x-caddy: ""
`,
want: nil,
},
{
name: "no x-caddy extension",
composeYAML: `
services:
web:
image: nginx
`,
want: nil,
},
{
name: "x-caddy with empty object",
composeYAML: `
services:
web:
image: nginx
x-caddy: {}
`,
want: nil,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
project, err := loadProjectFromContent(t, tt.composeYAML)
require.NoError(t, err)
spec, err := ServiceSpecFromCompose(project, "web")
require.NoError(t, err)
assert.Equal(t, tt.want, spec.Caddy)
})
}
}
func TestServiceSpecFromCompose_XMachinesPlacement(t *testing.T) { func TestServiceSpecFromCompose_XMachinesPlacement(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
+3
View File
@@ -0,0 +1,3 @@
test.example.com {
reverse_proxy test:8000
}
+12
View File
@@ -31,6 +31,18 @@ services:
target: /tmpfs target: /tmpfs
tmpfs: tmpfs:
size: 10485760 size: 10485760
x-ports:
- test.example.com:80/https
- 8000/http
- 5000:3000@host
test-caddy-config:
image: myapp:1.2.3
# x-ports and x-caddy are mutually exclusive.
x-caddy: |
test-caddy-config.example.com {
reverse_proxy {{ upstreams 80 }}
}
volumes: volumes:
data1: data1:
+41 -1
View File
@@ -4,12 +4,17 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"os"
"strings" "strings"
dockercommand "github.com/docker/cli/cli/command"
dockerconfig "github.com/docker/cli/cli/config"
"github.com/docker/compose/v2/pkg/progress" "github.com/docker/compose/v2/pkg/progress"
"github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/registry"
dockerclient "github.com/docker/docker/client" dockerclient "github.com/docker/docker/client"
"github.com/docker/docker/pkg/jsonmessage" "github.com/docker/docker/pkg/jsonmessage"
"github.com/psviderski/uncloud/internal/machine/docker"
"github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/internal/secret"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
@@ -89,7 +94,14 @@ func (cli *Client) pullImageWithProgress(ctx context.Context, image, machineName
StatusText: "Pulling", StatusText: "Pulling",
}) })
pullCh, err := cli.Docker.PullImage(ctx, image) opts := docker.PullOptions{}
// Try to retrieve the authentication token for the image from the default local Docker config file.
if encodedAuth, err := retrieveRegistryAuthFromDocker(image); err == nil && encodedAuth != "" {
// If RegistryAuth is empty, Uncloud daemon will try to retrieve the credentials from its own Docker config.
opts.RegistryAuth = encodedAuth
}
pullCh, err := cli.Docker.PullImage(ctx, image, opts)
if err != nil { if err != nil {
statusErr := status.Convert(err) statusErr := status.Convert(err)
pw.Event(progress.Event{ pw.Event(progress.Event{
@@ -143,6 +155,34 @@ func (cli *Client) pullImageWithProgress(ctx context.Context, image, machineName
return nil return nil
} }
// retrieveRegistryAuthFromDocker retrieves the authentication token for the specified image from the local Docker
// config file. It returns the encoded authentication token if it contains any credentials, or an empty string if
// no credentials are found.
func retrieveRegistryAuthFromDocker(image string) (string, error) {
// Try to retrieve the authentication token for the image from the default local Docker config file.
dockerConfig := dockerconfig.LoadDefaultConfigFile(os.Stderr)
encodedAuth, err := dockercommand.RetrieveAuthTokenFromImage(dockerConfig, image)
if err != nil {
return "", err
}
// The encodedAuth can be a base64-encoded "{}" (empty JSON object) or include a server address but no credentials.
// Return encodedAuth only if it contains any credentials.
auth, err := registry.DecodeAuthConfig(encodedAuth)
if err != nil {
return "", fmt.Errorf("decode auth config: %w", err)
}
if auth.Username == "" &&
auth.Password == "" &&
auth.Auth == "" &&
auth.IdentityToken == "" &&
auth.RegistryToken == "" {
return "", nil
}
return encodedAuth, nil
}
// toPullProgressEvent converts a JSON progress message from the Docker API to a progress event. // toPullProgressEvent converts a JSON progress message from the Docker API to a progress event.
// It's based on toPullProgressEvent from Docker Compose. // It's based on toPullProgressEvent from Docker Compose.
func toPullProgressEvent(jm jsonmessage.JSONMessage) *progress.Event { func toPullProgressEvent(jm jsonmessage.JSONMessage) *progress.Event {
+4
View File
@@ -72,6 +72,10 @@ func EvalContainerSpecChange(current api.ServiceSpec, new api.ServiceSpec) Conta
} }
// Check if any mutable properties changed. // Check if any mutable properties changed.
if !current.Caddy.Equals(new.Caddy) {
return ContainerNeedsRecreate
}
if !reflect.DeepEqual(current.Container.Resources, newResources) { if !reflect.DeepEqual(current.Container.Resources, newResources) {
return ContainerNeedsUpdate return ContainerNeedsUpdate
} }
+17 -5
View File
@@ -25,7 +25,8 @@ type Strategy interface {
// RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time // RollingStrategy implements a rolling update deployment pattern where containers are updated one at a time
// to minimize service disruption. // to minimize service disruption.
type RollingStrategy struct { type RollingStrategy struct {
State *scheduler.ClusterState State *scheduler.ClusterState
ForceRecreate bool
} }
func (s *RollingStrategy) Type() string { func (s *RollingStrategy) Type() string {
@@ -92,7 +93,12 @@ func (s *RollingStrategy) planReplicated(svc *api.Service, spec api.ServiceSpec)
continue continue
} }
status := EvalContainerSpecChange(c.Container.ServiceSpec, spec) var status ContainerSpecStatus
if s.ForceRecreate {
status = ContainerNeedsRecreate
} else {
status = EvalContainerSpecChange(c.Container.ServiceSpec, spec)
}
containerSpecStatuses[c.Container.ID] = status containerSpecStatuses[c.Container.ID] = status
if status == ContainerUpToDate { if status == ContainerUpToDate {
@@ -225,7 +231,7 @@ func (s *RollingStrategy) planGlobal(svc *api.Service, spec api.ServiceSpec) (Pl
for _, m := range availableMachines { for _, m := range availableMachines {
containers := containersOnMachine[m.Info.Id] containers := containersOnMachine[m.Info.Id]
ops, err := reconcileGlobalContainer(containers, spec, plan.ServiceID, m.Info.Id) ops, err := reconcileGlobalContainer(containers, spec, plan.ServiceID, m.Info.Id, s.ForceRecreate)
if err != nil { if err != nil {
return plan, err return plan, err
} }
@@ -252,7 +258,7 @@ func (s *RollingStrategy) planGlobal(svc *api.Service, spec api.ServiceSpec) (Pl
// It ensures exactly one container with the desired spec is running on the machine by creating a new container and // It ensures exactly one container with the desired spec is running on the machine by creating a new container and
// removing old ones. If there is a host port conflict, it stops the old container before starting a new one. // removing old ones. If there is a host port conflict, it stops the old container before starting a new one.
func reconcileGlobalContainer( func reconcileGlobalContainer(
containers []api.MachineServiceContainer, spec api.ServiceSpec, serviceID, machineID string, containers []api.MachineServiceContainer, spec api.ServiceSpec, serviceID, machineID string, forceRecreate bool,
) ([]Operation, error) { ) ([]Operation, error) {
var ops []Operation var ops []Operation
@@ -274,7 +280,13 @@ func reconcileGlobalContainer(
continue continue
} }
status := EvalContainerSpecChange(c.Container.ServiceSpec, spec) var status ContainerSpecStatus
if forceRecreate {
status = ContainerNeedsRecreate
} else {
status = EvalContainerSpecChange(c.Container.ServiceSpec, spec)
}
if status == ContainerUpToDate { if status == ContainerUpToDate {
// The container is already running with the same spec. // The container is already running with the same spec.
upToDate = true upToDate = true
+1 -1
View File
@@ -166,7 +166,7 @@ RestartSec=2
NoNewPrivileges=true NoNewPrivileges=true
ProtectSystem=full ProtectSystem=full
ProtectControlGroups=true ProtectControlGroups=true
ProtectHome=true ProtectHome=read-only
ProtectKernelTunables=true ProtectKernelTunables=true
PrivateTmp=true PrivateTmp=true
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK
+1 -1
View File
@@ -100,7 +100,7 @@ func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api
assert.Equal(t, portBindings, ctr.HostConfig.PortBindings) assert.Equal(t, portBindings, ctr.HostConfig.PortBindings)
assert.Equal(t, container.RestartPolicy{ assert.Equal(t, container.RestartPolicy{
Name: container.RestartPolicyAlways, Name: container.RestartPolicyUnlessStopped,
MaximumRetryCount: 0, MaximumRetryCount: 0,
}, ctr.HostConfig.RestartPolicy) }, ctr.HostConfig.RestartPolicy)
+151 -19
View File
@@ -9,6 +9,7 @@ import (
"github.com/psviderski/uncloud/internal/ucind" "github.com/psviderski/uncloud/internal/ucind"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/psviderski/uncloud/pkg/client/compose" "github.com/psviderski/uncloud/pkg/client/compose"
"github.com/psviderski/uncloud/pkg/client/deploy"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -34,14 +35,14 @@ func TestComposeDeployment(t *testing.T) {
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-basic.yaml"}) project, err := compose.LoadProject(ctx, []string{"fixtures/compose-basic.yaml"})
require.NoError(t, err) require.NoError(t, err)
deploy, err := compose.NewDeployment(ctx, cli, project) deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
plan, err := deploy.Plan(ctx) plan, err := deployment.Plan(ctx)
require.NoError(t, err) require.NoError(t, err)
assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy") assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy")
err = deploy.Run(ctx) err = deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
svc, err := cli.InspectService(ctx, name) svc, err := cli.InspectService(ctx, name)
@@ -71,6 +72,137 @@ func TestComposeDeployment(t *testing.T) {
assertServiceMatchesSpec(t, svc, expectedSpec) assertServiceMatchesSpec(t, svc, expectedSpec)
}) })
t.Run("multi-service deployment with redeploy and recreate", func(t *testing.T) {
t.Parallel()
serviceNames := []string{
"test-compose-multi-web",
"test-compose-multi-api",
"test-compose-multi-worker",
}
t.Cleanup(func() {
removeServices(t, cli, serviceNames...)
})
// Initial deployment.
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-multi-service.yaml"})
require.NoError(t, err)
deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err)
plan, err := deployment.Plan(ctx)
require.NoError(t, err)
assert.Len(t, plan.Operations, 3, "Expected 3 services to deploy")
err = deployment.Run(ctx)
require.NoError(t, err)
// Verify web service.
webSvc, err := cli.InspectService(ctx, "test-compose-multi-web")
require.NoError(t, err)
expectedWebSpec := api.ServiceSpec{
Name: "test-compose-multi-web",
Mode: api.ServiceModeReplicated,
Container: api.ContainerSpec{
Env: map[string]string{
"SERVICE": "web",
"VERSION": "1.0",
},
Image: "portainer/pause:3.9",
},
Ports: []api.PortSpec{
{
Hostname: "multi.example.com",
ContainerPort: 80,
Protocol: api.ProtocolHTTPS,
Mode: api.PortModeIngress,
},
},
Replicas: 2,
}
assertServiceMatchesSpec(t, webSvc, expectedWebSpec)
// Verify api service.
apiSvc, err := cli.InspectService(ctx, "test-compose-multi-api")
require.NoError(t, err)
expectedApiSpec := api.ServiceSpec{
Name: "test-compose-multi-api",
Mode: api.ServiceModeReplicated,
Container: api.ContainerSpec{
Env: map[string]string{
"SERVICE": "api",
"PORT": "8080",
},
Image: "portainer/pause:3.9",
},
Replicas: 3,
}
assertServiceMatchesSpec(t, apiSvc, expectedApiSpec)
// Verify worker service.
workerSvc, err := cli.InspectService(ctx, "test-compose-multi-worker")
require.NoError(t, err)
expectedWorkerSpec := api.ServiceSpec{
Name: "test-compose-multi-worker",
Mode: api.ServiceModeReplicated,
Container: api.ContainerSpec{
Env: map[string]string{
"SERVICE": "worker",
"CONCURRENCY": "5",
},
Image: "portainer/pause:3.9",
},
Replicas: 1,
}
assertServiceMatchesSpec(t, workerSvc, expectedWorkerSpec)
// Save container IDs for later verification.
containers := serviceContainerIDs(webSvc).
Union(serviceContainerIDs(apiSvc)).
Union(serviceContainerIDs(workerSvc))
// Redeploy without changes - should be up to date.
redeploy, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err)
redeployPlan, err := redeploy.Plan(ctx)
require.NoError(t, err)
assert.Len(t, redeployPlan.Operations, 0, "Expected no operations - deployment should be up to date")
// Deploy with ForceRecreate - should recreate all service containers.
strategy := &deploy.RollingStrategy{ForceRecreate: true}
recreateDeploy, err := compose.NewDeploymentWithStrategy(ctx, cli, project, strategy)
require.NoError(t, err)
recreatePlan, err := recreateDeploy.Plan(ctx)
require.NoError(t, err)
assert.Len(t, recreatePlan.Operations, 3, "Expected 3 services to be recreated")
err = recreateDeploy.Run(ctx)
require.NoError(t, err)
// Verify services match the expected specs after recreate.
webSvcAfter, err := cli.InspectService(ctx, "test-compose-multi-web")
require.NoError(t, err)
assertServiceMatchesSpec(t, webSvcAfter, expectedWebSpec)
apiSvcAfter, err := cli.InspectService(ctx, "test-compose-multi-api")
require.NoError(t, err)
assertServiceMatchesSpec(t, apiSvcAfter, expectedApiSpec)
workerSvcAfter, err := cli.InspectService(ctx, "test-compose-multi-worker")
require.NoError(t, err)
assertServiceMatchesSpec(t, workerSvcAfter, expectedWorkerSpec)
// Verify that all containers have been recreated.
afterContainers := serviceContainerIDs(webSvcAfter).
Union(serviceContainerIDs(apiSvcAfter)).
Union(serviceContainerIDs(workerSvcAfter))
assert.NotEqual(t, containers.ToSlice(), afterContainers.ToSlice(),
"Expected containers to be recreated after deployment with ForceRecreate strategy")
})
t.Run("multiple services with volumes", func(t *testing.T) { t.Run("multiple services with volumes", func(t *testing.T) {
t.Parallel() t.Parallel()
@@ -94,10 +226,10 @@ func TestComposeDeployment(t *testing.T) {
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-volumes.yaml"}) project, err := compose.LoadProject(ctx, []string{"fixtures/compose-volumes.yaml"})
require.NoError(t, err) require.NoError(t, err)
deploy, err := compose.NewDeployment(ctx, cli, project) deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
_, err = deploy.Plan(ctx) _, err = deployment.Plan(ctx)
require.ErrorContains(t, err, "external volumes not found: 'test-compose-volumes-external'") require.ErrorContains(t, err, "external volumes not found: 'test-compose-volumes-external'")
externalVolumeOpts := volume.CreateOptions{Name: "test-compose-volumes-external"} externalVolumeOpts := volume.CreateOptions{Name: "test-compose-volumes-external"}
@@ -105,14 +237,14 @@ func TestComposeDeployment(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
// Recreate the deployment as it caches the cluster state. // Recreate the deployment as it caches the cluster state.
deploy, err = compose.NewDeployment(ctx, cli, project) deployment, err = compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
plan, err := deploy.Plan(ctx) plan, err := deployment.Plan(ctx)
require.NoError(t, err) require.NoError(t, err)
assert.Len(t, plan.Operations, 5, "Expected 2 volumes creation and 3 services to deploy") assert.Len(t, plan.Operations, 5, "Expected 2 volumes creation and 3 services to deploy")
err = deploy.Run(ctx) err = deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
// Verify data1 and data2 volumes have been created. // Verify data1 and data2 volumes have been created.
@@ -247,10 +379,10 @@ func TestComposeDeployment(t *testing.T) {
"service3 should be on the same machine as external volume") "service3 should be on the same machine as external volume")
// Verify deployment is up-to-date. // Verify deployment is up-to-date.
deploy, err = compose.NewDeployment(ctx, cli, project) deployment, err = compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
plan, err = deploy.Plan(ctx) plan, err = deployment.Plan(ctx)
require.NoError(t, err) require.NoError(t, err)
assert.Len(t, plan.Operations, 0, "Expected no new operations after deployment") assert.Len(t, plan.Operations, 0, "Expected no new operations after deployment")
}) })
@@ -266,14 +398,14 @@ func TestComposeDeployment(t *testing.T) {
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-placement.yaml"}) project, err := compose.LoadProject(ctx, []string{"fixtures/compose-placement.yaml"})
require.NoError(t, err) require.NoError(t, err)
deploy, err := compose.NewDeployment(ctx, cli, project) deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
plan, err := deploy.Plan(ctx) plan, err := deployment.Plan(ctx)
require.NoError(t, err) require.NoError(t, err)
assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy") assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy")
err = deploy.Run(ctx) err = deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
svc, err := cli.InspectService(ctx, name) svc, err := cli.InspectService(ctx, name)
@@ -319,14 +451,14 @@ func TestComposeDeployment(t *testing.T) {
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-placement-nonexistent.yaml"}) project, err := compose.LoadProject(ctx, []string{"fixtures/compose-placement-nonexistent.yaml"})
require.NoError(t, err) require.NoError(t, err)
deploy, err := compose.NewDeployment(ctx, cli, project) deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
plan, err := deploy.Plan(ctx) plan, err := deployment.Plan(ctx)
require.NoError(t, err) require.NoError(t, err)
assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy") assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy")
err = deploy.Run(ctx) err = deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
svc, err := cli.InspectService(ctx, name) svc, err := cli.InspectService(ctx, name)
@@ -367,14 +499,14 @@ func TestComposeDeployment(t *testing.T) {
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-placement-comma.yaml"}) project, err := compose.LoadProject(ctx, []string{"fixtures/compose-placement-comma.yaml"})
require.NoError(t, err) require.NoError(t, err)
deploy, err := compose.NewDeployment(ctx, cli, project) deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err) require.NoError(t, err)
plan, err := deploy.Plan(ctx) plan, err := deployment.Plan(ctx)
require.NoError(t, err) require.NoError(t, err)
assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy") assert.Len(t, plan.Operations, 1, "Expected 1 service to deploy")
err = deploy.Run(ctx) err = deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
svc, err := cli.InspectService(ctx, name) svc, err := cli.InspectService(ctx, name)
@@ -0,0 +1,24 @@
services:
test-compose-multi-web:
image: portainer/pause:3.9
environment:
SERVICE: web
VERSION: "1.0"
deploy:
replicas: 2
x-ports:
- multi.example.com:80/https
test-compose-multi-api:
image: portainer/pause:3.9
environment:
SERVICE: api
PORT: "8080"
deploy:
replicas: 3
test-compose-multi-worker:
image: portainer/pause:3.9
environment:
SERVICE: worker
CONCURRENCY: "5"
+1 -25
View File
@@ -279,35 +279,11 @@ func TestDeployment(t *testing.T) {
svc, err := cli.InspectService(ctx, client.CaddyServiceName) svc, err := cli.InspectService(ctx, client.CaddyServiceName)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, client.CaddyServiceName, svc.Name)
assert.Equal(t, api.ServiceModeGlobal, svc.Mode)
assert.Len(t, svc.Containers, 3) assert.Len(t, svc.Containers, 3)
assertServiceMatchesSpec(t, svc, deployment.Spec)
ctr := svc.Containers[0].Container ctr := svc.Containers[0].Container
assert.Regexp(t, `^caddy:2\.\d+\.\d+$`, ctr.Config.Image) assert.Regexp(t, `^caddy:2\.\d+\.\d+$`, ctr.Config.Image)
ports, err := ctr.ServicePorts()
require.NoError(t, err)
expectedPorts := []api.PortSpec{
{
PublishedPort: 80,
ContainerPort: 80,
Protocol: api.ProtocolTCP,
Mode: api.PortModeHost,
},
{
PublishedPort: 443,
ContainerPort: 443,
Protocol: api.ProtocolTCP,
Mode: api.PortModeHost,
},
}
assert.Equal(t, expectedPorts, ports)
assert.Equal(t, container.RestartPolicy{
Name: container.RestartPolicyAlways,
MaximumRetryCount: 0,
}, ctr.HostConfig.RestartPolicy)
}) })
t.Run("caddy with machine placement", func(t *testing.T) { t.Run("caddy with machine placement", func(t *testing.T) {