feat: Add immediate option to ReplicationSource

api/v1alpha1/replicationsource_types.go

@@ -283,6 +283,9 @@ type ReplicationSourceSpec struct {
 	// paused can be used to temporarily stop replication. Defaults to "false".
 	//+optional
 	Paused bool `json:"paused,omitempty"`
+	// immediate can be used to stop replication when a new ReplicationSource is created. Defaults to "true".
+	//+optional
+	Immediate *bool `json:"immediate,omitempty"`
 }
 
 type ReplicationSourceRsyncStatus struct {

bundle/manifests/volsync.backube_replicationsources.yaml

@@ -66,6 +66,10 @@ spec:
                       provider. The name should be of the form: domain.com/provider.'
                     type: string
                 type: object
+              immediate:
+                description: immediate can be used to stop replication when a new
+                  ReplicationSource is created. Defaults to "true".
+                type: boolean
               paused:
                 description: paused can be used to temporarily stop replication. Defaults
                   to "false".

bundle/manifests/volsync.clusterserviceversion.yaml

@@ -55,7 +55,7 @@ metadata:
         }
       ]
     capabilities: Basic Install
-    createdAt: "2023-07-13T13:28:33Z"
+    createdAt: "2023-07-27T21:09:50Z"
     olm.skipRange: '>=0.4.0 <0.8.0'
     operators.operatorframework.io/builder: operator-sdk-v1.26.0
     operators.operatorframework.io/project_layout: go.kubebuilder.io/v3

config/crd/bases/volsync.backube_replicationsources.yaml

@@ -67,6 +67,10 @@ spec:
                       provider. The name should be of the form: domain.com/provider.'
                     type: string
                 type: object
+              immediate:
+                description: immediate can be used to stop replication when a new
+                  ReplicationSource is created. Defaults to "true".
+                type: boolean
               paused:
                 description: paused can be used to temporarily stop replication. Defaults
                   to "false".

controllers/mover/rclone/builder.go

@@ -87,6 +87,7 @@
 	return rb.viper.GetString(rcloneContainerImageFlag)
 }
 
+// nolint: funlen
 func (rb *Builder) FromSource(client client.Client, logger logr.Logger,
 	eventRecorder events.EventRecorder,
 	source *volsyncv1alpha1.ReplicationSource, privileged bool) (mover.Mover, error) {
@@ -114,6 +115,11 @@
 	saHandler := utils.NewSAHandler(client, source, isSource, privileged,
 		source.Spec.Rclone.MoverServiceAccount)
 
+	isImmediate := true
+	if source.Spec.Immediate != nil {
+		isImmediate = *source.Spec.Immediate
+	}
+
 	return &Mover{
 		client:               client,
 		logger:               logger.WithValues("method", "Rclone"),
@@ -132,6 +138,7 @@
 		privileged:           privileged,
 		moverSecurityContext: source.Spec.Rclone.MoverSecurityContext,
 		latestMoverStatus:    source.Status.LatestMoverStatus,
+		isImmediate:          isImmediate,
 	}, nil
 }
 

controllers/mover/rclone/mover.go

@@ -65,6 +65,7 @@ type Mover struct {
 	privileged           bool // true if the mover should have elevated privileges
 	moverSecurityContext *corev1.PodSecurityContext
 	latestMoverStatus    *volsyncv1alpha1.MoverStatus
+	isImmediate          bool
 }
 
 var _ mover.Mover = &Mover{}
