chore: make ssh+cli connections reuse one SSH connection via control socket. Fix image push

This commit is contained in:
Pasha Sviderski
2026-01-28 16:49:08 +10:00
parent e0a63a3f49
commit c91a964513
5 changed files with 225 additions and 70 deletions
+90 -24
View File
@@ -4,6 +4,8 @@ import (
"context"
"fmt"
"net"
"os"
"path/filepath"
"strconv"
"github.com/docker/cli/cli/connhelper/commandconn"
@@ -18,26 +20,72 @@ import (
type SSHCLIConnector struct {
config SSHConnectorConfig
conn net.Conn
// Path to SSH control socket for connection reuse.
controlSockPath string
}
func NewSSHCLIConnector(cfg *SSHConnectorConfig) *SSHCLIConnector {
return &SSHCLIConnector{config: *cfg}
return &SSHCLIConnector{
config: *cfg,
controlSockPath: controlSocketPath(),
}
}
// controlSocketPath returns a unique control socket path for the SSH connection.
// Returns an empty string if unable to find or create a suitable path.
func controlSocketPath() string {
// %C is expanded by `ssh` to a hash of user, local and remote hostnames, port, and the contents
// of the ProxyJump option. This ensures that shared connections are uniquely identified.
sockName := fmt.Sprintf("uc_control_%%C.sock")
// Prefer XDG_RUNTIME_DIR if set, fall back to ~/.ssh if it exists.
if dir := os.Getenv("XDG_RUNTIME_DIR"); dir != "" {
return filepath.Join(dir, sockName)
}
if home, err := os.UserHomeDir(); err == nil {
sshDir := filepath.Join(home, ".ssh")
if fi, sErr := os.Stat(sshDir); sErr == nil && fi.IsDir() {
return filepath.Join(sshDir, sockName)
}
}
// Last resort: create a subdirectory in temp with restricted permissions.
tmpDir := filepath.Join(os.TempDir(), fmt.Sprintf("uncloud-%d", os.Getuid()))
path := filepath.Join(tmpDir, sockName)
if len(path)-2+40 < 104 { // 40 chars for %C hash, 104 is typical UNIX socket path limit
if err := os.MkdirAll(tmpDir, 0o700); err == nil {
return path
}
}
return ""
}
// 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 {
args := []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 non-standard.
if d.config.Port != 0 && d.config.Port != 22 {
// Add port if specified.
if d.config.Port != 0 {
args = append(args, "-p", strconv.Itoa(d.config.Port))
}
@@ -49,8 +97,8 @@ func (d *sshCLIDialer) buildDialArgs(address string) []string {
// Add -W flag for stdin/stdout forwarding to target address.
args = append(args, "-W", address)
// Add user@host.
args = append(args, d.config.User+"@"+d.config.Host)
// Add [user@]host destination.
args = append(args, d.config.Destination())
return args
}
@@ -65,28 +113,28 @@ func (d *sshCLIDialer) DialContext(ctx context.Context, network, address string)
// Build SSH command arguments.
args := d.buildDialArgs(address)
// Create connection using commandconn.
// Create connection using docker's commandconn.
conn, err := commandconn.New(ctx, "ssh", args...)
if err != nil {
return nil, fmt.Errorf("SSH connection to %s@%s for dialing %s: %w", d.config.User, d.config.Host, address, err)
return nil, fmt.Errorf("SSH connection to %s for dialing %s: %w", d.config.Destination(), address, err)
}
return conn, nil
}
func (c *SSHCLIConnector) Connect(ctx context.Context) (*grpc.ClientConn, error) {
// Build SSH command arguments.
args := c.buildSSHArgs()
if c.conn == nil {
args := c.buildSSHArgs()
// Create connection using commandconn.
conn, err := commandconn.New(ctx, "ssh", args...)
if err != nil {
return nil, fmt.Errorf("SSH connection to %s@%s: %w", c.config.User, c.config.Host, err)
// Create connection using docker's commandconn.
conn, err := commandconn.New(ctx, "ssh", args...)
if err != nil {
return nil, fmt.Errorf("SSH connection to %s: %w", c.config.Destination(), err)
}
c.conn = conn
}
c.conn = conn
// Create gRPC client over the connection.
// Use a custom dialer that returns our existing connection.
// Create gRPC client over the connection. Use a custom dialer that returns our existing connection.
grpcConn, err := grpc.NewClient(
"passthrough:///", // Dummy target since we're using a custom dialer.
grpc.WithTransportCredentials(insecure.NewCredentials()),
@@ -103,15 +151,32 @@ func (c *SSHCLIConnector) Connect(ctx context.Context) (*grpc.ClientConn, error)
return grpcConn, nil
}
// buildSSHArgs constructs the SSH command arguments.
// buildSSHArgs constructs the SSH command arguments to run `uncloudd dial-stdio` on the remote machine reusing
// the established connection via control socket.
func (c *SSHCLIConnector) buildSSHArgs() []string {
args := []string{}
var args []string
// Add control socket options for connection reuse if available.
if c.controlSockPath != "" {
args = append(args, "-o", "ControlMaster=auto")
args = append(args, "-o", "ControlPath="+c.controlSockPath)
// Keep the established connection alive for a short duration after the last session closes to allow reuse.
controlPersist := "10m"
// Override the default duration with the UNCLOUD_SSH_CONTROL_PERSIST env variable.
if v := os.Getenv("UNCLOUD_SSH_CONTROL_PERSIST"); v != "" {
controlPersist = v
}
args = append(args, "-o", "ControlPersist="+controlPersist)
}
// 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 non-standard.
if c.config.Port != 0 && c.config.Port != 22 {
// Add port if specified.
if c.config.Port != 0 {
args = append(args, "-p", strconv.Itoa(c.config.Port))
}
@@ -120,8 +185,8 @@ func (c *SSHCLIConnector) buildSSHArgs() []string {
args = append(args, "-i", c.config.KeyPath)
}
// Add user@host.
args = append(args, c.config.User+"@"+c.config.Host)
// Add [user@]host destination.
args = append(args, c.config.Destination())
// Add remote command: uncloudd dial-stdio
args = append(args, "uncloudd", "dial-stdio")
@@ -141,7 +206,8 @@ func (c *SSHCLIConnector) Dialer() (proxy.ContextDialer, error) {
}
return &sshCLIDialer{
config: c.config,
config: c.config,
controlSockPath: c.controlSockPath,
}, nil
}