Compare commits

...
Author SHA1 Message Date
Pasha Sviderski 07e380b95f chore(pre-deploy): warn about depends_on: service_completed_successfully to use pre-deploy hook instead 2026-04-10 17:08:38 +10:00
Pasha Sviderski 4e060914ec docs: add themed diagram to Pre-deploy hooks doc 2026-04-10 15:10:07 +10:00
Pasha Sviderski f317dde168 docs: move Compose extensions to a separate reference page 2026-04-09 20:24:48 +10:00
Pasha Sviderski f2f6fee89a docs: new pre-deploy hooks page and x-pre_deploy extension in the reference page 2026-04-09 20:24:19 +10:00
Pasha Sviderski b3b33a82ba chore(e2e): increase ucind cluster init timeouts, decrease max backoff delay for TCP connection to retry faster 2026-04-09 14:39:12 +10:00
Pasha Sviderski cd149e4506 fix: spec validation when name is empty, race in e2e tests 2026-04-09 14:39:12 +10:00
Miek GiebenandGitHub eff1bc88d8 feat: uc docs --manual flag to generate manual pages (#302)
This adds 'doc --manual' that generates the manual pages in the current
directory.

This allows for packaging the manual pages in the same way as the
completion, e.g. https://github.com/miekg/unpkg/blob/main/deb/uncloud/debian/rules#L28
(For debian packages local manpages should be included)

Signed-off-by: Miek Gieben <miek@miek.nl>
2026-04-09 13:41:38 +10:00
Pasha Sviderski ab644f6a47 fix: race on warned flag in versioncheck pkg, make consistent use of client/server terms 2026-04-08 20:42:27 +10:00
Pasha Sviderski 883f184e37 lint 2026-04-08 19:53:26 +10:00
897f30fd36 feat: add client/server version check mechanism to gRPC calls (#260)
* Add version check mechanism to gRPC calls

* Use semver not semver/v3

* Give dev builds a special "infinite" version number (999.0.0-dev)

* Handle no metadata on grpc call correctly for version check

* Move versioncheck package to root pkg/ from pkg/api/ since it is shared by both pkg/api/ and pkg/client/

* Only show "no daemon version" warning once

* Append version headers to metadata, not overwrite...

* Unit tests on versioncheck logic

* Move SetHeader to the ServerStream in ServerStreamInterceptor

* Go modernizer nits: interface{} -> any

* Use more conventional gRPC header names for version/min-versions

* Add TODO notes on checkDaemonVersionInResponse and related code that can be removed eventually after transition to version checking client/daemons

* Use testify for testing assertions

* Add explanatory comments on MinCLIVersion and MinDaemonVersion

---------

Co-authored-by: Pasha Sviderski <me@psviderski.name>
2026-04-08 19:52:42 +10:00
Pasha Sviderski 943fea0515 refactor: simplify journal logs following, fix corrosion unit name 2026-04-08 19:38:02 +10:00
Pasha Sviderski e96cce88c1 fix(pre-deploy): exclude hook containers from caddy and dns controllers 2026-04-08 19:16:42 +10:00
Pasha Sviderski 1930bb5766 feat(pre-deploy): list hook containers in 'uc ps' and 'uc inspect' 2026-04-08 19:16:42 +10:00
Pasha Sviderski a844cc6f67 fix(pre-deploy): include hook containers in logs, ls, stop commands 2026-04-08 19:16:42 +10:00
Pasha Sviderski 043b85ab70 feat(pre-deploy): include timeout information in pre-deploy hook formatting in plan 2026-04-08 19:16:42 +10:00
Pasha Sviderski fe813302b7 refactor: event ID generation for images, machines, and volumes to align with plan formatting 2026-04-08 19:16:42 +10:00
Pasha Sviderski 3e30a59d4c fix(pre-deploy): progress event for stopping running hooks 2026-04-08 19:16:42 +10:00
Pasha Sviderski 0da9df17a4 feat(pre-deploy): correct progress events for pre-deploy hooks, make other container events more aligned with plan 2026-04-08 19:16:42 +10:00
Pasha Sviderski 0e820f9c52 feat(pre-deploy): add support for x-pre_deploy extension in Compose + e2e test 2026-04-08 19:16:42 +10:00
Pasha Sviderski 772b31b57f feat(pre-deploy): implement pre-deploy hook operations for deployment 2026-04-08 19:16:42 +10:00
Pasha Sviderski 1c8b77054a feat(pre-deploy): add support for pre-deploy hook containers in service management, sync *all* containers to store 2026-04-08 19:16:28 +10:00
Miek GiebenandGitHub cef047221c feat(machine-logs): add server side of journal logs (#282)
* Add server side of journal logs

This add the server side and grpc methods to get a journal logs from a
machine. It repurposes ServiceLogEntry for these logs to keep the
changes somewhat to a minimum. And it lets us re-use the merging of the
various logs. In the protobufs ContainerLog has been renamed to just Log
and LogEntry, as these are now also used for journal logs.

It does api.LogOptions in more places to reduce the various logOpts that
were used.

It does not yet plumb it through to the uc client, that needs a follow
up pr. Following logs is also not yet implemented.

Signed-off-by: Miek Gieben <miek@miek.nl>

* Fix test too

Signed-off-by: Miek Gieben <miek@miek.nl>

* remove entire comment

Signed-off-by: Miek Gieben <miek@miek.nl>

* Implement the follow option, untested mind you

Signed-off-by: Miek Gieben <miek@miek.nl>

* update debug line

Signed-off-by: Miek Gieben <miek@miek.nl>

* internal/jounal: First batch of PR comments

Signed-off-by: Miek Gieben <miek@miek.nl>

* internal/journal: code review comments

Signed-off-by: Miek Gieben <miek@miek.nl>

* Manually apply suggestion

Signed-off-by: Miek Gieben <miek@miek.nl>

* apply comment manually

Signed-off-by: Miek Gieben <miek@miek.nl>

* internal/journal: add unit test

Signed-off-by: Miek Gieben <miek@miek.nl>

* Use testify

Signed-off-by: Miek Gieben <miek@miek.nl>

* -amFix scanner.Err checking

Signed-off-by: Miek Gieben <miek@miek.nl>

* Implement code review comments

Signed-off-by: Miek Gieben <miek@miek.nl>

---------

Signed-off-by: Miek Gieben <miek@miek.nl>
2026-04-08 18:51:52 +10:00
Miek GiebenandGitHub 4b34c42b76 refactor: modernize, remove AsPtr (#300)
modernize missed this in #278, but AsPtr can also be removed here and
replace with new()

Signed-off-by: Miek Gieben <miek@miek.nl>
2026-04-03 09:41:16 +10:00
NickandGitHub 7ec8879af3 feat: accept optional -- in uc exec (#301) 2026-04-03 09:38:24 +10:00
Pasha Sviderski cb77d42107 Merge branch 'docs/uncloud-config' 2026-04-02 15:34:25 +10:00
Pasha Sviderski d45b5a488e docs: 2 new pages: Connecting to a cluster and CLI configuration file 2026-04-02 15:32:58 +10:00
Pasha Sviderski f4581eac11 feat(pre-deploy): add pre-deploy hook to service spec 2026-04-01 19:24:53 +10:00
Pasha Sviderski 0651b9d0ac test: increase timeout waiting for reconciled michine in cluster lifecycle test 2026-04-01 19:16:10 +10:00
Pasha Sviderski 62976f3338 refactor: print warnings for all detected unsupported Compose features, not only first one 2026-03-31 17:08:53 +10:00
Miek GiebenandGitHub d38312fec6 feat: print warning on some unsupported compose features (#288)
* feat: error on unsupported compose features

This implements a check for the unimplemented features of the uncloud
support matrix and points to the matrix in the error.

Fixes: #237

This redos #277, as I messed up my branch, (basically rebase gone
wrong).

Signed-off-by: Miek Gieben <miek@miek.nl>

* Fix tests by capturing stderr

The error should be nil, but we _do_ want to know if the user saw an
warning. Redirect stderr to capture this as the error is emited by
printing to stderr.

Signed-off-by: Miek Gieben <miek@miek.nl>

---------

Signed-off-by: Miek Gieben <miek@miek.nl>
2026-03-31 16:37:35 +10:00
Pasha Sviderski 636307d3ea refactor: replace tabwriter with a Table for consistent table rendering across commands 2026-03-30 14:32:37 +10:00
Pasha Sviderski e22cdf0b18 refactor: NewTable for consistent CLI table rendering, style images for ls,ps 2026-03-30 14:08:50 +10:00
Pasha Sviderski 9aee75ba28 fix: tests 2026-03-27 21:25:47 +10:00
Pasha Sviderski 37c937a570 lint 2026-03-27 20:57:36 +10:00
Pasha Sviderski a1bde30b57 fix: disable interactive prompts for ssh connections to not interfere with TUI 2026-03-27 20:51:38 +10:00
Pasha Sviderski 06c2bc55f2 feat: make connection method using system 'ssh' the default (add ssh+go:// fallback) 2026-03-27 20:27:36 +10:00
Pasha Sviderski 0a9a1db815 fix: try other connections when ssh+cli connection fails, enable spinner for ssh+cli 2026-03-27 16:58:05 +10:00
159 changed files with 7069 additions and 1596 deletions
+1
View File
@@ -197,6 +197,7 @@ uc context use <name> # Switch context
- Integration tests in `test/e2e/` - Integration tests in `test/e2e/`
- Test fixtures in `test/fixtures/` - Test fixtures in `test/fixtures/`
- Use table driven tests whenever possible - Use table driven tests whenever possible
- Use the `testify` library for assertions (e.g., `require.Equal`, `assert.Nil`)
### Dependencies ### Dependencies
+2 -2
View File
@@ -88,12 +88,12 @@ func runDeploy(ctx context.Context, uncli *cli.CLI, opts deployOptions) error {
if len(currentImages) > 1 { if len(currentImages) > 1 {
formattedImages := make([]string, len(currentImages)) formattedImages := make([]string, len(currentImages))
for i, img := range currentImages { for i, img := range currentImages {
formattedImages[i] = tui.FormatImage(img, lipgloss.NewStyle()) formattedImages[i] = tui.FormatImage(img, tui.NoStyle)
} }
fmt.Println(tui.Faint.Render("current images (multiple versions detected): ") + fmt.Println(tui.Faint.Render("current images (multiple versions detected): ") +
strings.Join(formattedImages, tui.Faint.Render(", "))) strings.Join(formattedImages, tui.Faint.Render(", ")))
} else { } else {
fmt.Println(tui.Faint.Render("current image: ") + tui.FormatImage(currentImages[0], lipgloss.NewStyle())) fmt.Println(tui.Faint.Render("current image: ") + tui.FormatImage(currentImages[0], tui.NoStyle))
} }
} }
+6 -6
View File
@@ -3,11 +3,10 @@ package context
import ( import (
"fmt" "fmt"
"maps" "maps"
"os"
"slices" "slices"
"text/tabwriter"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -38,8 +37,8 @@ func list(uncli *cli.CLI) error {
contextNames := slices.Sorted(maps.Keys(uncli.Config.Contexts)) contextNames := slices.Sorted(maps.Keys(uncli.Config.Contexts))
currentContext := uncli.Config.CurrentContext currentContext := uncli.Config.CurrentContext
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) t := tui.NewTable()
fmt.Fprintln(tw, "NAME\tCURRENT\tCONNECTIONS") t.Headers("NAME", "CURRENT", "CONNECTIONS")
for _, name := range contextNames { for _, name := range contextNames {
current := "" current := ""
@@ -47,8 +46,9 @@ func list(uncli *cli.CLI) error {
current = "✓" current = "✓"
} }
connCount := len(uncli.Config.Contexts[name].Connections) connCount := len(uncli.Config.Contexts[name].Connections)
fmt.Fprintf(tw, "%s\t%s\t%d\n", name, current, connCount) t.Row(name, current, fmt.Sprintf("%d", connCount))
} }
return tw.Flush() fmt.Println(t)
return nil
} }
+29
View File
@@ -13,6 +13,10 @@ import (
const docsDir = "website/docs/9-cli-reference" const docsDir = "website/docs/9-cli-reference"
type docOptions struct {
manual bool
}
type cmdWrapper struct { type cmdWrapper struct {
cmd *cobra.Command cmd *cobra.Command
} }
@@ -20,6 +24,7 @@ type cmdWrapper struct {
// NewDocsCommand creates a new hidden command to generate CLI reference docs. // NewDocsCommand creates a new hidden command to generate CLI reference docs.
func NewDocsCommand() *cobra.Command { func NewDocsCommand() *cobra.Command {
wrapper := &cmdWrapper{} wrapper := &cmdWrapper{}
opts := docOptions{}
cmd := &cobra.Command{ cmd := &cobra.Command{
Use: "docs", Use: "docs",
Short: "Generate Uncloud CLI reference docs", Short: "Generate Uncloud CLI reference docs",
@@ -29,6 +34,28 @@ func NewDocsCommand() *cobra.Command {
Args: cobra.NoArgs, Args: cobra.NoArgs,
ValidArgsFunction: cobra.NoFileCompletions, ValidArgsFunction: cobra.NoFileCompletions,
RunE: func(cmd *cobra.Command, _ []string) error { RunE: func(cmd *cobra.Command, _ []string) error {
if opts.manual {
header := &doc.GenManHeader{
Title: "Uncloud",
Section: "1",
Source: "Uncloud https://uncloud.run",
}
if err := doc.GenManTree(cmd.Root(), header, "."); err != nil {
return fmt.Errorf("generate CLI manual pages: %w", err)
}
completionFiles, err := filepath.Glob("uc-completion*.1")
if err != nil {
return fmt.Errorf("list generated manual pages: %w", err)
}
for _, f := range completionFiles {
if err = os.Remove(f); err != nil {
return fmt.Errorf("remove '%s': %w", f, err)
}
}
return nil
}
// Remove existing markdown files. // Remove existing markdown files.
mdFiles, err := filepath.Glob(filepath.Join(docsDir, "*.md")) mdFiles, err := filepath.Glob(filepath.Join(docsDir, "*.md"))
if err != nil { if err != nil {
@@ -72,6 +99,8 @@ func NewDocsCommand() *cobra.Command {
return nil return nil
}, },
} }
cmd.Flags().BoolVar(&opts.manual, "manual", false,
"Generate Uncloud manual pages in the current directory.")
wrapper.cmd = cmd wrapper.cmd = cmd
return cmd return cmd
+3 -18
View File
@@ -10,13 +10,13 @@ import (
"time" "time"
"charm.land/lipgloss/v2" "charm.land/lipgloss/v2"
"charm.land/lipgloss/v2/table"
"github.com/charmbracelet/colorprofile" "github.com/charmbracelet/colorprofile"
"github.com/containerd/platforms" "github.com/containerd/platforms"
"github.com/docker/docker/api/types/image" "github.com/docker/docker/api/types/image"
"github.com/docker/go-units" "github.com/docker/go-units"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -260,22 +260,7 @@ func formatImageTable(rows []imageRow) string {
columns[5].hide = true columns[5].hide = true
} }
t := table.New(). t := tui.NewTable()
// Remove the default border.
Border(lipgloss.Border{}).
BorderTop(false).
BorderBottom(false).
BorderLeft(false).
BorderRight(false).
BorderHeader(false).
BorderColumn(false).
StyleFunc(func(row, col int) lipgloss.Style {
if row == table.HeaderRow {
return lipgloss.NewStyle().Bold(true).PaddingRight(3)
}
// Regular style for data rows with padding.
return lipgloss.NewStyle().PaddingRight(3)
})
var headers []string var headers []string
for _, col := range columns { for _, col := range columns {
@@ -288,7 +273,7 @@ func formatImageTable(rows []imageRow) string {
for _, row := range rows { for _, row := range rows {
values := []string{ values := []string{
row.id, row.id,
row.name, tui.FormatImage(row.name, tui.NoStyle),
row.platforms, row.platforms,
row.createdHuman, row.createdHuman,
row.size, row.size,
+10 -9
View File
@@ -40,17 +40,18 @@ func NewAddCommand() *cobra.Command {
Long: `Add a new machine to an existing Uncloud cluster. Long: `Add a new machine to an existing Uncloud cluster.
Connection methods: Connection methods:
ssh://user@host - Use built-in SSH library (default, no prefix required) [ssh://]user@host - Use system 'ssh' command with full SSH config support (default, no prefix required)
ssh+cli://user@host - Use system SSH command (supports ProxyJump, SSH config)`, ssh+go://user@host - Use Go's built-in SSH library`,
Args: cobra.ExactArgs(1), Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
cli.BindEnvToFlag(cmd, "yes", "UNCLOUD_AUTO_CONFIRM") cli.BindEnvToFlag(cmd, "yes", "UNCLOUD_AUTO_CONFIRM")
uncli := cmd.Context().Value("cli").(*cli.CLI) uncli := cmd.Context().Value("cli").(*cli.CLI)
// Determine if SSH CLI needs to be used and strip scheme // Determine connection mode and strip scheme.
destination := args[0] destination := args[0]
useSSHCLI := strings.HasPrefix(destination, "ssh+cli://") useSSHGo := strings.HasPrefix(destination, "ssh+go://")
destination = strings.TrimPrefix(destination, "ssh+go://")
destination = strings.TrimPrefix(destination, "ssh+cli://") destination = strings.TrimPrefix(destination, "ssh+cli://")
destination = strings.TrimPrefix(destination, "ssh://") destination = strings.TrimPrefix(destination, "ssh://")
@@ -59,11 +60,11 @@ Connection methods:
return fmt.Errorf("parse remote machine: %w", err) return fmt.Errorf("parse remote machine: %w", err)
} }
remoteMachine := &cli.RemoteMachine{ remoteMachine := &cli.RemoteMachine{
User: user, User: user,
Host: host, Host: host,
Port: port, Port: port,
KeyPath: opts.sshKey, KeyPath: opts.sshKey,
UseSSHCLI: useSSHCLI, UseSSHGo: useSSHGo,
} }
return add(cmd.Context(), uncli, remoteMachine, opts) return add(cmd.Context(), uncli, remoteMachine, opts)
+10 -9
View File
@@ -45,8 +45,8 @@ func NewInitCommand() *cobra.Command {
This command creates a new context in your Uncloud config to manage the cluster. This command creates a new context in your Uncloud config to manage the cluster.
Connection methods: Connection methods:
ssh://user@host - Use built-in SSH library (default, no prefix required) [ssh://]user@host - Use system 'ssh' command with full SSH config support (default, no prefix required)
ssh+cli://user@host - Use system SSH command (supports ProxyJump, SSH config)`, ssh+go://user@host - Use Go's built-in SSH library`,
Example: ` # Initialise a new cluster with default settings. Example: ` # Initialise a new cluster with default settings.
uc machine init root@<your-server-ip> uc machine init root@<your-server-ip>
@@ -68,9 +68,10 @@ Connection methods:
var remoteMachine *cli.RemoteMachine var remoteMachine *cli.RemoteMachine
if len(args) > 0 { if len(args) > 0 {
// Determine if SSH CLI is requested and strip scheme // Determine connection mode and strip scheme.
destination := args[0] destination := args[0]
useSSHCLI := strings.HasPrefix(destination, "ssh+cli://") useSSHGo := strings.HasPrefix(destination, "ssh+go://")
destination = strings.TrimPrefix(destination, "ssh+go://")
destination = strings.TrimPrefix(destination, "ssh+cli://") destination = strings.TrimPrefix(destination, "ssh+cli://")
destination = strings.TrimPrefix(destination, "ssh://") destination = strings.TrimPrefix(destination, "ssh://")
@@ -79,11 +80,11 @@ Connection methods:
return fmt.Errorf("parse remote machine: %w", err) return fmt.Errorf("parse remote machine: %w", err)
} }
remoteMachine = &cli.RemoteMachine{ remoteMachine = &cli.RemoteMachine{
User: user, User: user,
Host: host, Host: host,
Port: port, Port: port,
KeyPath: opts.sshKey, KeyPath: opts.sshKey,
UseSSHCLI: useSSHCLI, UseSSHGo: useSSHGo,
} }
} }
+15 -15
View File
@@ -4,11 +4,10 @@ import (
"context" "context"
"fmt" "fmt"
"net/netip" "net/netip"
"os"
"strings" "strings"
"text/tabwriter"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/machine/network"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -39,12 +38,9 @@ func list(ctx context.Context, uncli *cli.CLI) error {
} }
// Print the list of machines in a table format. // Print the list of machines in a table format.
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) t := tui.NewTable()
// Print header. t.Headers("NAME", "STATE", "ADDRESS", "PUBLIC IP", "WIREGUARD ENDPOINTS", "MACHINE ID")
if _, err = fmt.Fprintln(tw, "NAME\tSTATE\tADDRESS\tPUBLIC IP\tWIREGUARD ENDPOINTS\tMACHINE ID"); err != nil {
return fmt.Errorf("write header: %w", err)
}
// Print rows.
for _, member := range machines { for _, member := range machines {
m := member.Machine m := member.Machine
subnet, _ := m.Network.Subnet.ToPrefix() subnet, _ := m.Network.Subnet.ToPrefix()
@@ -62,14 +58,18 @@ func list(ctx context.Context, uncli *cli.CLI) error {
endpoints[i] = addrPort.String() endpoints[i] = addrPort.String()
} }
if _, err = fmt.Fprintf( t.Row(
tw, "%s\t%s\t%s\t%s\t%s\t%s\n", m.Name, capitalise(member.State.String()), subnet, publicIP, m.Name,
strings.Join(endpoints, ", "), member.Machine.Id, capitalise(member.State.String()),
); err != nil { subnet.String(),
return fmt.Errorf("write row: %w", err) publicIP,
} strings.Join(endpoints, tui.Faint.Render(", ")),
member.Machine.Id,
)
} }
return tw.Flush()
fmt.Println(t)
return nil
} }
// capitalise returns a string where the first character is upper case, and the rest is lower case. // capitalise returns a string where the first character is upper case, and the rest is lower case.
+1 -1
View File
@@ -206,7 +206,7 @@ func formatContainerTree(containers []api.ServiceContainer) string {
// Add containers as children. // Add containers as children.
for _, ctr := range ctrs { for _, ctr := range ctrs {
state, _ := ctr.HumanState() state, _ := ctr.HumanState()
info := fmt.Sprintf("%s • %s • %s", ctr.Name, ctr.Config.Image, state) info := fmt.Sprintf("%s • %s • %s", ctr.Name, tui.FormatImage(ctr.Config.Image, tui.NoStyle), state)
t.Child(info) t.Child(info)
} }
+12 -5
View File
@@ -47,23 +47,30 @@ func main() {
var conn *config.MachineConnection var conn *config.MachineConnection
if opts.connect != "" { if opts.connect != "" {
if strings.HasPrefix(opts.connect, "tcp://") { if strings.HasPrefix(opts.connect, "tcp://") {
addrPort, err := netip.ParseAddrPort(opts.connect[len("tcp://"):]) addrPort, err := netip.ParseAddrPort(strings.TrimPrefix(opts.connect, "tcp://"))
if err != nil { if err != nil {
return fmt.Errorf("parse TCP address: %w", err) return fmt.Errorf("parse TCP address: %w", err)
} }
conn = &config.MachineConnection{ conn = &config.MachineConnection{
TCP: &addrPort, TCP: &addrPort,
} }
} else if strings.HasPrefix(opts.connect, "ssh+cli://") { } else if strings.HasPrefix(opts.connect, "ssh+go://") {
dest := opts.connect[len("ssh+cli://"):] dest := strings.TrimPrefix(opts.connect, "ssh+go://")
conn = &config.MachineConnection{ conn = &config.MachineConnection{
SSHCLI: config.SSHDestination(dest), SSHGo: config.SSHDestination(dest),
}
} else if strings.HasPrefix(opts.connect, "ssh+cli://") {
// Backward-compatible alias for ssh://.
dest := strings.TrimPrefix(opts.connect, "ssh+cli://")
conn = &config.MachineConnection{
SSH: config.SSHDestination(dest),
} }
} else if strings.HasPrefix(opts.connect, "unix://") { } else if strings.HasPrefix(opts.connect, "unix://") {
conn = &config.MachineConnection{ conn = &config.MachineConnection{
Unix: opts.connect[len("unix://"):], Unix: opts.connect[len("unix://"):],
} }
} else { } else {
// Default: system ssh CLI command (no prefix or ssh:// prefix).
dest := strings.TrimPrefix(opts.connect, "ssh://") dest := strings.TrimPrefix(opts.connect, "ssh://")
conn = &config.MachineConnection{ conn = &config.MachineConnection{
SSH: config.SSHDestination(dest), SSH: config.SSHDestination(dest),
@@ -83,7 +90,7 @@ func main() {
cmd.PersistentFlags().StringVar(&opts.connect, "connect", "", cmd.PersistentFlags().StringVar(&opts.connect, "connect", "",
"Connect to a remote cluster machine without using the Uncloud configuration file. [$UNCLOUD_CONNECT]\n"+ "Connect to a remote cluster machine without using the Uncloud configuration file. [$UNCLOUD_CONNECT]\n"+
"Format: [ssh://]user@host[:port], ssh+cli://user@host[:port], tcp://host:port, or unix:///path/to/uncloud.sock") "Format: [ssh://]user@host[:port], ssh+go://user@host[:port], tcp://host:port, or unix:///path/to/uncloud.sock")
cmd.PersistentFlags().StringVar(&opts.configPath, "uncloud-config", "~/.config/uncloud/config.yaml", cmd.PersistentFlags().StringVar(&opts.configPath, "uncloud-config", "~/.config/uncloud/config.yaml",
"Path to the Uncloud configuration file. [$UNCLOUD_CONFIG]") "Path to the Uncloud configuration file. [$UNCLOUD_CONFIG]")
_ = cmd.MarkPersistentFlagFilename("uncloud-config", "yaml", "yml") _ = cmd.MarkPersistentFlagFilename("uncloud-config", "yaml", "yml")
+47 -34
View File
@@ -8,14 +8,13 @@ import (
"charm.land/huh/v2/spinner" "charm.land/huh/v2/spinner"
"charm.land/lipgloss/v2" "charm.land/lipgloss/v2"
"charm.land/lipgloss/v2/table"
"github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/container"
"github.com/docker/go-units" "github.com/docker/go-units"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/spf13/cobra"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api"
"github.com/psviderski/uncloud/pkg/client" "github.com/psviderski/uncloud/pkg/client"
"github.com/spf13/cobra"
) )
const ( const (
@@ -67,12 +66,13 @@ type containerInfo struct {
serviceName string serviceName string
machineName string machineName string
id string id string
name string
image string image string
status string status string
highlight containerHighlight highlight containerHighlight
created time.Time created time.Time
ip string ip string
// Hook type (e.g., "pre-deploy"), empty for regular containers.
hook string
} }
func runPs(ctx context.Context, uncli *cli.CLI, opts psOptions) error { func runPs(ctx context.Context, uncli *cli.CLI, opts psOptions) error {
@@ -132,24 +132,22 @@ func runPs(ctx context.Context, uncli *cli.CLI, opts psOptions) error {
} }
func printContainers(containers []containerInfo) error { func printContainers(containers []containerInfo) error {
t := table.New(). t := tui.NewTable()
// Remove the default border.
Border(lipgloss.Border{}).
BorderTop(false).
BorderBottom(false).
BorderLeft(false).
BorderRight(false).
BorderHeader(false).
BorderColumn(false).
StyleFunc(func(row, col int) lipgloss.Style {
if row == table.HeaderRow {
return lipgloss.NewStyle().Bold(true).PaddingRight(3)
}
// Regular style for data rows with padding.
return lipgloss.NewStyle().PaddingRight(3)
})
t.Headers("SERVICE", "CONTAINER ID", "CONTAINER NAME", "IMAGE", "CREATED", "STATUS", "IP ADDRESS", "MACHINE") // Show HOOK column only when hook containers are present.
hasHooks := false
for _, ctr := range containers {
if ctr.hook != "" {
hasHooks = true
break
}
}
if hasHooks {
t.Headers("SERVICE", "CONTAINER ID", "IMAGE", "CREATED", "STATUS", "HOOK", "IP ADDRESS", "MACHINE")
} else {
t.Headers("SERVICE", "CONTAINER ID", "IMAGE", "CREATED", "STATUS", "IP ADDRESS", "MACHINE")
}
for _, ctr := range containers { for _, ctr := range containers {
id := ctr.id id := ctr.id
@@ -171,16 +169,28 @@ func printContainers(containers []containerInfo) error {
statusStyle = lipgloss.NewStyle() // Default statusStyle = lipgloss.NewStyle() // Default
} }
t.Row( if hasHooks {
ctr.serviceName, t.Row(
id, ctr.serviceName,
ctr.name, id,
ctr.image, tui.FormatImage(ctr.image, tui.NoStyle),
created, created,
statusStyle.Render(ctr.status), statusStyle.Render(ctr.status),
ctr.ip, ctr.hook,
ctr.machineName, ctr.ip,
) ctr.machineName,
)
} else {
t.Row(
ctr.serviceName,
id,
tui.FormatImage(ctr.image, tui.NoStyle),
created,
statusStyle.Render(ctr.status),
ctr.ip,
ctr.machineName,
)
}
} }
fmt.Println(t) fmt.Println(t)
@@ -237,7 +247,7 @@ func collectContainers(ctx context.Context, cli *client.Client) ([]containerInfo
continue continue
} }
for _, ctr := range msc.Containers { for _, ctr := range append(msc.Containers, msc.HookContainers...) {
if ctr.Container.State == nil || ctr.Container.Config == nil { if ctr.Container.State == nil || ctr.Container.Config == nil {
continue continue
} }
@@ -259,6 +269,9 @@ func collectContainers(ctx context.Context, cli *client.Client) ([]containerInfo
highlight = highlightSuccess highlight = highlightSuccess
} else if ctr.Container.State.Status == "running" { } else if ctr.Container.State.Status == "running" {
highlight = highlightNormal highlight = highlightNormal
} else if ctr.IsHook() && ctr.Container.State.Status == "exited" && ctr.Container.State.ExitCode == 0 {
// Hook containers (e.g., pre-deploy) are expected to exit successfully.
highlight = highlightNormal
} else { // Other non-critical but noteworthy states } else { // Other non-critical but noteworthy states
highlight = highlightWarning highlight = highlightWarning
} }
@@ -276,12 +289,12 @@ func collectContainers(ctx context.Context, cli *client.Client) ([]containerInfo
serviceName: ctr.ServiceName(), serviceName: ctr.ServiceName(),
machineName: machineName, machineName: machineName,
id: ctr.Container.ID, id: ctr.Container.ID,
name: ctr.Container.Name,
image: ctr.Container.Config.Image, image: ctr.Container.Config.Image,
status: status, status: status,
highlight: highlight, highlight: highlight,
created: created, created: created,
ip: ipStr, ip: ipStr,
hook: ctr.Config.Labels[api.LabelHook],
} }
containers = append(containers, info) containers = append(containers, info)
} }
-1
View File
@@ -111,7 +111,6 @@ func TestCollectContainers_NilMetadata(t *testing.T) {
if len(containers) > 0 { if len(containers) > 0 {
c := containers[0] c := containers[0]
assert.Equal(t, "container1", c.id) assert.Equal(t, "container1", c.id)
assert.Equal(t, "test-container", c.name)
assert.Equal(t, "machine-1", c.machineName, "Should fall back to the single machine name when metadata is nil") assert.Equal(t, "machine-1", c.machineName, "Should fall back to the single machine name when metadata is nil")
} }
} }
+10 -2
View File
@@ -48,8 +48,7 @@ If the service has multiple replicas and no container ID is specified, the comma
Args: cobra.MinimumNArgs(1), Args: cobra.MinimumNArgs(1),
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
uncli := cmd.Context().Value("cli").(*cli.CLI) uncli := cmd.Context().Value("cli").(*cli.CLI)
serviceName := args[0] serviceName, command := normalizeExecArgs(args)
command := args[1:]
if len(command) == 0 { if len(command) == 0 {
command = DEFAULT_COMMAND command = DEFAULT_COMMAND
} }
@@ -82,6 +81,15 @@ If the service has multiple replicas and no container ID is specified, the comma
return execCmd return execCmd
} }
func normalizeExecArgs(args []string) (serviceName string, command []string) {
serviceName = args[0]
command = args[1:]
if len(command) > 0 && command[0] == "--" {
command = command[1:]
}
return serviceName, command
}
func runExec(ctx context.Context, uncli *cli.CLI, serviceName string, command []string, opts execCliOptions) error { func runExec(ctx context.Context, uncli *cli.CLI, serviceName string, command []string, opts execCliOptions) error {
// Disable TTY allocation if not connected to a terminal // Disable TTY allocation if not connected to a terminal
if !tui.IsStdoutTerminal() { if !tui.IsStdoutTerminal() {
+61
View File
@@ -0,0 +1,61 @@
package service
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestNormalizeExecArgs(t *testing.T) {
tests := []struct {
name string
args []string
wantServiceName string
wantCommand []string
}{
{
name: "service only",
args: []string{"test-service"},
wantServiceName: "test-service",
wantCommand: []string{},
},
{
name: "service with command",
args: []string{"test-service", "echo", "hello"},
wantServiceName: "test-service",
wantCommand: []string{"echo", "hello"},
},
{
name: "service with separator and command",
args: []string{"test-service", "--", "echo", "hello"},
wantServiceName: "test-service",
wantCommand: []string{"echo", "hello"},
},
{
name: "service with separator only",
args: []string{"test-service", "--"},
wantServiceName: "test-service",
wantCommand: []string{},
},
{
name: "separator preserves command flag",
args: []string{"test-service", "--", "--help"},
wantServiceName: "test-service",
wantCommand: []string{"--help"},
},
{
name: "only first separator is removed",
args: []string{"test-service", "--", "cmd", "--", "arg"},
wantServiceName: "test-service",
wantCommand: []string{"cmd", "--", "arg"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
gotServiceName, gotCommand := normalizeExecArgs(tt.args)
assert.Equal(t, tt.wantServiceName, gotServiceName)
assert.Equal(t, tt.wantCommand, gotCommand)
})
}
}
+38 -22
View File
@@ -3,14 +3,13 @@ package service
import ( import (
"context" "context"
"fmt" "fmt"
"os"
"slices" "slices"
"text/tabwriter"
"time" "time"
"github.com/docker/docker/pkg/stringid" "github.com/docker/docker/pkg/stringid"
"github.com/docker/go-units" "github.com/docker/go-units"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -61,25 +60,33 @@ func inspect(ctx context.Context, uncli *cli.CLI, opts inspectOptions) error {
fmt.Printf("Mode: %s\n", svc.Mode) fmt.Printf("Mode: %s\n", svc.Mode)
fmt.Println() fmt.Println()
// Combine regular and hook containers.
allContainers := append(svc.Containers, svc.HookContainers...)
// Parse created times for sorting and display. // Parse created times for sorting and display.
createdTimes := make(map[string]time.Time, len(svc.Containers)) createdTimes := make(map[string]time.Time, len(allContainers))
for _, ctr := range svc.Containers { for _, ctr := range allContainers {
createdTimes[ctr.Container.ID], _ = time.Parse(time.RFC3339Nano, ctr.Container.Created) createdTimes[ctr.Container.ID], _ = time.Parse(time.RFC3339Nano, ctr.Container.Created)
} }
// Sort containers by created time (newest first). // Sort containers by created time (newest first).
slices.SortFunc(svc.Containers, func(a, b api.MachineServiceContainer) int { slices.SortFunc(allContainers, func(a, b api.MachineServiceContainer) int {
return createdTimes[b.Container.ID].Compare(createdTimes[a.Container.ID]) return createdTimes[b.Container.ID].Compare(createdTimes[a.Container.ID])
}) })
// Print the list of containers in a table format. // Print the list of containers in a table format.
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) // Show HOOK column only when hook containers are present.
if _, err = fmt.Fprintln(tw, "CONTAINER ID\tIMAGE\tCREATED\tSTATUS\tIP ADDRESS\tMACHINE"); err != nil { hasHooks := len(svc.HookContainers) > 0
return fmt.Errorf("write header: %w", err)
t := tui.NewTable()
if hasHooks {
t.Headers("CONTAINER ID", "IMAGE", "CREATED", "STATUS", "HOOK", "IP ADDRESS", "MACHINE")
} else {
t.Headers("CONTAINER ID", "IMAGE", "CREATED", "STATUS", "IP ADDRESS", "MACHINE")
} }
now := time.Now().UTC() now := time.Now().UTC()
for _, ctr := range svc.Containers { for _, ctr := range allContainers {
created := units.HumanDuration(now.Sub(createdTimes[ctr.Container.ID])) + " ago" created := units.HumanDuration(now.Sub(createdTimes[ctr.Container.ID])) + " ago"
machine := machinesNamesByID[ctr.MachineID] machine := machinesNamesByID[ctr.MachineID]
@@ -98,19 +105,28 @@ func inspect(ctx context.Context, uncli *cli.CLI, opts inspectOptions) error {
ipStr = ip.String() ipStr = ip.String()
} }
_, err = fmt.Fprintf( if hasHooks {
tw, t.Row(
"%s\t%s\t%s\t%s\t%s\t%s\n", stringid.TruncateID(ctr.Container.ID),
stringid.TruncateID(ctr.Container.ID), tui.FormatImage(ctr.Container.Config.Image, tui.NoStyle),
ctr.Container.Config.Image, created,
created, state,
state, ctr.Container.Config.Labels[api.LabelHook],
ipStr, ipStr,
machine, machine,
) )
if err != nil { } else {
return fmt.Errorf("write row: %w", err) t.Row(
stringid.TruncateID(ctr.Container.ID),
tui.FormatImage(ctr.Container.Config.Image, tui.NoStyle),
created,
state,
ipStr,
machine,
)
} }
} }
return tw.Flush()
fmt.Println(t)
return nil
} }
+18 -19
View File
@@ -3,12 +3,11 @@ package service
import ( import (
"context" "context"
"fmt" "fmt"
"os"
"slices" "slices"
"strings" "strings"
"text/tabwriter"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -50,20 +49,22 @@ func list(ctx context.Context, uncli *cli.CLI) error {
}) })
// Print the list of services in a table format. // Print the list of services in a table format.
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) t := tui.NewTable()
// Include the ID column if there are duplicate service names to differentiate them. // Include the ID column if there are duplicate service names to differentiate them.
headers := []string{"NAME", "MODE", "REPLICAS", "IMAGE", "ENDPOINTS"}
if haveDuplicateNames { if haveDuplicateNames {
if _, err = fmt.Fprintf(tw, "ID\t"); err != nil { headers = append([]string{"ID"}, headers...)
return fmt.Errorf("write header: %w", err)
}
}
if _, err = fmt.Fprintln(tw, "NAME\tMODE\tREPLICAS\tIMAGE\tENDPOINTS"); err != nil {
return fmt.Errorf("write header: %w", err)
} }
t.Headers(headers...)
for _, s := range services { for _, s := range services {
images := strings.Join(s.Images(), ", ") images := s.Images()
endpoints := strings.Join(s.Endpoints(), ", ") for i, img := range images {
images[i] = tui.FormatImage(img, tui.NoStyle)
}
formattedImages := strings.Join(images, tui.Faint.Render(", "))
endpoints := strings.Join(s.Endpoints(), tui.Faint.Render(", "))
// If no endpoints from ports, check if the service uses custom Caddy config. // If no endpoints from ports, check if the service uses custom Caddy config.
if endpoints == "" { if endpoints == "" {
@@ -74,15 +75,13 @@ func list(ctx context.Context, uncli *cli.CLI) error {
} }
} }
row := []string{s.Name, s.Mode, fmt.Sprintf("%d", len(s.Containers)), formattedImages, endpoints}
if haveDuplicateNames { if haveDuplicateNames {
if _, err = fmt.Fprintf(tw, "%s\t", s.ID); err != nil { row = append([]string{s.ID}, row...)
return fmt.Errorf("write row: %w", err)
}
}
if _, err = fmt.Fprintf(tw, "%s\t%s\t%d\t%s\t%s\n",
s.Name, s.Mode, len(s.Containers), images, endpoints); err != nil {
return fmt.Errorf("write row: %w", err)
} }
t.Row(row...)
} }
return tw.Flush()
fmt.Println(t)
return nil
} }
+6 -10
View File
@@ -3,12 +3,11 @@ package volume
import ( import (
"context" "context"
"fmt" "fmt"
"os"
"slices" "slices"
"strings" "strings"
"text/tabwriter"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )
@@ -86,16 +85,13 @@ func list(ctx context.Context, uncli *cli.CLI, opts listOptions) error {
} }
// Print the volumes in a table format. // Print the volumes in a table format.
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) t := tui.NewTable()
fmt.Fprintln(tw, "NAME\tDRIVER\tMACHINE") t.Headers("NAME", "DRIVER", "MACHINE")
for _, v := range volumes { for _, v := range volumes {
fmt.Fprintf(tw, "%s\t%s\t%s\n", t.Row(v.Volume.Name, v.Volume.Driver, v.MachineName)
v.Volume.Name,
v.Volume.Driver,
v.MachineName,
)
} }
return tw.Flush() fmt.Println(t)
return nil
} }
+8 -14
View File
@@ -3,13 +3,12 @@ package wg
import ( import (
"context" "context"
"fmt" "fmt"
"os"
"strings" "strings"
"text/tabwriter"
"time" "time"
"github.com/docker/go-units" "github.com/docker/go-units"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/spf13/cobra" "github.com/spf13/cobra"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes" "golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
@@ -97,10 +96,8 @@ func runShow(ctx context.Context, uncli *cli.CLI, opts showOptions) error {
return nil return nil
} }
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) t := tui.NewTable()
if _, err = fmt.Fprintln(tw, "PEER\tPUBLIC KEY\tENDPOINT\tHANDSHAKE\tRECEIVED\tSENT\tALLOWED IPS"); err != nil { t.Headers("PEER", "PUBLIC KEY", "ENDPOINT", "HANDSHAKE", "RECEIVED", "SENT", "ALLOWED IPS")
return fmt.Errorf("write header: %w", err)
}
for _, peer := range resp.Peers { for _, peer := range resp.Peers {
machineName, ok := machinesNamesByPublicKey[wgtypes.Key(peer.PublicKey).String()] machineName, ok := machinesNamesByPublicKey[wgtypes.Key(peer.PublicKey).String()]
@@ -113,20 +110,17 @@ func runShow(ctx context.Context, uncli *cli.CLI, opts showOptions) error {
lastHandshake = time.Since(peer.LastHandshakeTime.AsTime()).Round(time.Second).String() + " ago" lastHandshake = time.Since(peer.LastHandshakeTime.AsTime()).Round(time.Second).String() + " ago"
} }
_, err = fmt.Fprintf( t.Row(
tw,
"%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
machineName, machineName,
wgtypes.Key(peer.PublicKey).String(), wgtypes.Key(peer.PublicKey).String(),
peer.Endpoint, peer.Endpoint,
lastHandshake, lastHandshake,
units.HumanSize(float64(peer.ReceiveBytes)), units.HumanSize(float64(peer.ReceiveBytes)),
units.HumanSize(float64(peer.TransmitBytes)), units.HumanSize(float64(peer.TransmitBytes)),
strings.Join(peer.AllowedIps, ", "), strings.Join(peer.AllowedIps, tui.Faint.Render(", ")),
) )
if err != nil {
return fmt.Errorf("write row: %w", err)
}
} }
return tw.Flush()
fmt.Println(t)
return nil
} }
+1 -1
View File
@@ -14,6 +14,7 @@ require (
github.com/caddyserver/caddy/v2 v2.8.4 github.com/caddyserver/caddy/v2 v2.8.4
github.com/cenkalti/backoff/v4 v4.3.0 github.com/cenkalti/backoff/v4 v4.3.0
github.com/charmbracelet/colorprofile v0.4.2 github.com/charmbracelet/colorprofile v0.4.2
github.com/charmbracelet/x/ansi v0.11.6
github.com/compose-spec/compose-go/v2 v2.9.0 github.com/compose-spec/compose-go/v2 v2.9.0
github.com/containerd/errdefs v1.0.0 github.com/containerd/errdefs v1.0.0
github.com/containerd/platforms v1.0.0-rc.1 github.com/containerd/platforms v1.0.0-rc.1
@@ -111,7 +112,6 @@ require (
github.com/cespare/xxhash v1.1.0 // indirect github.com/cespare/xxhash v1.1.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/charmbracelet/ultraviolet v0.0.0-20260205113103-524a6607adb8 // indirect github.com/charmbracelet/ultraviolet v0.0.0-20260205113103-524a6607adb8 // indirect
github.com/charmbracelet/x/ansi v0.11.6 // indirect
github.com/charmbracelet/x/exp/ordered v0.1.0 // indirect github.com/charmbracelet/x/exp/ordered v0.1.0 // indirect
github.com/charmbracelet/x/exp/strings v0.0.0-20240919170804-a4978c8e603a // indirect github.com/charmbracelet/x/exp/strings v0.0.0-20240919170804-a4978c8e603a // indirect
github.com/charmbracelet/x/term v0.2.2 // indirect github.com/charmbracelet/x/term v0.2.2 // indirect
+52 -50
View File
@@ -263,8 +263,8 @@ func (cli *CLI) initRemoteMachine(ctx context.Context, opts InitClusterOptions)
SSHKeyFile: opts.RemoteMachine.KeyPath, SSHKeyFile: opts.RemoteMachine.KeyPath,
MachineID: resp.Machine.Id, MachineID: resp.Machine.Id,
} }
if opts.RemoteMachine.UseSSHCLI { if opts.RemoteMachine.UseSSHGo {
connCfg.SSHCLI = config.NewSSHDestination( connCfg.SSHGo = config.NewSSHDestination(
opts.RemoteMachine.User, opts.RemoteMachine.User,
opts.RemoteMachine.Host, opts.RemoteMachine.Host,
opts.RemoteMachine.Port, opts.RemoteMachine.Port,
@@ -467,8 +467,8 @@ func (cli *CLI) AddMachine(ctx context.Context, opts AddMachineOptions) (*client
SSHKeyFile: opts.RemoteMachine.KeyPath, SSHKeyFile: opts.RemoteMachine.KeyPath,
MachineID: addResp.Machine.Id, MachineID: addResp.Machine.Id,
} }
if opts.RemoteMachine.UseSSHCLI { if opts.RemoteMachine.UseSSHGo {
connCfg.SSHCLI = config.NewSSHDestination( connCfg.SSHGo = config.NewSSHDestination(
opts.RemoteMachine.User, opts.RemoteMachine.User,
opts.RemoteMachine.Host, opts.RemoteMachine.Host,
opts.RemoteMachine.Port, opts.RemoteMachine.Port,
@@ -498,73 +498,75 @@ func (cli *CLI) AddMachine(ctx context.Context, opts AddMachineOptions) (*client
func provisionOrConnectRemoteMachine( func provisionOrConnectRemoteMachine(
ctx context.Context, remoteMachine *RemoteMachine, skipInstall bool, version string, ctx context.Context, remoteMachine *RemoteMachine, skipInstall bool, version string,
) (*client.Client, error) { ) (*client.Client, error) {
// Use SSH CLI // Use Go's built-in SSH library.
if remoteMachine.UseSSHCLI { if remoteMachine.UseSSHGo {
exec := sshexec.NewSSHCLIRemote( sshClient, err := sshexec.Connect(
remoteMachine.User, remoteMachine.User, remoteMachine.Host, remoteMachine.Port, remoteMachine.KeyPath,
remoteMachine.Host,
remoteMachine.Port,
remoteMachine.KeyPath,
) )
// If the SSH connection using SSH agent fails and no key path is provided, try to use the default SSH key.
if err != nil && remoteMachine.KeyPath == "" {
remoteMachine.KeyPath = DefaultSSHKeyPath
sshClient, err = sshexec.Connect(
remoteMachine.User, remoteMachine.Host, remoteMachine.Port, remoteMachine.KeyPath,
)
}
if err != nil {
return nil, fmt.Errorf(
"SSH login to remote machine %s: %w",
config.NewSSHDestination(remoteMachine.User, remoteMachine.Host, remoteMachine.Port), err,
)
}
if !skipInstall { if !skipInstall {
if err := provisionMachine(ctx, exec, version); err != nil { // Provision the remote machine by installing the Uncloud daemon and dependencies over SSH.
exec := sshexec.NewRemote(sshClient)
if err = provisionMachine(ctx, exec, version); err != nil {
return nil, fmt.Errorf("provision machine: %w", err) return nil, fmt.Errorf("provision machine: %w", err)
} }
} }
sshConfig := &connector.SSHConnectorConfig{ var machineClient *client.Client
User: remoteMachine.User, if remoteMachine.User == "root" || skipInstall {
Host: remoteMachine.Host, // Create a machine API client over the established SSH connection to the remote machine.
Port: remoteMachine.Port, machineClient, err = client.New(ctx, connector.NewSSHConnectorFromClient(sshClient))
KeyPath: remoteMachine.KeyPath, } else {
// Since the user is not root, we need to establish a new SSH connection to make the user's addition
// to the uncloud group effective, thus allowing access to the Uncloud daemon Unix socket.
sshConfig := &connector.SSHConnectorConfig{
User: remoteMachine.User,
Host: remoteMachine.Host,
Port: remoteMachine.Port,
KeyPath: remoteMachine.KeyPath,
}
machineClient, err = client.New(ctx, connector.NewSSHConnector(sshConfig))
} }
machineClient, err := client.New(ctx, connector.NewSSHCLIConnector(sshConfig))
if err != nil { if err != nil {
return nil, fmt.Errorf("connect to remote machine: %w", err) return nil, fmt.Errorf("connect to remote machine: %w", err)
} }
return machineClient, nil return machineClient, nil
} }
// Use Go SSH // Use the system 'ssh' command (default).
sshClient, err := sshexec.Connect(remoteMachine.User, remoteMachine.Host, remoteMachine.Port, remoteMachine.KeyPath) exec := sshexec.NewSSHCLIRemote(
// If the SSH connection using SSH agent fails and no key path is provided, try to use the default SSH key. remoteMachine.User,
if err != nil && remoteMachine.KeyPath == "" { remoteMachine.Host,
remoteMachine.KeyPath = DefaultSSHKeyPath remoteMachine.Port,
sshClient, err = sshexec.Connect( remoteMachine.KeyPath,
remoteMachine.User, remoteMachine.Host, remoteMachine.Port, remoteMachine.KeyPath, )
)
}
if err != nil {
return nil, fmt.Errorf(
"SSH login to remote machine %s: %w",
config.NewSSHDestination(remoteMachine.User, remoteMachine.Host, remoteMachine.Port), err,
)
}
if !skipInstall { if !skipInstall {
// Provision the remote machine by installing the Uncloud daemon and dependencies over SSH. if err := provisionMachine(ctx, exec, version); err != nil {
exec := sshexec.NewRemote(sshClient)
if err = provisionMachine(ctx, exec, version); err != nil {
return nil, fmt.Errorf("provision machine: %w", err) return nil, fmt.Errorf("provision machine: %w", err)
} }
} }
var machineClient *client.Client sshConfig := &connector.SSHConnectorConfig{
if remoteMachine.User == "root" || skipInstall { User: remoteMachine.User,
// Create a machine API client over the established SSH connection to the remote machine. Host: remoteMachine.Host,
machineClient, err = client.New(ctx, connector.NewSSHConnectorFromClient(sshClient)) Port: remoteMachine.Port,
} else { KeyPath: remoteMachine.KeyPath,
// Since the user is not root, we need to establish a new SSH connection to make the user's addition
// to the uncloud group effective, thus allowing access to the Uncloud daemon Unix socket.
sshConfig := &connector.SSHConnectorConfig{
User: remoteMachine.User,
Host: remoteMachine.Host,
Port: remoteMachine.Port,
KeyPath: remoteMachine.KeyPath,
}
machineClient, err = client.New(ctx, connector.NewSSHConnector(sshConfig))
} }
machineClient, err := client.New(ctx, connector.NewSSHCLIConnector(sshConfig))
if err != nil { if err != nil {
return nil, fmt.Errorf("connect to remote machine: %w", err) return nil, fmt.Errorf("connect to remote machine: %w", err)
} }
+14 -5
View File
@@ -12,8 +12,12 @@ import (
) )
type MachineConnection struct { type MachineConnection struct {
SSH SSHDestination `yaml:"ssh,omitempty"` // SSH uses the system ssh CLI command to connect. This is the default SSH connection method.
SSHCLI SSHDestination `yaml:"ssh_cli,omitempty"` SSH SSHDestination `yaml:"ssh,omitempty"`
// SSHCLI is a backward-compatible alias for SSH.
SSHCLI SSHDestination `yaml:"ssh_cli,omitempty"`
// SSHGo uses Go's built-in SSH library to connect.
SSHGo SSHDestination `yaml:"ssh_go,omitempty"`
SSHKeyFile string `yaml:"ssh_key_file,omitempty"` SSHKeyFile string `yaml:"ssh_key_file,omitempty"`
// TCP is the address and port of the machine's API server. // TCP is the address and port of the machine's API server.
// The pointer is used to omit the field when not set. Otherwise, yaml marshalling includes an empty object. // The pointer is used to omit the field when not set. Otherwise, yaml marshalling includes an empty object.
@@ -29,7 +33,9 @@ func (c *MachineConnection) String() string {
if c.SSH != "" { if c.SSH != "" {
return "ssh://" + string(c.SSH) return "ssh://" + string(c.SSH)
} else if c.SSHCLI != "" { } else if c.SSHCLI != "" {
return "ssh+cli://" + string(c.SSHCLI) return "ssh://" + string(c.SSHCLI)
} else if c.SSHGo != "" {
return "ssh+go://" + string(c.SSHGo)
} else if c.TCP != nil && c.TCP.IsValid() { } else if c.TCP != nil && c.TCP.IsValid() {
return fmt.Sprintf("tcp://%s", c.TCP) return fmt.Sprintf("tcp://%s", c.TCP)
} else if c.Unix != "" { } else if c.Unix != "" {
@@ -46,6 +52,9 @@ func (c *MachineConnection) Validate() error {
if c.SSHCLI != "" { if c.SSHCLI != "" {
setCount++ setCount++
} }
if c.SSHGo != "" {
setCount++
}
if c.TCP != nil && c.TCP.IsValid() { if c.TCP != nil && c.TCP.IsValid() {
setCount++ setCount++
} }
@@ -54,10 +63,10 @@ func (c *MachineConnection) Validate() error {
} }
if setCount == 0 { if setCount == 0 {
return errors.New("no connection method specified (ssh, ssh_cli, tcp, or unix required)") return errors.New("no connection method specified (ssh, ssh_go, tcp, or unix required)")
} }
if setCount > 1 { if setCount > 1 {
return errors.New("only one connection method allowed per connection (ssh, ssh_cli, tcp, or unix)") return errors.New("only one connection method allowed per connection (ssh, ssh_go, tcp, or unix)")
} }
return nil return nil
+35 -5
View File
@@ -30,18 +30,32 @@ func TestMachineConnection_String(t *testing.T) {
want: "ssh://user@host.com:2222", want: "ssh://user@host.com:2222",
}, },
{ {
name: "ssh_cli connection", name: "ssh_cli connection (backward compat alias for ssh)",
conn: MachineConnection{ conn: MachineConnection{
SSHCLI: "user@host.com", SSHCLI: "user@host.com",
}, },
want: "ssh+cli://user@host.com", want: "ssh://user@host.com",
}, },
{ {
name: "ssh_cli connection with port", name: "ssh_cli connection with port (backward compat alias for ssh)",
conn: MachineConnection{ conn: MachineConnection{
SSHCLI: "user@host.com:2222", SSHCLI: "user@host.com:2222",
}, },
want: "ssh+cli://user@host.com:2222", want: "ssh://user@host.com:2222",
},
{
name: "ssh_go connection",
conn: MachineConnection{
SSHGo: "user@host.com",
},
want: "ssh+go://user@host.com",
},
{
name: "ssh_go connection with port",
conn: MachineConnection{
SSHGo: "user@host.com:2222",
},
want: "ssh+go://user@host.com:2222",
}, },
{ {
name: "tcp connection", name: "tcp connection",
@@ -103,12 +117,19 @@ func TestMachineConnection_Validate(t *testing.T) {
wantErr: false, wantErr: false,
}, },
{ {
name: "ssh_cli only - valid", name: "ssh_cli only - valid (backward compat)",
conn: MachineConnection{ conn: MachineConnection{
SSHCLI: "user@host", SSHCLI: "user@host",
}, },
wantErr: false, wantErr: false,
}, },
{
name: "ssh_go only - valid",
conn: MachineConnection{
SSHGo: "user@host",
},
wantErr: false,
},
{ {
name: "tcp only - valid", name: "tcp only - valid",
conn: MachineConnection{ conn: MachineConnection{
@@ -141,6 +162,15 @@ func TestMachineConnection_Validate(t *testing.T) {
wantErr: true, wantErr: true,
errMsg: "only one connection method allowed", errMsg: "only one connection method allowed",
}, },
{
name: "ssh and ssh_go - error",
conn: MachineConnection{
SSH: "user@host",
SSHGo: "user@host",
},
wantErr: true,
errMsg: "only one connection method allowed",
},
{ {
name: "ssh and unix - error", name: "ssh and unix - error",
conn: MachineConnection{ conn: MachineConnection{
+12 -8
View File
@@ -57,9 +57,9 @@ func connectClusterWithProgress(ctx context.Context, conn config.MachineConnecti
} }
func connectCluster(ctx context.Context, conn config.MachineConnection) (*client.Client, error) { func connectCluster(ctx context.Context, conn config.MachineConnection) (*client.Client, error) {
// Determine which SSH type is configured // Determine which SSH type is configured.
var sshDest config.SSHDestination var sshDest config.SSHDestination
var useSSHCLI bool var useGoSSH bool
// Validate connection configuration early to provide clear error messages. // Validate connection configuration early to provide clear error messages.
if err := conn.Validate(); err != nil { if err := conn.Validate(); err != nil {
@@ -67,11 +67,15 @@ func connectCluster(ctx context.Context, conn config.MachineConnection) (*client
} }
if conn.SSH != "" { if conn.SSH != "" {
// SSH uses the system ssh CLI command (default).
sshDest = conn.SSH sshDest = conn.SSH
useSSHCLI = false
} else if conn.SSHCLI != "" { } else if conn.SSHCLI != "" {
// SSHCLI is a backward-compatible alias for SSH.
sshDest = conn.SSHCLI sshDest = conn.SSHCLI
useSSHCLI = true } else if conn.SSHGo != "" {
// SSHGo uses Go's built-in SSH library.
sshDest = conn.SSHGo
useGoSSH = true
} else if conn.TCP != nil && conn.TCP.IsValid() { } else if conn.TCP != nil && conn.TCP.IsValid() {
return client.New(ctx, connector.NewTCPConnector(*conn.TCP)) return client.New(ctx, connector.NewTCPConnector(*conn.TCP))
} else if conn.Unix != "" { } else if conn.Unix != "" {
@@ -95,11 +99,11 @@ func connectCluster(ctx context.Context, conn config.MachineConnection) (*client
KeyPath: keyPath, KeyPath: keyPath,
} }
// Create appropriate connector based on type // Create appropriate connector based on type.
if useSSHCLI { if useGoSSH {
return client.New(ctx, connector.NewSSHCLIConnector(sshConfig)) return client.New(ctx, connector.NewSSHConnector(sshConfig))
} }
return client.New(ctx, connector.NewSSHConnector(sshConfig)) return client.New(ctx, connector.NewSSHCLIConnector(sshConfig))
} }
// connectModel is a TUI model for connecting to a cluster with a progress spinner. // connectModel is a TUI model for connecting to a cluster with a progress spinner.
+5 -5
View File
@@ -23,11 +23,11 @@ const (
) )
type RemoteMachine struct { type RemoteMachine struct {
User string User string
Host string Host string
Port int Port int
KeyPath string KeyPath string
UseSSHCLI bool // indicates ssh+cli:// should be used UseSSHGo bool // Use Go's built-in SSH library instead of the system ssh CLI command.
} }
func installCmd(user string, version string) string { func installCmd(user string, version string) string {
+67
View File
@@ -0,0 +1,67 @@
package progress
import (
"context"
"github.com/docker/docker/pkg/stringid"
"github.com/psviderski/uncloud/internal/cli/tui"
)
type eventIDKey struct{}
// WithEventID returns a context that overrides the default event ID used by client methods.
func WithEventID(ctx context.Context, eventID string) context.Context {
return context.WithValue(ctx, eventIDKey{}, eventID)
}
// ContainerEventID returns a progress event ID for operations on existing containers using the canonical
// service_name/short_id format. Allows to override it using WithEventID.
func ContainerEventID(ctx context.Context, serviceName, containerID, machineName string) string {
if id, ok := ctx.Value(eventIDKey{}).(string); ok && id != "" {
return id
}
return tui.Faint.Render("Container ") +
serviceName + tui.Faint.Render("/") + stringid.TruncateID(containerID) +
tui.Faint.Render(" on ") + machineName
}
// NewContainerEventID returns a progress event ID for new container creation where the Docker container ID
// is not yet known. Allows to override it using WithEventID.
func NewContainerEventID(ctx context.Context, containerName, machineName string) string {
if id, ok := ctx.Value(eventIDKey{}).(string); ok && id != "" {
return id
}
return tui.Faint.Render("Container ") + containerName +
tui.Faint.Render(" on ") + machineName
}
// PreDeployHookEventID returns a progress event ID for pre-deploy hook operations.
func PreDeployHookEventID(serviceName, machineName string) string {
return tui.Faint.Render("Pre-deploy hook ") + serviceName +
tui.Faint.Render(" on ") + machineName
}
// OldPreDeployHookEventID returns a progress event ID for old pre-deploy hook container cleanup.
func OldPreDeployHookEventID(serviceName, containerID, machineName string) string {
return tui.Faint.Render("Old pre-deploy hook ") +
serviceName + tui.Faint.Render("/") + stringid.TruncateID(containerID) +
tui.Faint.Render(" on ") + machineName
}
// ImageEventID returns a progress event ID for image pull operations.
func ImageEventID(image, machineName string) string {
return tui.Faint.Render("Image ") + image +
tui.Faint.Render(" on ") + machineName
}
// VolumeEventID returns a progress event ID for volume operations.
func VolumeEventID(volumeName, machineName string) string {
return tui.Faint.Render("Volume ") + volumeName +
tui.Faint.Render(" on ") + machineName
}
// MachineEventID returns a progress event ID for machine operations.
func MachineEventID(machineName, publicIP string) string {
return tui.Faint.Render("Machine ") + machineName +
tui.Faint.Render(" (") + publicIP + tui.Faint.Render(")")
}
+10
View File
@@ -51,3 +51,13 @@ func IsStdinTerminal() bool {
func IsStdoutTerminal() bool { func IsStdoutTerminal() bool {
return term.IsTerminal(int(os.Stdout.Fd())) return term.IsTerminal(int(os.Stdout.Fd()))
} }
// TerminalWidth returns the width of the terminal.
// Returns 0 if stdout is not a terminal or the width cannot be determined.
func TerminalWidth() int {
width, _, err := term.GetSize(int(os.Stdout.Fd()))
if err != nil || width <= 0 {
return 0
}
return width
}
+2
View File
@@ -6,6 +6,8 @@ import (
) )
var ( var (
NoStyle = lipgloss.NewStyle()
Faint = lipgloss.NewStyle().Faint(true) Faint = lipgloss.NewStyle().Faint(true)
Red = lipgloss.NewStyle().Foreground(lipgloss.Red) Red = lipgloss.NewStyle().Foreground(lipgloss.Red)
Green = lipgloss.NewStyle().Foreground(lipgloss.Green) Green = lipgloss.NewStyle().Foreground(lipgloss.Green)
+24
View File
@@ -0,0 +1,24 @@
package tui
import (
"charm.land/lipgloss/v2"
"charm.land/lipgloss/v2/table"
)
// NewTable creates a borderless table with bold headers and consistent padding for CLI output.
func NewTable() *table.Table {
return table.New().
Border(lipgloss.Border{}).
BorderTop(false).
BorderBottom(false).
BorderLeft(false).
BorderRight(false).
BorderHeader(false).
BorderColumn(false).
StyleFunc(func(row, col int) lipgloss.Style {
if row == table.HeaderRow {
return Bold.PaddingRight(3)
}
return NoStyle.PaddingRight(3)
})
}
+184
View File
@@ -0,0 +1,184 @@
package grpcversion
import (
"context"
"fmt"
"io"
"os"
"sync/atomic"
"github.com/Masterminds/semver"
"github.com/psviderski/uncloud/internal/version"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
const (
MetadataKeyClientVersion = "uncloud-client-version"
MetadataKeyMinServerVersion = "uncloud-min-server-version"
MetadataKeyServerVersion = "uncloud-server-version"
// MinClientVersion is the minimum client version the daemon accepts. The daemon
// rejects requests from older clients, forcing them to upgrade. This provides
// a clean cut-off for dropping support for old clients.
//
// MinServerVersion is the minimum daemon version the client requires. The client
// sends this with each request so the daemon can immediately reject if it's too old,
// avoiding the need for a preflight request. This is useful when a new client feature
// requires daemon capabilities that didn't exist in older versions.
//
// The two minimums are independent: a client might require a newer daemon for new
// features, while that same daemon could still handle requests from older clients.
MinClientVersion = "0.0.0"
MinServerVersion = "0.0.0"
ReleaseURL = "https://github.com/psviderski/uncloud/releases/latest"
)
var (
// currentVersion is the version of this binary (CLI or daemon).
currentVersion = semver.MustParse(version.String())
// zeroVersion is used when no version is specified (treated as 0.0.0).
zeroVersion = semver.MustParse("0.0.0")
// Pre-parsed minimum versions for comparison.
minClientVersion = semver.MustParse(MinClientVersion)
minServerVersion = semver.MustParse(MinServerVersion)
// warned tracks if we've already printed the daemon version warning.
// TODO: Remove when checkServerVersionInResponse is no longer needed (see below).
warned atomic.Bool
// WarnWriter is the writer used for version mismatch warnings. Defaults to os.Stderr.
// Tests can override this to capture warning output.
WarnWriter io.Writer = os.Stderr
)
func extractVersion(md metadata.MD, key string) *semver.Version {
if md == nil {
return zeroVersion
}
values := md.Get(key)
if len(values) == 0 || values[0] == "" {
return zeroVersion
}
sv, err := semver.NewVersion(values[0])
if err != nil {
return zeroVersion
}
return sv
}
func checkClientVersionHeaders(ctx context.Context) error {
md, _ := metadata.FromIncomingContext(ctx)
actualClientVersion := extractVersion(md, MetadataKeyClientVersion)
if actualClientVersion.LessThan(minClientVersion) {
return status.Errorf(codes.FailedPrecondition,
"version check failed: client version is below minimum %s. Please upgrade: %s",
minClientVersion, ReleaseURL)
}
requiredMinServer := extractVersion(md, MetadataKeyMinServerVersion)
if currentVersion.LessThan(requiredMinServer) {
return status.Errorf(codes.FailedPrecondition,
"version check failed: daemon version %s is below client's minimum required version %s. Please upgrade the daemon: %s",
currentVersion, requiredMinServer, ReleaseURL)
}
return nil
}
func ServerUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
if err := checkClientVersionHeaders(ctx); err != nil {
return nil, err
}
if err := grpc.SetHeader(ctx, metadata.Pairs(MetadataKeyServerVersion, currentVersion.String())); err != nil {
return nil, err
}
return handler(ctx, req)
}
func ServerStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
if err := checkClientVersionHeaders(ss.Context()); err != nil {
return err
}
if err := ss.SetHeader(metadata.Pairs(MetadataKeyServerVersion, currentVersion.String())); err != nil {
return err
}
return handler(srv, ss)
}
func ClientUnaryInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
ctx = metadata.AppendToOutgoingContext(ctx,
MetadataKeyClientVersion, currentVersion.String(),
MetadataKeyMinServerVersion, MinServerVersion,
)
// TODO: Remove when checkServerVersionInResponse is no longer needed,
// as we'll no longer need to extract headers from the response here.
var respMD metadata.MD
opts = append(opts, grpc.Header(&respMD))
err := invoker(ctx, method, req, reply, cc, opts...)
if err != nil {
return err
}
// TODO: Remove eventually (see note on method below).
checkServerVersionInResponse(respMD)
return nil
}
// checkServerVersionInResponse warns the user when they communicated with a daemon that
// did not check the version requirements. This is only needed during the transition to
// version-checking releases.
// TODO: Remove this in some later release, after users have upgraded.
func checkServerVersionInResponse(md metadata.MD) {
serverVersion := extractVersion(md, MetadataKeyServerVersion)
if serverVersion.LessThan(minServerVersion) {
if warned.Swap(true) {
return
}
msg := fmt.Sprintf("daemon version is below minimum required version %s. The daemon did not verify this CLI's minimum version requirement, so the operation may not have behaved as intended. Please upgrade the daemon: %s",
minServerVersion, ReleaseURL)
fmt.Fprintf(WarnWriter, "WARNING: %s\n", msg)
}
}
func ClientStreamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
ctx = metadata.AppendToOutgoingContext(ctx,
MetadataKeyClientVersion, currentVersion.String(),
MetadataKeyMinServerVersion, MinServerVersion,
)
stream, err := streamer(ctx, desc, cc, method, opts...)
if err != nil {
return nil, err
}
// TODO: Wrapping the stream in versionedClientStream will no longer
// be necessary when we are ready to remove the temporary, transition
// safety check checkServerVersionInResponse (see note on method above).
return &versionedClientStream{ClientStream: stream}, nil
}
// TODO: Remove when checkServerVersionInResponse is no longer needed.
type versionedClientStream struct {
grpc.ClientStream
}
// TODO: Remove when checkServerVersionInResponse is no longer needed.
func (s *versionedClientStream) Header() (metadata.MD, error) {
md, err := s.ClientStream.Header()
if err != nil {
return nil, err
}
checkServerVersionInResponse(md)
return md, nil
}
+195
View File
@@ -0,0 +1,195 @@
package grpcversion
import (
"bytes"
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
func TestExtractVersion(t *testing.T) {
tests := []struct {
name string
md metadata.MD
key string
expected string
}{
{
name: "nil metadata",
md: nil,
key: MetadataKeyClientVersion,
expected: "0.0.0",
},
{
name: "missing key",
md: metadata.MD{},
key: MetadataKeyClientVersion,
expected: "0.0.0",
},
{
name: "empty value",
md: metadata.Pairs(MetadataKeyClientVersion, ""),
key: MetadataKeyClientVersion,
expected: "0.0.0",
},
{
name: "invalid version",
md: metadata.Pairs(MetadataKeyClientVersion, "not-a-version"),
key: MetadataKeyClientVersion,
expected: "0.0.0",
},
{
name: "valid version",
md: metadata.Pairs(MetadataKeyClientVersion, "1.2.3"),
key: MetadataKeyClientVersion,
expected: "1.2.3",
},
{
name: "version with prerelease",
md: metadata.Pairs(MetadataKeyClientVersion, "0.0.0-dev"),
key: MetadataKeyClientVersion,
expected: "0.0.0-dev",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := extractVersion(tt.md, tt.key)
assert.Equal(t, tt.expected, got.String())
})
}
}
func TestCheckClientVersionHeaders(t *testing.T) {
tests := []struct {
name string
md metadata.MD
wantErr bool
errCode codes.Code
errContain string
}{
{
name: "cli version below minimum",
md: metadata.Pairs(
MetadataKeyClientVersion, "0.0.0-dev",
),
wantErr: true,
errCode: codes.FailedPrecondition,
errContain: "client version is below minimum",
},
{
name: "cli version above minimum",
md: metadata.Pairs(
MetadataKeyClientVersion, "999.0.0",
),
wantErr: false,
},
{
name: "min daemon version above current daemon",
md: metadata.Pairs(
MetadataKeyClientVersion, "999.0.0",
MetadataKeyMinServerVersion, "999.0.0",
),
wantErr: true,
errCode: codes.FailedPrecondition,
errContain: "daemon version",
},
{
name: "min daemon version below current daemon",
md: metadata.Pairs(
MetadataKeyClientVersion, "999.0.0",
MetadataKeyMinServerVersion, "0.0.1",
),
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctx := context.Background()
if tt.md != nil {
ctx = metadata.NewIncomingContext(ctx, tt.md)
}
err := checkClientVersionHeaders(ctx)
if tt.wantErr {
require.Error(t, err)
st, ok := status.FromError(err)
require.True(t, ok, "expected gRPC status error, got %T", err)
assert.Equal(t, tt.errCode, st.Code())
assert.Contains(t, st.Message(), tt.errContain)
} else {
assert.NoError(t, err)
}
})
}
}
func captureWarnings(t *testing.T, fn func()) string {
t.Helper()
var buf bytes.Buffer
old := WarnWriter
WarnWriter = &buf
t.Cleanup(func() { WarnWriter = old })
fn()
return buf.String()
}
func TestCheckServerVersionInResponse(t *testing.T) {
tests := []struct {
name string
md metadata.MD
wantWarning bool
}{
{
name: "daemon version below minimum",
md: metadata.Pairs(MetadataKeyServerVersion, "0.0.0-dev"),
wantWarning: true,
},
{
name: "daemon version above minimum",
md: metadata.Pairs(MetadataKeyServerVersion, "999.0.0"),
wantWarning: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
warned.Store(false)
output := captureWarnings(t, func() {
checkServerVersionInResponse(tt.md)
})
if tt.wantWarning {
assert.Contains(t, output, "WARNING")
} else {
assert.Empty(t, output)
}
})
}
}
func TestCheckServerVersionInResponse_WarnOnce(t *testing.T) {
warned.Store(false)
md := metadata.Pairs(MetadataKeyServerVersion, "0.0.0-dev")
// First call should warn.
output1 := captureWarnings(t, func() {
checkServerVersionInResponse(md)
})
assert.Contains(t, output1, "WARNING", "first call should warn")
// Second call should not warn (warned flag is now true).
output2 := captureWarnings(t, func() {
checkServerVersionInResponse(md)
})
assert.Empty(t, output2, "second call should not warn")
}
+69
View File
@@ -0,0 +1,69 @@
package journal
import (
"bufio"
"context"
"fmt"
"io"
"os/exec"
"github.com/psviderski/uncloud/pkg/api"
)
const journalctl = "journalctl"
var commandContext = exec.CommandContext // overidable for the test
func logs(ctx context.Context, unit string, opts api.ServiceLogsOptions) (io.ReadCloser, error) {
args := []string{"-u", unit, "--no-hostname"}
args = append(args, "-n")
if opts.Tail > -1 {
args = append(args, fmt.Sprintf("%d", opts.Tail))
} else {
args = append(args, "all")
}
if opts.Follow {
args = append(args, "-f")
}
args = append(args, "-o")
args = append(args, "short-iso-precise")
if opts.Since != "" {
args = append(args, "-S")
args = append(args, opts.Since)
}
if opts.Until != "" {
args = append(args, "-U")
args = append(args, opts.Until)
}
cmd := commandContext(ctx, journalctl, args...)
p, err := cmd.StdoutPipe()
if err != nil {
return nil, err
}
if err := cmd.Start(); err != nil {
return nil, err
}
return p, nil
}
// follow synchronously follows the io.Reader, writing each new journal entry to channel.
// It stops when the reader is exhausted or the context is cancelled.
func follow(ctx context.Context, reader io.Reader, outCh chan api.LogEntry) {
scanner := bufio.NewScanner(reader)
for scanner.Scan() {
select {
case outCh <- entry(scanner.Bytes()):
case <-ctx.Done():
return
}
}
if err := scanner.Err(); err != nil {
outCh <- api.LogEntry{Err: fmt.Errorf("journal logs: %w", err)}
}
}
+60
View File
@@ -0,0 +1,60 @@
package journal
import (
"bytes"
"context"
"fmt"
"slices"
"time"
"github.com/psviderski/uncloud/pkg/api"
)
// Logs streams logs from a service and returns entries via a channel.
func Logs(ctx context.Context, unit string, opts api.ServiceLogsOptions) (<-chan api.LogEntry, error) {
// Hard code unit check for now
switch unit {
case "uncloud":
case "uncloud-corrosion":
case "docker":
default:
return nil, fmt.Errorf("journal logs: invalid unit: %s", unit)
}
reader, err := logs(ctx, unit, opts)
if err != nil {
return nil, err
}
outCh := make(chan api.LogEntry)
go func() {
defer close(outCh)
follow(ctx, reader, outCh)
}()
return outCh, nil
}
func entry(data []byte) api.LogEntry {
// 2025-10-12T11:03:27+02:00 systemd[1]:
timestamp := time.Time{}
message := data
if len(data) > 30 && data[4] == '-' && data[7] == '-' && data[10] == 'T' {
timestampPart, messagePart, found := bytes.Cut(data, []byte(" "))
var err error
if found {
timestamp, err = time.Parse(time.RFC3339Nano, string(timestampPart))
if err != nil {
timestamp = time.Time{}
}
message = messagePart
}
}
return api.LogEntry{
Timestamp: timestamp,
Message: slices.Clone(message), // scanner controls the buffer
Stream: api.LogStreamStdout,
}
}
+49
View File
@@ -0,0 +1,49 @@
package journal
import (
"context"
"os/exec"
"testing"
"time"
"github.com/psviderski/uncloud/pkg/api"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestLogs(t *testing.T) {
commandContext = func(ctx context.Context, _ string, _ ...string) *exec.Cmd {
return exec.CommandContext(ctx, "/usr/bin/tail", "testdata/logs")
}
ctx := context.Background()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
ch, err := Logs(ctx, "uncloud", api.ServiceLogsOptions{})
require.NoError(t, err)
i := 0
for range ch {
i++
}
assert.Equal(t, 6, i)
commandContext = func(ctx context.Context, _ string, _ ...string) *exec.Cmd {
return exec.CommandContext(ctx, "/usr/bin/tail", "-f", "testdata/logs")
}
ctx = context.Background()
ctx, cancel = context.WithCancel(ctx)
go func() { time.Sleep(1 * time.Second); cancel() }()
ch, err = Logs(ctx, "uncloud", api.ServiceLogsOptions{Tail: 3})
require.NoError(t, err)
i = 0
for range ch {
i++
}
// Still six because heartbeats are not written here and Tail is ignored as the command is overridden.
assert.Equal(t, 6, i)
}
+6
View File
@@ -0,0 +1,6 @@
2026-01-23T17:19:33.686964+01:00 fedora kernel: apple-dcp 271c00000.dcp: DCP index:1 dptx target phy: 5 dptx die: 0
2026-01-23T17:19:33.687155+01:00 fedora kernel: platform 271c00000.dcp:piodma: Adding to iommu group 9
2026-01-23T17:19:33.687343+01:00 fedora kernel: apple-dcp 271c00000.dcp: RTKit: Initializing (protocol version 12)
2026-01-23T17:19:33.687500+01:00 fedora kernel: apple-dcp 271c00000.dcp: shmem_setup: iova: 880255000 -> pa: be4f29000 -> iomem: ffff800082>
2026-01-23T17:19:33.687657+01:00 fedora kernel: apple-dcp 271c00000.dcp: shmem_setup: iova: ffffec000, buffer: ffff8000817cc000
2026-01-23T17:19:33.687826+01:00 fedora kernel: apple-dcp 271c00000.dcp: shmem_setup: iova: ffffe8000, buffer: ffff8000817d4000
+298 -46
View File
@@ -10,6 +10,7 @@ import (
status "google.golang.org/genproto/googleapis/rpc/status" status "google.golang.org/genproto/googleapis/rpc/status"
protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl" protoimpl "google.golang.org/protobuf/runtime/protoimpl"
timestamppb "google.golang.org/protobuf/types/known/timestamppb"
reflect "reflect" reflect "reflect"
sync "sync" sync "sync"
) )
@@ -21,6 +22,58 @@ const (
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
) )
type LogEntry_StreamType int32
const (
LogEntry_UNKNOWN LogEntry_StreamType = 0
LogEntry_STDOUT LogEntry_StreamType = 1
LogEntry_STDERR LogEntry_StreamType = 2
LogEntry_HEARTBEAT LogEntry_StreamType = 3
)
// Enum value maps for LogEntry_StreamType.
var (
LogEntry_StreamType_name = map[int32]string{
0: "UNKNOWN",
1: "STDOUT",
2: "STDERR",
3: "HEARTBEAT",
}
LogEntry_StreamType_value = map[string]int32{
"UNKNOWN": 0,
"STDOUT": 1,
"STDERR": 2,
"HEARTBEAT": 3,
}
)
func (x LogEntry_StreamType) Enum() *LogEntry_StreamType {
p := new(LogEntry_StreamType)
*p = x
return p
}
func (x LogEntry_StreamType) String() string {
return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
}
func (LogEntry_StreamType) Descriptor() protoreflect.EnumDescriptor {
return file_internal_machine_api_pb_common_proto_enumTypes[0].Descriptor()
}
func (LogEntry_StreamType) Type() protoreflect.EnumType {
return &file_internal_machine_api_pb_common_proto_enumTypes[0]
}
func (x LogEntry_StreamType) Number() protoreflect.EnumNumber {
return protoreflect.EnumNumber(x)
}
// Deprecated: Use LogEntry_StreamType.Descriptor instead.
func (LogEntry_StreamType) EnumDescriptor() ([]byte, []int) {
return file_internal_machine_api_pb_common_proto_rawDescGZIP(), []int{7, 0}
}
// Common metadata message nested in all reply message types, injected by the gRPC proxy to provide information // Common metadata message nested in all reply message types, injected by the gRPC proxy to provide information
// about the machine that responded to the request. // about the machine that responded to the request.
type Metadata struct { type Metadata struct {
@@ -343,6 +396,150 @@ func (x *IPPrefix) GetBits() uint32 {
return 0 return 0
} }
type LogsRequest struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
// Options for logs retrieval.
Follow bool `protobuf:"varint,2,opt,name=follow,proto3" json:"follow,omitempty"`
Tail int32 `protobuf:"varint,3,opt,name=tail,proto3" json:"tail,omitempty"` // -1 means all
Since string `protobuf:"bytes,4,opt,name=since,proto3" json:"since,omitempty"` // https://www.rfc-editor.org/rfc/rfc3339.html timestamp or Go duration string
Until string `protobuf:"bytes,5,opt,name=until,proto3" json:"until,omitempty"` // https://www.rfc-editor.org/rfc/rfc3339.html timestamp or Go duration string
}
func (x *LogsRequest) Reset() {
*x = LogsRequest{}
if protoimpl.UnsafeEnabled {
mi := &file_internal_machine_api_pb_common_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *LogsRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*LogsRequest) ProtoMessage() {}
func (x *LogsRequest) ProtoReflect() protoreflect.Message {
mi := &file_internal_machine_api_pb_common_proto_msgTypes[6]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use LogsRequest.ProtoReflect.Descriptor instead.
func (*LogsRequest) Descriptor() ([]byte, []int) {
return file_internal_machine_api_pb_common_proto_rawDescGZIP(), []int{6}
}
func (x *LogsRequest) GetId() string {
if x != nil {
return x.Id
}
return ""
}
func (x *LogsRequest) GetFollow() bool {
if x != nil {
return x.Follow
}
return false
}
func (x *LogsRequest) GetTail() int32 {
if x != nil {
return x.Tail
}
return 0
}
func (x *LogsRequest) GetSince() string {
if x != nil {
return x.Since
}
return ""
}
func (x *LogsRequest) GetUntil() string {
if x != nil {
return x.Until
}
return ""
}
type LogEntry struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Stream LogEntry_StreamType `protobuf:"varint,1,opt,name=stream,proto3,enum=api.LogEntry_StreamType" json:"stream,omitempty"`
Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
// Log line content. Empty for heartbeat entries.
Message []byte `protobuf:"bytes,3,opt,name=message,proto3" json:"message,omitempty"`
}
func (x *LogEntry) Reset() {
*x = LogEntry{}
if protoimpl.UnsafeEnabled {
mi := &file_internal_machine_api_pb_common_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *LogEntry) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*LogEntry) ProtoMessage() {}
func (x *LogEntry) ProtoReflect() protoreflect.Message {
mi := &file_internal_machine_api_pb_common_proto_msgTypes[7]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use LogEntry.ProtoReflect.Descriptor instead.
func (*LogEntry) Descriptor() ([]byte, []int) {
return file_internal_machine_api_pb_common_proto_rawDescGZIP(), []int{7}
}
func (x *LogEntry) GetStream() LogEntry_StreamType {
if x != nil {
return x.Stream
}
return LogEntry_UNKNOWN
}
func (x *LogEntry) GetTimestamp() *timestamppb.Timestamp {
if x != nil {
return x.Timestamp
}
return nil
}
func (x *LogEntry) GetMessage() []byte {
if x != nil {
return x.Message
}
return nil
}
var File_internal_machine_api_pb_common_proto protoreflect.FileDescriptor var File_internal_machine_api_pb_common_proto protoreflect.FileDescriptor
var file_internal_machine_api_pb_common_proto_rawDesc = []byte{ var file_internal_machine_api_pb_common_proto_rawDesc = []byte{
@@ -350,32 +547,55 @@ var file_internal_machine_api_pb_common_proto_rawDesc = []byte{
0x6e, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x62, 0x2f, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x6e, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x62, 0x2f, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e,
0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x03, 0x61, 0x70, 0x69, 0x1a, 0x17, 0x67, 0x6f, 0x6f, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x03, 0x61, 0x70, 0x69, 0x1a, 0x17, 0x67, 0x6f, 0x6f,
0x67, 0x6c, 0x65, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x2e, 0x70, 0x67, 0x6c, 0x65, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x2e, 0x70,
0x72, 0x6f, 0x74, 0x6f, 0x22, 0x66, 0x0a, 0x08, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f,
0x12, 0x18, 0x0a, 0x07, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e,
0x09, 0x52, 0x07, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x66, 0x0a, 0x08, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74,
0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x61, 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x18, 0x01, 0x20, 0x01,
0x12, 0x2a, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x28, 0x09, 0x52, 0x07, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65,
0x32, 0x12, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x72, 0x70, 0x63, 0x2e, 0x53, 0x74, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f,
0x61, 0x74, 0x75, 0x73, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x32, 0x0a, 0x05, 0x72, 0x12, 0x2a, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28,
0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x29, 0x0a, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x0b, 0x32, 0x12, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x72, 0x70, 0x63, 0x2e, 0x53,
0x61, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4d, 0x65, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x32, 0x0a,
0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x05, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x29, 0x0a, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61,
0x22, 0x37, 0x0a, 0x0d, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x74, 0x61, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4d,
0x65, 0x12, 0x26, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x18, 0x01, 0x20, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74,
0x03, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x52, 0x61, 0x22, 0x37, 0x0a, 0x0d, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x14, 0x0a, 0x02, 0x49, 0x50, 0x12, 0x73, 0x65, 0x12, 0x26, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x18, 0x01,
0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x02, 0x69, 0x70, 0x22, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79,
0x35, 0x0a, 0x06, 0x49, 0x50, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x52, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x14, 0x0a, 0x02, 0x49, 0x50,
0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52, 0x02, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x02, 0x69, 0x70,
0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x22, 0x35, 0x0a, 0x06, 0x49, 0x50, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70,
0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x22, 0x37, 0x0a, 0x08, 0x49, 0x50, 0x50, 0x72, 0x65, 0x66, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52,
0x69, 0x78, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28,
0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52, 0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x62, 0x0d, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x22, 0x37, 0x0a, 0x08, 0x49, 0x50, 0x50, 0x72, 0x65,
0x69, 0x74, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x62, 0x69, 0x74, 0x73, 0x42, 0x66, 0x69, 0x78, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32,
0x37, 0x5a, 0x35, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x07, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52, 0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04,
0x76, 0x69, 0x64, 0x65, 0x72, 0x73, 0x6b, 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x62, 0x69, 0x74, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x62, 0x69, 0x74, 0x73,
0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x22, 0x75, 0x0a, 0x0b, 0x4c, 0x6f, 0x67, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12,
0x16, 0x0a, 0x06, 0x66, 0x6f, 0x6c, 0x6c, 0x6f, 0x77, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52,
0x06, 0x66, 0x6f, 0x6c, 0x6c, 0x6f, 0x77, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x61, 0x69, 0x6c, 0x18,
0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x04, 0x74, 0x61, 0x69, 0x6c, 0x12, 0x14, 0x0a, 0x05, 0x73,
0x69, 0x6e, 0x63, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x73, 0x69, 0x6e, 0x63,
0x65, 0x12, 0x14, 0x0a, 0x05, 0x75, 0x6e, 0x74, 0x69, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09,
0x52, 0x05, 0x75, 0x6e, 0x74, 0x69, 0x6c, 0x22, 0xd2, 0x01, 0x0a, 0x08, 0x4c, 0x6f, 0x67, 0x45,
0x6e, 0x74, 0x72, 0x79, 0x12, 0x30, 0x0a, 0x06, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x18, 0x01,
0x20, 0x01, 0x28, 0x0e, 0x32, 0x18, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4c, 0x6f, 0x67, 0x45, 0x6e,
0x74, 0x72, 0x79, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, 0x79, 0x70, 0x65, 0x52, 0x06,
0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74,
0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67,
0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65,
0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
0x0c, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x40, 0x0a, 0x0a, 0x53, 0x74,
0x72, 0x65, 0x61, 0x6d, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x4e, 0x4b, 0x4e,
0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x54, 0x44, 0x4f, 0x55, 0x54, 0x10,
0x01, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x54, 0x44, 0x45, 0x52, 0x52, 0x10, 0x02, 0x12, 0x0d, 0x0a,
0x09, 0x48, 0x45, 0x41, 0x52, 0x54, 0x42, 0x45, 0x41, 0x54, 0x10, 0x03, 0x42, 0x37, 0x5a, 0x35,
0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x76, 0x69, 0x64,
0x65, 0x72, 0x73, 0x6b, 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x2f, 0x69, 0x6e,
0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x2f, 0x61,
0x70, 0x69, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
} }
var ( var (
@@ -390,27 +610,34 @@ func file_internal_machine_api_pb_common_proto_rawDescGZIP() []byte {
return file_internal_machine_api_pb_common_proto_rawDescData return file_internal_machine_api_pb_common_proto_rawDescData
} }
var file_internal_machine_api_pb_common_proto_msgTypes = make([]protoimpl.MessageInfo, 6) var file_internal_machine_api_pb_common_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
var file_internal_machine_api_pb_common_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
var file_internal_machine_api_pb_common_proto_goTypes = []any{ var file_internal_machine_api_pb_common_proto_goTypes = []any{
(*Metadata)(nil), // 0: api.Metadata (LogEntry_StreamType)(0), // 0: api.LogEntry.StreamType
(*Empty)(nil), // 1: api.Empty (*Metadata)(nil), // 1: api.Metadata
(*EmptyResponse)(nil), // 2: api.EmptyResponse (*Empty)(nil), // 2: api.Empty
(*IP)(nil), // 3: api.IP (*EmptyResponse)(nil), // 3: api.EmptyResponse
(*IPPort)(nil), // 4: api.IPPort (*IP)(nil), // 4: api.IP
(*IPPrefix)(nil), // 5: api.IPPrefix (*IPPort)(nil), // 5: api.IPPort
(*status.Status)(nil), // 6: google.rpc.Status (*IPPrefix)(nil), // 6: api.IPPrefix
(*LogsRequest)(nil), // 7: api.LogsRequest
(*LogEntry)(nil), // 8: api.LogEntry
(*status.Status)(nil), // 9: google.rpc.Status
(*timestamppb.Timestamp)(nil), // 10: google.protobuf.Timestamp
} }
var file_internal_machine_api_pb_common_proto_depIdxs = []int32{ var file_internal_machine_api_pb_common_proto_depIdxs = []int32{
6, // 0: api.Metadata.status:type_name -> google.rpc.Status 9, // 0: api.Metadata.status:type_name -> google.rpc.Status
0, // 1: api.Empty.metadata:type_name -> api.Metadata 1, // 1: api.Empty.metadata:type_name -> api.Metadata
1, // 2: api.EmptyResponse.messages:type_name -> api.Empty 2, // 2: api.EmptyResponse.messages:type_name -> api.Empty
3, // 3: api.IPPort.ip:type_name -> api.IP 4, // 3: api.IPPort.ip:type_name -> api.IP
3, // 4: api.IPPrefix.ip:type_name -> api.IP 4, // 4: api.IPPrefix.ip:type_name -> api.IP
5, // [5:5] is the sub-list for method output_type 0, // 5: api.LogEntry.stream:type_name -> api.LogEntry.StreamType
5, // [5:5] is the sub-list for method input_type 10, // 6: api.LogEntry.timestamp:type_name -> google.protobuf.Timestamp
5, // [5:5] is the sub-list for extension type_name 7, // [7:7] is the sub-list for method output_type
5, // [5:5] is the sub-list for extension extendee 7, // [7:7] is the sub-list for method input_type
0, // [0:5] is the sub-list for field type_name 7, // [7:7] is the sub-list for extension type_name
7, // [7:7] is the sub-list for extension extendee
0, // [0:7] is the sub-list for field type_name
} }
func init() { file_internal_machine_api_pb_common_proto_init() } func init() { file_internal_machine_api_pb_common_proto_init() }
@@ -491,19 +718,44 @@ func file_internal_machine_api_pb_common_proto_init() {
return nil return nil
} }
} }
file_internal_machine_api_pb_common_proto_msgTypes[6].Exporter = func(v any, i int) any {
switch v := v.(*LogsRequest); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_internal_machine_api_pb_common_proto_msgTypes[7].Exporter = func(v any, i int) any {
switch v := v.(*LogEntry); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
} }
type x struct{} type x struct{}
out := protoimpl.TypeBuilder{ out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{ File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(), GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_internal_machine_api_pb_common_proto_rawDesc, RawDescriptor: file_internal_machine_api_pb_common_proto_rawDesc,
NumEnums: 0, NumEnums: 1,
NumMessages: 6, NumMessages: 8,
NumExtensions: 0, NumExtensions: 0,
NumServices: 0, NumServices: 0,
}, },
GoTypes: file_internal_machine_api_pb_common_proto_goTypes, GoTypes: file_internal_machine_api_pb_common_proto_goTypes,
DependencyIndexes: file_internal_machine_api_pb_common_proto_depIdxs, DependencyIndexes: file_internal_machine_api_pb_common_proto_depIdxs,
EnumInfos: file_internal_machine_api_pb_common_proto_enumTypes,
MessageInfos: file_internal_machine_api_pb_common_proto_msgTypes, MessageInfos: file_internal_machine_api_pb_common_proto_msgTypes,
}.Build() }.Build()
File_internal_machine_api_pb_common_proto = out.File File_internal_machine_api_pb_common_proto = out.File
+23
View File
@@ -6,6 +6,7 @@ option go_package = "github.com/psviderski/uncloud/internal/machine/api/pb";
// Vendored at internal/machine/api/vendor/google/rpc/status.proto. // Vendored at internal/machine/api/vendor/google/rpc/status.proto.
import "google/rpc/status.proto"; import "google/rpc/status.proto";
import "google/protobuf/timestamp.proto";
// Common metadata message nested in all reply message types, injected by the gRPC proxy to provide information // Common metadata message nested in all reply message types, injected by the gRPC proxy to provide information
// about the machine that responded to the request. // about the machine that responded to the request.
@@ -42,3 +43,25 @@ message IPPrefix {
IP ip = 1; IP ip = 1;
uint32 bits = 2; uint32 bits = 2;
} }
message LogsRequest {
string id = 1;
// Options for logs retrieval.
bool follow = 2;
int32 tail = 3; // -1 means all
string since = 4; // https://www.rfc-editor.org/rfc/rfc3339.html timestamp or Go duration string
string until = 5; // https://www.rfc-editor.org/rfc/rfc3339.html timestamp or Go duration string
}
message LogEntry {
enum StreamType {
UNKNOWN = 0;
STDOUT = 1;
STDERR = 2;
HEARTBEAT = 3;
}
StreamType stream = 1;
google.protobuf.Timestamp timestamp = 2;
// Log line content. Empty for heartbeat entries.
bytes message = 3;
}
File diff suppressed because it is too large Load Diff
+12 -24
View File
@@ -5,7 +5,6 @@ package api;
option go_package = "github.com/psviderski/uncloud/internal/machine/api/pb"; option go_package = "github.com/psviderski/uncloud/internal/machine/api/pb";
import "google/protobuf/empty.proto"; import "google/protobuf/empty.proto";
import "google/protobuf/timestamp.proto";
import "internal/machine/api/pb/common.proto"; import "internal/machine/api/pb/common.proto";
service Docker { service Docker {
@@ -17,7 +16,7 @@ service Docker {
rpc RemoveContainer(RemoveContainerRequest) returns (google.protobuf.Empty); rpc RemoveContainer(RemoveContainerRequest) returns (google.protobuf.Empty);
rpc ExecContainer(stream ExecContainerRequest) returns (stream ExecContainerResponse); rpc ExecContainer(stream ExecContainerRequest) returns (stream ExecContainerResponse);
rpc ContainerLogs(ContainerLogsRequest) returns (stream ContainerLogEntry); rpc ContainerLogs(LogsRequest) returns (stream LogEntry);
rpc PullImage(PullImageRequest) returns (stream JSONMessage); rpc PullImage(PullImageRequest) returns (stream JSONMessage);
rpc InspectImage(InspectImageRequest) returns (InspectImageResponse); rpc InspectImage(InspectImageRequest) returns (InspectImageResponse);
@@ -132,28 +131,6 @@ message ExecContainerResponse {
} }
} }
message ContainerLogsRequest {
string container_id = 1;
// Options for logs retrieval.
bool follow = 2;
int32 tail = 3; // -1 means all
string since = 4; // https://www.rfc-editor.org/rfc/rfc3339.html timestamp or Go duration string
string until = 5; // https://www.rfc-editor.org/rfc/rfc3339.html timestamp or Go duration string
}
message ContainerLogEntry {
enum StreamType {
UNKNOWN = 0;
STDOUT = 1;
STDERR = 2;
HEARTBEAT = 3;
}
StreamType stream = 1;
google.protobuf.Timestamp timestamp = 2;
// Log line content. Empty for heartbeat entries.
bytes message = 3;
}
message PullImageRequest { message PullImageRequest {
string image = 1; string image = 1;
// JSON serialised image.PullOptions. // JSON serialised image.PullOptions.
@@ -251,6 +228,15 @@ message CreateServiceContainerRequest {
// JSON serialised api.ServiceSpec. // JSON serialised api.ServiceSpec.
bytes service_spec = 2; bytes service_spec = 2;
string container_name = 3; string container_name = 3;
// ContainerType specifies the purpose of the container being created.
enum ContainerType {
// SERVICE is the default type for long-running service containers.
SERVICE = 0;
// PRE_DEPLOY is a one-shot container for a pre-deploy hook to run before service deployment.
PRE_DEPLOY = 1;
}
ContainerType container_type = 4;
} }
message ServiceContainer { message ServiceContainer {
@@ -275,4 +261,6 @@ message ListServiceContainersResponse {
message MachineServiceContainers { message MachineServiceContainers {
Metadata metadata = 1; Metadata metadata = 1;
repeated ServiceContainer containers = 2; repeated ServiceContainer containers = 2;
// One-shot containers for deployment hooks (e.g. pre-deploy).
repeated ServiceContainer hook_containers = 3;
} }
+9 -9
View File
@@ -52,7 +52,7 @@ type DockerClient interface {
ListContainers(ctx context.Context, in *ListContainersRequest, opts ...grpc.CallOption) (*ListContainersResponse, error) ListContainers(ctx context.Context, in *ListContainersRequest, opts ...grpc.CallOption) (*ListContainersResponse, error)
RemoveContainer(ctx context.Context, in *RemoveContainerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) RemoveContainer(ctx context.Context, in *RemoveContainerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
ExecContainer(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[ExecContainerRequest, ExecContainerResponse], error) ExecContainer(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[ExecContainerRequest, ExecContainerResponse], error)
ContainerLogs(ctx context.Context, in *ContainerLogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ContainerLogEntry], error) ContainerLogs(ctx context.Context, in *LogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[LogEntry], error)
PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JSONMessage], error) PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JSONMessage], error)
InspectImage(ctx context.Context, in *InspectImageRequest, opts ...grpc.CallOption) (*InspectImageResponse, error) InspectImage(ctx context.Context, in *InspectImageRequest, opts ...grpc.CallOption) (*InspectImageResponse, error)
// InspectRemoteImage returns the image metadata for an image in a remote registry using the machine's // InspectRemoteImage returns the image metadata for an image in a remote registry using the machine's
@@ -149,13 +149,13 @@ func (c *dockerClient) ExecContainer(ctx context.Context, opts ...grpc.CallOptio
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type Docker_ExecContainerClient = grpc.BidiStreamingClient[ExecContainerRequest, ExecContainerResponse] type Docker_ExecContainerClient = grpc.BidiStreamingClient[ExecContainerRequest, ExecContainerResponse]
func (c *dockerClient) ContainerLogs(ctx context.Context, in *ContainerLogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ContainerLogEntry], error) { func (c *dockerClient) ContainerLogs(ctx context.Context, in *LogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[LogEntry], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &Docker_ServiceDesc.Streams[1], Docker_ContainerLogs_FullMethodName, cOpts...) stream, err := c.cc.NewStream(ctx, &Docker_ServiceDesc.Streams[1], Docker_ContainerLogs_FullMethodName, cOpts...)
if err != nil { if err != nil {
return nil, err return nil, err
} }
x := &grpc.GenericClientStream[ContainerLogsRequest, ContainerLogEntry]{ClientStream: stream} x := &grpc.GenericClientStream[LogsRequest, LogEntry]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil { if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err return nil, err
} }
@@ -166,7 +166,7 @@ func (c *dockerClient) ContainerLogs(ctx context.Context, in *ContainerLogsReque
} }
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type Docker_ContainerLogsClient = grpc.ServerStreamingClient[ContainerLogEntry] type Docker_ContainerLogsClient = grpc.ServerStreamingClient[LogEntry]
func (c *dockerClient) PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JSONMessage], error) { func (c *dockerClient) PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JSONMessage], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
@@ -298,7 +298,7 @@ type DockerServer interface {
ListContainers(context.Context, *ListContainersRequest) (*ListContainersResponse, error) ListContainers(context.Context, *ListContainersRequest) (*ListContainersResponse, error)
RemoveContainer(context.Context, *RemoveContainerRequest) (*emptypb.Empty, error) RemoveContainer(context.Context, *RemoveContainerRequest) (*emptypb.Empty, error)
ExecContainer(grpc.BidiStreamingServer[ExecContainerRequest, ExecContainerResponse]) error ExecContainer(grpc.BidiStreamingServer[ExecContainerRequest, ExecContainerResponse]) error
ContainerLogs(*ContainerLogsRequest, grpc.ServerStreamingServer[ContainerLogEntry]) error ContainerLogs(*LogsRequest, grpc.ServerStreamingServer[LogEntry]) error
PullImage(*PullImageRequest, grpc.ServerStreamingServer[JSONMessage]) error PullImage(*PullImageRequest, grpc.ServerStreamingServer[JSONMessage]) error
InspectImage(context.Context, *InspectImageRequest) (*InspectImageResponse, error) InspectImage(context.Context, *InspectImageRequest) (*InspectImageResponse, error)
// InspectRemoteImage returns the image metadata for an image in a remote registry using the machine's // InspectRemoteImage returns the image metadata for an image in a remote registry using the machine's
@@ -343,7 +343,7 @@ func (UnimplementedDockerServer) RemoveContainer(context.Context, *RemoveContain
func (UnimplementedDockerServer) ExecContainer(grpc.BidiStreamingServer[ExecContainerRequest, ExecContainerResponse]) error { func (UnimplementedDockerServer) ExecContainer(grpc.BidiStreamingServer[ExecContainerRequest, ExecContainerResponse]) error {
return status.Errorf(codes.Unimplemented, "method ExecContainer not implemented") return status.Errorf(codes.Unimplemented, "method ExecContainer not implemented")
} }
func (UnimplementedDockerServer) ContainerLogs(*ContainerLogsRequest, grpc.ServerStreamingServer[ContainerLogEntry]) error { func (UnimplementedDockerServer) ContainerLogs(*LogsRequest, grpc.ServerStreamingServer[LogEntry]) error {
return status.Errorf(codes.Unimplemented, "method ContainerLogs not implemented") return status.Errorf(codes.Unimplemented, "method ContainerLogs not implemented")
} }
func (UnimplementedDockerServer) PullImage(*PullImageRequest, grpc.ServerStreamingServer[JSONMessage]) error { func (UnimplementedDockerServer) PullImage(*PullImageRequest, grpc.ServerStreamingServer[JSONMessage]) error {
@@ -516,15 +516,15 @@ func _Docker_ExecContainer_Handler(srv interface{}, stream grpc.ServerStream) er
type Docker_ExecContainerServer = grpc.BidiStreamingServer[ExecContainerRequest, ExecContainerResponse] type Docker_ExecContainerServer = grpc.BidiStreamingServer[ExecContainerRequest, ExecContainerResponse]
func _Docker_ContainerLogs_Handler(srv interface{}, stream grpc.ServerStream) error { func _Docker_ContainerLogs_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(ContainerLogsRequest) m := new(LogsRequest)
if err := stream.RecvMsg(m); err != nil { if err := stream.RecvMsg(m); err != nil {
return err return err
} }
return srv.(DockerServer).ContainerLogs(m, &grpc.GenericServerStream[ContainerLogsRequest, ContainerLogEntry]{ServerStream: stream}) return srv.(DockerServer).ContainerLogs(m, &grpc.GenericServerStream[LogsRequest, LogEntry]{ServerStream: stream})
} }
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type Docker_ContainerLogsServer = grpc.ServerStreamingServer[ContainerLogEntry] type Docker_ContainerLogsServer = grpc.ServerStreamingServer[LogEntry]
func _Docker_PullImage_Handler(srv interface{}, stream grpc.ServerStream) error { func _Docker_PullImage_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(PullImageRequest) m := new(PullImageRequest)
+24 -17
View File
@@ -1146,7 +1146,7 @@ var file_internal_machine_api_pb_machine_proto_rawDesc = []byte{
0x03, 0x52, 0x0d, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x6d, 0x69, 0x74, 0x42, 0x79, 0x74, 0x65, 0x73, 0x03, 0x52, 0x0d, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x6d, 0x69, 0x74, 0x42, 0x79, 0x74, 0x65, 0x73,
0x12, 0x1f, 0x0a, 0x0b, 0x61, 0x6c, 0x6c, 0x6f, 0x77, 0x65, 0x64, 0x5f, 0x69, 0x70, 0x73, 0x18, 0x12, 0x1f, 0x0a, 0x0b, 0x61, 0x6c, 0x6c, 0x6f, 0x77, 0x65, 0x64, 0x5f, 0x69, 0x70, 0x73, 0x18,
0x06, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0a, 0x61, 0x6c, 0x6c, 0x6f, 0x77, 0x65, 0x64, 0x49, 0x70, 0x06, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0a, 0x61, 0x6c, 0x6c, 0x6f, 0x77, 0x65, 0x64, 0x49, 0x70,
0x73, 0x32, 0xe3, 0x04, 0x0a, 0x07, 0x4d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x12, 0x4d, 0x0a, 0x73, 0x32, 0x95, 0x05, 0x0a, 0x07, 0x4d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x12, 0x4d, 0x0a,
0x12, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x50, 0x72, 0x65, 0x72, 0x65, 0x71, 0x75, 0x69, 0x73, 0x69, 0x12, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x50, 0x72, 0x65, 0x72, 0x65, 0x71, 0x75, 0x69, 0x73, 0x69,
0x74, 0x65, 0x73, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x65, 0x73, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x1f, 0x2e, 0x61, 0x70, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x1f, 0x2e, 0x61, 0x70,
@@ -1184,11 +1184,14 @@ var file_internal_machine_api_pb_machine_proto_rawDesc = []byte{
0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x6e, 0x73, 0x70, 0x65, 0x63, 0x74, 0x53, 0x65, 0x72, 0x76, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x6e, 0x73, 0x70, 0x65, 0x63, 0x74, 0x53, 0x65, 0x72, 0x76,
0x69, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, 0x61, 0x70, 0x69, 0x69, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, 0x61, 0x70, 0x69,
0x2e, 0x49, 0x6e, 0x73, 0x70, 0x65, 0x63, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x52, 0x2e, 0x49, 0x6e, 0x73, 0x70, 0x65, 0x63, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x52,
0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x37, 0x5a, 0x35, 0x67, 0x69, 0x74, 0x68, 0x75, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x30, 0x0a, 0x0b, 0x4d, 0x61, 0x63, 0x68, 0x69,
0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x76, 0x69, 0x64, 0x65, 0x72, 0x73, 0x6b, 0x69, 0x6e, 0x65, 0x4c, 0x6f, 0x67, 0x73, 0x12, 0x10, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4c, 0x6f, 0x67,
0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0d, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4c,
0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x62, 0x6f, 0x67, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x30, 0x01, 0x42, 0x37, 0x5a, 0x35, 0x67, 0x69, 0x74,
0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x76, 0x69, 0x64, 0x65, 0x72, 0x73,
0x6b, 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72,
0x6e, 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f,
0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
} }
var ( var (
@@ -1227,6 +1230,8 @@ var file_internal_machine_api_pb_machine_proto_goTypes = []any{
(*Metadata)(nil), // 19: api.Metadata (*Metadata)(nil), // 19: api.Metadata
(*timestamppb.Timestamp)(nil), // 20: google.protobuf.Timestamp (*timestamppb.Timestamp)(nil), // 20: google.protobuf.Timestamp
(*emptypb.Empty)(nil), // 21: google.protobuf.Empty (*emptypb.Empty)(nil), // 21: google.protobuf.Empty
(*LogsRequest)(nil), // 22: api.LogsRequest
(*LogEntry)(nil), // 23: api.LogEntry
} }
var file_internal_machine_api_pb_machine_proto_depIdxs = []int32{ var file_internal_machine_api_pb_machine_proto_depIdxs = []int32{
1, // 0: api.MachineInfo.network:type_name -> api.NetworkConfig 1, // 0: api.MachineInfo.network:type_name -> api.NetworkConfig
@@ -1256,17 +1261,19 @@ var file_internal_machine_api_pb_machine_proto_depIdxs = []int32{
21, // 24: api.Machine.InspectWireGuardNetwork:input_type -> google.protobuf.Empty 21, // 24: api.Machine.InspectWireGuardNetwork:input_type -> google.protobuf.Empty
9, // 25: api.Machine.Reset:input_type -> api.ResetRequest 9, // 25: api.Machine.Reset:input_type -> api.ResetRequest
11, // 26: api.Machine.InspectService:input_type -> api.InspectServiceRequest 11, // 26: api.Machine.InspectService:input_type -> api.InspectServiceRequest
2, // 27: api.Machine.CheckPrerequisites:output_type -> api.CheckPrerequisitesResponse 22, // 27: api.Machine.MachineLogs:input_type -> api.LogsRequest
4, // 28: api.Machine.InitCluster:output_type -> api.InitClusterResponse 2, // 28: api.Machine.CheckPrerequisites:output_type -> api.CheckPrerequisitesResponse
21, // 29: api.Machine.JoinCluster:output_type -> google.protobuf.Empty 4, // 29: api.Machine.InitCluster:output_type -> api.InitClusterResponse
8, // 30: api.Machine.Token:output_type -> api.TokenResponse 21, // 30: api.Machine.JoinCluster:output_type -> google.protobuf.Empty
0, // 31: api.Machine.Inspect:output_type -> api.MachineInfo 8, // 31: api.Machine.Token:output_type -> api.TokenResponse
6, // 32: api.Machine.InspectMachine:output_type -> api.InspectMachineResponse 0, // 32: api.Machine.Inspect:output_type -> api.MachineInfo
13, // 33: api.Machine.InspectWireGuardNetwork:output_type -> api.InspectWireGuardNetworkResponse 6, // 33: api.Machine.InspectMachine:output_type -> api.InspectMachineResponse
21, // 34: api.Machine.Reset:output_type -> google.protobuf.Empty 13, // 34: api.Machine.InspectWireGuardNetwork:output_type -> api.InspectWireGuardNetworkResponse
12, // 35: api.Machine.InspectService:output_type -> api.InspectServiceResponse 21, // 35: api.Machine.Reset:output_type -> google.protobuf.Empty
27, // [27:36] is the sub-list for method output_type 12, // 36: api.Machine.InspectService:output_type -> api.InspectServiceResponse
18, // [18:27] is the sub-list for method input_type 23, // 37: api.Machine.MachineLogs:output_type -> api.LogEntry
28, // [28:38] is the sub-list for method output_type
18, // [18:28] is the sub-list for method input_type
18, // [18:18] is the sub-list for extension type_name 18, // [18:18] is the sub-list for extension type_name
18, // [18:18] is the sub-list for extension extendee 18, // [18:18] is the sub-list for extension extendee
0, // [0:18] is the sub-list for field type_name 0, // [0:18] is the sub-list for field type_name
+2
View File
@@ -24,6 +24,8 @@ service Machine {
rpc Reset(ResetRequest) returns (google.protobuf.Empty); rpc Reset(ResetRequest) returns (google.protobuf.Empty);
rpc InspectService(InspectServiceRequest) returns (InspectServiceResponse); rpc InspectService(InspectServiceRequest) returns (InspectServiceResponse);
rpc MachineLogs(LogsRequest) returns (stream LogEntry);
} }
message MachineInfo { message MachineInfo {
+43 -1
View File
@@ -29,6 +29,7 @@ const (
Machine_InspectWireGuardNetwork_FullMethodName = "/api.Machine/InspectWireGuardNetwork" Machine_InspectWireGuardNetwork_FullMethodName = "/api.Machine/InspectWireGuardNetwork"
Machine_Reset_FullMethodName = "/api.Machine/Reset" Machine_Reset_FullMethodName = "/api.Machine/Reset"
Machine_InspectService_FullMethodName = "/api.Machine/InspectService" Machine_InspectService_FullMethodName = "/api.Machine/InspectService"
Machine_MachineLogs_FullMethodName = "/api.Machine/MachineLogs"
) )
// MachineClient is the client API for Machine service. // MachineClient is the client API for Machine service.
@@ -49,6 +50,7 @@ type MachineClient interface {
// Reset restores the machine to a clean state, removing all cluster-related configuration and data. // Reset restores the machine to a clean state, removing all cluster-related configuration and data.
Reset(ctx context.Context, in *ResetRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) Reset(ctx context.Context, in *ResetRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
InspectService(ctx context.Context, in *InspectServiceRequest, opts ...grpc.CallOption) (*InspectServiceResponse, error) InspectService(ctx context.Context, in *InspectServiceRequest, opts ...grpc.CallOption) (*InspectServiceResponse, error)
MachineLogs(ctx context.Context, in *LogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[LogEntry], error)
} }
type machineClient struct { type machineClient struct {
@@ -149,6 +151,25 @@ func (c *machineClient) InspectService(ctx context.Context, in *InspectServiceRe
return out, nil return out, nil
} }
func (c *machineClient) MachineLogs(ctx context.Context, in *LogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[LogEntry], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &Machine_ServiceDesc.Streams[0], Machine_MachineLogs_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[LogsRequest, LogEntry]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type Machine_MachineLogsClient = grpc.ServerStreamingClient[LogEntry]
// MachineServer is the server API for Machine service. // MachineServer is the server API for Machine service.
// All implementations must embed UnimplementedMachineServer // All implementations must embed UnimplementedMachineServer
// for forward compatibility. // for forward compatibility.
@@ -167,6 +188,7 @@ type MachineServer interface {
// Reset restores the machine to a clean state, removing all cluster-related configuration and data. // Reset restores the machine to a clean state, removing all cluster-related configuration and data.
Reset(context.Context, *ResetRequest) (*emptypb.Empty, error) Reset(context.Context, *ResetRequest) (*emptypb.Empty, error)
InspectService(context.Context, *InspectServiceRequest) (*InspectServiceResponse, error) InspectService(context.Context, *InspectServiceRequest) (*InspectServiceResponse, error)
MachineLogs(*LogsRequest, grpc.ServerStreamingServer[LogEntry]) error
mustEmbedUnimplementedMachineServer() mustEmbedUnimplementedMachineServer()
} }
@@ -204,6 +226,9 @@ func (UnimplementedMachineServer) Reset(context.Context, *ResetRequest) (*emptyp
func (UnimplementedMachineServer) InspectService(context.Context, *InspectServiceRequest) (*InspectServiceResponse, error) { func (UnimplementedMachineServer) InspectService(context.Context, *InspectServiceRequest) (*InspectServiceResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method InspectService not implemented") return nil, status.Errorf(codes.Unimplemented, "method InspectService not implemented")
} }
func (UnimplementedMachineServer) MachineLogs(*LogsRequest, grpc.ServerStreamingServer[LogEntry]) error {
return status.Errorf(codes.Unimplemented, "method MachineLogs not implemented")
}
func (UnimplementedMachineServer) mustEmbedUnimplementedMachineServer() {} func (UnimplementedMachineServer) mustEmbedUnimplementedMachineServer() {}
func (UnimplementedMachineServer) testEmbeddedByValue() {} func (UnimplementedMachineServer) testEmbeddedByValue() {}
@@ -387,6 +412,17 @@ func _Machine_InspectService_Handler(srv interface{}, ctx context.Context, dec f
return interceptor(ctx, in, info, handler) return interceptor(ctx, in, info, handler)
} }
func _Machine_MachineLogs_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(LogsRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(MachineServer).MachineLogs(m, &grpc.GenericServerStream[LogsRequest, LogEntry]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type Machine_MachineLogsServer = grpc.ServerStreamingServer[LogEntry]
// Machine_ServiceDesc is the grpc.ServiceDesc for Machine service. // Machine_ServiceDesc is the grpc.ServiceDesc for Machine service.
// It's only intended for direct use with grpc.RegisterService, // It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy) // and not to be introspected or modified (even as a copy)
@@ -431,6 +467,12 @@ var Machine_ServiceDesc = grpc.ServiceDesc{
Handler: _Machine_InspectService_Handler, Handler: _Machine_InspectService_Handler,
}, },
}, },
Streams: []grpc.StreamDesc{}, Streams: []grpc.StreamDesc{
{
StreamName: "MachineLogs",
Handler: _Machine_MachineLogs_Handler,
ServerStreams: true,
},
},
Metadata: "internal/machine/api/pb/machine.proto", Metadata: "internal/machine/api/pb/machine.proto",
} }
+4 -1
View File
@@ -94,13 +94,16 @@ func (c *Controller) Run(ctx context.Context) error {
} }
} }
// filterHealthyContainers filters out containers that are not healthy. // filterHealthyContainers filters out unhealthy and hook containers.
// TODO: Filters out containers from this machine that are likely unavailable. The availability can be determined // TODO: Filters out containers from this machine that are likely unavailable. The availability can be determined
// by the cluster membership state of the machine that the container is running on. Implement machine membership // by the cluster membership state of the machine that the container is running on. Implement machine membership
// check using Corrossion Admin client. // check using Corrossion Admin client.
func filterHealthyContainers(containers []store.ContainerRecord) []store.ContainerRecord { func filterHealthyContainers(containers []store.ContainerRecord) []store.ContainerRecord {
healthy := make([]store.ContainerRecord, 0, len(containers)) healthy := make([]store.ContainerRecord, 0, len(containers))
for _, cr := range containers { for _, cr := range containers {
if cr.Container.IsHook() {
continue
}
if cr.Container.Healthy() { if cr.Container.Healthy() {
healthy = append(healthy, cr) healthy = append(healthy, cr)
} }
+3
View File
@@ -72,6 +72,9 @@ func (r *ClusterResolver) updateServiceIPs(containers []store.ContainerRecord) {
containersCount := 0 containersCount := 0
for _, record := range containers { for _, record := range containers {
if record.Container.IsHook() {
continue
}
if !record.Container.Healthy() { if !record.Container.Healthy() {
continue continue
} }
+24 -38
View File
@@ -420,34 +420,6 @@ func (c *Client) RemoveVolume(ctx context.Context, id string, force bool) error
return err return err
} }
// CreateServiceContainer creates a new container for the service with the given specifications.
func (c *Client) CreateServiceContainer(
ctx context.Context, serviceID string, spec api.ServiceSpec, containerName string,
) (container.CreateResponse, error) {
var resp container.CreateResponse
specBytes, err := json.Marshal(spec)
if err != nil {
return resp, fmt.Errorf("marshal service spec: %w", err)
}
grpcResp, err := c.GRPCClient.CreateServiceContainer(ctx, &pb.CreateServiceContainerRequest{
ServiceId: serviceID,
ServiceSpec: specBytes,
ContainerName: containerName,
})
if err != nil {
if status.Convert(err).Code() == codes.NotFound {
return resp, errdefs.NotFound(err)
}
return resp, err
}
if err = json.Unmarshal(grpcResp.Response, &resp); err != nil {
return resp, fmt.Errorf("unmarshal gRPC response: %w", err)
}
return resp, nil
}
// InspectServiceContainer returns the container information and service specification that was used to create the // InspectServiceContainer returns the container information and service specification that was used to create the
// container with the given ID. // container with the given ID.
func (c *Client) InspectServiceContainer(ctx context.Context, id string) (api.ServiceContainer, error) { func (c *Client) InspectServiceContainer(ctx context.Context, id string) (api.ServiceContainer, error) {
@@ -474,10 +446,13 @@ func (c *Client) InspectServiceContainer(ctx context.Context, id string) (api.Se
type MachineServiceContainers struct { type MachineServiceContainers struct {
Metadata *pb.Metadata Metadata *pb.Metadata
Containers []api.ServiceContainer Containers []api.ServiceContainer
// HookContainers are one-shot containers for deployment hooks (e.g. pre-deploy).
HookContainers []api.ServiceContainer
} }
// ListServiceContainers returns all containers on requested machines that belong to the service with the given // ListServiceContainers returns all containers on requested machines that belong to the service with the given
// name or ID. If serviceNameOrID is empty, all service containers are returned. // name or ID. If serviceNameOrID is empty, all service containers are returned.
// Set opts.All to true to include hook containers.
func (c *Client) ListServiceContainers( func (c *Client) ListServiceContainers(
ctx context.Context, serviceNameOrID string, opts container.ListOptions, ctx context.Context, serviceNameOrID string, opts container.ListOptions,
) ([]MachineServiceContainers, error) { ) ([]MachineServiceContainers, error) {
@@ -501,22 +476,33 @@ func (c *Client) ListServiceContainers(
continue continue
} }
containers := make([]api.ServiceContainer, len(msg.Containers)) machineContainers[i].Containers, err = serviceContainersFromProto(msg.Containers)
for j, sc := range msg.Containers { if err != nil {
if err = json.Unmarshal(sc.Container, &containers[j].Container); err != nil { return nil, err
return nil, fmt.Errorf("unmarshal container: %w", err) }
} machineContainers[i].HookContainers, err = serviceContainersFromProto(msg.HookContainers)
if err = json.Unmarshal(sc.ServiceSpec, &containers[j].ServiceSpec); err != nil { if err != nil {
return nil, fmt.Errorf("unmarshal service spec: %w", err) return nil, err
}
} }
machineContainers[i].Containers = containers
} }
return machineContainers, nil return machineContainers, nil
} }
// serviceContainersFromProto converts a slice of protobuf service containers to api.ServiceContainer.
func serviceContainersFromProto(pbContainers []*pb.ServiceContainer) ([]api.ServiceContainer, error) {
containers := make([]api.ServiceContainer, len(pbContainers))
for i, sc := range pbContainers {
if err := json.Unmarshal(sc.Container, &containers[i].Container); err != nil {
return nil, fmt.Errorf("unmarshal container: %w", err)
}
if err := json.Unmarshal(sc.ServiceSpec, &containers[i].ServiceSpec); err != nil {
return nil, fmt.Errorf("unmarshal service spec: %w", err)
}
}
return containers, nil
}
// RemoveServiceContainer stops (kills after grace period) and removes a service container with the given ID. // RemoveServiceContainer stops (kills after grace period) and removes a service container with the given ID.
// A service container is a container that has been created with CreateServiceContainer. // A service container is a container that has been created with CreateServiceContainer.
func (c *Client) RemoveServiceContainer(ctx context.Context, id string, opts container.RemoveOptions) error { func (c *Client) RemoveServiceContainer(ctx context.Context, id string, opts container.RemoveOptions) error {
+5 -1
View File
@@ -152,12 +152,16 @@ func (c *Controller) syncContainersToStore(ctx context.Context) error {
return fmt.Errorf("list containers from store: %w", err) return fmt.Errorf("list containers from store: %w", err)
} }
containers, err := c.service.ListServiceContainers(ctx, "", container.ListOptions{}) // List containers for all services, including stopped ones and deployment hooks.
result, err := c.service.ListServiceContainers(ctx, "", container.ListOptions{All: true})
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 service containers: %w", err) return fmt.Errorf("list service containers: %w", err)
} }
// Sync both regular and one-off hook containers to the store.
containers := append(result.Containers, result.HookContainers...)
// 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.
var deleteIDs []string var deleteIDs []string
for _, sc := range storeContainers { for _, sc := range storeContainers {
+67 -23
View File
@@ -664,6 +664,39 @@ func (s *Server) CreateServiceContainer(
} }
} }
// Strip the container configuration that doesn't make sense for the pre-deploy hook and apply its overrides.
if spec.PreDeploy != nil && req.ContainerType == pb.CreateServiceContainerRequest_PRE_DEPLOY {
config.Labels = map[string]string{
api.LabelServiceID: req.ServiceId,
api.LabelServiceName: spec.Name,
api.LabelHook: api.LabelHookPreDeploy,
api.LabelManaged: "",
}
config.Healthcheck = &container.HealthConfig{
Test: []string{"NONE"},
}
hostConfig.PortBindings = nil
hostConfig.RestartPolicy = container.RestartPolicy{
Name: container.RestartPolicyDisabled,
}
// Apply the pre-deploy hook overrides.
config.Cmd = spec.PreDeploy.Command
for k, v := range spec.PreDeploy.Env {
envVars[k] = v
}
envVars["UNCLOUD_HOOK_PRE_DEPLOY"] = "true"
config.Env = envVars.ToSlice()
if spec.PreDeploy.Privileged != nil {
hostConfig.Privileged = *spec.PreDeploy.Privileged
}
if spec.PreDeploy.User != "" {
config.User = spec.PreDeploy.User
}
}
networkConfig := &network.NetworkingConfig{ networkConfig := &network.NetworkingConfig{
EndpointsConfig: map[string]*network.EndpointSettings{ EndpointsConfig: map[string]*network.EndpointSettings{
NetworkName: {}, NetworkName: {},
@@ -1009,37 +1042,49 @@ func (s *Server) ListServiceContainers(
} }
} }
containers, err := s.service.ListServiceContainers(ctx, req.ServiceId, opts) result, 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())
} }
// Convert to protobuf format. pbContainers, err := serviceContainersToProto(result.Containers)
if err != nil {
return nil, err
}
pbHookContainers, err := serviceContainersToProto(result.HookContainers)
if err != nil {
return nil, err
}
return &pb.ListServiceContainersResponse{
Messages: []*pb.MachineServiceContainers{
{
Containers: pbContainers,
HookContainers: pbHookContainers,
},
},
}, nil
}
// serviceContainersToProto converts a slice of service containers to protobuf format.
func serviceContainersToProto(containers []api.ServiceContainer) ([]*pb.ServiceContainer, error) {
pbContainers := make([]*pb.ServiceContainer, 0, len(containers)) pbContainers := make([]*pb.ServiceContainer, 0, len(containers))
for _, ctr := range containers { for _, ctr := range containers {
ctrBytes, err := json.Marshal(ctr.Container) ctrBytes, err := json.Marshal(ctr.Container)
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)
} }
specBytes, err := json.Marshal(ctr.ServiceSpec) specBytes, err := json.Marshal(ctr.ServiceSpec)
if err != nil { if err != nil {
return nil, status.Errorf(codes.Internal, "marshal service spec: %v", err) return nil, status.Errorf(codes.Internal, "marshal service spec: %v", err)
} }
pbContainers = append(pbContainers, &pb.ServiceContainer{ pbContainers = append(pbContainers, &pb.ServiceContainer{
Container: ctrBytes, Container: ctrBytes,
ServiceSpec: specBytes, ServiceSpec: specBytes,
}) })
} }
return &pb.ListServiceContainersResponse{ return pbContainers, nil
Messages: []*pb.MachineServiceContainers{
{
Containers: pbContainers,
},
},
}, nil
} }
// RemoveServiceContainer stops (kills after grace period) and removes a service container with the given ID. // RemoveServiceContainer stops (kills after grace period) and removes a service container with the given ID.
@@ -1078,20 +1123,19 @@ const logsHeartbeatInterval = 200 * time.Millisecond
// ContainerLogs streams logs from a container. // ContainerLogs streams logs from a container.
func (s *Server) ContainerLogs( func (s *Server) ContainerLogs(
req *pb.ContainerLogsRequest, stream grpc.ServerStreamingServer[pb.ContainerLogEntry], req *pb.LogsRequest, stream grpc.ServerStreamingServer[pb.LogEntry],
) error { ) error {
// Stream context is cancelled when the client has disconnected or the stream has ended. // Stream context is cancelled when the client has disconnected or the stream has ended.
ctx := stream.Context() ctx := stream.Context()
opts := ContainerLogsOptions{ opts := api.ServiceLogsOptions{
ContainerID: req.ContainerId, Follow: req.Follow,
Follow: req.Follow, Tail: int(req.Tail),
Tail: int(req.Tail), Since: req.Since,
Since: req.Since, Until: req.Until,
Until: req.Until,
} }
logsCh, err := s.service.ContainerLogs(ctx, opts) logsCh, err := s.service.ContainerLogs(ctx, req.Id, opts)
if err != nil { if err != nil {
if errdefs.IsNotFound(err) { if errdefs.IsNotFound(err) {
return status.Error(codes.NotFound, err.Error()) return status.Error(codes.NotFound, err.Error())
@@ -1099,7 +1143,7 @@ func (s *Server) ContainerLogs(
return status.Errorf(codes.Internal, "get container logs: %v", err) return status.Errorf(codes.Internal, "get container logs: %v", err)
} }
log := slog.With("container_id", req.ContainerId, "stream_id", fmt.Sprintf("%p", stream)[2:]) log := slog.With("container_id", req.Id, "stream_id", fmt.Sprintf("%p", stream)[2:])
log.Debug("Starting container logs streaming.", log.Debug("Starting container logs streaming.",
"follow", req.Follow, "tail", req.Tail, "since", req.Since, "until", req.Until) "follow", req.Follow, "tail", req.Tail, "since", req.Since, "until", req.Until)
@@ -1127,7 +1171,7 @@ func (s *Server) ContainerLogs(
return status.Error(codes.Internal, entry.Err.Error()) return status.Error(codes.Internal, entry.Err.Error())
} }
pbEntry := &pb.ContainerLogEntry{ pbEntry := &pb.LogEntry{
Stream: api.LogStreamTypeToProto(entry.Stream), Stream: api.LogStreamTypeToProto(entry.Stream),
Timestamp: timestamppb.New(entry.Timestamp), Timestamp: timestamppb.New(entry.Timestamp),
Message: entry.Message, Message: entry.Message,
@@ -1148,8 +1192,8 @@ func (s *Server) ContainerLogs(
// Use the timestamp one heartbeat in the past to be conservative. This reduces the chance of sending // Use the timestamp one heartbeat in the past to be conservative. This reduces the chance of sending
// a timestamp that is greater than a log entry currently being parsed but not yet sent, which would // a timestamp that is greater than a log entry currently being parsed but not yet sent, which would
// cause the client to incorrectly believe it has received all logs up to that point. // cause the client to incorrectly believe it has received all logs up to that point.
heartbeat := &pb.ContainerLogEntry{ heartbeat := &pb.LogEntry{
Stream: pb.ContainerLogEntry_HEARTBEAT, Stream: pb.LogEntry_HEARTBEAT,
Timestamp: timestamppb.New(now.Add(-logsHeartbeatInterval)), Timestamp: timestamppb.New(now.Add(-logsHeartbeatInterval)),
} }
if err = stream.Send(heartbeat); err != nil { if err = stream.Send(heartbeat); err != nil {
+24 -20
View File
@@ -74,11 +74,20 @@ func (s *Service) InspectServiceContainer(ctx context.Context, nameOrID string)
return serviceCtr, nil return serviceCtr, nil
} }
// ListServiceContainersResult holds the result of listing service containers, split into regular
// service containers and one-off hook containers.
type ListServiceContainersResult struct {
Containers []api.ServiceContainer
HookContainers []api.ServiceContainer
}
// ListServiceContainers lists Docker containers that belong to the service with the given name or ID. // 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. // If serviceIDOrName is empty, all service containers are returned. The opts parameter allows additional filtering.
func (s *Service) ListServiceContainers( func (s *Service) ListServiceContainers(
ctx context.Context, serviceNameOrID string, opts container.ListOptions, ctx context.Context, serviceNameOrID string, opts container.ListOptions,
) ([]api.ServiceContainer, error) { ) (ListServiceContainersResult, error) {
var result ListServiceContainersResult
if opts.Filters.Len() == 0 { if opts.Filters.Len() == 0 {
opts.Filters = filters.NewArgs() opts.Filters = filters.NewArgs()
} }
@@ -88,10 +97,9 @@ func (s *Service) ListServiceContainers(
containerSummaries, err := s.Client.ContainerList(ctx, opts) containerSummaries, err := s.Client.ContainerList(ctx, opts)
if err != nil { if err != nil {
return nil, err return result, err
} }
var containers []api.ServiceContainer
for _, cs := range containerSummaries { for _, cs := range containerSummaries {
// Filter by service name or ID if provided. // Filter by service name or ID if provided.
if serviceNameOrID != "" && if serviceNameOrID != "" &&
@@ -106,10 +114,15 @@ func (s *Service) ListServiceContainers(
slog.Error("Failed to inspect service container.", "service", serviceNameOrID, "id", cs.ID, "err", err) slog.Error("Failed to inspect service container.", "service", serviceNameOrID, "id", cs.ID, "err", err)
continue continue
} }
containers = append(containers, ctr)
if ctr.IsHook() {
result.HookContainers = append(result.HookContainers, ctr)
} else {
result.Containers = append(result.Containers, ctr)
}
} }
return containers, nil return result, nil
} }
// IsContainerdImageStoreEnabled checks if Docker is configured to use the containerd image store: // IsContainerdImageStoreEnabled checks if Docker is configured to use the containerd image store:
@@ -155,18 +168,9 @@ func (s *Service) ListImages(ctx context.Context, opts image.ListOptions) (Image
return imagesResp, nil return imagesResp, nil
} }
// ContainerLogsOptions specifies parameters for ContainerLogs.
type ContainerLogsOptions struct {
ContainerID string
Follow bool
Tail int
Since string
Until string
}
// ContainerLogs streams logs from a container and returns demultiplexed entries via a channel. // ContainerLogs streams logs from a container and returns demultiplexed entries via a channel.
// The channel is closed when streaming completes or context is cancelled. // The channel is closed when streaming completes or context is cancelled.
func (s *Service) ContainerLogs(ctx context.Context, opts ContainerLogsOptions) (<-chan api.ContainerLogEntry, error) { func (s *Service) ContainerLogs(ctx context.Context, containerID string, opts api.ServiceLogsOptions) (<-chan api.LogEntry, error) {
dockerOpts := container.LogsOptions{ dockerOpts := container.LogsOptions{
ShowStdout: true, ShowStdout: true,
ShowStderr: true, ShowStderr: true,
@@ -177,12 +181,12 @@ func (s *Service) ContainerLogs(ctx context.Context, opts ContainerLogsOptions)
Timestamps: true, Timestamps: true,
} }
reader, err := s.Client.ContainerLogs(ctx, opts.ContainerID, dockerOpts) reader, err := s.Client.ContainerLogs(ctx, containerID, dockerOpts)
if err != nil { if err != nil {
return nil, err return nil, err
} }
outCh := make(chan api.ContainerLogEntry) outCh := make(chan api.LogEntry)
stdoutWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: false} stdoutWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: false}
stderrWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: true} stderrWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: true}
@@ -198,7 +202,7 @@ func (s *Service) ContainerLogs(ctx context.Context, opts ContainerLogsOptions)
if _, err := stdcopy.StdCopy(stdoutWriter, stderrWriter, reader); err != nil { if _, err := stdcopy.StdCopy(stdoutWriter, stderrWriter, reader); err != nil {
// Send error as the last entry. // Send error as the last entry.
select { select {
case outCh <- api.ContainerLogEntry{Err: fmt.Errorf("demultiplex container logs: %w", err)}: case outCh <- api.LogEntry{Err: fmt.Errorf("demultiplex container logs: %w", err)}:
case <-ctx.Done(): case <-ctx.Done():
} }
} }
@@ -216,7 +220,7 @@ func (s *Service) ContainerLogs(ctx context.Context, opts ContainerLogsOptions)
// logsChannelWriter is a writer for stdcopy.StdCopy that sends demultiplexed container logs to a channel. // logsChannelWriter is a writer for stdcopy.StdCopy that sends demultiplexed container logs to a channel.
type logsChannelWriter struct { type logsChannelWriter struct {
ctx context.Context ctx context.Context
ch chan<- api.ContainerLogEntry ch chan<- api.LogEntry
isStderr bool isStderr bool
} }
@@ -236,7 +240,7 @@ func (w *logsChannelWriter) Write(data []byte) (n int, err error) {
} }
} }
entry := api.ContainerLogEntry{ entry := api.LogEntry{
Timestamp: timestamp, Timestamp: timestamp,
// Clone is required because message is a slice into data, which stdcopy.StdCopy may reuse // Clone is required because message is a slice into data, which stdcopy.StdCopy may reuse
// after Write returns but before the entry is consumed from the channel. // after Write returns but before the entry is consumed from the channel.
+99
View File
@@ -14,12 +14,16 @@ import (
"slices" "slices"
"strconv" "strconv"
"sync" "sync"
"time"
"github.com/containerd/errdefs"
"github.com/docker/docker/client" "github.com/docker/docker/client"
"github.com/docker/go-connections/sockets" "github.com/docker/go-connections/sockets"
"github.com/psviderski/uncloud/internal/corrosion" "github.com/psviderski/uncloud/internal/corrosion"
"github.com/psviderski/uncloud/internal/docker" "github.com/psviderski/uncloud/internal/docker"
"github.com/psviderski/uncloud/internal/fs" "github.com/psviderski/uncloud/internal/fs"
"github.com/psviderski/uncloud/internal/grpcversion"
"github.com/psviderski/uncloud/internal/journal"
"github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/api/pb"
apiproxy "github.com/psviderski/uncloud/internal/machine/api/proxy" apiproxy "github.com/psviderski/uncloud/internal/machine/api/proxy"
"github.com/psviderski/uncloud/internal/machine/caddyconfig" "github.com/psviderski/uncloud/internal/machine/caddyconfig"
@@ -30,6 +34,7 @@ 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/psviderski/unregistry" "github.com/psviderski/unregistry"
"github.com/siderolabs/grpc-proxy/proxy" "github.com/siderolabs/grpc-proxy/proxy"
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
@@ -268,6 +273,8 @@ func NewMachine(config *Config) (*Machine, error) {
proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort) proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort)
localProxyServer := grpc.NewServer( localProxyServer := grpc.NewServer(
grpc.ForceServerCodecV2(proxy.Codec()), grpc.ForceServerCodecV2(proxy.Codec()),
grpc.UnaryInterceptor(grpcversion.ServerUnaryInterceptor),
grpc.StreamInterceptor(grpcversion.ServerStreamInterceptor),
grpc.UnknownServiceHandler( grpc.UnknownServiceHandler(
proxy.TransparentHandler(proxyDirector.Director), proxy.TransparentHandler(proxyDirector.Director),
), ),
@@ -421,6 +428,8 @@ func (m *Machine) Run(ctx context.Context) error {
m.proxyDirector.UpdateLocalAddress(m.state.Network.ManagementIP.String()) m.proxyDirector.UpdateLocalAddress(m.state.Network.ManagementIP.String())
proxyServer := grpc.NewServer( proxyServer := grpc.NewServer(
grpc.ForceServerCodecV2(proxy.Codec()), grpc.ForceServerCodecV2(proxy.Codec()),
grpc.UnaryInterceptor(grpcversion.ServerUnaryInterceptor),
grpc.StreamInterceptor(grpcversion.ServerStreamInterceptor),
grpc.UnknownServiceHandler( grpc.UnknownServiceHandler(
proxy.TransparentHandler(m.proxyDirector.Director), proxy.TransparentHandler(m.proxyDirector.Director),
), ),
@@ -1073,3 +1082,93 @@ func (m *Machine) InspectService(
} }
return &pb.InspectServiceResponse{Service: svc}, nil return &pb.InspectServiceResponse{Service: svc}, nil
} }
// logsHeartbeatInterval is the interval at which heartbeat entries are sent when there are no logs to stream.
const logsHeartbeatInterval = 200 * time.Millisecond
// MachineLogs streams logs from a systemd service.
func (s *Machine) MachineLogs(
req *pb.LogsRequest, stream grpc.ServerStreamingServer[pb.LogEntry],
) error {
// TODO(miek): almost duplicate of docker/server.ContainerLogs
ctx := stream.Context()
opts := api.ServiceLogsOptions{
Follow: req.Follow,
Tail: int(req.Tail),
Since: req.Since,
Until: req.Until,
}
logsCh, err := journal.Logs(ctx, req.Id, opts)
if err != nil {
if errdefs.IsNotFound(err) {
return status.Error(codes.NotFound, err.Error())
}
return status.Errorf(codes.Internal, "get journal logs: %v", err)
}
log := slog.With("unit", req.Id, "stream_id", fmt.Sprintf("%p", stream)[2:])
log.Debug("Starting systemd service logs streaming.",
"follow", req.Follow, "tail", req.Tail, "since", req.Since, "until", req.Until)
// Heartbeats are needed only when following logs to let the client know when there are no new log entries
// to allow it to advance the watermark of last received log timestamp.
var heartbeatCh <-chan time.Time
if req.Follow {
heartbeatTicker := time.NewTicker(logsHeartbeatInterval)
defer heartbeatTicker.Stop()
heartbeatCh = heartbeatTicker.C
}
started := time.Now()
lastSent := time.Time{}
for {
select {
case entry, ok := <-logsCh:
if !ok {
// Channel closed, no more log entries.
return nil
}
if entry.Err != nil {
return status.Error(codes.Internal, entry.Err.Error())
}
pbEntry := &pb.LogEntry{
Stream: api.LogStreamTypeToProto(entry.Stream),
Timestamp: timestamppb.New(entry.Timestamp),
Message: entry.Message,
}
if err = stream.Send(pbEntry); err != nil {
return status.Errorf(codes.Internal, "send log entry: %v", err)
}
lastSent = entry.Timestamp
case now := <-heartbeatCh:
// Only send heartbeat if no log entries have been sent since the last heartbeat interval or
// if no log entries have been sent at all for at least a heartbeat interval since starting.
if now.Sub(lastSent) < logsHeartbeatInterval ||
(lastSent.IsZero() && now.Sub(started) < logsHeartbeatInterval) {
continue
}
// Use the timestamp one heartbeat in the past to be conservative. This reduces the chance of sending
// a timestamp that is greater than a log entry currently being parsed but not yet sent, which would
// cause the client to incorrectly believe it has received all logs up to that point.
heartbeat := &pb.LogEntry{
Stream: pb.LogEntry_HEARTBEAT,
Timestamp: timestamppb.New(now.Add(-logsHeartbeatInterval)),
}
if err = stream.Send(heartbeat); err != nil {
return status.Errorf(codes.Internal, "send log stream heartbeat: %v", err)
}
lastSent = heartbeat.Timestamp.AsTime()
log.Debug("Sent log stream heartbeat.", "timestamp", lastSent)
case <-ctx.Done():
return status.Error(codes.Canceled, ctx.Err().Error())
}
}
}
+3 -3
View File
@@ -125,7 +125,7 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
} }
defer initClient.Close() defer initClient.Close()
if err := initClient.WaitMachineReady(ctx, 30*time.Second); err != nil { if err := initClient.WaitMachineReady(ctx, 90*time.Second); err != nil {
return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err) return fmt.Errorf("wait for machine %q to be ready: %w", initMachine.Name, err)
} }
@@ -140,7 +140,7 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
fmt.Printf("Cluster %q initialised with machine %q\n", initMachine.ClusterName, initResp.Machine.Name) fmt.Printf("Cluster %q initialised with machine %q\n", initMachine.ClusterName, initResp.Machine.Name)
fmt.Printf("Waiting for cluster to be ready...") fmt.Printf("Waiting for cluster to be ready...")
if err = initClient.WaitClusterReady(ctx, 30*time.Second); err != nil { if err = initClient.WaitClusterReady(ctx, 90*time.Second); err != nil {
return fmt.Errorf("wait for cluster to be ready: %w", err) return fmt.Errorf("wait for cluster to be ready: %w", err)
} }
fmt.Println(" done.") fmt.Println(" done.")
@@ -160,7 +160,7 @@ func (p *Provisioner) initCluster(ctx context.Context, machines []Machine) error
//goland:noinspection GoDeferInLoop //goland:noinspection GoDeferInLoop
defer cli.Close() defer cli.Close()
if err := cli.WaitMachineReady(ctx, 30*time.Second); err != nil { if err := cli.WaitMachineReady(ctx, 90*time.Second); err != nil {
return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err) return fmt.Errorf("wait for machine %q to be ready: %w", m.Name, err)
} }
+3
View File
@@ -3,5 +3,8 @@ package version
var version string var version string
func String() string { func String() string {
if version == "" {
return "999.0.0-dev"
}
return version return version
} }
+4 -8
View File
@@ -20,7 +20,10 @@ type Client interface {
type ContainerClient interface { type ContainerClient interface {
CreateContainer( CreateContainer(
ctx context.Context, serviceID string, spec ServiceSpec, machineID string, ctx context.Context, serviceID string, spec ServiceSpec, machineID string,
) (container.CreateResponse, error) ) (CreateContainerResponse, error)
CreatePreDeployHookContainer(
ctx context.Context, serviceID string, spec ServiceSpec, machineID string,
) (CreateContainerResponse, error)
ExecContainer(ctx context.Context, serviceNameOrID, containerNameOrID string, config ExecOptions) (int, error) ExecContainer(ctx context.Context, serviceNameOrID, containerNameOrID string, config ExecOptions) (int, error)
InspectContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) (MachineServiceContainer, error) InspectContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) (MachineServiceContainer, error)
StartContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) error StartContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) error
@@ -60,10 +63,3 @@ type VolumeClient interface {
ListVolumes(ctx context.Context, filter *VolumeFilter) ([]MachineVolume, error) ListVolumes(ctx context.Context, filter *VolumeFilter) ([]MachineVolume, error)
RemoveVolume(ctx context.Context, machineNameOrID, volumeName string, force bool) error RemoveVolume(ctx context.Context, machineNameOrID, volumeName string, force bool) error
} }
// AsPtr returns a pointer to the given value. Useful for optional fields in API structs.
//
//go:fix inline
func AsPtr[T any](v T) *T {
return new(v)
}
+20 -1
View File
@@ -18,11 +18,16 @@ const (
// DockerNetworkName is the name of the Docker network used by uncloud. Keep the value in sync with NetworkName // DockerNetworkName is the name of the Docker network used by uncloud. Keep the value in sync with NetworkName
// in internal/machine/docker/manager.go. // in internal/machine/docker/manager.go.
DockerNetworkName = "uncloud" DockerNetworkName = "uncloud"
LabelManaged = "uncloud.managed" LabelManaged = "uncloud.managed"
LabelServiceID = "uncloud.service.id" LabelServiceID = "uncloud.service.id"
LabelServiceName = "uncloud.service.name" LabelServiceName = "uncloud.service.name"
LabelServiceMode = "uncloud.service.mode" LabelServiceMode = "uncloud.service.mode"
LabelServicePorts = "uncloud.service.ports" LabelServicePorts = "uncloud.service.ports"
// LabelHook marks a container as a deployment hook. The value indicates the hook type (e.g. LabelHookPreDeploy).
LabelHook = "uncloud.service.hook"
// LabelHookPreDeploy indicates that the container is a pre-deploy hook that runs before deploying the service.
LabelHookPreDeploy = "pre-deploy"
) )
type Container struct { type Container struct {
@@ -159,6 +164,13 @@ func (c *Container) UnmarshalJSON(data []byte) error {
return nil return nil
} }
// CreateContainerResponse wraps a container creation response with the container name assigned during creation.
type CreateContainerResponse struct {
container.CreateResponse
// Name is the container name assigned during creation.
Name string
}
type ServiceContainer struct { type ServiceContainer struct {
Container Container
ServiceSpec ServiceSpec ServiceSpec ServiceSpec
@@ -181,10 +193,17 @@ func (c *ServiceContainer) ServiceName() string {
// ServiceMode returns the replication mode of the service this container belongs to. // ServiceMode returns the replication mode of the service this container belongs to.
func (c *ServiceContainer) ServiceMode() string { func (c *ServiceContainer) ServiceMode() string {
return c.Config.Labels[LabelServiceMode] return c.ServiceSpec.Mode
}
// IsHook returns true if the container is a deployment hook (e.g. pre-deploy).
func (c *ServiceContainer) IsHook() bool {
_, ok := c.Config.Labels[LabelHook]
return ok
} }
// ServicePorts returns the ports this container publishes as part of its service. // ServicePorts returns the ports this container publishes as part of its service.
// TODO: return ports from ServiceSpec to allow updating ingress ports without recreating containers.
func (c *ServiceContainer) ServicePorts() ([]PortSpec, error) { func (c *ServiceContainer) ServicePorts() ([]PortSpec, error) {
encoded, ok := c.Config.Labels[LabelServicePorts] encoded, ok := c.Config.Labels[LabelServicePorts]
if !ok { if !ok {
+14 -14
View File
@@ -18,31 +18,31 @@ const (
type LogStreamType int type LogStreamType int
// LogStreamTypeFromProto converts a protobuf ContainerLogEntry.StreamType to the internal LogStreamType. // LogStreamTypeFromProto converts a protobuf LogEntry.StreamType to the internal LogStreamType.
func LogStreamTypeFromProto(s pb.ContainerLogEntry_StreamType) LogStreamType { func LogStreamTypeFromProto(s pb.LogEntry_StreamType) LogStreamType {
switch s { switch s {
case pb.ContainerLogEntry_STDOUT: case pb.LogEntry_STDOUT:
return LogStreamStdout return LogStreamStdout
case pb.ContainerLogEntry_STDERR: case pb.LogEntry_STDERR:
return LogStreamStderr return LogStreamStderr
case pb.ContainerLogEntry_HEARTBEAT: case pb.LogEntry_HEARTBEAT:
return LogStreamHeartbeat return LogStreamHeartbeat
default: default:
return LogStreamUnknown return LogStreamUnknown
} }
} }
// LogStreamTypeToProto converts LogStreamType to protobuf ContainerLogEntry.StreamType. // LogStreamTypeToProto converts LogStreamType to protobuf LogEntry.StreamType.
func LogStreamTypeToProto(s LogStreamType) pb.ContainerLogEntry_StreamType { func LogStreamTypeToProto(s LogStreamType) pb.LogEntry_StreamType {
switch s { switch s {
case LogStreamStdout: case LogStreamStdout:
return pb.ContainerLogEntry_STDOUT return pb.LogEntry_STDOUT
case LogStreamStderr: case LogStreamStderr:
return pb.ContainerLogEntry_STDERR return pb.LogEntry_STDERR
case LogStreamHeartbeat: case LogStreamHeartbeat:
return pb.ContainerLogEntry_HEARTBEAT return pb.LogEntry_HEARTBEAT
default: default:
return pb.ContainerLogEntry_UNKNOWN return pb.LogEntry_UNKNOWN
} }
} }
@@ -61,7 +61,7 @@ type ServiceLogsOptions struct {
type ServiceLogEntry struct { type ServiceLogEntry struct {
// Metadata may not be set if an error occurred (Err is not nil). // Metadata may not be set if an error occurred (Err is not nil).
Metadata ServiceLogEntryMetadata Metadata ServiceLogEntryMetadata
ContainerLogEntry LogEntry
} }
// ServiceLogEntryMetadata contains metadata about the source of a log entry. // ServiceLogEntryMetadata contains metadata about the source of a log entry.
@@ -73,8 +73,8 @@ type ServiceLogEntryMetadata struct {
MachineName string MachineName string
} }
// ContainerLogEntry represents a single log entry from a container. // LogEntry represents a single log entry from a container or a service.
type ContainerLogEntry struct { type LogEntry struct {
Stream LogStreamType Stream LogStreamType
Timestamp time.Time Timestamp time.Time
Message []byte Message []byte
+57 -4
View File
@@ -55,7 +55,7 @@ type ServiceSpec struct {
// Caddy is the optional Caddy reverse proxy configuration for the service. // Caddy is the optional Caddy reverse proxy configuration for the service.
// Caddy and Ports cannot be specified simultaneously. // Caddy and Ports cannot be specified simultaneously.
Caddy *CaddySpec `json:",omitempty"` Caddy *CaddySpec `json:",omitempty"`
// Configs is list of configuration objects that can be mounted into the container. // Configs is a list of configuration objects that can be mounted into the container.
Configs []ConfigSpec Configs []ConfigSpec
// Container defines the desired state of each container in the service. // Container defines the desired state of each container in the service.
Container ContainerSpec Container ContainerSpec
@@ -67,6 +67,9 @@ type ServiceSpec struct {
// 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. // Caddy and Ports cannot be specified simultaneously.
Ports []PortSpec Ports []PortSpec
// PreDeploy is an optional hook that runs a command in a temporary container before deploying the service.
// The container uses the service's image and inherits its configuration.
PreDeploy *PreDeployHook `json:",omitempty"`
// 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"`
// UpdateConfig configures how the service is updated during a deployment. // UpdateConfig configures how the service is updated during a deployment.
@@ -205,6 +208,12 @@ func (s *ServiceSpec) Validate() error {
return fmt.Errorf("validate service configs and mounts: %w", err) return fmt.Errorf("validate service configs and mounts: %w", err)
} }
if s.PreDeploy != nil {
if err := s.PreDeploy.Validate(); err != nil {
return err
}
}
return nil return nil
} }
@@ -216,6 +225,7 @@ func (s *ServiceSpec) Clone() ServiceSpec {
spec.Caddy = &caddyCopy spec.Caddy = &caddyCopy
} }
spec.Container = s.Container.Clone() spec.Container = s.Container.Clone()
spec.PreDeploy = s.PreDeploy.Clone()
if s.Ports != nil { if s.Ports != nil {
spec.Ports = make([]PortSpec, len(s.Ports)) spec.Ports = make([]PortSpec, len(s.Ports))
@@ -436,6 +446,46 @@ type LogDriver struct {
Options map[string]string Options map[string]string
} }
// PreDeployHook defines a command to run in a temporary container before deploying the service.
// The container uses the service's image and inherits its configuration (env, volumes, placement).
// It must exit successfully (code 0) for the deployment to proceed.
type PreDeployHook struct {
// Command to execute in the container.
Command []string
// Env defines additional environment variables for the container, merged with the service's environment variables.
Env EnvVars `json:",omitempty"`
// Privileged overrides the container's privileged mode. nil means inherit from the service.
Privileged *bool `json:",omitempty"`
// Timeout is the maximum duration to wait for the command to complete. On timeout, the container is stopped
// and the deployment fails. If nil, a default timeout is used.
Timeout *time.Duration `json:",omitempty"`
// User to run the command as. If empty, the service user is used. Format: user|UID[:group|GID].
User string `json:",omitempty"`
}
func (h *PreDeployHook) Validate() error {
if len(h.Command) == 0 {
return fmt.Errorf("pre-deploy hook command is required")
}
return nil
}
func (h *PreDeployHook) Clone() *PreDeployHook {
if h == nil {
return nil
}
hook := *h
hook.Command = slices.Clone(h.Command)
hook.Env = maps.Clone(h.Env)
return &hook
}
func (h *PreDeployHook) Equals(other *PreDeployHook) bool {
return cmp.Equal(h, other, cmpopts.EquateEmpty())
}
// UpdateConfig configures how a service is updated during a deployment. // UpdateConfig configures how a service is updated during a deployment.
type UpdateConfig struct { type UpdateConfig struct {
// Order specifies the order of operations during an update. // Order specifies the order of operations during an update.
@@ -455,10 +505,13 @@ type RunServiceResponse struct {
} }
type Service struct { type Service struct {
ID string ID string
Name string Name string
Mode string Mode string
// Containers is the regular long-running service containers.
Containers []MachineServiceContainer Containers []MachineServiceContainer
// HookContainers are one-shot containers for deployment hooks (e.g. pre-deploy).
HookContainers []MachineServiceContainer
} }
type MachineServiceContainer struct { type MachineServiceContainer struct {
+26
View File
@@ -0,0 +1,26 @@
package compose
import (
"fmt"
"github.com/compose-spec/compose-go/v2/types"
)
const PreDeployHookExtensionKey = "x-pre_deploy"
// PreDeployHook represents the parsed x-pre_deploy extension config.
type PreDeployHook struct {
Command types.ShellCommand `yaml:"command" json:"command"`
Environment types.MappingWithEquals `yaml:"environment,omitempty" json:"environment,omitempty"`
Privileged *bool `yaml:"privileged,omitempty" json:"privileged,omitempty"`
Timeout *types.Duration `yaml:"timeout,omitempty" json:"timeout,omitempty"`
User string `yaml:"user,omitempty" json:"user,omitempty"`
}
// Validate checks that the pre-deploy hook configuration is valid.
func (p *PreDeployHook) Validate() error {
if len(p.Command) == 0 {
return fmt.Errorf("missing required attribute 'command' in %s extension", PreDeployHookExtensionKey)
}
return nil
}
+149
View File
@@ -0,0 +1,149 @@
package compose
import (
"context"
"testing"
"time"
"github.com/compose-spec/compose-go/v2/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestPreDeployHookExtension(t *testing.T) {
tests := []struct {
name string
yaml string
want PreDeployHook
wantErr string
}{
{
name: "command only",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
command: ["echo", "hello"]
`,
want: PreDeployHook{
Command: types.ShellCommand{"echo", "hello"},
},
},
{
name: "command as string",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
command: echo hello
`,
want: PreDeployHook{
Command: types.ShellCommand{"echo", "hello"},
},
},
// TODO: explore ways to error on unknown attributes instead of ignoring them.
{
name: "all attributes",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
command: ["sh", "-c", "migrate up"]
environment:
DB_HOST: localhost
DB_PORT: "5432"
privileged: true
timeout: 2m30s
user: root
unknown_attribute: should be ignored
`,
want: PreDeployHook{
Command: types.ShellCommand{"sh", "-c", "migrate up"},
Environment: types.MappingWithEquals{
"DB_HOST": new("localhost"),
"DB_PORT": new("5432"),
},
Privileged: new(true),
Timeout: new(types.Duration(2*time.Minute + 30*time.Second)),
User: "root",
},
},
{
name: "timeout as seconds",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
command: ["true"]
timeout: 30s
`,
want: PreDeployHook{
Command: types.ShellCommand{"true"},
Timeout: new(types.Duration(30 * time.Second)),
},
},
{
name: "privileged false",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
command: ["true"]
privileged: false
`,
want: PreDeployHook{
Command: types.ShellCommand{"true"},
Privileged: new(false),
},
},
{
name: "missing command should fail",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
user: root
`,
wantErr: "missing required attribute 'command'",
},
{
name: "empty command should fail",
yaml: `
services:
web:
image: nginx
x-pre_deploy:
command: []
`,
wantErr: "missing required attribute 'command'",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
project, err := LoadProjectFromContent(context.Background(), tt.yaml)
if tt.wantErr != "" {
require.ErrorContains(t, err, tt.wantErr)
return
}
require.NoError(t, err)
service, err := project.GetService("web")
require.NoError(t, err)
ext, ok := service.Extensions[PreDeployHookExtensionKey]
require.True(t, ok, "x-pre_deploy extension not found")
hook, ok := ext.(PreDeployHook)
require.True(t, ok, "x-pre_deploy extension is not PreDeployHook type")
assert.Equal(t, tt.want, hook)
})
}
}
+6
View File
@@ -12,6 +12,7 @@ import (
"github.com/compose-spec/compose-go/v2/transform" "github.com/compose-spec/compose-go/v2/transform"
"github.com/compose-spec/compose-go/v2/tree" "github.com/compose-spec/compose-go/v2/tree"
"github.com/compose-spec/compose-go/v2/types" "github.com/compose-spec/compose-go/v2/types"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
) )
@@ -38,6 +39,7 @@ func LoadProject(ctx context.Context, paths []string, opts ...composecli.Project
composecli.WithExtension(CaddyExtensionKey, Caddy{}), composecli.WithExtension(CaddyExtensionKey, Caddy{}),
composecli.WithExtension(MachinesExtensionKey, MachinesSource{}), composecli.WithExtension(MachinesExtensionKey, MachinesSource{}),
composecli.WithExtension(PortsExtensionKey, PortsSource{}), composecli.WithExtension(PortsExtensionKey, PortsSource{}),
composecli.WithExtension(PreDeployHookExtensionKey, PreDeployHook{}),
} }
options, err := composecli.NewProjectOptions( options, err := composecli.NewProjectOptions(
@@ -66,6 +68,10 @@ func LoadProject(ctx context.Context, paths []string, opts ...composecli.Project
return nil, err return nil, err
} }
for _, err = range validateServicesFeatures(project) {
tui.PrintWarning(err.Error())
}
// Process image templates in services to expand Go template expressions using git repo state. // Process image templates in services to expand Go template expressions using git repo state.
if project, err = ProcessImageTemplates(project); err != nil { if project, err = ProcessImageTemplates(project); err != nil {
return nil, err return nil, err
+131
View File
@@ -2,8 +2,10 @@ package compose
import ( import (
"context" "context"
"io"
"os" "os"
"path/filepath" "path/filepath"
"strings"
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -186,3 +188,132 @@ REDIS_URL=redis://localhost:6379
}) })
} }
} }
// TestLoadProject_Unsupported checks that unsupported features lead to warnings.
func TestLoadProject_Unsupported(t *testing.T) {
// captureStderr runs fn while capturing stderr output and returns what was written.
captureStderr := func(t *testing.T, fn func()) string {
t.Helper()
old := os.Stderr
r, w, err := os.Pipe()
require.NoError(t, err)
os.Stderr = w
defer func() { os.Stderr = old }()
fn()
w.Close()
out, err := io.ReadAll(r)
require.NoError(t, err)
r.Close()
return string(out)
}
tests := []struct {
name string
composeYAML string
warnCount int
warnContains []string
}{
{
name: "unsupported dns",
composeYAML: `services:
app:
image: myapp:latest
dns: 8.8.8.8
`,
warnCount: 1,
warnContains: []string{"dns"},
},
{
name: "unsupported networks",
composeYAML: `services:
app:
image: myapp:latest
networks:
- frontend
networks:
frontend:
`,
warnCount: 1,
warnContains: []string{"networks"},
},
{
name: "unsupported depends_on service_completed_successfully",
composeYAML: `services:
migrate:
image: alpine
command: ["true"]
app:
image: nginx
depends_on:
migrate:
condition: service_completed_successfully
`,
warnCount: 1,
warnContains: []string{
"service_completed_successfully",
"pre-deploy hook",
"https://uncloud.run/docs/guides/deployments/pre-deploy-hooks",
},
},
{
name: "multiple unsupported features",
composeYAML: `services:
app:
image: myapp:latest
dns: 8.8.8.8
links:
- db
db:
image: postgres:latest
secrets:
- db_password
secrets:
db_password:
file: ./secret.txt
`,
warnCount: 3,
warnContains: []string{"dns", "links", "secrets"},
},
{
name: "supported networks",
composeYAML: `services:
app:
image: myapp:latest
networks:
default: {}
web:
image: nginx
networks:
- default
networks:
default:
`,
warnCount: 0,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
stderr := captureStderr(t, func() {
_, err := LoadProjectFromContent(context.Background(), tt.composeYAML)
require.NoError(t, err)
})
if tt.warnCount == 0 {
assert.Empty(t, stderr)
} else {
assert.Equal(t, tt.warnCount, strings.Count(stderr, "WARNING:"),
"expected %d warnings, got stderr: %s", tt.warnCount, stderr)
for _, substr := range tt.warnContains {
assert.Contains(t, stderr, substr)
}
}
})
}
}
+93 -1
View File
@@ -33,7 +33,7 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser
return api.ServiceSpec{}, fmt.Errorf("unsupported pull policy: '%s'", service.PullPolicy) return api.ServiceSpec{}, fmt.Errorf("unsupported pull policy: '%s'", service.PullPolicy)
} }
env := make(map[string]string, len(service.Environment)) env := make(api.EnvVars, len(service.Environment))
for k, v := range service.Environment { for k, v := range service.Environment {
if v == nil { if v == nil {
// nil value means the variable misses a value in the compose file, and it hasn't been resolved // nil value means the variable misses a value in the compose file, and it hasn't been resolved
@@ -142,6 +142,27 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser
spec.Configs = configSpecs spec.Configs = configSpecs
spec.Container.ConfigMounts = configMounts spec.Container.ConfigMounts = configMounts
if h, ok := service.Extensions[PreDeployHookExtensionKey].(PreDeployHook); ok {
hook := &api.PreDeployHook{
Command: h.Command,
Privileged: h.Privileged,
User: h.User,
}
if h.Environment != nil {
hook.Env = make(api.EnvVars)
for k, v := range h.Environment {
if v != nil {
hook.Env[k] = *v
}
}
}
if h.Timeout != nil {
d := time.Duration(*h.Timeout)
hook.Timeout = &d
}
spec.PreDeploy = hook
}
return spec, nil return spec, nil
} }
@@ -418,7 +439,78 @@ func validateServicesExtensions(project *types.Project) error {
"Host mode ports in 'x-caddy' can be used with 'x-caddy'", service.Name) "Host mode ports in 'x-caddy' can be used with 'x-caddy'", service.Name)
} }
} }
if hook, ok := service.Extensions[PreDeployHookExtensionKey].(PreDeployHook); ok {
if err := hook.Validate(); err != nil {
return fmt.Errorf("service '%s': %w", service.Name, err)
}
}
} }
return nil return nil
} }
// validateServicesFeatures checks services for unsupported features and returns all found.
func validateServicesFeatures(project *types.Project) []error {
err := func(service, feature string) error {
return fmt.Errorf("service '%s': unsupported feature '%s', see %s",
service, feature, "https://uncloud.run/docs/compose-file-reference/support-matrix")
}
// TODO: check other commonly used but unsupported features.
var errs []error
for _, service := range project.Services {
if service.SecurityOpt != nil {
errs = append(errs, err(service.Name, "security_opt"))
}
if service.DNS != nil {
errs = append(errs, err(service.Name, "dns"))
}
if service.DNSSearch != nil {
errs = append(errs, err(service.Name, "dns_search"))
}
if service.Labels != nil {
errs = append(errs, err(service.Name, "labels"))
}
if service.Links != nil {
errs = append(errs, err(service.Name, "links"))
}
if service.MemSwappiness > 0 {
errs = append(errs, err(service.Name, "mem_swappiness"))
}
if service.MemSwapLimit > 0 {
errs = append(errs, err(service.Name, "memswap_limit"))
}
if service.Secrets != nil {
errs = append(errs, err(service.Name, "secrets"))
}
if service.StorageOpt != nil {
errs = append(errs, err(service.Name, "storage_opt"))
}
// we only allow the 'default' network, nothing else.
if x := service.Networks; x != nil {
if len(x) != 1 {
errs = append(errs, err(service.Name, "networks"))
} else if _, ok := x["default"]; !ok {
errs = append(errs, err(service.Name, "networks"))
}
}
// Err about depends_on conditions 'service_completed_successfully' that we do not support because services in
// Uncloud are long-running. Services that run to completion need a separate abstraction, e.g. a Job.
// Plus we don't want the lifecycle of a service to control the lifecycle of another one. It should be fully
// owned. Otherwise, it's hard to make the behaviour deterministic when each service could also be deployed and
// managed independently.
for depName, dep := range service.DependsOn {
if dep.Condition == types.ServiceConditionCompletedSuccessfully {
errs = append(errs, fmt.Errorf(
"service '%s': depends_on condition '%s' on service '%s' is not supported, "+
"use a pre-deploy hook instead: %s",
service.Name, types.ServiceConditionCompletedSuccessfully, depName,
"https://uncloud.run/docs/guides/deployments/pre-deploy-hooks"))
}
}
}
return errs
}
+10 -3
View File
@@ -208,6 +208,13 @@ func TestServiceSpecFromCompose(t *testing.T) {
Placement: api.Placement{ Placement: api.Placement{
Machines: []string{"machine-1", "machine-2"}, Machines: []string{"machine-1", "machine-2"},
}, },
PreDeploy: &api.PreDeployHook{
Command: []string{"sh", "-c", "migrate"},
Env: api.EnvVars{"DB_HOST": "localhost"},
Privileged: new(false),
Timeout: new(2*time.Minute + 30*time.Second),
User: "root",
},
Replicas: 3, Replicas: 3,
UpdateConfig: api.UpdateConfig{ UpdateConfig: api.UpdateConfig{
Order: api.UpdateOrderStopFirst, Order: api.UpdateOrderStopFirst,
@@ -968,7 +975,7 @@ services:
monitor: 10s monitor: 10s
`, `,
expected: api.UpdateConfig{ expected: api.UpdateConfig{
MonitorPeriod: api.AsPtr(10 * time.Second), MonitorPeriod: new(10 * time.Second),
}, },
}, },
{ {
@@ -984,7 +991,7 @@ services:
`, `,
expected: api.UpdateConfig{ expected: api.UpdateConfig{
Order: api.UpdateOrderStartFirst, Order: api.UpdateOrderStartFirst,
MonitorPeriod: api.AsPtr(30 * time.Second), MonitorPeriod: new(30 * time.Second),
}, },
}, },
{ {
@@ -998,7 +1005,7 @@ services:
monitor: 0s monitor: 0s
`, `,
expected: api.UpdateConfig{ expected: api.UpdateConfig{
MonitorPeriod: api.AsPtr(time.Duration(0)), MonitorPeriod: new(time.Duration(0)),
}, },
}, },
} }
+7
View File
@@ -78,6 +78,13 @@ services:
- test.example.com:80/https - test.example.com:80/https
- 8000/http - 8000/http
- 5000:3000@host - 5000:3000@host
x-pre_deploy:
command: ["sh", "-c", "migrate"]
environment:
DB_HOST: localhost
privileged: false
timeout: 2m30s
user: root
test-caddy-config: test-caddy-config:
image: myapp:1.2.3 image: myapp:1.2.3
+3
View File
@@ -7,6 +7,7 @@ import (
"net" "net"
"strings" "strings"
"github.com/psviderski/uncloud/internal/grpcversion"
"github.com/psviderski/uncloud/internal/machine" "github.com/psviderski/uncloud/internal/machine"
"github.com/psviderski/uncloud/internal/sshexec" "github.com/psviderski/uncloud/internal/sshexec"
"golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh"
@@ -74,6 +75,8 @@ func (c *SSHConnector) Connect(ctx context.Context) (*grpc.ClientConn, error) {
"unix://"+sockPath, "unix://"+sockPath,
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithDefaultServiceConfig(defaultServiceConfig), grpc.WithDefaultServiceConfig(defaultServiceConfig),
grpc.WithUnaryInterceptor(grpcversion.ClientUnaryInterceptor),
grpc.WithStreamInterceptor(grpcversion.ClientStreamInterceptor),
grpc.WithContextDialer( grpc.WithContextDialer(
func(ctx context.Context, addr string) (net.Conn, error) { func(ctx context.Context, addr string) (net.Conn, error) {
addr = strings.TrimPrefix(addr, "unix://") addr = strings.TrimPrefix(addr, "unix://")
+45 -81
View File
@@ -5,11 +5,13 @@ import (
"fmt" "fmt"
"net" "net"
"os" "os"
"os/exec"
"path/filepath" "path/filepath"
"strconv" "strconv"
"strings"
"github.com/docker/cli/cli/connhelper/commandconn" "github.com/docker/cli/cli/connhelper/commandconn"
"github.com/psviderski/uncloud/internal/machine" "github.com/psviderski/uncloud/internal/grpcversion"
"golang.org/x/net/proxy" "golang.org/x/net/proxy"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
@@ -61,17 +63,32 @@ func controlSocketPath() string {
} }
func (c *SSHCLIConnector) Connect(ctx context.Context) (*grpc.ClientConn, error) { func (c *SSHCLIConnector) Connect(ctx context.Context) (*grpc.ClientConn, error) {
// Create gRPC client with a dialer that spawns a new SSH connection on demand. // Validate SSH connectivity by running a no-op command on the remote machine. This also
// Each dial attempt runs `ssh ... uncloudd dial-stdio`, reusing the control socket if available. // establishes the control socket (ControlMaster=auto) so subsequent connections reuse it.
probeArgs := append(c.buildSSHArgs(), "true")
probe := exec.CommandContext(ctx, "ssh", probeArgs...)
if output, err := probe.CombinedOutput(); err != nil {
return nil, fmt.Errorf("SSH connection to '%s': %w: %s",
c.config.Destination(), err, strings.TrimSpace(string(output)))
}
// Create gRPC client with a dialer that spawns new SSH connections on demand,
// reusing the control socket established above.
grpcConn, err := grpc.NewClient( grpcConn, err := grpc.NewClient(
"passthrough:///", // Dummy target since we're using a custom dialer. "passthrough:///", // Dummy target since we're using a custom dialer.
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithDefaultServiceConfig(defaultServiceConfig), grpc.WithDefaultServiceConfig(defaultServiceConfig),
grpc.WithUnaryInterceptor(grpcversion.ClientUnaryInterceptor),
grpc.WithStreamInterceptor(grpcversion.ClientStreamInterceptor),
grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
args := c.buildSSHArgs() dialArgs := append(c.buildSSHArgs(), "uncloudd", "dial-stdio")
conn, err := commandconn.New(ctx, "ssh", args...) if c.config.SockPath != "" {
dialArgs = append(dialArgs, "--socket", c.config.SockPath)
}
conn, err := commandconn.New(ctx, "ssh", dialArgs...)
if err != nil { if err != nil {
return nil, fmt.Errorf("SSH connection to %s: %w", c.config.Destination(), err) return nil, fmt.Errorf("SSH connection to '%s': %w", c.config.Destination(), err)
} }
return conn, nil return conn, nil
}), }),
@@ -83,8 +100,9 @@ func (c *SSHCLIConnector) Connect(ctx context.Context) (*grpc.ClientConn, error)
return grpcConn, nil return grpcConn, nil
} }
// buildSSHArgs constructs the SSH command arguments to run `uncloudd dial-stdio` on the remote machine reusing // buildSSHArgs constructs the SSH command arguments with connection options and destination. The options
// the established connection via control socket. // include control socket settings for connection reuse if necessary.
// The remote command is not included and should be appended by the caller.
func (c *SSHCLIConnector) buildSSHArgs() []string { func (c *SSHCLIConnector) buildSSHArgs() []string {
var args []string var args []string
@@ -104,6 +122,9 @@ func (c *SSHCLIConnector) buildSSHArgs() []string {
// Add connection timeout to fail fast when node is down. // Add connection timeout to fail fast when node is down.
args = append(args, "-o", "ConnectTimeout=5") args = append(args, "-o", "ConnectTimeout=5")
// Disable interactive prompts (e.g., passphrase input) to prevent interference with the TUI.
// Authentication must succeed non-interactively via SSH agent or unencrypted key.
args = append(args, "-o", "BatchMode=yes")
// Disable pseudo-terminal allocation to prevent SSH from executing as a login shell. // Disable pseudo-terminal allocation to prevent SSH from executing as a login shell.
args = append(args, "-T") args = append(args, "-T")
@@ -120,14 +141,6 @@ func (c *SSHCLIConnector) buildSSHArgs() []string {
// Add [user@]host destination. // Add [user@]host destination.
args = append(args, c.config.Destination()) args = append(args, c.config.Destination())
// Add remote command: uncloudd dial-stdio
args = append(args, "uncloudd", "dial-stdio")
// Add socket path if non-default.
if c.config.SockPath != "" && c.config.SockPath != machine.DefaultUncloudSockPath {
args = append(args, "--socket", c.config.SockPath)
}
return args return args
} }
@@ -137,10 +150,22 @@ func (c *SSHCLIConnector) Dialer() (proxy.ContextDialer, error) {
return nil, fmt.Errorf("SSH connector not configured") return nil, fmt.Errorf("SSH connector not configured")
} }
return &sshCLIDialer{ return c, nil
config: c.config, }
controlSockPath: c.controlSockPath,
}, nil // DialContext establishes a connection to the target address through an SSH tunnel using -W flag.
func (c *SSHCLIConnector) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
if network != "tcp" {
return nil, fmt.Errorf("unsupported network type: %s", network)
}
args := append(c.buildSSHArgs(), "-W", address)
conn, err := commandconn.New(ctx, "ssh", args...)
if err != nil {
return nil, fmt.Errorf("SSH connection to '%s' for dialing '%s': %w", c.config.Destination(), address, err)
}
return conn, nil
} }
func (c *SSHCLIConnector) Close() error { func (c *SSHCLIConnector) Close() error {
@@ -148,64 +173,3 @@ func (c *SSHCLIConnector) Close() error {
// The SSH control socket may persist for connection reuse across CLI invocations. // The SSH control socket may persist for connection reuse across CLI invocations.
return nil return nil
} }
// sshCLIDialer implements proxy.ContextDialer by spawning SSH processes with -W flag.
type sshCLIDialer struct {
config SSHConnectorConfig
// Shared control socket path from SSHCLIConnector for connection reuse.
controlSockPath string
}
// buildDialArgs constructs SSH command arguments for -W flag dialing.
func (d *sshCLIDialer) buildDialArgs(address string) []string {
var args []string
if d.controlSockPath != "" {
// Try to reuse the existing control connection without initiating a new one.
// Falls back to direct connection if the control socket is not available.
args = append(args, "-o", "ControlMaster=no")
args = append(args, "-o", "ControlPath="+d.controlSockPath)
}
// Add connection timeout to fail fast when node is down.
args = append(args, "-o", "ConnectTimeout=5")
// Disable pseudo-terminal allocation to prevent SSH from executing as a login shell.
args = append(args, "-T")
// Add port if specified.
if d.config.Port != 0 {
args = append(args, "-p", strconv.Itoa(d.config.Port))
}
// Add identity file if specified.
if d.config.KeyPath != "" {
args = append(args, "-i", d.config.KeyPath)
}
// Add -W flag for stdin/stdout forwarding to target address.
args = append(args, "-W", address)
// Add [user@]host destination.
args = append(args, d.config.Destination())
return args
}
// DialContext establishes a connection to the target address through an SSH tunnel using -W flag.
func (d *sshCLIDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
// Only support TCP connections.
if network != "tcp" {
return nil, fmt.Errorf("unsupported network type: %s", network)
}
// Build SSH command arguments.
args := d.buildDialArgs(address)
// Create connection using docker's commandconn.
conn, err := commandconn.New(ctx, "ssh", args...)
if err != nil {
return nil, fmt.Errorf("SSH connection to %s for dialing %s: %w", d.config.Destination(), address, err)
}
return conn, nil
}
+7 -117
View File
@@ -4,7 +4,6 @@ import (
"strings" "strings"
"testing" "testing"
"github.com/psviderski/uncloud/internal/machine"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
) )
@@ -24,7 +23,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
Host: "example.com", Host: "example.com",
}, },
controlSockPath: "/tmp/test.sock", controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "root@example.com", "uncloudd", "dial-stdio"}, expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "root@example.com"},
}, },
{ {
name: "basic connection without control socket", name: "basic connection without control socket",
@@ -33,7 +32,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
Host: "example.com", Host: "example.com",
}, },
controlSockPath: "", controlSockPath: "",
expected: []string{"-o", "ConnectTimeout=5", "-T", "root@example.com", "uncloudd", "dial-stdio"}, expected: []string{"-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "root@example.com"},
}, },
{ {
name: "with custom port", name: "with custom port",
@@ -43,7 +42,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
Port: 2222, Port: 2222,
}, },
controlSockPath: "/tmp/test.sock", controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "-p", "2222", "root@example.com", "uncloudd", "dial-stdio"}, expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "-p", "2222", "root@example.com"},
}, },
{ {
name: "with identity file", name: "with identity file",
@@ -53,27 +52,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
KeyPath: "/path/to/key", KeyPath: "/path/to/key",
}, },
controlSockPath: "/tmp/test.sock", controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "-i", "/path/to/key", "root@example.com", "uncloudd", "dial-stdio"}, expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "-i", "/path/to/key", "root@example.com"},
},
{
name: "with custom socket path",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
SockPath: "/custom/path/uncloud.sock",
},
controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "root@example.com", "uncloudd", "dial-stdio", "--socket", "/custom/path/uncloud.sock"},
},
{
name: "with default socket path (not included)",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
SockPath: machine.DefaultUncloudSockPath,
},
controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "root@example.com", "uncloudd", "dial-stdio"},
}, },
{ {
name: "all options combined", name: "all options combined",
@@ -85,7 +64,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
SockPath: "/custom/path/uncloud.sock", SockPath: "/custom/path/uncloud.sock",
}, },
controlSockPath: "/tmp/test.sock", controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "-p", "2222", "-i", "/path/to/key", "root@example.com", "uncloudd", "dial-stdio", "--socket", "/custom/path/uncloud.sock"}, expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "-p", "2222", "-i", "/path/to/key", "root@example.com"},
}, },
{ {
name: "port 0 not included", name: "port 0 not included",
@@ -95,7 +74,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
Port: 0, Port: 0,
}, },
controlSockPath: "/tmp/test.sock", controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "root@example.com", "uncloudd", "dial-stdio"}, expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "root@example.com"},
}, },
{ {
name: "port 22 included when explicit", name: "port 22 included when explicit",
@@ -105,7 +84,7 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
Port: 22, Port: 22,
}, },
controlSockPath: "/tmp/test.sock", controlSockPath: "/tmp/test.sock",
expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-T", "-p", "22", "root@example.com", "uncloudd", "dial-stdio"}, expected: []string{"-o", "ControlMaster=auto", "-o", "ControlPath=/tmp/test.sock", "-o", "ControlPersist=10m", "-o", "ConnectTimeout=5", "-o", "BatchMode=yes", "-T", "-p", "22", "root@example.com"},
}, },
} }
@@ -120,95 +99,6 @@ func TestSSHCLIConnector_buildSSHArgs(t *testing.T) {
} }
} }
func TestSSHCLIDialer_buildDialArgs(t *testing.T) {
t.Parallel()
tests := []struct {
name string
config SSHConnectorConfig
controlSockPath string
address string
expected []string
}{
{
name: "basic connection without control socket",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
},
controlSockPath: "",
address: "10.210.1.1:5000",
expected: []string{"-o", "ConnectTimeout=5", "-T", "-W", "10.210.1.1:5000", "root@example.com"},
},
{
name: "basic connection with control socket",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
},
controlSockPath: "/tmp/test.sock",
address: "10.210.1.1:5000",
expected: []string{"-o", "ControlMaster=no", "-o", "ControlPath=/tmp/test.sock", "-o", "ConnectTimeout=5", "-T", "-W", "10.210.1.1:5000", "root@example.com"},
},
{
name: "custom port with control socket",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
Port: 2222,
},
controlSockPath: "/tmp/test.sock",
address: "10.210.1.1:5000",
expected: []string{"-o", "ControlMaster=no", "-o", "ControlPath=/tmp/test.sock", "-o", "ConnectTimeout=5", "-T", "-p", "2222", "-W", "10.210.1.1:5000", "root@example.com"},
},
{
name: "with identity file and control socket",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
Port: 22,
KeyPath: "/home/user/.ssh/id_rsa",
},
controlSockPath: "/tmp/test.sock",
address: "10.210.1.1:5000",
expected: []string{"-o", "ControlMaster=no", "-o", "ControlPath=/tmp/test.sock", "-o", "ConnectTimeout=5", "-T", "-p", "22", "-i", "/home/user/.ssh/id_rsa", "-W", "10.210.1.1:5000", "root@example.com"},
},
{
name: "custom port with identity file and control socket",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
Port: 2222,
KeyPath: "/home/user/.ssh/id_rsa",
},
controlSockPath: "/tmp/test.sock",
address: "10.210.1.1:5000",
expected: []string{"-o", "ControlMaster=no", "-o", "ControlPath=/tmp/test.sock", "-o", "ConnectTimeout=5", "-T", "-p", "2222", "-i", "/home/user/.ssh/id_rsa", "-W", "10.210.1.1:5000", "root@example.com"},
},
{
name: "port 0 not included",
config: SSHConnectorConfig{
User: "root",
Host: "example.com",
Port: 0,
},
controlSockPath: "/tmp/test.sock",
address: "10.210.1.1:5000",
expected: []string{"-o", "ControlMaster=no", "-o", "ControlPath=/tmp/test.sock", "-o", "ConnectTimeout=5", "-T", "-W", "10.210.1.1:5000", "root@example.com"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
d := &sshCLIDialer{config: tt.config, controlSockPath: tt.controlSockPath}
got := d.buildDialArgs(tt.address)
assert.Equal(t, tt.expected, got)
})
}
}
func TestControlSocketPath(t *testing.T) { func TestControlSocketPath(t *testing.T) {
// Note: Cannot use t.Parallel() because a subtest uses t.Setenv(). // Note: Cannot use t.Parallel() because a subtest uses t.Setenv().
+14
View File
@@ -4,9 +4,12 @@ import (
"context" "context"
"fmt" "fmt"
"net/netip" "net/netip"
"time"
"github.com/psviderski/uncloud/internal/grpcversion"
"golang.org/x/net/proxy" "golang.org/x/net/proxy"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/backoff"
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
) )
@@ -20,10 +23,21 @@ func NewTCPConnector(apiAddr netip.AddrPort) *TCPConnector {
} }
func (c *TCPConnector) Connect(_ context.Context) (*grpc.ClientConn, error) { func (c *TCPConnector) Connect(_ context.Context) (*grpc.ClientConn, error) {
// Use a faster connection backoff than the default (which grows up to 120s). TCP connections are typically to
// local Docker containers (ucind) or LAN machines where long backoff delays are unnecessary.
backoffConfig := backoff.DefaultConfig
backoffConfig.MaxDelay = 5 * time.Second
conn, err := grpc.NewClient( conn, err := grpc.NewClient(
c.apiAddr.String(), c.apiAddr.String(),
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithDefaultServiceConfig(defaultServiceConfig), grpc.WithDefaultServiceConfig(defaultServiceConfig),
grpc.WithConnectParams(grpc.ConnectParams{
Backoff: backoffConfig,
MinConnectTimeout: 5 * time.Second,
}),
grpc.WithUnaryInterceptor(grpcversion.ClientUnaryInterceptor),
grpc.WithStreamInterceptor(grpcversion.ClientStreamInterceptor),
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("create machine API client: %w", err) return nil, fmt.Errorf("create machine API client: %w", err)
+3
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"fmt" "fmt"
"github.com/psviderski/uncloud/internal/grpcversion"
"golang.org/x/net/proxy" "golang.org/x/net/proxy"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
@@ -26,6 +27,8 @@ func (c *UnixConnector) Connect(_ context.Context) (*grpc.ClientConn, error) {
target, target,
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithDefaultServiceConfig(defaultServiceConfig), grpc.WithDefaultServiceConfig(defaultServiceConfig),
grpc.WithUnaryInterceptor(grpcversion.ClientUnaryInterceptor),
grpc.WithStreamInterceptor(grpcversion.ClientStreamInterceptor),
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("create machine API client: %w", err) return nil, fmt.Errorf("create machine API client: %w", err)
+3
View File
@@ -8,6 +8,7 @@ import (
"strconv" "strconv"
"github.com/psviderski/uncloud/internal/cli/config" "github.com/psviderski/uncloud/internal/cli/config"
"github.com/psviderski/uncloud/internal/grpcversion"
"github.com/psviderski/uncloud/internal/machine/constants" "github.com/psviderski/uncloud/internal/machine/constants"
"github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/machine/network"
"github.com/psviderski/uncloud/internal/machine/network/tunnel" "github.com/psviderski/uncloud/internal/machine/network/tunnel"
@@ -70,6 +71,8 @@ func (c *WireGuardConnector) Connect(ctx context.Context) (*grpc.ClientConn, err
grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) { grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
return c.tun.DialContext(ctx, "tcp", addr) return c.tun.DialContext(ctx, "tcp", addr)
}), }),
grpc.WithUnaryInterceptor(grpcversion.ClientUnaryInterceptor),
grpc.WithStreamInterceptor(grpcversion.ClientStreamInterceptor),
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("connect to machine API through WireGuard tunnel: %w", err) return nil, fmt.Errorf("connect to machine API through WireGuard tunnel: %w", err)
+61 -13
View File
@@ -2,19 +2,22 @@ package client
import ( import (
"context" "context"
"encoding/json"
"errors" "errors"
"fmt" "fmt"
"strings" "strings"
"time" "time"
"github.com/containerd/errdefs"
"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/pkg/jsonmessage" "github.com/docker/docker/pkg/jsonmessage"
cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
"github.com/psviderski/uncloud/internal/docker" "github.com/psviderski/uncloud/internal/docker"
"github.com/psviderski/uncloud/internal/machine/api/pb"
machinedocker "github.com/psviderski/uncloud/internal/machine/docker" machinedocker "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/codes"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
) )
@@ -24,8 +27,29 @@ import (
// CreateContainer creates a new container for the given service on the specified machine. // CreateContainer creates a new container for the given service on the specified machine.
func (cli *Client) CreateContainer( func (cli *Client) CreateContainer(
ctx context.Context, serviceID string, spec api.ServiceSpec, machineID string, ctx context.Context, serviceID string, spec api.ServiceSpec, machineID string,
) (container.CreateResponse, error) { ) (api.CreateContainerResponse, error) {
var resp container.CreateResponse return cli.createServiceContainerWithPull(ctx, serviceID, spec, machineID, pb.CreateServiceContainerRequest_SERVICE)
}
// CreatePreDeployHookContainer creates a one-shot container for a pre-deploy hook for the given service
// on the specified machine.
func (cli *Client) CreatePreDeployHookContainer(
ctx context.Context, serviceID string, spec api.ServiceSpec, machineID string,
) (api.CreateContainerResponse, error) {
return cli.createServiceContainerWithPull(
ctx, serviceID, spec, machineID, pb.CreateServiceContainerRequest_PRE_DEPLOY)
}
// createServiceContainerWithPull creates a regular or deployment hook container for the service
// on the specified machine, pulling the image if needed.
func (cli *Client) createServiceContainerWithPull(
ctx context.Context,
serviceID string,
spec api.ServiceSpec,
machineID string,
containerType pb.CreateServiceContainerRequest_ContainerType,
) (api.CreateContainerResponse, error) {
var resp api.CreateContainerResponse
spec = spec.SetDefaults() spec = spec.SetDefaults()
if err := spec.Validate(); err != nil { if err := spec.Validate(); err != nil {
@@ -42,13 +66,18 @@ func (cli *Client) CreateContainer(
if err != nil { if err != nil {
return resp, fmt.Errorf("generate random suffix: %w", err) return resp, fmt.Errorf("generate random suffix: %w", err)
} }
containerName := fmt.Sprintf("%s-%s", spec.Name, suffix) containerName := fmt.Sprintf("%s-%s", spec.Name, suffix)
if containerType == pb.CreateServiceContainerRequest_PRE_DEPLOY {
containerName = fmt.Sprintf("%s-%s-%s", spec.Name, api.LabelHookPreDeploy, suffix)
}
resp.Name = containerName
// Proxy Docker gRPC requests to the selected machine. // Proxy Docker gRPC requests to the selected machine.
ctx = proxyToMachine(ctx, machine.Machine) ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", containerName, machine.Machine.Name) eventID := cliprogress.NewContainerEventID(ctx, containerName, machine.Machine.Name)
pw.Event(progress.CreatingEvent(eventID)) pw.Event(progress.CreatingEvent(eventID))
if spec.Container.PullPolicy == api.PullPolicyAlways { if spec.Container.PullPolicy == api.PullPolicyAlways {
@@ -57,7 +86,18 @@ func (cli *Client) CreateContainer(
} }
} }
resp, err = cli.Docker.CreateServiceContainer(ctx, serviceID, spec, containerName) specBytes, err := json.Marshal(spec)
if err != nil {
return resp, fmt.Errorf("marshal service spec: %w", err)
}
req := &pb.CreateServiceContainerRequest{
ServiceId: serviceID,
ServiceSpec: specBytes,
ContainerName: containerName,
ContainerType: containerType,
}
grpcResp, err := cli.Docker.GRPCClient.CreateServiceContainer(ctx, req)
if err != nil { if err != nil {
switch spec.Container.PullPolicy { switch spec.Container.PullPolicy {
case api.PullPolicyAlways, api.PullPolicyNever: case api.PullPolicyAlways, api.PullPolicyNever:
@@ -68,7 +108,7 @@ func (cli *Client) CreateContainer(
} }
// NotFound (No such image) error is expected if the image is missing. // NotFound (No such image) error is expected if the image is missing.
if !errdefs.IsNotFound(err) || !strings.Contains(err.Error(), "No such image") { if status.Code(err) != codes.NotFound || !strings.Contains(err.Error(), "No such image") {
return resp, err return resp, err
} }
@@ -76,10 +116,14 @@ func (cli *Client) CreateContainer(
if err = cli.pullImageWithProgress(ctx, spec.Container.Image, machine.Machine.Name, eventID); err != nil { if err = cli.pullImageWithProgress(ctx, spec.Container.Image, machine.Machine.Name, eventID); err != nil {
return resp, err return resp, err
} }
if resp, err = cli.Docker.CreateServiceContainer(ctx, serviceID, spec, containerName); err != nil { if grpcResp, err = cli.Docker.GRPCClient.CreateServiceContainer(ctx, req); err != nil {
return resp, err return resp, err
} }
} }
if err = json.Unmarshal(grpcResp.Response, &resp.CreateResponse); err != nil {
return resp, fmt.Errorf("unmarshal gRPC response: %w", err)
}
pw.Event(progress.CreatedEvent(eventID)) pw.Event(progress.CreatedEvent(eventID))
return resp, nil return resp, nil
@@ -87,7 +131,7 @@ func (cli *Client) CreateContainer(
func (cli *Client) pullImageWithProgress(ctx context.Context, image, machineName, parentEventID string) error { func (cli *Client) pullImageWithProgress(ctx context.Context, image, machineName, parentEventID string) error {
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Image %s on %s", image, machineName) eventID := cliprogress.ImageEventID(image, machineName)
pw.Event(progress.Event{ pw.Event(progress.Event{
ID: eventID, ID: eventID,
ParentID: parentEventID, ParentID: parentEventID,
@@ -213,7 +257,7 @@ func (cli *Client) InspectContainer(
} }
prefixMatchCandidates := []api.MachineServiceContainer{} prefixMatchCandidates := []api.MachineServiceContainer{}
for _, c := range svc.Containers { for _, c := range append(svc.Containers, svc.HookContainers...) {
if c.Container.ID == containerNameOrID || if c.Container.ID == containerNameOrID ||
c.Container.Name == containerNameOrID { c.Container.Name == containerNameOrID {
return c, nil return c, nil
@@ -248,7 +292,7 @@ func (cli *Client) StartContainer(ctx context.Context, serviceNameOrID, containe
ctx = proxyToMachine(ctx, machine.Machine) ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.Name, machine.Machine.Name) eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, machine.Machine.Name)
pw.Event(progress.StartingEvent(eventID)) pw.Event(progress.StartingEvent(eventID))
if err = cli.Docker.StartContainer(ctx, ctr.Container.ID, container.StartOptions{}); err != nil { if err = cli.Docker.StartContainer(ctx, ctr.Container.ID, container.StartOptions{}); err != nil {
@@ -260,6 +304,8 @@ func (cli *Client) StartContainer(ctx context.Context, serviceNameOrID, containe
} }
// StopContainer stops the specified container within the service. // StopContainer stops the specified container within the service.
//
//nolint:dupl // Structurally similar to RemoveContainer but performs a different operation.
func (cli *Client) StopContainer( func (cli *Client) StopContainer(
ctx context.Context, serviceNameOrID, containerNameOrID string, opts container.StopOptions, ctx context.Context, serviceNameOrID, containerNameOrID string, opts container.StopOptions,
) error { ) error {
@@ -275,7 +321,7 @@ func (cli *Client) StopContainer(
ctx = proxyToMachine(ctx, machine.Machine) ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.Name, machine.Machine.Name) eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, machine.Machine.Name)
pw.Event(progress.StoppingEvent(eventID)) pw.Event(progress.StoppingEvent(eventID))
if err = cli.Docker.StopContainer(ctx, ctr.Container.ID, opts); err != nil { if err = cli.Docker.StopContainer(ctx, ctr.Container.ID, opts); err != nil {
@@ -287,6 +333,8 @@ func (cli *Client) StopContainer(
} }
// RemoveContainer removes the specified container within the service. // RemoveContainer removes the specified container within the service.
//
//nolint:dupl // Structurally similar to StopContainer but performs a different operation.
func (cli *Client) RemoveContainer( func (cli *Client) RemoveContainer(
ctx context.Context, serviceNameOrID, containerNameOrID string, opts container.RemoveOptions, ctx context.Context, serviceNameOrID, containerNameOrID string, opts container.RemoveOptions,
) error { ) error {
@@ -302,7 +350,7 @@ func (cli *Client) RemoveContainer(
ctx = proxyToMachine(ctx, machine.Machine) ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", ctr.Container.Name, machine.Machine.Name) eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, machine.Machine.Name)
pw.Event(progress.RemovingEvent(eventID)) pw.Event(progress.RemovingEvent(eventID))
if err = cli.Docker.RemoveServiceContainer(ctx, ctr.Container.ID, opts); err != nil { if err = cli.Docker.RemoveServiceContainer(ctx, ctr.Container.ID, opts); err != nil {
@@ -382,7 +430,7 @@ func (cli *Client) WaitContainerHealthy(
} }
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Container %s on %s", mc.Container.Name, machine.Machine.Name) eventID := cliprogress.ContainerEventID(ctx, mc.Container.ServiceSpec.Name, mc.Container.ID, machine.Machine.Name)
var monitor time.Duration var monitor time.Duration
if opts.MonitorPeriod == nil { if opts.MonitorPeriod == nil {
+3 -3
View File
@@ -114,7 +114,7 @@ func (sp *ServicePlan) Format() string {
if sp.IsNewService { if sp.IsNewService {
specTable.Row("", "image:", tui.FormatImage(sp.Spec.Container.Image, tui.Green)) specTable.Row("", "image:", tui.FormatImage(sp.Spec.Container.Image, tui.Green))
} else { } else {
specTable.Row("", "image:", tui.FormatImage(sp.Spec.Container.Image, lipgloss.NewStyle())) specTable.Row("", "image:", tui.FormatImage(sp.Spec.Container.Image, tui.NoStyle))
} }
} else { } else {
mod := "" mod := ""
@@ -233,7 +233,7 @@ func formatImageDiff(oldImage, newImage string) string {
} }
if oldImage == newImage { if oldImage == newImage {
return tui.FormatImage(newImage, lipgloss.NewStyle()) return tui.FormatImage(newImage, tui.NoStyle)
} }
oldRef, _ := reference.ParseDockerRef(oldImage) oldRef, _ := reference.ParseDockerRef(oldImage)
@@ -337,7 +337,7 @@ func (d *Deployment) Validate(ctx context.Context) error {
return fmt.Errorf("invalid service spec: %w", err) return fmt.Errorf("invalid service spec: %w", err)
} }
if d.Service == nil { if d.Service == nil && d.Spec.Name != "" {
svc, err := d.cli.InspectService(ctx, d.Spec.Name) svc, err := d.cli.InspectService(ctx, d.Spec.Name)
if err == nil { if err == nil {
d.Service = &svc d.Service = &svc
+9 -2
View File
@@ -8,6 +8,7 @@ import (
"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/pkg/stringid" "github.com/docker/docker/pkg/stringid"
cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
"github.com/psviderski/uncloud/internal/cli/tui" "github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
) )
@@ -28,6 +29,10 @@ func (o *RunContainerOperation) Execute(ctx context.Context, cli Client) error {
if err != nil { if err != nil {
return fmt.Errorf("create container: %w", err) return fmt.Errorf("create container: %w", err)
} }
// Override event ID so StartContainer and WaitContainerHealthy update the same progress line as creation.
// TODO: This is a hack to work around the limitations of the compose progress library.
// We likely need to fork or create our own to decouple event IDs from presentation layer.
ctx = cliprogress.WithEventID(ctx, cliprogress.NewContainerEventID(ctx, resp.Name, o.MachineName))
if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil { if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil {
return fmt.Errorf("start container: %w", err) return fmt.Errorf("start container: %w", err)
} }
@@ -171,13 +176,15 @@ func (o *ReplaceContainerOperation) Execute(ctx context.Context, cli Client) err
if err != nil { if err != nil {
return fmt.Errorf("create new container: %w", err) return fmt.Errorf("create new container: %w", err)
} }
if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil { // Override event ID so StartContainer and WaitContainerHealthy update the same progress line as creation.
newCtx := cliprogress.WithEventID(ctx, cliprogress.NewContainerEventID(ctx, resp.Name, o.MachineName))
if err = cli.StartContainer(newCtx, o.ServiceID, resp.ID); err != nil {
return fmt.Errorf("start new container: %w", err) return fmt.Errorf("start new container: %w", err)
} }
if !o.SkipHealthMonitor { if !o.SkipHealthMonitor {
opts := api.WaitContainerHealthyOptions{MonitorPeriod: o.Spec.UpdateConfig.MonitorPeriod} opts := api.WaitContainerHealthyOptions{MonitorPeriod: o.Spec.UpdateConfig.MonitorPeriod}
if err = cli.WaitContainerHealthy(ctx, o.ServiceID, resp.ID, opts); err != nil { if err = cli.WaitContainerHealthy(newCtx, o.ServiceID, resp.ID, opts); err != nil {
// New container failed to become healthy. Stop it and roll back to the previous container. // New container failed to become healthy. Stop it and roll back to the previous container.
// Don't remove the new stopped container to allow users to inspect logs and state. // Don't remove the new stopped container to allow users to inspect logs and state.
// TODO: collect logs from the new container and include in the error message to speed up debugging. // TODO: collect logs from the new container and include in the error message to speed up debugging.
+200
View File
@@ -0,0 +1,200 @@
package operation
import (
"context"
"fmt"
"strings"
"time"
"charm.land/lipgloss/v2"
"github.com/charmbracelet/x/ansi"
"github.com/containerd/errdefs"
"github.com/docker/compose/v2/pkg/progress"
"github.com/docker/docker/api/types/container"
cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
"github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/pkg/api"
)
// DefaultPreDeployTimeout is the maximum duration to wait for a pre-deploy hook container to complete.
const DefaultPreDeployTimeout = 5 * time.Minute
// StopPreDeployOperation stops a running pre-deploy hook container from a previous deployment.
type StopPreDeployOperation struct {
MachineID string
// MachineName is used for formatting the operation as part of the deployment plan.
MachineName string
Container api.ServiceContainer
}
func (o *StopPreDeployOperation) Execute(ctx context.Context, cli Client) error {
ctx = cliprogress.WithEventID(ctx,
cliprogress.OldPreDeployHookEventID(o.Container.ServiceName(), o.Container.ID, o.MachineName))
if err := cli.StopContainer(ctx, o.Container.ServiceID(), o.Container.ID, container.StopOptions{}); err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("stop pre-deploy hook container '%s': %w", o.Container.ID, err)
}
}
return nil
}
func (o *StopPreDeployOperation) Format() string {
displayName := o.Container.ServiceSpec.Name + tui.Faint.Render("/") + o.Container.ShortID()
status, _ := o.Container.HumanState()
return tui.BoldRed.Render("⏹") + " " +
tui.Faint.Render("stop pre-deploy hook") + " " +
displayName + " " +
tui.Faint.Render("("+status+")") + " " +
tui.Faint.Render("on") + " " +
o.MachineName
}
func (o *StopPreDeployOperation) String() string {
return fmt.Sprintf("StopPreDeployOperation[machine_id=%s container_id=%s]",
o.MachineID, o.Container.ID)
}
// RunPreDeployOperation runs a one-shot pre-deploy hook container before service deployment.
type RunPreDeployOperation struct {
ServiceID string
Spec api.ServiceSpec
MachineID string
// MachineName is used for formatting the operation as part of the deployment plan.
MachineName string
// OldContainerIDs are pre-deploy hook containers from previous deployments to remove.
OldContainerIDs []string
}
func (o *RunPreDeployOperation) Execute(ctx context.Context, cli Client) error {
// Remove old pre-deploy containers.
for _, id := range o.OldContainerIDs {
oldCtx := cliprogress.WithEventID(ctx, cliprogress.OldPreDeployHookEventID(o.Spec.Name, id, o.MachineName))
_ = cli.StopContainer(oldCtx, o.ServiceID, id, container.StopOptions{})
err := cli.RemoveContainer(oldCtx, o.ServiceID, id, container.RemoveOptions{RemoveVolumes: true})
if err != nil && !errdefs.IsNotFound(err) {
return fmt.Errorf("remove old pre-deploy hook container '%s': %w", id, err)
}
}
// Set the event ID override so all downstream client methods (create, start, stop) use the same
// pre-deploy hook progress event instead of the generic container one.
ctx = cliprogress.WithEventID(ctx, cliprogress.PreDeployHookEventID(o.Spec.Name, o.MachineName))
resp, err := cli.CreatePreDeployHookContainer(ctx, o.ServiceID, o.Spec, o.MachineID)
if err != nil {
return fmt.Errorf("create pre-deploy hook container: %w", err)
}
if err = cli.StartContainer(ctx, o.ServiceID, resp.ID); err != nil {
return fmt.Errorf("start pre-deploy hook container: %w", err)
}
timeout := DefaultPreDeployTimeout
if o.Spec.PreDeploy.Timeout != nil {
timeout = *o.Spec.PreDeploy.Timeout
}
return o.waitForExit(ctx, cli, resp.ID, timeout)
}
// waitForExit polls the container state until it exits or the timeout is reached.
func (o *RunPreDeployOperation) waitForExit(
ctx context.Context, cli Client, containerID string, timeout time.Duration,
) error {
pw := progress.ContextWriter(ctx)
eventID := cliprogress.PreDeployHookEventID(o.Spec.Name, o.MachineName)
pw.Event(progress.Waiting(eventID))
timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
var ctr *api.ServiceContainer
for {
select {
case <-timeoutCtx.Done():
// Stop the container on timeout or context cancellation.
_ = cli.StopContainer(ctx, o.ServiceID, containerID, container.StopOptions{})
ctrID := containerID
if ctr != nil {
ctrID = fmt.Sprintf("%s/%s", o.Spec.Name, ctr.ShortID())
}
if ctx.Err() != nil {
// The parent context has been cancelled before the timeout.
pw.Event(progress.NewEvent(eventID, progress.Error, "Cancelled"))
return fmt.Errorf("pre-deploy hook container '%s': %w", ctrID, ctx.Err())
}
pw.Event(progress.NewEvent(eventID, progress.Error, fmt.Sprintf("Timeout (%s)", timeout)))
return fmt.Errorf("pre-deploy hook container '%s' timed out after %s. "+
"It's stopped and available for inspection. Fetch logs with 'uc logs %s'",
ctrID, timeout, o.Spec.Name)
case <-ticker.C:
mc, err := cli.InspectContainer(ctx, o.ServiceID, containerID)
if err != nil {
return fmt.Errorf("inspect pre-deploy hook container: %w", err)
}
ctr = &mc.Container
if ctr.State.Running {
if state, err := ctr.Container.HumanState(); err == nil {
pw.Event(progress.NewEvent(eventID, progress.Working, fmt.Sprintf("Waiting (%s)", state)))
}
continue
}
// Hook container has exited successfully.
if ctr.State.ExitCode == 0 {
pw.Event(progress.Event{
ID: eventID,
Status: progress.Done,
})
return nil
}
pw.Event(progress.ErrorEvent(eventID))
ctrID := fmt.Sprintf("%s/%s", o.Spec.Name, ctr.ShortID())
return fmt.Errorf("pre-deploy hook container '%s' failed with exit code: %d. "+
"It's stopped and available for inspection. Fetch logs with 'uc logs %s'",
ctrID, ctr.State.ExitCode, o.Spec.Name)
}
}
}
func (o *RunPreDeployOperation) Format() string {
cmd := strings.Join(o.Spec.PreDeploy.Command, " ")
timeout := DefaultPreDeployTimeout
if o.Spec.PreDeploy.Timeout != nil {
timeout = *o.Spec.PreDeploy.Timeout
}
prefix := tui.BoldGreen.Render("▶") + " " +
tui.Faint.Render("run pre-deploy hook") + " " +
o.Spec.Name + " ["
suffix := "] " +
tui.Faint.Render("on") + " " +
o.MachineName + " " +
tui.Yellow.Render(fmt.Sprintf("(timeout %s)", timeout))
// Truncate the command to fit within the terminal width.
termWidth := tui.TerminalWidth()
if termWidth > 0 {
buffer := 10
maxCmdWidth := termWidth - lipgloss.Width(prefix) - lipgloss.Width(suffix) - buffer
if maxCmdWidth <= 0 {
maxCmdWidth = 10
}
cmd = ansi.Truncate(cmd, maxCmdWidth, "…")
}
return prefix + cmd + suffix
}
func (o *RunPreDeployOperation) String() string {
return fmt.Sprintf("RunPreDeployOperation[machine_id=%s service_id=%s cmd=%v]",
o.MachineID, o.ServiceID, o.Spec.PreDeploy.Command)
}
+8
View File
@@ -67,3 +67,11 @@ func (s *ClusterState) Machine(nameOrID string) (*Machine, bool) {
} }
return nil, false return nil, false
} }
// MachineName returns the machine name by ID from the cluster state. If the id is not found, ("", false) is returned.
func (s *ClusterState) MachineName(id string) (string, bool) {
if m, ok := s.Machine(id); ok {
return m.Info.Name, true
}
return "", false
}
+67
View File
@@ -199,6 +199,10 @@ func (s *RollingStrategy) planReplicated(svc *api.Service, spec api.ServiceSpec)
} }
} }
if ops := s.preDeployOperations(svc, plan); len(ops) > 0 {
plan.Operations = append(ops, plan.Operations...)
}
return plan, nil return plan, nil
} }
@@ -265,6 +269,10 @@ func (s *RollingStrategy) planGlobal(svc *api.Service, spec api.ServiceSpec) (Se
} }
} }
if ops := s.preDeployOperations(svc, plan); len(ops) > 0 {
plan.Operations = append(ops, plan.Operations...)
}
return plan, nil return plan, nil
} }
@@ -324,6 +332,7 @@ func reconcileGlobalContainer(
break break
} }
// TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported. // TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported.
// Make sure to update preDeployOperation accordingly.
} }
if upToDate { if upToDate {
return ops, nil return ops, nil
@@ -435,6 +444,64 @@ func determineUpdateOrder(oldContainer api.ServiceContainer, spec api.ServiceSpe
return api.UpdateOrderStartFirst return api.UpdateOrderStartFirst
} }
// preDeployOperations returns operations for the pre-deploy hook if the spec has one and the plan updates the service.
// It prepends StopPreDeployOperations for any running hook containers, followed by a RunPreDeployOperation on the same
// machine as the first run/replace operation.
func (s *RollingStrategy) preDeployOperations(svc *api.Service, plan ServicePlan) []operation.Operation {
if plan.Spec.PreDeploy == nil {
return nil
}
// Find the first run or replace operation to determine the target machine.
var machineID, machineName string
for _, op := range plan.Operations {
switch o := op.(type) {
case *operation.RunContainerOperation:
machineID = o.MachineID
machineName = o.MachineName
case *operation.ReplaceContainerOperation:
machineID = o.MachineID
machineName = o.MachineName
default:
continue
}
break
}
// Skip the hook as there are no run or replace operations.
if machineID == "" {
return nil
}
var ops []operation.Operation
// Collect old pre-deploy container IDs to clean up and stop any that are still running.
var oldContainerIDs []string
if svc != nil {
for _, c := range svc.HookContainers {
oldContainerIDs = append(oldContainerIDs, c.Container.ID)
if c.Container.State.Running {
hookMachineName, _ := s.state.MachineName(c.MachineID)
ops = append(ops, &operation.StopPreDeployOperation{
MachineID: c.MachineID,
MachineName: hookMachineName,
Container: c.Container,
})
}
}
}
ops = append(ops, &operation.RunPreDeployOperation{
ServiceID: plan.ServiceID,
Spec: plan.Spec,
MachineID: machineID,
MachineName: machineName,
OldContainerIDs: oldContainerIDs,
})
return ops
}
// newEmptyServicePlan creates a new empty plan for a service deployment with initialised service ID and name. // newEmptyServicePlan creates a new empty plan for a service deployment with initialised service ID and name.
func newEmptyServicePlan(svc *api.Service, spec api.ServiceSpec) (ServicePlan, error) { func newEmptyServicePlan(svc *api.Service, spec api.ServiceSpec) (ServicePlan, error) {
plan := ServicePlan{ plan := ServicePlan{
+242
View File
@@ -9,6 +9,7 @@ import (
"github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/api/pb"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/psviderski/uncloud/pkg/client/deploy/operation" "github.com/psviderski/uncloud/pkg/client/deploy/operation"
"github.com/psviderski/uncloud/pkg/client/deploy/scheduler"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
) )
@@ -423,6 +424,246 @@ func TestReconcileGlobalContainer(t *testing.T) {
} }
} }
func TestPreDeployOperations(t *testing.T) {
hook := &api.PreDeployHook{
Command: []string{"db", "migrate"},
}
strategy := &RollingStrategy{
state: &scheduler.ClusterState{
Machines: []*scheduler.Machine{
{Info: &pb.MachineInfo{Id: "m-1", Name: "machine-1"}},
{Info: &pb.MachineInfo{Id: "m-2", Name: "machine-2"}},
{Info: &pb.MachineInfo{Id: "m-3", Name: "machine-3"}},
},
},
}
runningHook1 := newServiceContainer("running-hook-1", container.State{Running: true, Status: "running"})
runningHook2 := newServiceContainer("running-hook-2", container.State{Running: true, Status: "running"})
tests := []struct {
name string
plan ServicePlan
svc *api.Service
expected []operation.Operation
}{
{
name: "no pre-deploy hook in spec",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
},
},
},
expected: nil,
},
{
name: "plan has RunContainerOperation",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
},
},
},
expected: []operation.Operation{
&operation.RunPreDeployOperation{
ServiceID: "svc-1",
MachineID: "m-1",
MachineName: "machine-1",
},
},
},
{
name: "plan has ReplaceContainerOperation",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.ReplaceContainerOperation{MachineID: "m-2", MachineName: "machine-2"},
},
},
},
expected: []operation.Operation{
&operation.RunPreDeployOperation{
ServiceID: "svc-1",
MachineID: "m-2",
MachineName: "machine-2",
},
},
},
{
name: "plan has only RemoveContainerOperation",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.RemoveContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
},
},
},
expected: nil,
},
{
name: "plan has only StopContainerOperation",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.StopContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
},
},
},
expected: nil,
},
{
name: "empty plan with no operations",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
},
expected: nil,
},
{
name: "stopped hook containers are collected for cleanup",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
},
},
},
svc: &api.Service{
HookContainers: []api.MachineServiceContainer{
{MachineID: "m-1", Container: newServiceContainer("old-hook-1", container.State{Status: "exited"})},
},
},
expected: []operation.Operation{
&operation.RunPreDeployOperation{
ServiceID: "svc-1",
MachineID: "m-1",
MachineName: "machine-1",
OldContainerIDs: []string{"old-hook-1"},
},
},
},
{
name: "running hook containers are stopped before run",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.RunContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
},
},
},
svc: &api.Service{
HookContainers: []api.MachineServiceContainer{
{MachineID: "m-2", Container: runningHook1},
},
},
expected: []operation.Operation{
&operation.StopPreDeployOperation{
MachineID: "m-2",
MachineName: "machine-2",
Container: runningHook1,
},
&operation.RunPreDeployOperation{
ServiceID: "svc-1",
MachineID: "m-1",
MachineName: "machine-1",
OldContainerIDs: []string{"running-hook-1"},
},
},
},
{
name: "mixed stopped and running hook containers and mixed container operations",
plan: ServicePlan{
ServiceID: "svc-1",
ServiceName: "app",
Spec: api.ServiceSpec{PreDeploy: hook},
SequenceOperation: operation.SequenceOperation{
Operations: []operation.Operation{
&operation.StopContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
&operation.RemoveContainerOperation{MachineID: "m-1", MachineName: "machine-1"},
&operation.ReplaceContainerOperation{MachineID: "m-2", MachineName: "machine-2"},
&operation.RunContainerOperation{MachineID: "m-3", MachineName: "machine-3"},
&operation.RemoveContainerOperation{MachineID: "m-3", MachineName: "machine-3"},
},
},
},
svc: &api.Service{
HookContainers: []api.MachineServiceContainer{
{MachineID: "m-1", Container: newServiceContainer("stopped-hook", container.State{Status: "exited"})},
{MachineID: "m-1", Container: runningHook1},
{MachineID: "m-3", Container: runningHook2},
},
},
expected: []operation.Operation{
&operation.StopPreDeployOperation{
MachineID: "m-1",
MachineName: "machine-1",
Container: runningHook1,
},
&operation.StopPreDeployOperation{
MachineID: "m-3",
MachineName: "machine-3",
Container: runningHook2,
},
&operation.RunPreDeployOperation{
ServiceID: "svc-1",
MachineID: "m-2",
MachineName: "machine-2",
OldContainerIDs: []string{"stopped-hook", "running-hook-1", "running-hook-2"},
},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result := strategy.preDeployOperations(tt.svc, tt.plan)
if tt.expected == nil {
assert.Nil(t, result)
return
}
assertOperationsEqual(t, tt.expected, result)
})
}
}
// newServiceContainer creates an api.ServiceContainer with the given ID and state.
func newServiceContainer(id string, state container.State) api.ServiceContainer {
return api.ServiceContainer{Container: api.Container{
InspectResponse: container.InspectResponse{
ContainerJSONBase: &container.ContainerJSONBase{
ID: id,
State: &state,
},
},
}}
}
// assertOperationsEqual compares expected and actual operations, ignoring the Spec field // assertOperationsEqual compares expected and actual operations, ignoring the Spec field
// which is passed separately to the function and not the focus of these tests. // which is passed separately to the function and not the focus of these tests.
func assertOperationsEqual(t *testing.T, expected, actual []operation.Operation) { func assertOperationsEqual(t *testing.T, expected, actual []operation.Operation) {
@@ -430,6 +671,7 @@ func assertOperationsEqual(t *testing.T, expected, actual []operation.Operation)
opts := cmp.Options{ opts := cmp.Options{
cmpopts.IgnoreFields(operation.RunContainerOperation{}, "Spec"), cmpopts.IgnoreFields(operation.RunContainerOperation{}, "Spec"),
cmpopts.IgnoreFields(operation.ReplaceContainerOperation{}, "Spec"), cmpopts.IgnoreFields(operation.ReplaceContainerOperation{}, "Spec"),
cmpopts.IgnoreFields(operation.RunPreDeployOperation{}, "Spec"),
cmpopts.IgnoreUnexported(api.Container{}), cmpopts.IgnoreUnexported(api.Container{}),
} }
if diff := cmp.Diff(expected, actual, opts); diff != "" { if diff := cmp.Diff(expected, actual, opts); diff != "" {
+2 -1
View File
@@ -13,6 +13,7 @@ import (
"github.com/cenkalti/backoff/v4" "github.com/cenkalti/backoff/v4"
"github.com/docker/compose/v2/pkg/progress" "github.com/docker/compose/v2/pkg/progress"
cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
"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/pkg/api" "github.com/psviderski/uncloud/pkg/api"
@@ -148,7 +149,7 @@ func verifyCaddyReachable(ctx context.Context, m *pb.MachineInfo) error {
publicIP, _ := m.PublicIp.ToAddr() publicIP, _ := m.PublicIp.ToAddr()
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Machine %s (%s)", m.Name, publicIP) eventID := cliprogress.MachineEventID(m.Name, publicIP.String())
pw.Event(progress.NewEvent(eventID, progress.Working, "Querying")) pw.Event(progress.NewEvent(eventID, progress.Working, "Querying"))
verifyURL := getVerifyURL(publicIP) verifyURL := getVerifyURL(publicIP)
+1 -1
View File
@@ -214,7 +214,7 @@ func (m *LogMerger) run() {
for _, s := range stalled { for _, s := range stalled {
errEntry := api.ServiceLogEntry{ errEntry := api.ServiceLogEntry{
ContainerLogEntry: api.ContainerLogEntry{ LogEntry: api.LogEntry{
Err: api.ErrLogStreamStalled, Err: api.ErrLogStreamStalled,
}, },
} }
+3 -3
View File
@@ -12,7 +12,7 @@ import (
// testEntry creates a ServiceLogEntry for testing. // testEntry creates a ServiceLogEntry for testing.
func testEntry(stream api.LogStreamType, ts time.Time, msg string) api.ServiceLogEntry { func testEntry(stream api.LogStreamType, ts time.Time, msg string) api.ServiceLogEntry {
return api.ServiceLogEntry{ return api.ServiceLogEntry{
ContainerLogEntry: api.ContainerLogEntry{ LogEntry: api.LogEntry{
Stream: stream, Stream: stream,
Timestamp: ts, Timestamp: ts,
Message: []byte(msg), Message: []byte(msg),
@@ -91,7 +91,7 @@ func TestLogMerger_PreservesData(t *testing.T) {
e := api.ServiceLogEntry{ e := api.ServiceLogEntry{
Metadata: metadata, Metadata: metadata,
ContainerLogEntry: api.ContainerLogEntry{ LogEntry: api.LogEntry{
Stream: api.LogStreamStdout, Stream: api.LogStreamStdout,
Timestamp: time.Now(), Timestamp: time.Now(),
Message: []byte("test"), Message: []byte("test"),
@@ -199,7 +199,7 @@ func TestLogMerger_ErrorForwarding(t *testing.T) {
ch1 <- testEntry(api.LogStreamStdout, t1, "ch1-first") ch1 <- testEntry(api.LogStreamStdout, t1, "ch1-first")
// Send an error entry. // Send an error entry.
ch1 <- api.ServiceLogEntry{ ch1 <- api.ServiceLogEntry{
ContainerLogEntry: api.ContainerLogEntry{ LogEntry: api.LogEntry{
Err: assert.AnError, Err: assert.AnError,
}, },
} }
+17 -16
View File
@@ -24,7 +24,8 @@ func (cli *Client) ServiceLogs(
return svc, nil, fmt.Errorf("inspect service: %w", err) return svc, nil, fmt.Errorf("inspect service: %w", err)
} }
if len(svc.Containers) == 0 { allContainers := append(svc.Containers, svc.HookContainers...)
if len(allContainers) == 0 {
return svc, nil, fmt.Errorf("no containers found for service: %s", serviceNameOrID) return svc, nil, fmt.Errorf("no containers found for service: %s", serviceNameOrID)
} }
@@ -35,8 +36,8 @@ func (cli *Client) ServiceLogs(
return svc, nil, fmt.Errorf("list machines: %w", err) return svc, nil, fmt.Errorf("list machines: %w", err)
} }
ctrStreams := make([]<-chan api.ServiceLogEntry, 0, len(svc.Containers)) ctrStreams := make([]<-chan api.ServiceLogEntry, 0, len(allContainers))
for _, ctr := range svc.Containers { for _, ctr := range allContainers {
// Skip containers not running on the specified machines. // Skip containers not running on the specified machines.
m := machines.FindByNameOrID(ctr.MachineID) m := machines.FindByNameOrID(ctr.MachineID)
if len(opts.Machines) > 0 && m == nil { if len(opts.Machines) > 0 && m == nil {
@@ -81,18 +82,18 @@ func (cli *Client) ServiceLogs(
// ContainerLogs streams log entries from a single container on a specified machine. // ContainerLogs streams log entries from a single container on a specified machine.
func (cli *Client) ContainerLogs( func (cli *Client) ContainerLogs(
ctx context.Context, machineNameOrID string, containerID string, opts api.ServiceLogsOptions, ctx context.Context, machineNameOrID string, containerID string, opts api.ServiceLogsOptions,
) (<-chan api.ContainerLogEntry, error) { ) (<-chan api.LogEntry, error) {
proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineNameOrID}) proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineNameOrID})
if err != nil { if err != nil {
return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineNameOrID, err) return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineNameOrID, err)
} }
req := &pb.ContainerLogsRequest{ req := &pb.LogsRequest{
ContainerId: containerID, Id: containerID,
Follow: opts.Follow, Follow: opts.Follow,
Tail: int32(opts.Tail), Tail: int32(opts.Tail),
Since: opts.Since, Since: opts.Since,
Until: opts.Until, Until: opts.Until,
} }
if !opts.Follow && opts.Tail == 0 { if !opts.Follow && opts.Tail == 0 {
// If not following and tail is 0, set tail to -1 to return all logs. // If not following and tail is 0, set tail to -1 to return all logs.
@@ -105,7 +106,7 @@ func (cli *Client) ContainerLogs(
return nil, err return nil, err
} }
ch := make(chan api.ContainerLogEntry) ch := make(chan api.LogEntry)
go func() { go func() {
defer close(ch) defer close(ch)
@@ -116,13 +117,13 @@ func (cli *Client) ContainerLogs(
return return
} }
if err != nil { if err != nil {
ch <- api.ContainerLogEntry{ ch <- api.LogEntry{
Err: err, Err: err,
} }
return return
} }
entry := api.ContainerLogEntry{ entry := api.LogEntry{
Stream: api.LogStreamTypeFromProto(pbEntry.Stream), Stream: api.LogStreamTypeFromProto(pbEntry.Stream),
Message: pbEntry.Message, Message: pbEntry.Message,
Timestamp: pbEntry.Timestamp.AsTime(), Timestamp: pbEntry.Timestamp.AsTime(),
@@ -141,15 +142,15 @@ func (cli *Client) ContainerLogs(
// logsStreamWithServiceMetadata wraps a container logs stream and enriches each log entry with service metadata. // logsStreamWithServiceMetadata wraps a container logs stream and enriches each log entry with service metadata.
func logsStreamWithServiceMetadata( func logsStreamWithServiceMetadata(
stream <-chan api.ContainerLogEntry, metadata api.ServiceLogEntryMetadata, stream <-chan api.LogEntry, metadata api.ServiceLogEntryMetadata,
) <-chan api.ServiceLogEntry { ) <-chan api.ServiceLogEntry {
out := make(chan api.ServiceLogEntry) out := make(chan api.ServiceLogEntry)
go func() { go func() {
for entry := range stream { for entry := range stream {
out <- api.ServiceLogEntry{ out <- api.ServiceLogEntry{
Metadata: metadata, Metadata: metadata,
ContainerLogEntry: entry, LogEntry: entry,
} }
} }
close(out) close(out)
+28 -21
View File
@@ -109,7 +109,7 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
} }
listCtx := metadata.NewOutgoingContext(ctx, md) listCtx := metadata.NewOutgoingContext(ctx, md)
// List all service containers including stopped ones. // List all service containers including stopped ones and deployment hooks.
opts := container.ListOptions{All: true} opts := container.ListOptions{All: true}
machineContainers, err := cli.Docker.ListServiceContainers(listCtx, nameOrID, opts) machineContainers, err := cli.Docker.ListServiceContainers(listCtx, nameOrID, opts)
if err != nil { if err != nil {
@@ -146,16 +146,15 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
} }
} }
for _, ctr := range mc.Containers { // Collect both regular and hook containers for the service.
if ctr.ServiceID() == nameOrID || ctr.ServiceName() == nameOrID { for _, ctr := range append(mc.Containers, mc.HookContainers...) {
containers = append(containers, api.MachineServiceContainer{ containers = append(containers, api.MachineServiceContainer{
MachineID: machineID, MachineID: machineID,
Container: ctr, Container: ctr,
}) })
if ctr.ServiceID() == nameOrID { if ctr.ServiceID() == nameOrID {
foundByID = true foundByID = true
}
} }
} }
} }
@@ -181,14 +180,22 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
} }
} }
svc = api.Service{ // Partition containers into regular service containers and hook containers.
ID: containers[0].Container.ServiceID(), var serviceContainers, hookContainers []api.MachineServiceContainer
Name: containers[0].Container.ServiceName(), for _, mc := range containers {
Mode: containers[0].Container.ServiceMode(), if mc.Container.IsHook() {
Containers: containers, hookContainers = append(hookContainers, mc)
} else {
serviceContainers = append(serviceContainers, mc)
}
} }
if svc.Mode == "" {
svc.Mode = api.ServiceModeReplicated svc = api.Service{
ID: containers[0].Container.ServiceID(),
Name: containers[0].Container.ServiceName(),
Mode: containers[0].Container.ServiceMode(),
Containers: serviceContainers,
HookContainers: hookContainers,
} }
return svc, nil return svc, nil
@@ -239,7 +246,7 @@ func (cli *Client) RemoveService(ctx context.Context, id string) error {
errCh := make(chan error) errCh := make(chan error)
// Remove all containers on all machines that belong to the service. // Remove all containers on all machines that belong to the service.
for _, mc := range svc.Containers { for _, mc := range append(svc.Containers, svc.HookContainers...) {
wg.Go(func() { wg.Go(func() {
err := cli.StopContainer(ctx, svc.ID, mc.Container.ID, container.StopOptions{}) err := cli.StopContainer(ctx, svc.ID, mc.Container.ID, container.StopOptions{})
if err != nil { if err != nil {
@@ -280,8 +287,8 @@ func (cli *Client) StopService(ctx context.Context, id string, opts container.St
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
errCh := make(chan error) errCh := make(chan error)
// Stop all containers on all machines that belong to the service. // Stop all containers on all machines that belong to the service, including hook containers.
for _, mc := range svc.Containers { for _, mc := range append(svc.Containers, svc.HookContainers...) {
wg.Go(func() { wg.Go(func() {
err := cli.StopContainer(ctx, svc.ID, mc.Container.ID, opts) err := cli.StopContainer(ctx, svc.ID, mc.Container.ID, opts)
if err != nil { if err != nil {
@@ -371,7 +378,7 @@ func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) {
continue continue
} }
for _, ctr := range mc.Containers { for _, ctr := range append(mc.Containers, mc.HookContainers...) {
if _, ok := servicesByID[ctr.ServiceID()]; ok { if _, ok := servicesByID[ctr.ServiceID()]; ok {
continue continue
} }
+3 -2
View File
@@ -7,6 +7,7 @@ import (
"github.com/containerd/errdefs" "github.com/containerd/errdefs"
"github.com/docker/compose/v2/pkg/progress" "github.com/docker/compose/v2/pkg/progress"
"github.com/docker/docker/api/types/volume" "github.com/docker/docker/api/types/volume"
cliprogress "github.com/psviderski/uncloud/internal/cli/progress"
"github.com/psviderski/uncloud/internal/cli/tui" "github.com/psviderski/uncloud/internal/cli/tui"
"github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/api/pb"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
@@ -30,7 +31,7 @@ func (cli *Client) CreateVolume(
ctx = proxyToMachine(ctx, machine.Machine) ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Volume %s on %s", opts.Name, machine.Machine.Name) eventID := cliprogress.VolumeEventID(opts.Name, machine.Machine.Name)
pw.Event(progress.CreatingEvent(eventID)) pw.Event(progress.CreatingEvent(eventID))
vol, err := cli.Docker.CreateVolume(ctx, opts) vol, err := cli.Docker.CreateVolume(ctx, opts)
@@ -120,7 +121,7 @@ func (cli *Client) RemoveVolume(ctx context.Context, machineNameOrID, volumeName
ctx = proxyToMachine(ctx, machine.Machine) ctx = proxyToMachine(ctx, machine.Machine)
pw := progress.ContextWriter(ctx) pw := progress.ContextWriter(ctx)
eventID := fmt.Sprintf("Volume %s on %s", volumeName, machine.Machine.Name) eventID := cliprogress.VolumeEventID(volumeName, machine.Machine.Name)
pw.Event(progress.RemovingEvent(eventID)) pw.Event(progress.RemovingEvent(eventID))
if err = cli.Docker.RemoveVolume(ctx, volumeName, force); err != nil { if err = cli.Docker.RemoveVolume(ctx, volumeName, force); err != nil {
+1 -1
View File
@@ -25,7 +25,7 @@ trap cleanup INT TERM EXIT
dind dockerd & dind dockerd &
echo "Waiting for Docker in Docker to be ready..." echo "Waiting for Docker in Docker to be ready..."
timeout 5s sh -c "until docker info &> /dev/null; do sleep 0.1; done" timeout 60s sh -c "until docker info &> /dev/null; do sleep 0.5; done"
echo "Docker in Docker is ready." echo "Docker in Docker is ready."
echo "Loading corrosion image from /images/corrosion.tar..." echo "Loading corrosion image from /images/corrosion.tar..."
+76
View File
@@ -31,6 +31,10 @@ func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpe
for _, mc := range svc.Containers { for _, mc := range svc.Containers {
assertContainerMatchesSpec(t, mc.Container, spec) assertContainerMatchesSpec(t, mc.Container, spec)
} }
if spec.PreDeploy != nil {
assertHookContainersMatchSpec(t, svc, spec)
}
} }
func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api.ServiceSpec) { func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api.ServiceSpec) {
@@ -139,6 +143,78 @@ func assertContainerMatchesSpec(t *testing.T, ctr api.ServiceContainer, spec api
assert.Contains(t, ctr.NetworkSettings.Networks, machinedocker.NetworkName) assert.Contains(t, ctr.NetworkSettings.Networks, machinedocker.NetworkName)
} }
// assertHookContainersMatchSpec validates that hook containers in the service match the pre-deploy hook spec.
func assertHookContainersMatchSpec(t *testing.T, svc api.Service, spec api.ServiceSpec) {
t.Helper()
require.NotEmpty(t, svc.HookContainers, "Expected at least one hook container")
for _, mc := range svc.HookContainers {
ctr := mc.Container
// Verify labels.
assert.True(t, api.ValidateServiceID(ctr.Config.Labels[api.LabelServiceID]))
assert.Equal(t, spec.Name, ctr.Config.Labels[api.LabelServiceName])
assert.Equal(t, api.LabelHookPreDeploy, ctr.Config.Labels[api.LabelHook])
assert.Contains(t, ctr.Config.Labels, api.LabelManaged)
assert.NotContains(t, ctr.Config.Labels, api.LabelServiceMode,
"Hook containers should not have the service mode label")
assert.EqualValues(t, spec.PreDeploy.Command, ctr.Config.Cmd)
if spec.Container.Entrypoint != nil {
assert.EqualValues(t, spec.Container.Entrypoint, ctr.Config.Entrypoint)
}
// Service env vars are inherited by hook containers.
for _, env := range spec.Container.Env.ToSlice() {
assert.Contains(t, ctr.Config.Env, env)
}
// Hook-specific env vars.
for _, env := range spec.PreDeploy.Env.ToSlice() {
assert.Contains(t, ctr.Config.Env, env)
}
assert.Contains(t, ctr.Config.Env, "UNCLOUD_HOOK_PRE_DEPLOY=true")
assert.Equal(t, spec.Container.Image, ctr.Config.Image)
assert.Equal(t, spec.Container.Init, ctr.HostConfig.Init)
assert.True(t, strings.HasPrefix(ctr.Name, spec.Name+"-pre-deploy-"),
"Hook container name %q should start with %q", ctr.Name, spec.Name+"-pre-deploy-")
// Privileged is overridden by PreDeploy.Privileged if set, otherwise inherited from the service.
if spec.PreDeploy.Privileged != nil {
assert.Equal(t, *spec.PreDeploy.Privileged, ctr.HostConfig.Privileged)
} else {
assert.Equal(t, spec.Container.Privileged, ctr.HostConfig.Privileged)
}
// User is overridden by PreDeploy.User if set.
if spec.PreDeploy.User != "" {
assert.Equal(t, spec.PreDeploy.User, ctr.Config.User)
} else if spec.Container.User != "" {
assert.Equal(t, spec.Container.User, ctr.Config.User)
}
// Compute resources.
assert.Equal(t, spec.Container.Resources.CPU, ctr.HostConfig.Resources.NanoCPUs)
assert.Equal(t, spec.Container.Resources.Memory, ctr.HostConfig.Resources.Memory)
assert.Equal(t, spec.Container.Resources.MemoryReservation, ctr.HostConfig.Resources.MemoryReservation)
// Hook-specific overrides: disabled restart, disabled healthcheck, no ports.
assert.Equal(t, container.RestartPolicy{Name: container.RestartPolicyDisabled}, ctr.HostConfig.RestartPolicy)
require.NotNil(t, ctr.Config.Healthcheck)
assert.Equal(t, []string{"NONE"}, ctr.Config.Healthcheck.Test)
assert.Empty(t, ctr.HostConfig.PortBindings)
assert.False(t, ctr.State.Running, "Hook container should not be running")
assert.Equal(t, 0, ctr.State.ExitCode, "Hook container should exit with code 0")
assertContainerMountsMatchSpec(t, ctr.HostConfig.Mounts, spec)
// Verify network settings.
assert.Len(t, ctr.NetworkSettings.Networks, 1)
assert.Contains(t, ctr.NetworkSettings.Networks, machinedocker.NetworkName)
}
}
func assertContainerMountsMatchSpec(t *testing.T, mounts []mount.Mount, spec api.ServiceSpec) { func assertContainerMountsMatchSpec(t *testing.T, mounts []mount.Mount, spec api.ServiceSpec) {
expectedMounts, err := machinedocker.ToDockerMounts(spec.Volumes, spec.Container.VolumeMounts) expectedMounts, err := machinedocker.ToDockerMounts(spec.Volumes, spec.Container.VolumeMounts)
require.NoError(t, err) require.NoError(t, err)
+2 -2
View File
@@ -55,7 +55,7 @@ func createTestCluster(
}) })
if waitReady { if waitReady {
require.NoError(t, p.WaitClusterReady(ctx, c, 60*time.Second)) require.NoError(t, p.WaitClusterReady(ctx, c, 90*time.Second))
} }
return c, p return c, p
@@ -104,7 +104,7 @@ func TestClusterLifecycle(t *testing.T) {
} }
return true return true
}, 15*time.Second, 50*time.Millisecond, "cluster store not reconciled on machine #%d", i+1) }, 30*time.Second, 50*time.Millisecond, "cluster store not reconciled on machine #%d", i+1)
} }
}) })
+56
View File
@@ -601,4 +601,60 @@ volumes:
assert.ElementsMatch(t, machines.ToSlice(), expectedMachines, assert.ElementsMatch(t, machines.ToSlice(), expectedMachines,
"Containers should be distributed across all machines") "Containers should be distributed across all machines")
}) })
t.Run("pre-deploy hook", func(t *testing.T) {
t.Parallel()
name := "test-compose-predeploy"
volumeName := "test-compose-predeploy-data"
t.Cleanup(func() {
removeServices(t, cli, name)
removeVolumes(t, cli, volumeName)
})
project, err := compose.LoadProject(ctx, []string{"fixtures/compose-predeploy.yaml"})
require.NoError(t, err)
deployment, err := compose.NewDeployment(ctx, cli, project)
require.NoError(t, err)
err = deployment.Run(ctx)
require.NoError(t, err)
svc, err := cli.InspectService(ctx, name)
require.NoError(t, err)
assertServiceMatchesSpec(t, svc, api.ServiceSpec{
Name: name,
Container: api.ContainerSpec{
Image: "busybox:1.37.0-uclibc",
Command: []string{"sleep", "600"},
Env: api.EnvVars{"SERVICE_VAR": "from-service"},
VolumeMounts: []api.VolumeMount{
{
VolumeName: volumeName,
ContainerPath: "/data",
},
},
},
Volumes: []api.VolumeSpec{
{
Name: volumeName,
Type: api.VolumeTypeVolume,
},
},
PreDeploy: &api.PreDeployHook{
Command: []string{"sh", "-c", "echo hello-from-predeploy > /data/predeploy.txt"},
Env: api.EnvVars{"HOOK_VAR": "from-hook"},
User: "root",
},
})
// Verify the pre-deploy hook wrote to the shared volume by reading the file from the running container.
containerID := svc.Containers[0].Container.ID
output, err := execInContainerAndReadOutput(t, ctx, cli, name, containerID,
[]string{"cat", "/data/predeploy.txt"})
require.NoError(t, err)
assert.Equal(t, "hello-from-predeploy\n", output)
})
} }
+17
View File
@@ -0,0 +1,17 @@
services:
test-compose-predeploy:
image: busybox:1.37.0-uclibc
command: ["sleep", "600"]
environment:
SERVICE_VAR: from-service
volumes:
- test-compose-predeploy-data:/data
x-pre_deploy:
command: [sh, -c, echo hello-from-predeploy > /data/predeploy.txt]
environment:
HOOK_VAR: from-hook
user: root
timeout: 30s
volumes:
test-compose-predeploy-data:
+15 -10
View File
@@ -40,14 +40,18 @@ func TestDeployment(t *testing.T) {
ctx := context.Background() ctx := context.Background()
c, _ := createTestCluster(t, clusterName, ucind.CreateClusterOptions{Machines: 3}, true) c, _ := createTestCluster(t, clusterName, ucind.CreateClusterOptions{Machines: 3}, true)
cli, err := c.Machines[0].Connect(ctx) cli, cErr := c.Machines[0].Connect(ctx)
require.NoError(t, err) require.NoError(t, cErr)
t.Run("global auto-generated name", func(t *testing.T) { t.Run("global auto-generated name", func(t *testing.T) {
t.Parallel() t.Parallel()
name := "" // auto-generated and updated name := "" // auto-generated and updated
t.Cleanup(func() { t.Cleanup(func() {
if name == "" {
return
}
err := cli.RemoveService(ctx, name) err := cli.RemoveService(ctx, name)
if !errors.Is(err, api.ErrNotFound) { if !errors.Is(err, api.ErrNotFound) {
require.NoError(t, err) require.NoError(t, err)
@@ -70,7 +74,7 @@ func TestDeployment(t *testing.T) {
} }
deployment := cli.NewDeployment(spec, nil) deployment := cli.NewDeployment(spec, nil)
err = deployment.Validate(ctx) err := deployment.Validate(ctx)
require.NoError(t, err) require.NoError(t, err)
plan, err := deployment.Plan(ctx) plan, err := deployment.Plan(ctx)
@@ -236,7 +240,7 @@ func TestDeployment(t *testing.T) {
} }
deployment := cli.NewDeployment(spec, nil) deployment := cli.NewDeployment(spec, nil)
_, err = deployment.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)
@@ -542,7 +546,7 @@ myapp.example.com {
} }
deployment := cli.NewDeployment(spec, nil) deployment := cli.NewDeployment(spec, nil)
err = deployment.Validate(ctx) err := deployment.Validate(ctx)
require.NoError(t, err) require.NoError(t, err)
plan, err := deployment.Plan(ctx) plan, err := deployment.Plan(ctx)
@@ -709,7 +713,7 @@ myapp.example.com {
deployment := cli.NewDeployment(spec, nil) deployment := cli.NewDeployment(spec, nil)
_, err = deployment.Run(ctx) _, err := deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
// Verify service has 2 containers on machines 0 and 1. // Verify service has 2 containers on machines 0 and 1.
@@ -1090,7 +1094,7 @@ myapp.example.com {
} }
d := deploy.NewDeployment(cli, spec, nil) d := deploy.NewDeployment(cli, spec, nil)
_, err = d.Run(ctx) _, err := d.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
svc, err := cli.InspectService(ctx, serviceName) svc, err := cli.InspectService(ctx, serviceName)
@@ -1128,7 +1132,7 @@ myapp.example.com {
} }
d := deploy.NewDeployment(cli, spec, nil) d := deploy.NewDeployment(cli, spec, nil)
_, err = d.Run(ctx) _, err := d.Run(ctx)
require.Error(t, err, "Global deployment should fail when required volume doesn't exist") require.Error(t, err, "Global deployment should fail when required volume doesn't exist")
require.Contains(t, err.Error(), "no machines available") require.Contains(t, err.Error(), "no machines available")
}) })
@@ -1270,7 +1274,7 @@ myapp.example.com {
} }
deployment := cli.NewDeployment(spec, nil) deployment := cli.NewDeployment(spec, nil)
_, err = deployment.Run(ctx) _, err := deployment.Run(ctx)
require.NoError(t, err) require.NoError(t, err)
svc, err := cli.InspectService(ctx, serviceName) svc, err := cli.InspectService(ctx, serviceName)
@@ -2086,7 +2090,8 @@ func TestServiceLifecycle(t *testing.T) {
} }
assertNoDNSErrors := func(t *testing.T, dnsOutput string) { assertNoDNSErrors := func(t *testing.T, dnsOutput string) {
assert.NotContains(t, dnsOutput, "server can't find", "DNS query should not contain NXDOMAIN/SERVFAIL errors") assert.NotContains(t, dnsOutput, "server can't find",
"DNS query should not contain NXDOMAIN/SERVFAIL errors")
} }
t.Run("service name resolves to all container IPs", func(t *testing.T) { t.Run("service name resolves to all container IPs", func(t *testing.T) {
@@ -446,6 +446,6 @@ Note: Docker installation was preserved. If you want to completely remove Docker
## Further reading ## Further reading
- **[Add more machines](../9-cli-reference/uc_machine_add.md)**: Scale horizontally by creating a cluster of machines - **[Add more machines](../9-cli-reference/uc_machine_add.md)**: Scale horizontally by creating a cluster of machines
- **[Ingress & HTTP](../3-concepts/1-ingress/1-overview.md)**: Learn how Uncloud handles incoming traffic and how to - **[Ingress & HTTP](../3-concepts/2-ingress/1-overview.md)**: Learn how Uncloud handles incoming traffic and how to
expose your services to the internet expose your services to the internet
- **[CLI reference](../9-cli-reference/uc.md)**: Explore all available commands and options - **[CLI reference](../9-cli-reference/uc.md)**: Explore all available commands and options
@@ -0,0 +1,90 @@
# Connecting to a cluster
`uc` only needs to reach one machine to work with the entire cluster. That machine acts as an **entry point** and
forwards requests to other machines as needed.
`uc` stores **cluster contexts** and **connection details** in a [configuration file](../../7-cli-config-reference.md)
(default location is `~/.config/uncloud/config.yaml`).
When you initialise a new cluster with `uc machine init` or add a machine to an existing cluster with `uc machine add`,
they automatically save the SSH addresses of your machines to the config so you don't have to specify them every time.
## Cluster contexts
The [config file](../../7-cli-config-reference.md) organises connections into **contexts**. Each context represents a
cluster. It has a name and a list of connection details for the machines in that cluster.
A context is not the same thing as a cluster. It is your local view of a cluster: which machines you can connect through
and in what order to try them. Different people or environments may need to reach the same cluster in different ways.
You can also manually create multiple contexts for the same cluster. For example, one that connects through
a machine with a public IP when you're not in the office, and another that connects through a private machine on the
office network when you're on-site to reduce latency. You can switch between them depending on where you are.
### Managing contexts
Use these commands to manage the contexts in your config:
- [`uc ctx`](../../9-cli-reference/uc_ctx.md): Switch contexts using an interactive TUI
- [`uc ctx ls`](../../9-cli-reference/uc_ctx_ls.md): List all contexts and see which one is current
- [`uc ctx use`](../../9-cli-reference/uc_ctx_use.md): Switch the current context by name
- [`uc ctx conn`](../../9-cli-reference/uc_ctx_connection.md): Change the default connection for the current context
using an interactive TUI
You can also set `x-context` in your Compose file to pin a specific context for deployments. See
[Deploy to a specific cluster context](../../4-guides/1-deployments/1-deploy-app.md#deploy-to-a-specific-cluster-context)
for details.
## Connection resolution
When you run a `uc` command, it determines which cluster to connect to using this priority:
1. If `--connect` is set, `uc` connects directly to that machine and ignores the config file entirely.
2. If `--context` is set, `uc` uses that context from the config.
3. Otherwise, `uc` uses `current_context` from the config.
Once the context is resolved, `uc` tries each connection in the context's `connections` list in order until one
succeeds.
## Global flags and environment variables
These flags are available on every `uc` command. They can also be set with an environment variable. The flag takes
priority if both are set.
| Flag | Environment variable | Description |
|--------------------|----------------------|-------------------------------------------------------------------|
| `--uncloud-config` | `UNCLOUD_CONFIG` | Path to the config file |
| `--context` | `UNCLOUD_CONTEXT` | Use a specific context instead of `current_context` in the config |
| `--connect` | `UNCLOUD_CONNECT` | Bypass the config file and connect directly |
### Connecting directly without a config
The `--connect` flag or `UNCLOUD_CONNECT` environment variable let you run one-off commands against a cluster without
using a config. This is useful for CI pipelines and scripts where you don't want to set up a config file.
It accepts these formats:
```shell
# System 'ssh' command with full SSH config support
uc --connect root@203.0.113.1 ls
# System 'ssh' command (explicit scheme, same as above)
uc --connect ssh://root@203.0.113.1 ls
# Go's built-in SSH library (no SSH config support, useful when the system ssh is not available)
uc --connect ssh+go://root@203.0.113.1 ls
# Direct connection to machine gRPC API over TCP (for advanced users with custom setups)
uc --connect tcp://[fdcc:4439:f545:3ca:5d17:66e5:7c96:40bd]:51000 ls
# Direct connection to machine gRPC API over a Unix socket (for running uc locally on a cluster machine)
uc --connect unix:///run/uncloud/uncloud.sock ls
```
:::info
Don't use `--connect` with `uc machine init`. `--connect` is for specifying or overriding the connection to an existing
cluster, but `uc machine init` creates a new one and writes the new cluster context to the config file. You can discard
the config when initialising a cluster with `--uncloud-config /dev/null` if you don't want to save it.
:::
@@ -0,0 +1,4 @@
label: Clusters
collapsed: true # keep the category closed by default
link:
type: generated-index
@@ -141,7 +141,7 @@ This generates tags like:
`uc deploy` renders the image templates when it loads the Compose file and then uses the resulting names for the build `uc deploy` renders the image templates when it loads the Compose file and then uses the resulting names for the build
and deploy stages. and deploy stages.
See the [Image tag template](../../8-compose-file-reference/2-image-tag-template.md) reference for all available See the [Image tag template](../../8-compose-file-reference/3-image-tag-template.md) reference for all available
template variables and functions. template variables and functions.
### Separate build and deploy steps ### Separate build and deploy steps
@@ -338,7 +338,7 @@ services:
With this configuration, `uc deploy` and other commands using the Compose file will always target the `prod` context, With this configuration, `uc deploy` and other commands using the Compose file will always target the `prod` context,
regardless of your currently active context. You can still override it with the `--context` flag if needed. regardless of your currently active context. You can still override it with the `--context` flag if needed.
See [`x-context`](../../8-compose-file-reference/1-support-matrix.md#x-context) for more details. See [`x-context`](../../8-compose-file-reference/2-extensions.md#x-context) for more details.
## Use a different Compose file location ## Use a different Compose file location
@@ -1,13 +1,13 @@
# Deploy to specific machines # Deploy to specific machines
Deploy services to specific machines in your cluster using the Deploy services to specific machines in your cluster using the
[`x-machines`](../../8-compose-file-reference/1-support-matrix.md#x-machines) extension in your Compose file. [`x-machines`](../../8-compose-file-reference/2-extensions.md#x-machines) extension in your Compose file.
## When to target specific machines ## When to target specific machines
By default, Uncloud randomly chooses available machines to run your services on, evenly spreading multiple replicas of a By default, Uncloud randomly chooses available machines to run your services on, evenly spreading multiple replicas of a
service across all machines for high availability. You can restrict which machines can run your service using the service across all machines for high availability. You can restrict which machines can run your service using the
[`x-machines`](../../8-compose-file-reference/1-support-matrix.md#x-machines) extension in your Compose file. [`x-machines`](../../8-compose-file-reference/2-extensions.md#x-machines) extension in your Compose file.
This is useful when you want to: This is useful when you want to:
@@ -78,7 +78,7 @@ See [Push local images to cluster machines](1-deploy-app.md#push-local-images-to
## See also ## See also
- [Deploy an app](1-deploy-app.md): Build and deploy from source code or prebuilt images - [Deploy an app](1-deploy-app.md): Build and deploy from source code or pre-built images
- [Deploy a global service](3-deploy-global-services.md): Deploy one service replica on each cluster machine - [Deploy a global service](3-deploy-global-services.md): Deploy one service replica on each cluster machine
- [Compose support matrix](../../8-compose-file-reference/1-support-matrix.md): Supported Compose features and Uncloud - [Compose support matrix](../../8-compose-file-reference/1-support-matrix.md): Supported Compose features and Uncloud
extensions extensions
@@ -33,7 +33,7 @@ Uncloud doesn't automatically scale global services to new machines.
## Deploy to a subset of machines ## Deploy to a subset of machines
You can combine the `global` mode with [`x-machines`](../../8-compose-file-reference/1-support-matrix.md#x-machines) You can combine the `global` mode with [`x-machines`](../../8-compose-file-reference/2-extensions.md#x-machines)
to deploy one container to each specified machine: to deploy one container to each specified machine:
```yaml title="compose.yaml" ```yaml title="compose.yaml"
@@ -63,7 +63,7 @@ The default mode is `replicated`, where you specify the number of replicas.
## See also ## See also
- [Deploy an app](1-deploy-app.md): Deploy from source code or prebuilt images - [Deploy an app](1-deploy-app.md): Deploy from source code or pre-built images
- [Deploy to specific machines](2-deploy-specific-machines.md): Deploy services to specific machines in your cluster - [Deploy to specific machines](2-deploy-specific-machines.md): Deploy services to specific machines in your cluster
- [Compose Specification: deploy.mode](https://github.com/compose-spec/compose-spec/blob/main/deploy.md#mode): - [Compose Specification: deploy.mode](https://github.com/compose-spec/compose-spec/blob/main/deploy.md#mode):
Compose specification for deployment modes Compose specification for deployment modes
@@ -131,7 +131,7 @@ services:
:::info important :::info important
If a health check fails after the deployment, Uncloud automatically removes the unhealthy container from the If a health check fails after the deployment, Uncloud automatically removes the unhealthy container from the
[Caddy](../../3-concepts/1-ingress/1-overview.md) configuration to prevent routing traffic to that container. But it [Caddy](../../3-concepts/2-ingress/1-overview.md) configuration to prevent routing traffic to that container. But it
doesn't automatically restart or roll it back. doesn't automatically restart or roll it back.
Uncloud automatically adds it back to Caddy when it recovers and becomes healthy again. You can inspect the health Uncloud automatically adds it back to Caddy when it recovers and becomes healthy again. You can inspect the health
@@ -168,6 +168,7 @@ configuration hasn't changed and only redeploy the remaining ones.
## See also ## See also
- [Pre-deploy hooks](5-pre-deploy-hooks.md): Run a command before deploying service containers
- [Deploy an app](1-deploy-app.md): Build and deploy from source code or pre-built images - [Deploy an app](1-deploy-app.md): Build and deploy from source code or pre-built images
- [Compose support matrix](../../8-compose-file-reference/1-support-matrix.md): Supported Compose features and Uncloud - [Compose support matrix](../../8-compose-file-reference/1-support-matrix.md): Supported Compose features and Uncloud
extensions extensions
@@ -0,0 +1,212 @@
# Pre-deploy hooks
Run a one-off command before deploying a service.
Pre-deploy hooks are useful for **one-off tasks** such as:
- Database schema and data migrations
- Uploading static assets to a CDN
- Cache invalidation
- Any setup task that needs to run **once** before new code goes live, not on every container startup
## How it works
When you run `uc deploy` for a service with a pre-deploy hook configured, the hook command runs after building and
pushing the new image (if [building from source](1-deploy-app.md#deploy-from-source-code)) but **before** rolling out
any new containers.
`uc deploy` runs your hook command inside a new container and waits for it to finish or time out (**5 minutes** by
default). This container **inherits** most of the **service's configuration**, including the image, environment
variables, volumes, placement, and compute resources.
If the command exits with code 0, the deployment continues with a normal [rolling update](4-rolling-deployments.md). If
the command fails or times out, the deployment stops immediately with an error. `uc deploy` will display the latest logs
from the hook container to help you diagnose the issue.
![Deploy flow with a pre-deploy hook](img/pre-deploy-hook-light.svg#gh-light-mode-only)
![Deploy flow with a pre-deploy hook](img/pre-deploy-hook-dark.svg#gh-dark-mode-only)
The hook runs on one of the machines where the service will be deployed. Similar to service containers, hook containers
can reach other services over the network, connect to databases, and read or write shared volumes. Note that file system
changes do not persist after the hook finishes, except for changes written to shared volumes.
## Usage
Add the [`x-pre_deploy`](../../8-compose-file-reference/2-extensions.md#x-pre_deploy) extension to a service in your
Compose file. The only required attribute is `command`, which can be a string or a list of strings, just like the
service's [`command`](https://github.com/compose-spec/compose-spec/blob/main/05-services.md#command).
```yaml title="compose.yaml"
services:
web:
build: .
x-pre_deploy:
command: python manage.py migrate
```
:::info note
The `command` replaces the image's default command ([`CMD`](https://docs.docker.com/reference/dockerfile/#cmd)) but the
[`ENTRYPOINT`](https://docs.docker.com/reference/dockerfile/#entrypoint) still runs. If your image has an entrypoint,
the hook command is passed as arguments to it. You can override the entrypoint for the service using
[`entrypoint`](https://github.com/compose-spec/compose-spec/blob/main/05-services.md#entrypoint) which applies to both
hook and regular service containers.
:::
Since your command runs in the same image as the service, any tools or dependencies it needs must be installed in that
image.
See [`x-pre_deploy`](../../8-compose-file-reference/2-extensions.md#x-pre_deploy) for all available attributes and
their defaults.
### Database migrations
The most common use case for pre-deploy hooks is running database migrations before deploying a new app version:
```yaml title="compose.yaml"
services:
web:
build: .
environment:
DATABASE_URL: postgres://postgres:${DB_PASSWORD}@db:5432/postgres
x-ports:
- app.example.com:8000/https
x-pre_deploy:
# Apply Django migrations from the built image before deploying new app containers
command: python manage.py migrate
depends_on:
- db
db:
image: postgres:18
environment:
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
- db-data:/var/lib/postgresql
volumes:
db-data:
```
When you run `uc deploy`, the migration runs first inside a new container created with the same image and environment
variables as the `web` service. Only after it succeeds, the deployment starts replacing service containers with the new
image.
### Running multiple commands
If you need to run several tasks before deployment, you can wrap them in a single shell command:
```yaml title="compose.yaml"
services:
web:
build: .
x-pre_deploy:
# Apply Django migrations, collect static files, and upload them to S3 bucket
command: sh -c "python manage.py migrate && python manage.py collectstatic --no-input"
```
For more complex scenarios, create a dedicated script and use it as the hook command:
<Tabs>
<TabItem value="compose.yaml">
```yaml
services:
web:
build: .
x-pre_deploy:
command: ./scripts/pre_deploy.sh
```
</TabItem>
<TabItem value="scripts/pre_deploy.sh">
```bash
#!/bin/bash
# Ensure the script exits immediately with a non-zero code if any command fails.
set -e
python manage.py migrate
python manage.py collectstatic --no-input
python manage.py clear_cache
```
</TabItem>
</Tabs>
Make sure the script is included in your service image and exits with a non-zero code on any command failure (`set -e`).
### Custom environment and user
The hook container inherits environment variables from the service. You can add hook-specific variables or override
existing ones with `environment`. Use `user` to run the command as a different user.
For example, if your service runs as a non-root user but the hook needs root to fix file permissions on a shared volume:
```yaml title="compose.yaml"
services:
web:
build: .
user: app
volumes:
- data:/data
x-pre_deploy:
command: chown -R app:app /data/uploads
user: root
volumes:
data:
```
:::tip
Uncloud automatically sets `UNCLOUD_HOOK_PRE_DEPLOY=true` in the hook container. You can check this variable in a shared
entrypoint script or your command to detect when it's running as a pre-deploy hook versus a regular service container.
:::
## Failure handling
### Non-zero exit code
When the hook command exits with a non-zero code, the deployment stops immediately. No service containers are created or
replaced. `uc deploy` prints the latest logs from the hook container to help you diagnose the issue.
The failed hook container is not automatically removed, so you can inspect it with `uc inspect` and `uc ps`, and fetch
its full logs as part of the service logs:
```shell
uc logs web
```
Fix the issue and run `uc deploy` again to retry.
### Timeout
If the hook doesn't finish within the timeout (default **5 minutes**), Uncloud kills the container and fails the
deployment. The stopped container is kept for inspection, same as with a non-zero exit code.
You can increase the timeout for long-running tasks like large database migrations or data uploads:
```yaml title="compose.yaml"
services:
web:
x-pre_deploy:
command: python manage.py migrate
timeout: 30m
```
### Idempotency
Design your hook commands to be **idempotent** when possible. If a deployment fails after the hook succeeds (for
example, a new container crashes on startup) and you retry with `uc deploy`, the hook runs again.
Most database migration tools handle this naturally since they track which migrations have already been applied.
## See also
- [`x-pre_deploy` reference](../../8-compose-file-reference/2-extensions.md#x-pre_deploy): All available attributes
and their defaults
- [Rolling deployments](4-rolling-deployments.md): How Uncloud updates containers with zero downtime
- [Deploy an app](1-deploy-app.md): Build and deploy from source code or pre-built images
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 25 KiB

Some files were not shown because too many files have changed in this diff Show More