mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 19:13:34 +00:00
Compare commits
4
Commits
b99abb7c39
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac56754281 | ||
|
|
b7e224a1ef | ||
|
|
351698c280 | ||
|
|
fa77edf53e |
@@ -24,6 +24,7 @@ import (
|
||||
"github.com/psviderski/uncloud/internal/machine"
|
||||
"github.com/psviderski/uncloud/internal/version"
|
||||
"github.com/spf13/cobra"
|
||||
"github.com/spf13/pflag"
|
||||
)
|
||||
|
||||
type globalOptions struct {
|
||||
@@ -42,6 +43,13 @@ func main() {
|
||||
SilenceUsage: true,
|
||||
SilenceErrors: true,
|
||||
PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
|
||||
// Shell completion runs through the hidden __complete command which has flag parsing disabled,
|
||||
// so the global flags from the completed command line are never parsed. Apply them manually to make
|
||||
// completion work with --connect, --context, and --uncloud-config.
|
||||
if cmd.Name() == cobra.ShellCompRequestCmd {
|
||||
applyGlobalFlagsFromCompletionArgs(cmd.Root().PersistentFlags(), os.Args[1:])
|
||||
}
|
||||
|
||||
cli.BindEnvToFlag(cmd, "connect", "UNCLOUD_CONNECT")
|
||||
cli.BindEnvToFlag(cmd, "context", "UNCLOUD_CONTEXT")
|
||||
cli.BindEnvToFlag(cmd, "uncloud-config", "UNCLOUD_CONFIG")
|
||||
@@ -159,3 +167,27 @@ func main() {
|
||||
cobra.CheckErr(err)
|
||||
}
|
||||
}
|
||||
|
||||
// applyGlobalFlagsFromCompletionArgs parses the global flags from the raw arguments of a __complete command and applies
|
||||
// the ones found to flags. The trailing word being completed, unknown flags, and positional arguments are ignored.
|
||||
func applyGlobalFlagsFromCompletionArgs(flags *pflag.FlagSet, args []string) {
|
||||
// The shell always passes the word being completed as the last argument, even if it's empty.
|
||||
// Exclude it from parsing as its value may not be complete yet.
|
||||
if len(args) == 0 {
|
||||
return
|
||||
}
|
||||
args = args[:len(args)-1]
|
||||
|
||||
fset := pflag.NewFlagSet("global", pflag.ContinueOnError)
|
||||
fset.ParseErrorsAllowlist.UnknownFlags = true
|
||||
fset.String("connect", "", "")
|
||||
fset.StringP("context", "c", "", "")
|
||||
fset.String("uncloud-config", "", "")
|
||||
// Parsing an incomplete command line may fail, apply the flags parsed so far anyway.
|
||||
_ = fset.Parse(args)
|
||||
|
||||
fset.Visit(func(f *pflag.Flag) {
|
||||
// Setting the flag marks it as changed so it takes precedence over environment variables.
|
||||
_ = flags.Set(f.Name, f.Value.String())
|
||||
})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/pflag"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestApplyGlobalFlagsFromCompletionArgs(t *testing.T) {
|
||||
defaultConfigPath := "~/.config/uncloud/config.yaml"
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
args []string
|
||||
wantConnect string
|
||||
wantContext string
|
||||
wantConfigPath string
|
||||
// Flag names expected to be marked as changed on the target flag set.
|
||||
wantChanged []string
|
||||
}{
|
||||
{
|
||||
name: "no flags",
|
||||
args: []string{"__complete", "inspect", ""},
|
||||
},
|
||||
{
|
||||
name: "connect with space",
|
||||
args: []string{"__complete", "--connect", "ssh://user@host", "inspect", ""},
|
||||
wantConnect: "ssh://user@host",
|
||||
wantChanged: []string{"connect"},
|
||||
},
|
||||
{
|
||||
name: "connect with equals",
|
||||
args: []string{"__complete", "--connect=tcp://127.0.0.1:51000", "inspect", ""},
|
||||
wantConnect: "tcp://127.0.0.1:51000",
|
||||
wantChanged: []string{"connect"},
|
||||
},
|
||||
{
|
||||
name: "context shorthand",
|
||||
args: []string{"__complete", "-c", "prod", "inspect", ""},
|
||||
wantContext: "prod",
|
||||
wantChanged: []string{"context"},
|
||||
},
|
||||
{
|
||||
name: "all flags",
|
||||
args: []string{"__complete", "--connect", "user@host", "-c", "prod", "--uncloud-config", "/tmp/uncloud.yaml", "inspect", ""},
|
||||
wantConnect: "user@host",
|
||||
wantContext: "prod",
|
||||
wantConfigPath: "/tmp/uncloud.yaml",
|
||||
wantChanged: []string{"connect", "context", "uncloud-config"},
|
||||
},
|
||||
{
|
||||
name: "unknown flags are ignored",
|
||||
args: []string{"__complete", "--quiet", "-n", "5", "--connect", "user@host", "logs", ""},
|
||||
wantConnect: "user@host",
|
||||
wantChanged: []string{"connect"},
|
||||
},
|
||||
{
|
||||
name: "flags after double dash are ignored",
|
||||
args: []string{"__complete", "exec", "svc", "--", "sh", "--connect", "user@host"},
|
||||
},
|
||||
{
|
||||
name: "flags before double dash are applied",
|
||||
args: []string{"__complete", "--connect", "user@host", "exec", "svc", "--", "sh", "-c", "env"},
|
||||
wantConnect: "user@host",
|
||||
wantChanged: []string{"connect"},
|
||||
},
|
||||
{
|
||||
name: "partial flag name being completed is excluded",
|
||||
args: []string{"__complete", "--connect", "user@host", "inspect", "--context"},
|
||||
wantConnect: "user@host",
|
||||
wantChanged: []string{"connect"},
|
||||
},
|
||||
{
|
||||
name: "partial flag value being completed is excluded",
|
||||
args: []string{"__complete", "--uncloud-config", "/tmp/"},
|
||||
},
|
||||
{
|
||||
name: "partial connect value being completed is excluded",
|
||||
args: []string{"__complete", "--connect", "tcp://127.0.0.1:5"},
|
||||
},
|
||||
{
|
||||
name: "completed flag value with partial command word",
|
||||
args: []string{"__complete", "--uncloud-config", "/tmp/uncloud.yaml", "insp"},
|
||||
wantConfigPath: "/tmp/uncloud.yaml",
|
||||
wantChanged: []string{"uncloud-config"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Mirror the global persistent flags defined on the root command.
|
||||
var opts globalOptions
|
||||
flags := pflag.NewFlagSet("test", pflag.ContinueOnError)
|
||||
flags.StringVar(&opts.connect, "connect", "", "")
|
||||
flags.StringVarP(&opts.context, "context", "c", "", "")
|
||||
flags.StringVar(&opts.configPath, "uncloud-config", defaultConfigPath, "")
|
||||
|
||||
applyGlobalFlagsFromCompletionArgs(flags, tt.args)
|
||||
|
||||
assert.Equal(t, tt.wantConnect, opts.connect)
|
||||
assert.Equal(t, tt.wantContext, opts.context)
|
||||
wantConfigPath := tt.wantConfigPath
|
||||
if wantConfigPath == "" {
|
||||
wantConfigPath = defaultConfigPath
|
||||
}
|
||||
assert.Equal(t, wantConfigPath, opts.configPath)
|
||||
|
||||
for _, name := range []string{"connect", "context", "uncloud-config"} {
|
||||
assert.Equal(t, slices.Contains(tt.wantChanged, name), flags.Changed(name),
|
||||
"changed status of flag '%s'", name)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+15
-16
@@ -132,35 +132,34 @@ func runProxy(ctx context.Context, uncli *cli.CLI, opts proxyOptions) error {
|
||||
// endpoint and shuffles the data, *it* will actually experience errors.
|
||||
remoteAddr := net.JoinHostPort(ip.String(), strconv.Itoa(opts.remotePort))
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
p := &proxy.Proxy{
|
||||
Listener: listener,
|
||||
RemoteAddr: remoteAddr,
|
||||
DialContext: dialer.DialContext,
|
||||
OnError: func(err error) {
|
||||
fmt.Printf("Failed to proxy to '%s': %v\n", remoteAddr, err)
|
||||
cancel()
|
||||
if proxy.IsConnectionClosedError(err) {
|
||||
return
|
||||
}
|
||||
// A more actionable error instead of the cryptic [ssh -W] command error.
|
||||
if strings.Contains(err.Error(), "Session open refused by peer") {
|
||||
fmt.Printf("Could not connect to '%s': connection refused. "+
|
||||
"Check that the service is running and listening on port %d inside the container.\n",
|
||||
remoteAddr, opts.remotePort)
|
||||
return
|
||||
}
|
||||
|
||||
fmt.Printf("Failed to proxy a connection to '%s': %v\n", remoteAddr, err)
|
||||
},
|
||||
}
|
||||
|
||||
// Run the proxy in the background and signal when it has fully shut down.
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
p.Run(ctx)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
// Prefix the local address with the scheme for common HTTP ports so it becomes control-clickable in most
|
||||
// terminals. We assume plain HTTP since TLS is typically terminated by Caddy in front of the service.
|
||||
fmt.Printf("%s%s → %s (%s%s%s)\n", schemeForPort(opts.remotePort), p.Listener.Addr().String(),
|
||||
remoteAddr, opts.service, tui.Faint.Render("/"), containerID)
|
||||
|
||||
<-ctx.Done()
|
||||
// Wait for the proxy to drain in-flight connections and shut down gracefully.
|
||||
<-done
|
||||
|
||||
if err = p.Run(ctx); err != nil {
|
||||
return fmt.Errorf("run proxy to '%s': %w", remoteAddr, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -11,6 +11,11 @@ import (
|
||||
)
|
||||
|
||||
func Contexts(ctx context.Context, uncli *cli.CLI, args []string, toComplete string) ([]cobra.Completion, cobra.ShellCompDirective) {
|
||||
// There are no contexts to complete when the CLI uses a direct machine connection (--connect) without a config.
|
||||
if uncli.Config == nil {
|
||||
return nil, cobra.ShellCompDirectiveNoFileComp
|
||||
}
|
||||
|
||||
contexts := slices.Sorted(maps.Keys(uncli.Config.Contexts))
|
||||
|
||||
names := []cobra.Completion{}
|
||||
@@ -21,7 +26,6 @@ func Contexts(ctx context.Context, uncli *cli.CLI, args []string, toComplete str
|
||||
if strings.HasPrefix(context, toComplete) {
|
||||
names = append(names, context)
|
||||
}
|
||||
names = append(names, context)
|
||||
}
|
||||
|
||||
return names, cobra.ShellCompDirectiveNoFileComp
|
||||
|
||||
@@ -10,7 +10,8 @@ import (
|
||||
)
|
||||
|
||||
func Machines(ctx context.Context, uncli *cli.CLI, args []string, toComplete string) ([]cobra.Completion, cobra.ShellCompDirective) {
|
||||
client, err := uncli.ConnectCluster(ctx)
|
||||
// Disable the connection progress output to not interfere with the shell completion output.
|
||||
client, err := uncli.ConnectClusterWithOptions(ctx, cli.ConnectOptions{})
|
||||
if err != nil {
|
||||
return nil, cobra.ShellCompDirectiveError
|
||||
}
|
||||
|
||||
@@ -11,7 +11,8 @@ import (
|
||||
)
|
||||
|
||||
func Services(ctx context.Context, uncli *cli.CLI, args []string, toComplete string) ([]cobra.Completion, cobra.ShellCompDirective) {
|
||||
client, err := uncli.ConnectCluster(ctx)
|
||||
// Disable the connection progress output to not interfere with the shell completion output.
|
||||
client, err := uncli.ConnectClusterWithOptions(ctx, cli.ConnectOptions{})
|
||||
if err != nil {
|
||||
return nil, cobra.ShellCompDirectiveError
|
||||
}
|
||||
|
||||
@@ -11,7 +11,8 @@ import (
|
||||
)
|
||||
|
||||
func Volumes(ctx context.Context, uncli *cli.CLI, args []string, toComplete string) ([]cobra.Completion, cobra.ShellCompDirective) {
|
||||
client, err := uncli.ConnectCluster(ctx)
|
||||
// Disable the connection progress output to not interfere with the shell completion output.
|
||||
client, err := uncli.ConnectClusterWithOptions(ctx, cli.ConnectOptions{})
|
||||
if err != nil {
|
||||
return nil, cobra.ShellCompDirectiveError
|
||||
}
|
||||
|
||||
@@ -562,7 +562,9 @@ func (s *Server) CreateServiceContainer(
|
||||
api.LabelServiceMode: spec.Mode,
|
||||
api.LabelManaged: "",
|
||||
},
|
||||
User: spec.Container.User,
|
||||
User: spec.Container.User,
|
||||
Tty: spec.Container.Tty,
|
||||
OpenStdin: spec.Container.OpenStdin,
|
||||
}
|
||||
if spec.Mode == "" {
|
||||
config.Labels[api.LabelServiceMode] = api.ServiceModeReplicated
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
package docker
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -233,11 +235,17 @@ func (s *Service) ListImages(ctx context.Context, opts image.ListOptions) (Image
|
||||
return imagesResp, nil
|
||||
}
|
||||
|
||||
// ContainerLogs streams logs from a container and returns demultiplexed entries via a channel.
|
||||
// ContainerLogs streams logs from a container and returns entries via a channel.
|
||||
// The channel is closed when streaming completes or context is cancelled.
|
||||
func (s *Service) ContainerLogs(
|
||||
ctx context.Context, containerID string, opts api.ServiceLogsOptions,
|
||||
) (<-chan api.LogEntry, error) {
|
||||
ctr, err := s.Client.ContainerInspect(ctx, containerID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("inspect container '%s': %w", containerID, err)
|
||||
}
|
||||
isTTY := ctr.Config != nil && ctr.Config.Tty
|
||||
|
||||
dockerOpts := container.LogsOptions{
|
||||
ShowStdout: true,
|
||||
ShowStderr: true,
|
||||
@@ -257,25 +265,31 @@ func (s *Service) ContainerLogs(
|
||||
stdoutWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: false}
|
||||
stderrWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: true}
|
||||
|
||||
// Wrap the context in a cancellable one to unblock the second goroutine below when StdCopy completes.
|
||||
// Wrap the context in a cancellable one to unblock the second goroutine when log copying completes.
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
|
||||
// Run StdCopy in a goroutine to be able to handle context cancellation.
|
||||
// Copy logs in a goroutine to be able to handle context cancellation.
|
||||
go func() {
|
||||
defer close(outCh)
|
||||
defer cancel()
|
||||
|
||||
// StdCopy is blocking and will return when the reader is closed in another goroutine below or on error.
|
||||
if _, err := stdcopy.StdCopy(stdoutWriter, stderrWriter, reader); err != nil {
|
||||
// Docker returns raw stdout for TTY containers and multiplexed stdout/stderr otherwise.
|
||||
var err error
|
||||
if isTTY {
|
||||
_, err = copyRawContainerLogs(stdoutWriter, reader)
|
||||
} else {
|
||||
_, err = stdcopy.StdCopy(stdoutWriter, stderrWriter, reader)
|
||||
}
|
||||
if err != nil {
|
||||
// Send error as the last entry.
|
||||
select {
|
||||
case outCh <- api.LogEntry{Err: fmt.Errorf("demultiplex container logs: %w", err)}:
|
||||
case outCh <- api.LogEntry{Err: fmt.Errorf("copy container logs: %w", err)}:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Close the reader when the context is done to cancel StdCopy if it's still running.
|
||||
// Close the reader when the context is done to cancel log copying if it's still running.
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
reader.Close()
|
||||
@@ -284,7 +298,32 @@ func (s *Service) ContainerLogs(
|
||||
return outCh, nil
|
||||
}
|
||||
|
||||
// logsChannelWriter is a writer for stdcopy.StdCopy that sends demultiplexed container logs to a channel.
|
||||
// copyRawContainerLogs copies a raw TTY log stream one line at a time so each write produces one log entry.
|
||||
func copyRawContainerLogs(dst io.Writer, src io.Reader) (written int64, _ error) {
|
||||
reader := bufio.NewReader(src)
|
||||
for {
|
||||
line, readErr := reader.ReadBytes('\n')
|
||||
if len(line) > 0 {
|
||||
n, writeErr := dst.Write(line)
|
||||
written += int64(n)
|
||||
if writeErr != nil {
|
||||
return written, writeErr
|
||||
}
|
||||
if n != len(line) {
|
||||
return written, io.ErrShortWrite
|
||||
}
|
||||
}
|
||||
|
||||
if readErr != nil {
|
||||
if errors.Is(readErr, io.EOF) {
|
||||
return written, nil
|
||||
}
|
||||
return written, readErr
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// logsChannelWriter sends container log writes to a channel.
|
||||
type logsChannelWriter struct {
|
||||
ctx context.Context
|
||||
ch chan<- api.LogEntry
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
package docker
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/docker/docker/api/types/container"
|
||||
dockerclient "github.com/docker/docker/client"
|
||||
"github.com/docker/docker/pkg/stdcopy"
|
||||
"github.com/psviderski/uncloud/pkg/api"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestServiceContainerLogs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
containerID = "container-id"
|
||||
firstLog = "2025-01-01T00:00:00.000000000Z first message\n"
|
||||
secondLog = "2025-01-01T00:00:01.000000000Z second message\n"
|
||||
)
|
||||
|
||||
var multiplexedLogs bytes.Buffer
|
||||
_, err := stdcopy.NewStdWriter(&multiplexedLogs, stdcopy.Stdout).Write([]byte(firstLog))
|
||||
require.NoError(t, err)
|
||||
_, err = stdcopy.NewStdWriter(&multiplexedLogs, stdcopy.Stderr).Write([]byte(secondLog))
|
||||
require.NoError(t, err)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
tty bool
|
||||
logs []byte
|
||||
streams []api.LogStreamType
|
||||
messages []string
|
||||
}{
|
||||
{
|
||||
name: "TTY raw stream",
|
||||
tty: true,
|
||||
logs: []byte(firstLog + secondLog),
|
||||
streams: []api.LogStreamType{api.LogStreamStdout, api.LogStreamStdout},
|
||||
messages: []string{"first message\n", "second message\n"},
|
||||
},
|
||||
{
|
||||
name: "non-TTY multiplexed stream",
|
||||
logs: multiplexedLogs.Bytes(),
|
||||
streams: []api.LogStreamType{api.LogStreamStdout, api.LogStreamStderr},
|
||||
messages: []string{"first message\n", "second message\n"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dockerClient := newLogsTestClient(t, tt.tty, tt.logs)
|
||||
service := NewService(dockerClient, nil)
|
||||
|
||||
logsCh, err := service.ContainerLogs(context.Background(), containerID, api.ServiceLogsOptions{})
|
||||
require.NoError(t, err)
|
||||
|
||||
var entries []api.LogEntry
|
||||
for entry := range logsCh {
|
||||
require.NoError(t, entry.Err)
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
|
||||
require.Len(t, entries, len(tt.messages))
|
||||
for i := range entries {
|
||||
assert.Equal(t, tt.streams[i], entries[i].Stream)
|
||||
assert.Equal(t, tt.messages[i], string(entries[i].Message))
|
||||
assert.False(t, entries[i].Timestamp.IsZero())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func newLogsTestClient(t *testing.T, tty bool, logs []byte) *dockerclient.Client {
|
||||
t.Helper()
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/containers/container-id/json"):
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if err := json.NewEncoder(w).Encode(container.InspectResponse{
|
||||
ContainerJSONBase: &container.ContainerJSONBase{ID: "container-id"},
|
||||
Config: &container.Config{Tty: tty},
|
||||
}); err != nil {
|
||||
t.Errorf("encode inspect response: %v", err)
|
||||
}
|
||||
case strings.HasSuffix(r.URL.Path, "/containers/container-id/logs"):
|
||||
w.Header().Set("Content-Type", "application/vnd.docker.raw-stream")
|
||||
if _, err := w.Write(logs); err != nil {
|
||||
t.Errorf("write logs response: %v", err)
|
||||
}
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
serverURL, err := url.Parse(server.URL)
|
||||
require.NoError(t, err)
|
||||
|
||||
dockerClient, err := dockerclient.NewClientWithOpts(
|
||||
dockerclient.WithHost("tcp://"+serverURL.Host),
|
||||
dockerclient.WithHTTPClient(server.Client()),
|
||||
dockerclient.WithVersion("1.48"),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
require.NoError(t, dockerClient.Close())
|
||||
})
|
||||
|
||||
return dockerClient
|
||||
}
|
||||
+61
-48
@@ -2,11 +2,12 @@ package proxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -15,58 +16,53 @@ type Proxy struct {
|
||||
Listener net.Listener
|
||||
RemoteAddr string
|
||||
DialContext func(ctx context.Context, network, address string) (net.Conn, error)
|
||||
// OnError is called for errors that occur during proxying individual connections. It may be called concurrently
|
||||
// for different connections.
|
||||
OnError func(error)
|
||||
activeConns sync.WaitGroup
|
||||
}
|
||||
|
||||
// deadliner is an interface for listeners that support setting deadlines.
|
||||
type deadliner interface {
|
||||
SetDeadline(t time.Time) error
|
||||
}
|
||||
|
||||
// halfCloser is an interface for connections that support half-close.
|
||||
type halfCloser interface {
|
||||
CloseWrite() error
|
||||
}
|
||||
|
||||
// Run starts the proxy and runs until the context is canceled.
|
||||
func (p *Proxy) Run(ctx context.Context) {
|
||||
// IsConnectionClosedError reports whether err indicates that a connection was closed or aborted by either peer.
|
||||
// Callers can use it to ignore routine connection shutdown or broken pipe errors reported to Proxy.OnError.
|
||||
func IsConnectionClosedError(err error) bool {
|
||||
return errors.Is(err, net.ErrClosed) || errors.Is(err, io.ErrClosedPipe) ||
|
||||
errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET)
|
||||
}
|
||||
|
||||
// Run starts the proxy and runs until the context is canceled or the listener fails. It returns nil when the context
|
||||
// is canceled. Errors handling individual connections are reported to OnError and do not stop the proxy.
|
||||
func (p *Proxy) Run(ctx context.Context) error {
|
||||
if p.DialContext == nil {
|
||||
p.DialContext = (&net.Dialer{}).DialContext
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
defer p.Listener.Close()
|
||||
|
||||
// Handle incoming connections until context is canceled.
|
||||
Loop:
|
||||
// Closing the listener unblocks Accept when the context is canceled. This works for both TCP and Unix listeners
|
||||
// and avoids polling with listener deadlines.
|
||||
stopClose := context.AfterFunc(ctx, func() {
|
||||
p.Listener.Close()
|
||||
})
|
||||
defer stopClose()
|
||||
|
||||
var runErr error
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break Loop
|
||||
default:
|
||||
}
|
||||
|
||||
// Set a deadline on the listener if supported to check context periodically.
|
||||
if dl, ok := p.Listener.(deadliner); ok {
|
||||
dl.SetDeadline(time.Now().Add(1 * time.Second))
|
||||
}
|
||||
|
||||
conn, err := p.Listener.Accept()
|
||||
if err != nil {
|
||||
if os.IsTimeout(err) {
|
||||
// Just a timeout, continue to check context and accept again.
|
||||
continue
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break Loop
|
||||
default:
|
||||
if p.OnError != nil {
|
||||
p.OnError(fmt.Errorf("accept local connection: %w", err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
runErr = fmt.Errorf("accept local connection: %w", err)
|
||||
cancel()
|
||||
break
|
||||
}
|
||||
|
||||
p.activeConns.Add(1)
|
||||
@@ -75,6 +71,7 @@ Loop:
|
||||
|
||||
// Wait for all connections to finish.
|
||||
p.activeConns.Wait()
|
||||
return runErr
|
||||
}
|
||||
|
||||
func (p *Proxy) handleConnection(ctx context.Context, localConn net.Conn) {
|
||||
@@ -87,46 +84,62 @@ func (p *Proxy) handleConnection(ctx context.Context, localConn net.Conn) {
|
||||
|
||||
remoteConn, err := p.DialContext(dialCtx, "tcp", p.RemoteAddr)
|
||||
if err != nil {
|
||||
if p.OnError != nil {
|
||||
if ctx.Err() == nil && p.OnError != nil {
|
||||
p.OnError(fmt.Errorf("connect remote address '%s': %w", p.RemoteAddr, err))
|
||||
}
|
||||
return
|
||||
}
|
||||
defer remoteConn.Close()
|
||||
|
||||
// Bidirectional copy with proper half-close handling.
|
||||
// Closing both connections aborts both copies after cancellation or a copy error. A clean EOF still uses
|
||||
// half-close so the other direction can finish sending any remaining data.
|
||||
closeConnections := func() {
|
||||
localConn.Close()
|
||||
remoteConn.Close()
|
||||
}
|
||||
stopClose := context.AfterFunc(ctx, closeConnections)
|
||||
defer stopClose()
|
||||
|
||||
done := make(chan error, 2)
|
||||
|
||||
go func() {
|
||||
_, err := io.Copy(remoteConn, localConn)
|
||||
if err != nil {
|
||||
done <- err
|
||||
closeConnections()
|
||||
return
|
||||
}
|
||||
// Close write half of remote connection if supported.
|
||||
if hc, ok := remoteConn.(halfCloser); ok {
|
||||
hc.CloseWrite()
|
||||
}
|
||||
done <- err
|
||||
done <- nil
|
||||
}()
|
||||
|
||||
go func() {
|
||||
_, err := io.Copy(localConn, remoteConn)
|
||||
if err != nil {
|
||||
done <- err
|
||||
closeConnections()
|
||||
return
|
||||
}
|
||||
// Close write half of local connection if supported.
|
||||
if hc, ok := localConn.(halfCloser); ok {
|
||||
hc.CloseWrite()
|
||||
}
|
||||
done <- err
|
||||
done <- nil
|
||||
}()
|
||||
|
||||
// Wait for both copies to complete or context cancel.
|
||||
// Wait for both copies to complete. The first error is the original failure because a copy reports it before
|
||||
// closing the connections to unblock the other copy.
|
||||
var copyErr error
|
||||
for range 2 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// Close connections to abort ongoing copies.
|
||||
localConn.Close()
|
||||
remoteConn.Close()
|
||||
return
|
||||
case err = <-done:
|
||||
if err != nil && p.OnError != nil {
|
||||
p.OnError(fmt.Errorf("data copy: %w", err))
|
||||
}
|
||||
if err = <-done; err != nil && copyErr == nil {
|
||||
copyErr = err
|
||||
}
|
||||
}
|
||||
|
||||
if copyErr != nil && ctx.Err() == nil && p.OnError != nil {
|
||||
p.OnError(fmt.Errorf("data copy: %w", copyErr))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestIsConnectionClosedError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{name: "closed network connection", err: net.ErrClosed, want: true},
|
||||
{name: "closed pipe", err: io.ErrClosedPipe, want: true},
|
||||
{name: "broken pipe", err: syscall.EPIPE, want: true},
|
||||
{name: "connection reset", err: syscall.ECONNRESET, want: true},
|
||||
{name: "wrapped connection error", err: fmt.Errorf("copy data: %w", syscall.EPIPE), want: true},
|
||||
{name: "other error", err: errors.New("copy failed"), want: false},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
require.Equal(t, tt.want, IsConnectionClosedError(tt.err))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunContinuesAfterClosedConnectionError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
listener := newTestListener()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
closedErrCh := make(chan error, 1)
|
||||
unexpectedErrCh := make(chan error, 1)
|
||||
var dialCount atomic.Int32
|
||||
|
||||
p := &Proxy{
|
||||
Listener: listener,
|
||||
RemoteAddr: "remote:80",
|
||||
DialContext: func(context.Context, string, string) (net.Conn, error) {
|
||||
if dialCount.Add(1) == 1 {
|
||||
return readErrorConn{err: syscall.EPIPE}, nil
|
||||
}
|
||||
|
||||
proxyConn, upstreamConn := net.Pipe()
|
||||
go func() {
|
||||
defer upstreamConn.Close()
|
||||
_, _ = upstreamConn.Write([]byte("ok"))
|
||||
}()
|
||||
return proxyConn, nil
|
||||
},
|
||||
OnError: func(err error) {
|
||||
if IsConnectionClosedError(err) {
|
||||
closedErrCh <- err
|
||||
return
|
||||
}
|
||||
unexpectedErrCh <- err
|
||||
},
|
||||
}
|
||||
|
||||
runErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
runErrCh <- p.Run(ctx)
|
||||
}()
|
||||
|
||||
firstConn := listener.connect()
|
||||
defer firstConn.Close()
|
||||
|
||||
select {
|
||||
case connErr := <-closedErrCh:
|
||||
require.ErrorIs(t, connErr, syscall.EPIPE)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for closed connection error")
|
||||
}
|
||||
|
||||
secondConn := listener.connect()
|
||||
defer secondConn.Close()
|
||||
require.NoError(t, secondConn.SetReadDeadline(time.Now().Add(time.Second)))
|
||||
|
||||
got := make([]byte, 2)
|
||||
_, err := io.ReadFull(secondConn, got)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "ok", string(got))
|
||||
|
||||
select {
|
||||
case unexpectedErr := <-unexpectedErrCh:
|
||||
t.Fatalf("unexpected connection error: %v", unexpectedErr)
|
||||
default:
|
||||
}
|
||||
|
||||
cancel()
|
||||
select {
|
||||
case runErr := <-runErrCh:
|
||||
require.NoError(t, runErr)
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for proxy to stop")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunReturnsListenerError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
listenerErr := errors.New("listener failed")
|
||||
listener := errorListener{err: listenerErr}
|
||||
|
||||
p := &Proxy{Listener: listener}
|
||||
err := p.Run(context.Background())
|
||||
require.Error(t, err)
|
||||
require.ErrorContains(t, err, "accept local connection")
|
||||
require.ErrorIs(t, err, listenerErr)
|
||||
}
|
||||
|
||||
type testListener struct {
|
||||
conns chan net.Conn
|
||||
closed chan struct{}
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
func newTestListener() *testListener {
|
||||
return &testListener{
|
||||
conns: make(chan net.Conn),
|
||||
closed: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (l *testListener) connect() net.Conn {
|
||||
clientConn, proxyConn := net.Pipe()
|
||||
l.conns <- proxyConn
|
||||
return clientConn
|
||||
}
|
||||
|
||||
func (l *testListener) Accept() (net.Conn, error) {
|
||||
select {
|
||||
case conn := <-l.conns:
|
||||
return conn, nil
|
||||
case <-l.closed:
|
||||
return nil, net.ErrClosed
|
||||
}
|
||||
}
|
||||
|
||||
func (l *testListener) Close() error {
|
||||
l.closeOnce.Do(func() {
|
||||
close(l.closed)
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *testListener) Addr() net.Addr {
|
||||
return &net.TCPAddr{}
|
||||
}
|
||||
|
||||
type errorListener struct {
|
||||
err error
|
||||
}
|
||||
|
||||
func (l errorListener) Accept() (net.Conn, error) { return nil, l.err }
|
||||
func (errorListener) Close() error { return nil }
|
||||
func (errorListener) Addr() net.Addr { return &net.TCPAddr{} }
|
||||
|
||||
// readErrorConn fails reads immediately so tests can deterministically exercise a proxy copy failure.
|
||||
type readErrorConn struct {
|
||||
err error
|
||||
}
|
||||
|
||||
func (c readErrorConn) Read([]byte) (int, error) { return 0, c.err }
|
||||
func (readErrorConn) Write(p []byte) (int, error) { return len(p), nil }
|
||||
func (readErrorConn) Close() error { return nil }
|
||||
func (readErrorConn) LocalAddr() net.Addr { return &net.TCPAddr{} }
|
||||
func (readErrorConn) RemoteAddr() net.Addr { return &net.TCPAddr{} }
|
||||
func (readErrorConn) SetDeadline(time.Time) error { return nil }
|
||||
func (readErrorConn) SetReadDeadline(time.Time) error { return nil }
|
||||
func (readErrorConn) SetWriteDeadline(time.Time) error { return nil }
|
||||
@@ -265,6 +265,11 @@ type ContainerSpec struct {
|
||||
LogDriver *LogDriver
|
||||
// PidMode sets the PID namespace mode for the container. Currently only "" or "host" is supported.
|
||||
PidMode string
|
||||
// Tty allocates a pseudo-TTY and connects the container's standard streams to it.
|
||||
// Standard output and standard error share one stream.
|
||||
Tty bool
|
||||
// OpenStdin allocates standard input and keeps it open.
|
||||
OpenStdin bool
|
||||
// Privileged gives extended privileges to the container. This is a security risk and should be used with caution.
|
||||
Privileged bool
|
||||
// PullPolicy determines when to pull the image from the registry or use the image already available in the cluster.
|
||||
|
||||
@@ -54,6 +54,8 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser
|
||||
Image: service.Image,
|
||||
Init: service.Init,
|
||||
PidMode: service.Pid,
|
||||
Tty: service.Tty,
|
||||
OpenStdin: service.StdinOpen,
|
||||
Privileged: service.Privileged,
|
||||
PullPolicy: pullPolicy,
|
||||
Resources: resourcesFromCompose(service),
|
||||
|
||||
@@ -128,6 +128,8 @@ func TestServiceSpecFromCompose(t *testing.T) {
|
||||
},
|
||||
},
|
||||
PidMode: "host",
|
||||
Tty: true,
|
||||
OpenStdin: true,
|
||||
Privileged: true,
|
||||
PullPolicy: api.PullPolicyAlways,
|
||||
Resources: api.ContainerResources{
|
||||
|
||||
@@ -7,6 +7,8 @@ services:
|
||||
command: ["nginx", "updated", "command"]
|
||||
cpus: 0.5
|
||||
pid: host
|
||||
tty: true
|
||||
stdin_open: true
|
||||
deploy:
|
||||
update_config:
|
||||
order: stop-first
|
||||
|
||||
@@ -66,6 +66,26 @@ func TestEvalContainerSpecChange_ContainerPidMode(t *testing.T) {
|
||||
assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(newSpec, currentSpec))
|
||||
}
|
||||
|
||||
func TestEvalContainerSpecChange_ContainerTty(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
currentSpec := api.ServiceSpec{
|
||||
Container: api.ContainerSpec{
|
||||
Image: "nginx:latest",
|
||||
},
|
||||
}
|
||||
newSpec := api.ServiceSpec{
|
||||
Container: api.ContainerSpec{
|
||||
Image: "nginx:latest",
|
||||
Tty: true,
|
||||
OpenStdin: true,
|
||||
},
|
||||
}
|
||||
|
||||
assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(currentSpec, newSpec))
|
||||
assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(newSpec, currentSpec))
|
||||
}
|
||||
|
||||
func TestEvalContainerSpecChange_ContainerResources(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
+12
-2
@@ -252,13 +252,19 @@ func (cli *Client) pushImageToMachine(
|
||||
// The proxy runs in a goroutine. Capture the first error in a channel
|
||||
// so we can surface it alongside the push error if push fails.
|
||||
proxyErrCh := make(chan error, 1)
|
||||
onProxyError := func(err error) {
|
||||
recordProxyError := func(err error) {
|
||||
select {
|
||||
case proxyErrCh <- fmt.Errorf("proxy to unregistry: %w", err):
|
||||
default:
|
||||
}
|
||||
pw.Event(progress.NewEvent(proxyEventID, progress.Error, err.Error()))
|
||||
}
|
||||
onProxyError := func(err error) {
|
||||
if proxy.IsConnectionClosedError(err) {
|
||||
return
|
||||
}
|
||||
recordProxyError(err)
|
||||
}
|
||||
|
||||
// socketPath is set for plain rootless Docker (not running inside a VM): the Go proxy listens on a unix
|
||||
// socket that is bind-mounted into the socat container, bypassing slirp4netns network routing entirely.
|
||||
@@ -315,7 +321,11 @@ func (cli *Client) pushImageToMachine(
|
||||
}
|
||||
defer cleanup()
|
||||
|
||||
go unregProxy.Run(proxyCtx)
|
||||
go func() {
|
||||
if err := unregProxy.Run(proxyCtx); err != nil {
|
||||
recordProxyError(err)
|
||||
}
|
||||
}()
|
||||
|
||||
if dockerEnv.Virtualised {
|
||||
// VM-based Docker (Docker Desktop, Rancher Desktop, etc.): run a socat container inside the VM
|
||||
|
||||
@@ -57,6 +57,12 @@ curl -fsS https://get.uncloud.run/install.sh | VERSION=nightly sh
|
||||
|
||||
More information about nightly builds of the CLI and daemon can be found [here](https://github.com/psviderski/uncloud/releases/tag/nightly).
|
||||
|
||||
## mise
|
||||
You can install uncloud with [mise](https://mise.jdx.dev/):
|
||||
```
|
||||
mise use github:psviderski/uncloud[exe=uc]
|
||||
```
|
||||
|
||||
## GitHub download (macOS, Linux)
|
||||
|
||||
You can manually download and use a pre-built binary from the
|
||||
|
||||
@@ -45,9 +45,11 @@ If you rely on a specific Compose feature that is not supported by Uncloud, plea
|
||||
| `secrets` | ⚠️ Limited | Reference secrets in `environment`, see [Secrets](../3-concepts/8-secrets.md). File mounts not supported |
|
||||
| `security_opt` | ❌ Not supported | |
|
||||
| `shm_size` | ✅ Supported | Shared memory size |
|
||||
| `stdin_open` | ✅ Supported | Allocate standard input and keep it open |
|
||||
| `stop_grace_period` | ✅ Supported | Time to wait after SIGTERM before SIGKILL |
|
||||
| `storage_opt` | ❌ Not supported | |
|
||||
| `sysctls` | ✅ Supported | Namespaced kernel parameters |
|
||||
| `tty` | ✅ Supported | Allocate a pseudo-TTY and connect the container's standard streams to it |
|
||||
| `ulimits` | ✅ Supported | Resource limits |
|
||||
| `user` | ✅ Supported | Set container user |
|
||||
| `volumes` | ✅ Supported | Named volumes, bind mounts, tmpfs |
|
||||
|
||||
Reference in New Issue
Block a user