Merge branch 'main' of https://github.com/qiuming-best/velero into support-restore-sparse

pkg/cmd/cli/install/install.go

@@ -73,6 +73,7 @@ type Options struct {
 	UseVolumeSnapshots              bool
 	DefaultRepoMaintenanceFrequency time.Duration
 	GarbageCollectionFrequency      time.Duration
+	PodVolumeOperationTimeout       time.Duration
 	Plugins                         flag.StringArray
 	NoDefaultBackupLocation         bool
 	CRDsOnly                        bool
@@ -116,6 +117,7 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
 	flags.BoolVar(&o.Wait, "wait", o.Wait, "Wait for Velero deployment to be ready. Optional.")
 	flags.DurationVar(&o.DefaultRepoMaintenanceFrequency, "default-repo-maintain-frequency", o.DefaultRepoMaintenanceFrequency, "How often 'maintain' is run for backup repositories by default. Optional.")
 	flags.DurationVar(&o.GarbageCollectionFrequency, "garbage-collection-frequency", o.GarbageCollectionFrequency, "How often the garbage collection runs for expired backups.(default 1h)")
+	flags.DurationVar(&o.PodVolumeOperationTimeout, "pod-volume-operation-timeout", o.PodVolumeOperationTimeout, "How long to wait for pod volume operations to complete before timing out(default 4h). Optional.")
 	flags.Var(&o.Plugins, "plugins", "Plugin container images to install into the Velero Deployment")
 	flags.BoolVar(&o.CRDsOnly, "crds-only", o.CRDsOnly, "Only generate CustomResourceDefinition resources. Useful for updating CRDs for an existing Velero install.")
 	flags.StringVar(&o.CACertFile, "cacert", o.CACertFile, "File containing a certificate bundle to use when verifying TLS connections to the object store. Optional.")
@@ -209,6 +211,7 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
 		VSLConfig:                       o.VolumeSnapshotConfig.Data(),
 		DefaultRepoMaintenanceFrequency: o.DefaultRepoMaintenanceFrequency,
 		GarbageCollectionFrequency:      o.GarbageCollectionFrequency,
+		PodVolumeOperationTimeout:       o.PodVolumeOperationTimeout,
 		Plugins:                         o.Plugins,
 		NoDefaultBackupLocation:         o.NoDefaultBackupLocation,
 		CACertData:                      caCertData,
@@ -426,5 +429,9 @@ func (o *Options) Validate(c *cobra.Command, args []string, f client.Factory) er
 		return errors.New("--garbage-collection-frequency must be non-negative")
 	}
 
+	if o.PodVolumeOperationTimeout < 0 {
+		return errors.New("--pod-volume-operation-timeout must be non-negative")
+	}
+
 	return nil
 }

pkg/cmd/cli/nodeagent/server.go

@@ -194,7 +194,6 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
 		Scheme:   scheme,
 		NewCache: cache.BuilderWithOptions(cacheOption),
 	})
-
 	if err != nil {
 		cancelFunc()
 		return nil, err
@@ -286,13 +285,13 @@ func (s *nodeAgentServer) run() {
 	}
 
 	dataUploadReconciler := controller.NewDataUploadReconciler(s.mgr.GetClient(), s.kubeClient, s.csiSnapshotClient.SnapshotV1(), s.dataPathMgr, repoEnsurer, clock.RealClock{}, credentialGetter, s.nodeName, s.fileSystem, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
-	s.markDataUploadsCancel(dataUploadReconciler)
+	s.attemptDataUploadResume(dataUploadReconciler)
 	if err = dataUploadReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the data upload controller")
 	}
 
 	dataDownloadReconciler := controller.NewDataDownloadReconciler(s.mgr.GetClient(), s.kubeClient, s.dataPathMgr, repoEnsurer, credentialGetter, s.nodeName, s.config.dataMoverPrepareTimeout, s.logger, s.metrics)
-	s.markDataDownloadsCancel(dataDownloadReconciler)
+	s.attemptDataDownloadResume(dataDownloadReconciler)
 	if err = dataDownloadReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the data download controller")
 	}
@@ -366,71 +365,34 @@ func (s *nodeAgentServer) markInProgressCRsFailed() {
 	s.markInProgressPVRsFailed(client)
 }
 