@@ -231,6 +232,9 @@ func (m *Mover) ensureJob(ctx context.Context, dataPVC *corev1.PersistentVolumeC
 		if m.paused {
 			parallelism = int32(0)
 		}
+		if !m.isImmediate && m.latestMoverStatus.Result == "" {
+			parallelism = int32(0)
+		}
 		job.Spec.Parallelism = &parallelism
 
 		envVars := []corev1.EnvVar{

controllers/mover/restic/builder.go

@@ -87,6 +87,7 @@
 	return rb.viper.GetString(resticContainerImageFlag)
 }
 
+// nolint: funlen
 func (rb *Builder) FromSource(client client.Client, logger logr.Logger,
 	eventRecorder events.EventRecorder,
 	source *volsyncv1alpha1.ReplicationSource, privileged bool) (mover.Mover, error) {
@@ -119,6 +120,11 @@
 	saHandler := utils.NewSAHandler(client, source, isSource, privileged,
 		source.Spec.Restic.MoverServiceAccount)
 
+	isImmediate := true
+	if source.Spec.Immediate != nil {
+		isImmediate = *source.Spec.Immediate
+	}
+
 	return &Mover{
 		client:                client,
 		logger:                logger.WithValues("method", "Restic"),
@@ -142,6 +148,7 @@
 		unlock:                source.Spec.Restic.Unlock,
 		sourceStatus:          source.Status.Restic,
 		latestMoverStatus:     source.Status.LatestMoverStatus,
+		isImmediate:           isImmediate,
 	}, nil
 }
 

controllers/mover/restic/mover.go

@@ -79,6 +79,7 @@ type Mover struct {
 	unlock        string
 	retainPolicy  *volsyncv1alpha1.ResticRetainPolicy
 	sourceStatus  *volsyncv1alpha1.ReplicationSourceResticStatus
+	isImmediate   bool
 	// Destination-only fields
 	previous    *int32
 	restoreAsOf *string
@@ -293,6 +294,9 @@ func (m *Mover) ensureJob(ctx context.Context, cachePVC *corev1.PersistentVolume
 		if m.paused {
 			parallelism = int32(0)
 		}
+		if !m.isImmediate && m.latestMoverStatus.Result == "" {
+			parallelism = int32(0)
+		}
 		job.Spec.Parallelism = &parallelism
 		forgetOptions := generateForgetOptions(m.retainPolicy)
 		// set default values

controllers/mover/rsync/builder.go

@@ -87,6 +87,7 @@
 	return rb.viper.GetString(rsyncContainerImageFlag)
 }
 
+// nolint: funlen
 func (rb *Builder) FromSource(client client.Client, logger logr.Logger,
 	eventRecorder events.EventRecorder,
 	source *volsyncv1alpha1.ReplicationSource, privileged bool) (mover.Mover, error) {
@@ -119,6 +120,11 @@
 	saHandler := utils.NewSAHandler(client, source, isSource, true, /*Rsync runs privileged only*/
 		source.Spec.Rsync.MoverServiceAccount)
 
+	isImmediate := true
+	if source.Spec.Immediate != nil {
+		isImmediate = *source.Spec.Immediate
+	}
+
 	return &Mover{
 		client:             client,
 		logger:             logger.WithValues("method", "Rsync"),
@@ -137,6 +143,7 @@
 		mainPVCName:        &source.Spec.SourcePVC,
 		sourceStatus:       source.Status.Rsync,
 		latestMoverStatus:  source.Status.LatestMoverStatus,
+		isImmediate:        isImmediate,
 	}, nil
 }
 

controllers/mover/rsync/mover.go

@@ -65,6 +65,7 @@ type Mover struct {
 	sourceStatus       *volsyncv1alpha1.ReplicationSourceRsyncStatus
 	destStatus         *volsyncv1alpha1.ReplicationDestinationRsyncStatus
 	latestMoverStatus  *volsyncv1alpha1.MoverStatus
+	isImmediate        bool
 }
 
 var _ mover.Mover = &Mover{}
@@ -356,6 +357,9 @@ func (m *Mover) ensureJob(ctx context.Context, dataPVC *corev1.PersistentVolumeC
 		if m.paused {
 			parallelism = int32(0)
 		}
+		if !m.isImmediate && m.latestMoverStatus.Result == "" {
+			parallelism = int32(0)
+		}
 		job.Spec.Parallelism = &parallelism
 
 		readOnlyVolume := false

controllers/mover/rsynctls/builder.go

@@ -87,6 +87,7 @@
 	return rb.viper.GetString(rsyncTLSContainerImageFlag)
 }
 
+// nolint: funlen
 func (rb *Builder) FromSource(client client.Client, logger logr.Logger,
 	eventRecorder events.EventRecorder,
 	source *volsyncv1alpha1.ReplicationSource, privileged bool) (mover.Mover, error) {
@@ -119,6 +120,11 @@
 	saHandler := utils.NewSAHandler(client, source, isSource, privileged,
 		source.Spec.RsyncTLS.MoverServiceAccount)
 
+	isImmediate := true
+	if source.Spec.Immediate != nil {
+		isImmediate = *source.Spec.Immediate
+	}
+
 	return &Mover{
 		client:               client,
 		logger:               logger.WithValues("method", "RsyncTLS"),
@@ -139,6 +145,7 @@
 		moverSecurityContext: source.Spec.RsyncTLS.MoverSecurityContext,
 		sourceStatus:         source.Status.RsyncTLS,
 		latestMoverStatus:    source.Status.LatestMoverStatus,
+		isImmediate:          isImmediate,
 	}, nil
 }
 

controllers/mover/rsynctls/mover.go

@@ -70,6 +70,7 @@ type Mover struct {
 	sourceStatus         *volsyncv1alpha1.ReplicationSourceRsyncTLSStatus
 	destStatus           *volsyncv1alpha1.ReplicationDestinationRsyncTLSStatus
 	latestMoverStatus    *volsyncv1alpha1.MoverStatus
+	isImmediate          bool
 }
 
 var _ mover.Mover = &Mover{}
@@ -363,6 +364,9 @@ func (m *Mover) ensureJob(ctx context.Context, dataPVC *corev1.PersistentVolumeC
 		if m.paused {
 			parallelism = int32(0)
 		}
+		if !m.isImmediate && m.latestMoverStatus.Result == "" {
+			parallelism = int32(0)
+		}
 		job.Spec.Parallelism = &parallelism
 
 		readOnlyVolume := false

controllers/mover/syncthing/builder.go

@@ -135,6 +135,7 @@ func (rb *Builder) FromSource(client client.Client, logger logr.Logger,
 		apiConfig:            api.APIConfig{},
 		privileged:           privileged,
 		moverSecurityContext: source.Spec.Syncthing.MoverSecurityContext,
+		isImmediate:          true,
 		// defer setting the VolumeHandler
 	}, nil
 }

controllers/mover/syncthing/mover.go

@@ -119,6 +119,7 @@ type Mover struct {
 	apiConfig            api.APIConfig
 	privileged           bool
 	moverSecurityContext *corev1.PodSecurityContext
+	isImmediate          bool
 }
 
 var _ mover.Mover = &Mover{}

helm/volsync/templates/volsync.backube_replicationsources.yaml

@@ -60,6 +60,9 @@ spec:
                       description: 'provider is the name of the external replication provider. The name should be of the form: domain.com/provider.'
                       type: string
                   type: object
+                immediate:
+                  description: immediate can be used to stop replication when a new ReplicationSource is created. Defaults to "true".
+                  type: boolean
                 paused:
                   description: paused can be used to temporarily stop replication. Defaults to "false".
                   type: boolean