Fix push MinIO and SFTP files once are ineffective

integration/integration_minio_test.go

@@ -17,6 +17,7 @@ func TestIntegration_MinIO(t *testing.T) {
 		{"gofs MinIO pull", "", "run-gofs-minio-pull-client.yaml", "test-gofs-minio-pull.yaml"},
 		{"gofs MinIO push partial", "", "run-gofs-minio-push-client-partial.yaml", "test-gofs-minio-push-partial.yaml"},
 		{"gofs MinIO pull partial", "", "run-gofs-minio-pull-client-partial.yaml", "test-gofs-minio-pull-partial.yaml"},
+		{"gofs MinIO push once", "", "run-gofs-minio-push-client-once.yaml", "test-gofs-minio-push-once.yaml"},
 	}
 
 	for _, tc := range testCases {

integration/integration_sftp_test.go

@@ -15,6 +15,7 @@ func TestIntegration_SFTP(t *testing.T) {
 	}{
 		{"gofs SFTP push", "", "run-gofs-sftp-push-client.yaml", "test-gofs-sftp-push.yaml"},
 		{"gofs SFTP pull", "", "run-gofs-sftp-pull-client.yaml", "test-gofs-sftp-pull.yaml"},
+		{"gofs SFTP push once", "", "run-gofs-sftp-push-client-once.yaml", "test-gofs-sftp-push-once.yaml"},
 	}
 
 	for _, tc := range testCases {

integration/testdata/conf/run-gofs-minio-push-client-once.yaml

@@ -0,0 +1,5 @@
+source: ./minio-push-client-once/source
+dest: minio://127.0.0.1:9000?secure=false&local_sync_disabled=false&path=./minio-push-client-once/dest&remote_path=minio-bucket
+log_dir: ./minio-push-client-once-logs/
+users: minio_user|minio_pwd
+sync_once: true

integration/testdata/conf/run-gofs-sftp-push-client-once.yaml

@@ -0,0 +1,4 @@
+source: ./sftp-push-client-once/source
+dest: sftp://127.0.0.1:22?local_sync_disabled=false&path=./sftp-push-client-once/dest&remote_path=/sftp-workspace&ssh_user=sftp_user&ssh_pass=sftp_pwd
+log_dir: ./sftp-push-client-once-logs/
+sync_once: true

integration/testdata/test/test-gofs-minio-push-once.yaml

@@ -0,0 +1,104 @@
+name: test for gofs MinIO push once
+init:
+  - mkdir:
+    source: ./minio-push-client-once/source
+  - mkdir:
+    source: ./minio-push-client-once/dest
+  - cp:
+    source: ./integration_test.go
+    dest: ./minio-push-client-once/source/integration_test.go.bak1
+  - cp:
+    source: ./integration_test.go
+    dest: ./minio-push-client-once/source/integration_test.go.bak2
+  - sleep: 2s
+  - rm:
+    source: ./minio-push-client-once/source/integration_test.go.bak2
+  - touch:
+    source: ./minio-push-client-once/source/hello
+  - echo:
+    source: ./minio-push-client-once/source/hello
+    input: Hello World
+    append: false
+  - echo:
+    source: ./minio-push-client-once/source/hello
+    input: Bye Bye
+    append: true
+    no-newline: false
+  - echo:
+    source: ./minio-push-client-once/source/empty1
+    no-newline: true
+  - echo:
+    source: ./minio-push-client-once/source/empty2
+    no-newline: true
+  - symlink:
+    link: ./minio-push-client-once/source/hello.symlink
+    dest: ./minio-push-client-once/source/hello
+    ignore-error: false
+  - symlink:
+    link: ./minio-push-client-once/source/hello.symlink.bak
+    dest: ./minio-push-client-once/source/hello
+    ignore-error: false
+  - symlink:
+    link: ./minio-push-client-once/source/not_exist.symlink
+    dest: ./minio-push-client-once/source/not_exist
+    ignore-error: false
+  - sleep: 2s
+  - rm:
+    source: ./minio-push-client-once/source/hello.symlink.bak
+actions:
+  - sleep: 10s
+  - is-equal:
+    source: ./minio-push-client-once/source/integration_test.go.bak1
+    dest: ./minio-data-mount/integration_test.go.bak1
+    expect: true
+    must-non-empty: true
+  - is-equal:
+    source: ./minio-push-client-once/source/integration_test.go.bak1
+    dest: ./minio-data-mount/integration_test.go.bak1
+    expect: true
+    must-non-empty: true
+    algorithm: sha1
+  - is-equal:
+    source: ./minio-push-client-once/source/integration_test.go.bak1
+    dest: ./minio-data-mount/empty1
+    expect: false
+  - is-equal:
+    source: ./minio-push-client-once/source/empty1
+    dest: ./minio-data-mount/empty2
+    expect: true
+    must-non-empty: false
+  - is-empty:
+    source: ./minio-data-mount/integration_test.go.bak1
+    expect: false
+  - is-exist:
+    source: ./minio-data-mount/integration_test.go.bak1
+    expect: true
+  - is-exist:
+    source: ./minio-data-mount/integration_test.go.bak2
+    expect: false
+  - is-dir:
+    source: ./minio-data-mount
+    expect: true
+  - is-dir:
+    source: ./minio-data-mount/integration_test.go.bak1
+    expect: false
+  - hash:
+    algorithm: md5
+    source: ./minio-data-mount/hello
+    expect: 856719c57653fa86008d49db895a5752
+  - hash:
+    algorithm: sha1
+    source: ./minio-data-mount/hello
+    expect: f343874b5df87e887d85df2e790df33584463162
+  - is-empty:
+    source: ./minio-data-mount/hello.symlink
+    expect: false
+  - is-empty:
+    source: ./minio-data-mount/not_exist.symlink
+    expect: false
+  - is-exist:
+    source: ./minio-data-mount/hello.symlink.bak
+    expect: false
+clear:
+  - rm:
+    source: ./minio-push-client-once

integration/testdata/test/test-gofs-sftp-push-once.yaml

@@ -0,0 +1,106 @@
+name: test for gofs SFTP push once
+init:
+  - mkdir:
+    source: ./sftp-push-client-once/source
+  - mkdir:
+    source: ./sftp-push-client-once/dest
+  - cp:
+    source: ./integration_test.go
+    dest: ./sftp-push-client-once/source/integration_test.go.bak1
+  - cp:
+    source: ./integration_test.go
+    dest: ./sftp-push-client-once/source/integration_test.go.bak2
+  - sleep: 2s
+  - rm:
+    source: ./sftp-push-client-once/source/integration_test.go.bak2
+  - touch:
+    source: ./sftp-push-client-once/source/hello
+  - echo:
+    source: ./sftp-push-client-once/source/hello
+    input: Hello World
+    append: false
+  - echo:
+    source: ./sftp-push-client-once/source/hello
+    input: Bye Bye
+    append: true
+    no-newline: false
+  - echo:
+    source: ./sftp-push-client-once/source/empty1
+    no-newline: true
+  - echo:
+    source: ./sftp-push-client-once/source/empty2
+    no-newline: true
+  - symlink:
+    link: ./sftp-push-client-once/source/hello.symlink
+    dest: ./sftp-push-client-once/source/hello
+    ignore-error: false
+  - symlink:
+    link: ./sftp-push-client-once/source/hello.symlink.bak
+    dest: ./sftp-push-client-once/source/hello
+    ignore-error: false
+  - symlink:
+    link: ./sftp-push-client-once/source/not_exist.symlink
+    dest: ./sftp-push-client-once/source/not_exist
+    ignore-error: false
+  - sleep: 2s
+  - rm:
+    source: ./sftp-push-client-once/source/hello.symlink.bak
+actions:
+  - sleep: 10s
+  - is-equal:
+    source: ./sftp-push-client-once/source/integration_test.go.bak1
+    dest: /sftp-workspace/integration_test.go.bak1
+    expect: true
+    must-non-empty: true
+  - is-equal:
+    source: ./sftp-push-client-once/source/integration_test.go.bak1
+    dest: /sftp-workspace/integration_test.go.bak1
+    expect: true
+    must-non-empty: true
+    algorithm: sha1
+  - is-equal:
+    source: ./sftp-push-client-once/source/integration_test.go.bak1
+    dest: /sftp-workspace/empty1
+    expect: false
+  - is-equal:
+    source: ./sftp-push-client-once/source/empty1
+    dest: /sftp-workspace/empty2
+    expect: true
+    must-non-empty: false
+  - is-empty:
+    source: /sftp-workspace/integration_test.go.bak1
+    expect: false
+  - is-exist:
+    source: /sftp-workspace/integration_test.go.bak1
+    expect: true
+  - is-exist:
+    source: /sftp-workspace/integration_test.go.bak2
+    expect: false
+  - is-dir:
+    source: /sftp-workspace
+    expect: true
+  - is-dir:
+    source: /sftp-workspace/integration_test.go.bak1
+    expect: false
+  - hash:
+    algorithm: md5
+    source: /sftp-workspace/hello
+    expect: 856719c57653fa86008d49db895a5752
+  - hash:
+    algorithm: sha1
+    source: /sftp-workspace/hello
+    expect: f343874b5df87e887d85df2e790df33584463162
+  - is-symlink:
+    link: /sftp-workspace/hello.symlink
+    expect: true
+    ignore-error: false
+  - is-symlink:
+    link: /sftp-workspace/not_exist.symlink
+    expect: true
+    ignore-error: false
+  - is-exist:
+    source: /sftp-workspace/hello.symlink.bak
+    expect: false
+clear:
+  - rm:
+    source: ./sftp-push-client-once

sync/driver_push_client_sync.go

@@ -27,9 +27,11 @@ func newDriverPushClientSync(ds diskSync, basePath string) driverPushClientSync
 	}
 }
 
