feat: use rsync instead of sftp

.github/workflows/tests.yml

@@ -25,6 +25,10 @@ jobs:
       TS_SERVER_OAUTH_SECRET: ${{ secrets.TS_SERVER_OAUTH_SECRET }}
     steps:
       - uses: actions/checkout@v3
+      - name: Connect to Tailscale
+        uses: tailscale/github-action@v4
+        with:
+          authkey: ${{ secrets.TS_CI_AUTH_KEY }}
       - name: Setup Go
         uses: actions/setup-go@v4
         with:
@@ -125,6 +129,10 @@ jobs:
       TS_SERVER_OAUTH_SECRET: ${{ secrets.TS_SERVER_OAUTH_SECRET }}
     steps:
       - uses: actions/checkout@v3
+      - name: Connect to Tailscale
+        uses: tailscale/github-action@v4
+        with:
+          authkey: ${{ secrets.TS_CI_AUTH_KEY }}
       - name: Setup Go
         uses: actions/setup-go@v4
         with:

go.mod

@@ -29,7 +29,6 @@ require (
 	github.com/opencontainers/image-spec v1.1.0
 	github.com/pelletier/go-toml/v2 v2.2.4
 	github.com/pkg/errors v0.9.1
-	github.com/pkg/sftp v1.13.7
 	github.com/prometheus/client_golang v1.22.0
 	github.com/skip-mev/catalyst v0.0.0-beta.15
 	github.com/spf13/afero v1.12.0
@@ -272,7 +271,6 @@ require (
 	github.com/kkHAIKE/contextcheck v1.1.4 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
-	github.com/kr/fs v0.1.0 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/kulti/thelper v0.6.3 // indirect
@@ -365,6 +363,7 @@ require (
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/sourcegraph/go-diff v0.7.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/spf13/afero v1.12.0 // indirect
 	github.com/spf13/cast v1.9.2 // indirect
 	github.com/spf13/cobra v1.9.1 // indirect
 	github.com/spf13/pflag v1.0.7 // indirect

go.sum

@@ -1666,7 +1666,6 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
-golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
 golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4=
 golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -1773,7 +1772,6 @@ golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.2.0/go.mod h1:KqCZLdyyvdV855qA2rE3GC2aiw5xGR5TEjj8smXukLY=
 golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE=
 golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -1880,9 +1878,7 @@ golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
 golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -1891,8 +1887,6 @@ golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.2.0/go.mod h1:TVmDHMZPmdnySmBfhjOoOdhjzdE1h4u1VwSiw2l1Nuc=
 golang.org/x/term v0.4.0/go.mod h1:9P2UbLfCdcvo3p/nzKvsmas4TnlujnuoV9hGgYzW1lQ=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
-golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
-golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
 golang.org/x/term v0.34.0 h1:O/2T7POpk0ZZ7MAzMeWFSg6S5IpWd/RXDlM9hgM3DR4=
 golang.org/x/term v0.34.0/go.mod h1:5jC53AEywhIVebHgPVeg0mj8OD3VO9OzclacVrqpaAw=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -1906,8 +1900,6 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
 golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=

petri/core/provider/digitalocean/droplet.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/digitalocean/godo"
 	"go.uber.org/zap"
-	"golang.org/x/crypto/ssh"
 
 	"github.com/skip-mev/ironbird/petri/core/provider"
 	"github.com/skip-mev/ironbird/petri/core/provider/clients"
@@ -140,31 +139,3 @@ func (t *Task) GetDroplet(ctx context.Context) (*godo.Droplet, error) {
 	}
 	return t.doClient.GetDroplet(ctx, dropletId)
 }
-
-func (t *Task) getDropletSSHClient(ctx context.Context) (*ssh.Client, error) {
-	if t.sshClient != nil {
-		status, _, err := t.sshClient.SendRequest("ping", true, []byte("ping"))
-
-		if err == nil && status {
-			return t.sshClient, nil
-		}
-	}
-
-	ip, err := t.GetIP(ctx)
-
-	if err != nil {
-		return nil, err
-	}
-
-	sshConfig := &ssh.ClientConfig{
-		User:            "root",
-		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
-	}
-
-	client, err := SSHDialWithCustomDial(ctx, "tcp", fmt.Sprintf("%s:22", ip), sshConfig, t.DialContext())
-	if err != nil {
-		return nil, err
-	}
-
-	return client, nil
-}

petri/core/provider/digitalocean/task.go

@@ -5,19 +5,16 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"os"
+	"os/exec"
 	"path"
 	"sync"
 	"time"
 
-	"golang.org/x/crypto/ssh"
-
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/mount"
 	dockerclient "github.com/docker/docker/client"
 	"github.com/docker/docker/pkg/stdcopy"
-	"github.com/pkg/sftp"
-	"github.com/spf13/afero"
-	"github.com/spf13/afero/sftpfs"
 	"go.uber.org/zap"
 
 	"github.com/skip-mev/ironbird/petri/core/provider"
@@ -40,7 +37,6 @@ type Task struct {
 
 	removeTask        provider.RemoveTaskFunc
 	logger            *zap.Logger
-	sshClient         *ssh.Client
 	doClient          DoClient
 	dockerClient      clients.DockerClient
 	tailscaleSettings TailscaleSettings
@@ -189,33 +185,52 @@ func (t *Task) GetStatus(ctx context.Context) (provider.TaskStatus, error) {
 func (t *Task) WriteFile(ctx context.Context, relPath string, content []byte) error {
 	absPath := path.Join("/docker_volumes", relPath)
 
-	sshClient, err := t.getDropletSSHClient(ctx)
+	tmpFile, err := os.CreateTemp("", "rsync-writefile-*")
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to create temp file: %w", err)
 	}
+	tmpPath := tmpFile.Name()
+	defer func() {
+		if err := os.Remove(tmpPath); err != nil {
+			t.logger.Warn("failed to remove temp file", zap.String("path", tmpPath), zap.Error(err))
+		}
+	}()
 
-	defer sshClient.Close()
+	if _, err := tmpFile.Write(content); err != nil {
+		if closeErr := tmpFile.Close(); closeErr != nil {
+			t.logger.Warn("failed to close temp file", zap.String("path", tmpPath), zap.Error(closeErr))
+		}
+		return fmt.Errorf("failed to write to temp file: %w", err)
+	}
+	if err := tmpFile.Close(); err != nil {
+		return fmt.Errorf("failed to close temp file: %w", err)
+	}
 
-	sftpClient, err := sftp.NewClient(sshClient)
-	if err != nil {
-		return err
+	if err := os.Chmod(tmpPath, 0644); err != nil {
+		return fmt.Errorf("failed to set temp file permissions: %w", err)
 	}
 
-	defer sftpClient.Close()
+	mkdirCmd := []string{"mkdir", "-p", path.Dir(absPath)}
+	if _, _, exitCode, err := t.RunCommand(ctx, mkdirCmd); err != nil || exitCode != 0 {
+		return fmt.Errorf("failed to create directory (exit code %d): %w", exitCode, err)
+	}
 
-	err = sftpClient.MkdirAll(path.Dir(absPath))
+	tailscaleIP, err := t.getTailscaleIp(ctx)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to get tailscale IP: %w", err)
 	}
 
-	file, err := sftpClient.Create(absPath)
-	if err != nil {
-		return err
+	rsyncArgs := []string{
+		"-Wp", // whole-file mode + preserve permissions
+		"--inplace",
+		"-e", "tailscale ssh",
+		tmpPath,
+		fmt.Sprintf("root@%s:%s", tailscaleIP, absPath),
 	}
 
-	_, err = file.Write(content)
-	if err != nil {
-		return err
+	cmd := exec.CommandContext(ctx, "rsync", rsyncArgs...)
+	if output, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("rsync failed: %w (output: %s)", err, string(output))
 	}
 
 	return nil
@@ -224,25 +239,41 @@ func (t *Task) WriteFile(ctx context.Context, relPath string, content []byte) er
 func (t *Task) ReadFile(ctx context.Context, relPath string) ([]byte, error) {
 	absPath := path.Join("/docker_volumes", relPath)
 
-	sshClient, err := t.getDropletSSHClient(ctx)
+	tmpFile, err := os.CreateTemp("", "rsync-readfile-*")
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to create temp file: %w", err)
 	}
+	tmpPath := tmpFile.Name()
+	if err := tmpFile.Close(); err != nil {
+		return nil, fmt.Errorf("failed to close temp file: %w", err)
+	}
+	defer func() {
+		if err := os.Remove(tmpPath); err != nil {
+			t.logger.Warn("failed to remove temp file", zap.String("path", tmpPath), zap.Error(err))
+		}
+	}()
 
-	defer sshClient.Close()
-
-	sftpClient, err := sftp.NewClient(sshClient)
+	tailscaleIP, err := t.getTailscaleIp(ctx)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to get tailscale IP: %w", err)
 	}
 
-	defer sftpClient.Close()
+	rsyncArgs := []string{
+		"-Wz", // whole-file mode + compression
+		"--inplace",
+		"-e", "tailscale ssh",
+		fmt.Sprintf("root@%s:%s", tailscaleIP, absPath),
+		tmpPath,
+	}
 
-	fs := sftpfs.New(sftpClient)
+	cmd := exec.CommandContext(ctx, "rsync", rsyncArgs...)
+	if output, err := cmd.CombinedOutput(); err != nil {
+		return nil, fmt.Errorf("rsync failed: %w (output: %s)", err, string(output))
+	}
 
-	content, err := afero.ReadFile(fs, absPath)
+	content, err := os.ReadFile(tmpPath)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to read temp file: %w", err)
 	}
 
 	return content, nil