run install.sh script when initialising a cluster on the init machine

internal/cli/cli.go

@@ -132,13 +132,17 @@ func (cli *CLI) InitCluster(
 func (cli *CLI) initRemoteMachine(
 	ctx context.Context, remoteMachine RemoteMachine, clusterName, machineName string, netPrefix netip.Prefix,
 ) error {
-	sshConfig := &connector.SSHConnectorConfig{
-		User:    remoteMachine.User,
-		Host:    remoteMachine.Host,
-		Port:    remoteMachine.Port,
-		KeyPath: remoteMachine.KeyPath,
+	// Create a command executor and a machine API client over the SSH connection to the remote machine.
+	sshClient, err := sshexec.Connect(remoteMachine.User, remoteMachine.Host, remoteMachine.Port, remoteMachine.KeyPath)
+	if err != nil {
+		return fmt.Errorf(
+			"SSH login to remote machine %s: %w",
+			config.NewSSHDestination(remoteMachine.User, remoteMachine.Host, remoteMachine.Port), err,
+		)
 	}
-	c, err := client.New(ctx, connector.NewSSHConnector(sshConfig))
+	exec := sshexec.NewRemote(sshClient)
+
+	c, err := client.New(ctx, connector.NewSSHConnectorFromClient(sshClient))
 	if err != nil {
 		return fmt.Errorf("connect to remote machine: %w", err)
 	}
@@ -157,9 +161,10 @@ func (cli *CLI) initRemoteMachine(
 		return fmt.Errorf("generate cluster user: %w", err)
 	}
 
-	// TODO: download and install the latest uncloudd binary by running the install shell script from GitHub.
-	//  For now upload the binary using scp manually.
-	// TODO: Check if the machine is already provisioned and ask the user to reset it first.
+	// Install and run the Uncloud daemon and dependencies on the remote machine.
+	if err = provisionMachine(ctx, exec); err != nil {
+		return fmt.Errorf("provision machine: %w", err)
+	}
 
 	req := &pb.InitClusterRequest{
 		MachineName: machineName,

internal/cli/machine.go

@@ -1,8 +1,48 @@
 package cli
 
+import (
+	"context"
+	"fmt"
+	"os"
+	"uncloud/internal/sshexec"
+)
+
+// TODO: support pinning the script version to the CLI version.
+const installScriptURL = "https://raw.githubusercontent.com/psviderski/uncloud/refs/heads/main/scripts/install.sh"
+
 type RemoteMachine struct {
 	User    string
 	Host    string
 	Port    int
 	KeyPath string
 }
+
+// provisionMachine provisions the remote machine by downloading the Uncloud install script from GitHub and running it.
+func provisionMachine(ctx context.Context, exec sshexec.Executor) error {
+	// TODO: Check if the machine is already provisioned and ask the user to reset it first.
+
+	user, err := exec.Run(ctx, "whoami")
+	if err != nil {
+		return fmt.Errorf("run whoami: %w", err)
+	}
+	sudoPrefix, env := "", ""
+	if user != "root" {
+		sudoPrefix = "sudo"
+		// Add the SSH user (non-root) to the uncloud group to allow access to the Uncloud daemon unix socket.
+		env = "UNCLOUD_GROUP_ADD_USER=" + user
+	}
+
+	fmt.Println("Downloading Uncloud install script:", installScriptURL)
+	curlBashCmd := fmt.Sprintf(
+		"curl -fsSL %s | %s %s bash", sshexec.Quote(installScriptURL), sudoPrefix, sshexec.Quote(env),
+	)
+	cmd := sshexec.QuoteCommand("bash", "-c", "set -o pipefail; "+curlBashCmd)
+	// TODO: figure out why sometimes the output of `docker version` within the script is intermixed with other
+	//  script output. Note, that the same behavior is observed when using exec.Run but not when running the script
+	//  using ssh CLI. Requesting a pseudo-terminal may help fix this issue:
+	//		session.RequestPty("xterm", 40, 80, ssh.TerminalModes{})
+	if err = exec.Stream(ctx, cmd, os.Stdout, os.Stderr); err != nil {
+		return fmt.Errorf("download and run install script: %w", err)
+	}
+	return nil
+}

internal/sshexec/executor.go

@@ -2,12 +2,14 @@ package sshexec
 
 import (
 	"context"
+	"io"
 	"regexp"
 	"strings"
 )
 
 type Executor interface {
 	Run(ctx context.Context, cmd string) (string, error)
+	Stream(ctx context.Context, cmd string, stdout, stderr io.Writer) error
 	Close() error
 }
 

internal/sshexec/remote.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"golang.org/x/crypto/ssh"
+	"io"
 	"strings"
 )
 
@@ -54,6 +55,37 @@ func (r *Remote) Run(ctx context.Context, cmd string) (string, error) {
 	}
 }
 
+// Stream runs the command on the remote host and streams its output to the provided writers.
+func (r *Remote) Stream(ctx context.Context, cmd string, stdout, stderr io.Writer) error {
+	session, err := r.client.NewSession()
+	if err != nil {
+		return fmt.Errorf("create session: %w", err)
+	}
+	defer func() {
+		_ = session.Close()
+	}()
+
+	session.Stdout, session.Stderr = stdout, stderr
+	// Run the command in a goroutine to be able to cancel it.
+	done := make(chan error)
+	go func() {
+		done <- session.Run(cmd)
+	}()
+
+	select {
+	case err = <-done:
+		if err != nil {
+			return fmt.Errorf("run command on remote host: %w", err)
+		}
+		return nil
+	case <-ctx.Done():
+		if err = session.Signal(ssh.SIGINT); err != nil {
+			return fmt.Errorf("send interrupt signal to remote process: %w", err)
+		}
+		return fmt.Errorf("canceled: %w", ctx.Err())
+	}
+}
+
 // Close closes the connection to the remote host.
 func (r *Remote) Close() error {
 	return r.client.Close()