-func (s *driverPushClientSync) start() error {
-	if err := s.initFileInfo(); err != nil {
-		return err
+func (s *driverPushClientSync) start(isSync bool) error {
+	if !isSync {
+		if err := s.initFileInfo(); err != nil {
+			return err
+		}
 	}
 	return s.driver.Connect()
 }

sync/minio_push_client_sync.go

@@ -23,6 +23,8 @@ func NewMinIOPushClientSync(opt Option) (Sync, error) {
 	maxTranRate := opt.MaxTranRate
 	r := opt.Retry
 	logger := opt.Logger
+	syncOnce := opt.SyncOnce
+	syncCron := opt.SyncCron
 
 	if chunkSize <= 0 {
 		return nil, errInvalidChunkSize
@@ -47,7 +49,8 @@ func NewMinIOPushClientSync(opt Option) (Sync, error) {
 
 	s.driver = minio.NewMinIODriver(s.endpoint, s.bucketName, s.secure, s.currentUser.UserName(), s.currentUser.Password(), true, r, maxTranRate, logger)
 
-	err = s.start()
+	isSync := syncOnce || len(syncCron) > 0
+	err = s.start(isSync)
 	if err != nil {
 		return nil, err
 	}

sync/option.go

@@ -39,6 +39,8 @@ type Option struct {
 	Reporter              report.Reporter
 	TaskConf              string
 	Logger                *logger.Logger
+	SyncOnce              bool
+	SyncCron              string
 }
 
 // NewSyncOption create an instance of the Option, store all the sync component options
@@ -70,6 +72,8 @@ func NewSyncOption(config conf.Config, users []*auth.User, r retry.Retry, pi ign
 		Reporter:              reporter,
 		TaskConf:              config.TaskConf,
 		Logger:                logger,
+		SyncOnce:              config.SyncOnce,
+		SyncCron:              config.SyncCron,
 	}
 	return opt
 }

sync/sftp_push_client_sync.go

@@ -18,6 +18,8 @@ func NewSftpPushClientSync(opt Option) (Sync, error) {
 	maxTranRate := opt.MaxTranRate
 	r := opt.Retry
 	logger := opt.Logger
+	syncOnce := opt.SyncOnce
+	syncCron := opt.SyncCron
 
 	if chunkSize <= 0 {
 		return nil, errInvalidChunkSize
@@ -35,7 +37,8 @@ func NewSftpPushClientSync(opt Option) (Sync, error) {
 
 	s.driver = sftp.NewSFTPDriver(s.remoteAddr, dest.SSHConfig(), true, r, maxTranRate, logger)
 
-	err = s.start()
+	isSync := syncOnce || len(syncCron) > 0
+	err = s.start(isSync)
 	if err != nil {
 		return nil, err
 	}