-func (s *nodeAgentServer) markDataUploadsCancel(r *controller.DataUploadReconciler) {
+func (s *nodeAgentServer) attemptDataUploadResume(r *controller.DataUploadReconciler) {
 	// the function is called before starting the controller manager, the embedded client isn't ready to use, so create a new one here
 	client, err := ctrlclient.New(s.mgr.GetConfig(), ctrlclient.Options{Scheme: s.mgr.GetScheme()})
 	if err != nil {
 		s.logger.WithError(errors.WithStack(err)).Error("failed to create client")
 		return
 	}
-	if dataUploads, err := r.FindDataUploads(s.ctx, client, s.namespace); err != nil {
-		s.logger.WithError(errors.WithStack(err)).Error("failed to find data uploads")
-	} else {
-		for i := range dataUploads {
-			du := dataUploads[i]
-			if du.Status.Phase == velerov2alpha1api.DataUploadPhaseAccepted ||
-				du.Status.Phase == velerov2alpha1api.DataUploadPhasePrepared ||
-				du.Status.Phase == velerov2alpha1api.DataUploadPhaseInProgress {
-				err = controller.UpdateDataUploadWithRetry(s.ctx, client, types.NamespacedName{Namespace: du.Namespace, Name: du.Name}, s.logger.WithField("dataupload", du.Name),
-					func(dataUpload *velerov2alpha1api.DataUpload) {
-						dataUpload.Spec.Cancel = true
-						dataUpload.Status.Message = fmt.Sprintf("found a dataupload with status %q during the node-agent starting, mark it as cancel", du.Status.Phase)
-					})
-
-				if err != nil {
-					s.logger.WithError(errors.WithStack(err)).Errorf("failed to mark dataupload %q cancel", du.GetName())
-					continue
-				}
-				s.logger.WithField("dataupload", du.GetName()).Warn(du.Status.Message)
-			}
-		}
+	if err := r.AttemptDataUploadResume(s.ctx, client, s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
+		s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data upload resume")
 	}
 }
 
-func (s *nodeAgentServer) markDataDownloadsCancel(r *controller.DataDownloadReconciler) {
+func (s *nodeAgentServer) attemptDataDownloadResume(r *controller.DataDownloadReconciler) {
 	// the function is called before starting the controller manager, the embedded client isn't ready to use, so create a new one here
 	client, err := ctrlclient.New(s.mgr.GetConfig(), ctrlclient.Options{Scheme: s.mgr.GetScheme()})
 	if err != nil {
 		s.logger.WithError(errors.WithStack(err)).Error("failed to create client")
 		return
 	}
-	if dataDownloads, err := r.FindDataDownloads(s.ctx, client, s.namespace); err != nil {
-		s.logger.WithError(errors.WithStack(err)).Error("failed to find data downloads")
-	} else {
-		for i := range dataDownloads {
-			dd := dataDownloads[i]
-			if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted ||
-				dd.Status.Phase == velerov2alpha1api.DataDownloadPhasePrepared ||
-				dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
-				err = controller.UpdateDataDownloadWithRetry(s.ctx, client, types.NamespacedName{Namespace: dd.Namespace, Name: dd.Name}, s.logger.WithField("datadownload", dd.Name),
-					func(dataDownload *velerov2alpha1api.DataDownload) {
-						dataDownload.Spec.Cancel = true
-						dataDownload.Status.Message = fmt.Sprintf("found a datadownload with status %q during the node-agent starting, mark it as cancel", dd.Status.Phase)
-					})
 
-				if err != nil {
-					s.logger.WithError(errors.WithStack(err)).Errorf("failed to mark datadownload %q cancel", dd.GetName())
-					continue
-				}
-				s.logger.WithField("datadownload", dd.GetName()).Warn(dd.Status.Message)
-			}
-		}
+	if err := r.AttemptDataDownloadResume(s.ctx, client, s.logger.WithField("node", s.nodeName), s.namespace); err != nil {
+		s.logger.WithError(errors.WithStack(err)).Error("failed to attempt data download resume")
 	}
 }
 
 func (s *nodeAgentServer) markInProgressPVBsFailed(client ctrlclient.Client) {
 	pvbs := &velerov1api.PodVolumeBackupList{}
-	if err := client.List(s.ctx, pvbs, &ctrlclient.MatchingFields{"metadata.namespace": s.namespace}); err != nil {
+	if err := client.List(s.ctx, pvbs, &ctrlclient.ListOptions{Namespace: s.namespace}); err != nil {
 		s.logger.WithError(errors.WithStack(err)).Error("failed to list podvolumebackups")
 		return
 	}
@@ -456,7 +418,7 @@ func (s *nodeAgentServer) markInProgressPVBsFailed(client ctrlclient.Client) {
 
 func (s *nodeAgentServer) markInProgressPVRsFailed(client ctrlclient.Client) {
 	pvrs := &velerov1api.PodVolumeRestoreList{}
-	if err := client.List(s.ctx, pvrs, &ctrlclient.MatchingFields{"metadata.namespace": s.namespace}); err != nil {
+	if err := client.List(s.ctx, pvrs, &ctrlclient.ListOptions{Namespace: s.namespace}); err != nil {
 		s.logger.WithError(errors.WithStack(err)).Error("failed to list podvolumerestores")
 		return
 	}

pkg/cmd/cli/restore/create.go

@@ -149,7 +149,7 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 
 	flags.StringVar(&o.ResourceModifierConfigMap, "resource-modifier-configmap", "", "Reference to the resource modifier configmap that restore will use")
 
-	f = flags.VarPF(&o.WriteSparseFiles, "write-sparse-files", "", "Whether to write sparse files when restoring volumes using restic or kopia")
+	f = flags.VarPF(&o.WriteSparseFiles, "write-sparse-files", "", "Whether to write sparse files when restoring volumes")
 	f.NoOptDefVal = cmd.FALSE
 }
 
@@ -323,7 +323,7 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 			ItemOperationTimeout: metav1.Duration{
 				Duration: o.ItemOperationTimeout,
 			},
-			RestoreConfig: &api.RestoreConfig{
+			UploaderConfigForRestore: &api.UploaderConfigForRestore{
 				WriteSparseFiles: boolptr.IsSetToTrue(o.WriteSparseFiles.Value),
 			},
 		},