chore: Handle sftp reconnections

Signed-off-by: Javier Aliaga <javier@diagrid.io>

Author: javier-aliaga

bindings/sftp/integration/README.md

@@ -0,0 +1,42 @@
+# SFTP with Toxiproxy Setup
+
+This setup includes an SFTP server and Toxiproxy for simulating network conditions during testing.
+
+## Services
+
+- **SFTP**: A simple SFTP server with a test user
+- **Toxiproxy**: A TCP proxy that allows simulating network conditions like latency, bandwidth restrictions, and connection failures
+
+## Getting Started
+
+1. Start the services:
+
+```bash
+docker-compose up -d
+```
+
+2. Connect to SFTP via the Toxiproxy port:
+
+```bash
+sftp -P 2222 foo@localhost
+# Password: pass
+```
+
+3. Control Toxiproxy via its API (port 8474):
+
+```bash
+# Add 1000ms latency to SFTP connections
+curl -X POST -H "Content-Type: application/json" \
+  http://localhost:8474/proxies/sftp/toxics \
+  -d '{"type":"latency", "attributes":{"latency":1000, "jitter":0}}'  
+
+# Simulate connection timeout
+curl -X POST -H "Content-Type: application/json" \
+  http://localhost:8474/proxies/sftp/toxics \
+  -d '{"type":"timeout", "attributes":{"timeout":1000}}'
+
+# Remove all toxics
+curl -X DELETE http://localhost:8474/proxies/sftp/toxics
+```
+
+See the [Toxiproxy documentation](https://github.com/Shopify/toxiproxy) for more information on available toxics and configuration options.

bindings/sftp/integration/docker-compose.yaml

@@ -0,0 +1,34 @@
+version: '3.8'
+
+services:
+  toxiproxy:
+    image: ghcr.io/shopify/toxiproxy:2.5.0
+    ports:
+      - "8474:8474"  # Toxiproxy API
+      - "2222:2222"  # Proxied SFTP port
+    environment:
+      - HOSTNAME=0.0.0.0
+    command: ["-host", "0.0.0.0"]
+    volumes:
+      - ./toxiproxy.json:/config/toxiproxy.json
+    depends_on:
+      - sftp
+    networks:
+      - sftp-network
+
+  sftp:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    environment:
+      - SFTP_USERS=foo:pass:1001:1001:upload
+    volumes:
+      - ./upload:/home/testuser/upload
+    expose:
+      - "2222:22"
+    networks:
+      - sftp-network
+
+networks:
+  sftp-network:
+    driver: bridge

bindings/sftp/sftp.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"reflect"
+	"sync"
 
 	sftpClient "github.com/pkg/sftp"
 	"golang.org/x/crypto/ssh"
@@ -25,9 +26,30 @@ const (
 
 // Sftp is a binding for file operations on sftp server.
 type Sftp struct {
-	metadata   *sftpMetadata
-	logger     logger.Logger
-	sftpClient *sftpClient.Client
+	metadata     *sftpMetadata
+	logger       logger.Logger
+	sftpClient   *sftpClient.Client
+	sshClient    *ssh.Client
+	clientConfig *ssh.ClientConfig
+	lock         sync.RWMutex
+}
+
+func (sftp *Sftp) Client() (*sftpClient.Client, error) {
+	sftp.lock.RLock()
+	current := sftp.sftpClient
+	sftp.lock.RUnlock()
+
+	if current != nil {
+		if _, err := current.Getwd(); err == nil {
+			return current, nil
+		}
+	}
+
+	err := sftp.handleReconnection()
+	if err != nil {
+		return nil, err
+	}
+	return sftp.sftpClient, nil
 }
 
 // sftpMetadata defines the sftp metadata.
@@ -122,9 +144,11 @@ func (sftp *Sftp) Init(_ context.Context, metadata bindings.Metadata) error {
 
 	newSftpClient, err := sftpClient.NewClient(sshClient)
 	if err != nil {
+		_ = sshClient.Close()
 		return fmt.Errorf("sftp binding error: error create sftp client: %w", err)
 	}
 
+	sftp.clientConfig = config
 	sftp.metadata = m
 	sftp.sftpClient = newSftpClient
 
@@ -163,12 +187,17 @@ func (sftp *Sftp) create(_ context.Context, req *bindings.InvokeRequest) (*bindi
 
 	dir, fileName := sftpClient.Split(path)
 
-	err = sftp.sftpClient.MkdirAll(dir)
+	c, err := sftp.Client()
+	if err != nil {
+		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
+	}
+
+	err = c.MkdirAll(dir)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error create dir %s: %w", dir, err)
 	}
 
-	file, err := sftp.sftpClient.Create(path)
+	file, err := c.Create(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error create file %s: %w", path, err)
 	}
@@ -211,7 +240,12 @@ func (sftp *Sftp) list(_ context.Context, req *bindings.InvokeRequest) (*binding
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	files, err := sftp.sftpClient.ReadDir(path)
+	c, err := sftp.Client()
+	if err != nil {
+		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
+	}
+
+	files, err := c.ReadDir(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error read dir %s: %w", path, err)
 	}
@@ -246,7 +280,12 @@ func (sftp *Sftp) get(_ context.Context, req *bindings.InvokeRequest) (*bindings
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	file, err := sftp.sftpClient.Open(path)
+	c, err := sftp.Client()
+	if err != nil {
+		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
+	}
+
+	file, err := c.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error open file %s: %w", path, err)
 	}
@@ -272,7 +311,11 @@ func (sftp *Sftp) delete(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	err = sftp.sftpClient.Remove(path)
+	c, err := sftp.Client()
+	if err != nil {
+		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
+	}
+	err = c.Remove(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error remove file %s: %w", path, err)
 	}
@@ -296,6 +339,8 @@ func (sftp *Sftp) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bin
 }
 
 func (sftp *Sftp) Close() error {
+	sftp.lock.Lock()
+	defer sftp.lock.Unlock()
 	return sftp.sftpClient.Close()
 }
 
@@ -330,3 +375,41 @@ func (sftp *Sftp) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 	metadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, metadata.BindingType)
 	return
 }
+
+func (sftp *Sftp) handleReconnection() error {
+	sftp.lock.Lock()
+	defer sftp.lock.Unlock()
+
+	// Re-check after acquiring the write lock
+	if sftp.sftpClient != nil {
+		if _, err := sftp.sftpClient.Getwd(); err == nil {
+			return nil
+		}
+		_ = sftp.sftpClient.Close()
+		sftp.sftpClient = nil
+	}
+	if sftp.sshClient != nil {
+		_ = sftp.sshClient.Close()
+		sftp.sshClient = nil
+	}
+
+	if sftp.metadata == nil || sftp.clientConfig == nil {
+		return fmt.Errorf("sftp binding error: client not initialized")
+	}
+
+	sshClient, err := ssh.Dial("tcp", sftp.metadata.Address, sftp.clientConfig)
+	if err != nil {
+		return fmt.Errorf("sftp binding error: error create ssh client: %w", err)
+	}
+
+	newSftpClient, err := sftpClient.NewClient(sshClient)
+	if err != nil {
+		_ = sshClient.Close()
+		return fmt.Errorf("sftp binding error: error create sftp client: %w", err)
+	}
+
+	sftp.sshClient = sshClient
+	sftp.sftpClient = newSftpClient
+
+	return nil
+}

bindings/sftp/sftp_integration_test.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"testing"
 
+	toxiproxy "github.com/Shopify/toxiproxy/client"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -14,11 +15,13 @@ import (
 var connectionStringEnvKey = "DAPR_TEST_SFTP_CONNSTRING"
 
 // Run docker from the file location as the upload folder is relative to the test
-// docker run -v ./upload:/home/foo/upload -p 2222:22 -d atmoz/sftp foo:pass:1001
+// cd integration
+// docker-compose up -d
+// export DAPR_TEST_SFTP_CONNSTRING=sftp:22
 func TestIntegrationCases(t *testing.T) {
 	connectionString := os.Getenv(connectionStringEnvKey)
 	if connectionString == "" {
-		t.Skipf(`sftp binding integration tests skipped. To enable this test, define the connection string using environment variable '%[1]s' (example 'export %[1]s="localhost:2222")'`, connectionStringEnvKey)
+		t.Skipf("sftp binding integration skipped. To enable this test, define the connection string using environment variable '%[1]s' (example 'export %[1]s=\"localhost:2222\")'", connectionStringEnvKey)
 	}
 
 	t.Run("List operation", testListOperation)
@@ -27,21 +30,40 @@ func TestIntegrationCases(t *testing.T) {
 
 func testListOperation(t *testing.T) {
 	c := Sftp{}
+
+	client := toxiproxy.NewClient("localhost:8474")
+	p, err := client.CreateProxy("sftp", "0.0.0.0:2222", os.Getenv(connectionStringEnvKey))
+	require.NoError(t, err)
+	defer p.Delete()
+
 	m := bindings.Metadata{}
+
 	m.Properties = map[string]string{
 		"rootPath":              "/upload",
-		"address":               os.Getenv(connectionStringEnvKey),
+		"address":               "0.0.0.0:2222",
 		"username":              "foo",
 		"password":              "pass",
 		"insecureIgnoreHostKey": "true",
 	}
-	err := c.Init(t.Context(), m)
+
+	err = c.Init(t.Context(), m)
 	require.NoError(t, err)
 
 	r, err := c.Invoke(t.Context(), &bindings.InvokeRequest{Operation: bindings.ListOperation})
 	require.NoError(t, err)
 	assert.NotNil(t, r.Data)
 
+	tx, err := p.AddToxic("reset", "reset_peer", "downstream", 1, toxiproxy.Attributes{})
+	require.NoError(t, err)
+	defer p.RemoveToxic(tx.Name)
+
+	_, err = c.Invoke(t.Context(), &bindings.InvokeRequest{Operation: bindings.ListOperation})
+	require.Error(t, err)
+
+	p.RemoveToxic(tx.Name)
+	r, err = c.Invoke(t.Context(), &bindings.InvokeRequest{Operation: bindings.ListOperation})
+	require.NoError(t, err)
+
 	var d []listResponse
 	err = json.Unmarshal(r.Data, &d)
 	require.NoError(t, err)

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/Azure/go-amqp v1.0.5
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/IBM/sarama v1.45.2
+	github.com/Shopify/toxiproxy v2.1.4+incompatible
 	github.com/aerospike/aerospike-client-go/v6 v6.12.0
 	github.com/alibaba/sentinel-golang v1.0.4
 	github.com/alibabacloud-go/darabonba-openapi v0.2.1

go.sum

@@ -156,6 +156,7 @@ github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAE
 github.com/RoaringBitmap/roaring v1.1.0 h1:b10lZrZXaY6Q6EKIRrmOF519FIyQQ5anPgGr3niw2yY=
 github.com/RoaringBitmap/roaring v1.1.0/go.mod h1:icnadbWcNyfEHlYdr+tDlOTih1Bf/h+rzPpv4sbomAA=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
+github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
 github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=