Kopia Pod Volume Backup/Restore

changelogs/unreleased/5259-lyndon

@@ -0,0 +1 @@
+Fill gaps for Kopia path of PVBR: integrate Repo Manager with Unified Repo; pass UploaderType to PVBR backupper and restorer; pass RepositoryType to BackupRepository controller and Repo Ensurer

pkg/apis/velero/v1/backup_repository_types.go

@@ -52,8 +52,8 @@ const (
 	BackupRepositoryPhaseReady    BackupRepositoryPhase = "Ready"
 	BackupRepositoryPhaseNotReady BackupRepositoryPhase = "NotReady"
 
-	BackupRepositoryTypeRestic  string = "restic"
-	BackupRepositoryTypeUnified string = "unified"
+	BackupRepositoryTypeRestic string = "restic"
+	BackupRepositoryTypeKopia  string = "kopia"
 )
 
 // BackupRepositoryStatus is the current status of a BackupRepository.

pkg/apis/velero/v1/labels_annotations.go

@@ -40,16 +40,19 @@ const (
 
 	// PodVolumeOperationTimeoutAnnotation is the annotation key used to apply
 	// a backup/restore-specific timeout value for pod volume operations (i.e.
-	// restic backups/restores).
+	// pod volume backups/restores).
 	PodVolumeOperationTimeoutAnnotation = "velero.io/pod-volume-timeout"
 
 	// StorageLocationLabel is the label key used to identify the storage
 	// location of a backup.
 	StorageLocationLabel = "velero.io/storage-location"
 
-	// ResticVolumeNamespaceLabel is the label key used to identify which
-	// namespace a restic repository stores pod volume backups for.
-	ResticVolumeNamespaceLabel = "velero.io/volume-namespace"
+	// VolumeNamespaceLabel is the label key used to identify which
+	// namespace a repository stores backups for.
+	VolumeNamespaceLabel = "velero.io/volume-namespace"
+
+	// RepositoryTypeLabel is the label key used to identify the type of a repository
+	RepositoryTypeLabel = "velero.io/repository-type"
 
 	// SourceClusterK8sVersionAnnotation is the label key used to identify the k8s
 	// git version of the backup , i.e. v1.16.4

pkg/backup/backup.go

@@ -78,6 +78,7 @@ type kubernetesBackupper struct {
 	resticTimeout          time.Duration
 	defaultVolumesToRestic bool
 	clientPageSize         int
+	uploaderType           string
 }
 
 func (i *itemKey) String() string {
@@ -104,6 +105,7 @@ func NewKubernetesBackupper(
 	resticTimeout time.Duration,
 	defaultVolumesToRestic bool,
 	clientPageSize int,
+	uploaderType string,
 ) (Backupper, error) {
 	return &kubernetesBackupper{
 		backupClient:           backupClient,
@@ -114,6 +116,7 @@ func NewKubernetesBackupper(
 		resticTimeout:          resticTimeout,
 		defaultVolumesToRestic: defaultVolumesToRestic,
 		clientPageSize:         clientPageSize,
+		uploaderType:           uploaderType,
 	}, nil
 }
 
@@ -236,7 +239,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(log logrus.FieldLogger,
 
 	var resticBackupper podvolume.Backupper
 	if kb.resticBackupperFactory != nil {
-		resticBackupper, err = kb.resticBackupperFactory.NewBackupper(ctx, backupRequest.Backup)
+		resticBackupper, err = kb.resticBackupperFactory.NewBackupper(ctx, backupRequest.Backup, kb.uploaderType)
 		if err != nil {
 			return errors.WithStack(err)
 		}

pkg/backup/backup_test.go

@@ -2595,7 +2595,7 @@ func TestBackupWithHooks(t *testing.T) {
 
 type fakeResticBackupperFactory struct{}
 
-func (f *fakeResticBackupperFactory) NewBackupper(context.Context, *velerov1.Backup) (podvolume.Backupper, error) {
+func (f *fakeResticBackupperFactory) NewBackupper(context.Context, *velerov1.Backup, string) (podvolume.Backupper, error) {
 	return &fakeResticBackupper{}, nil
 }
 

pkg/cmd/server/server.go

@@ -131,7 +131,7 @@ type serverConfig struct {
 	clientPageSize                                                          int
 	profilerAddress                                                         string
 	formatFlag                                                              *logging.FormatFlag
-	defaultResticMaintenanceFrequency                                       time.Duration
+	repoMaintenanceFrequency                                                time.Duration
 	garbageCollectionFrequency                                              time.Duration
 	defaultVolumesToRestic                                                  bool
 	uploaderType                                                            string
@@ -147,25 +147,24 @@ func NewCommand(f client.Factory) *cobra.Command {
 		volumeSnapshotLocations = flag.NewMap().WithKeyValueDelimiter(':')
 		logLevelFlag            = logging.LogLevelFlag(logrus.InfoLevel)
 		config                  = serverConfig{
-			pluginDir:                         "/plugins",
-			metricsAddress:                    defaultMetricsAddress,
-			defaultBackupLocation:             "default",
-			defaultVolumeSnapshotLocations:    make(map[string]string),
-			backupSyncPeriod:                  defaultBackupSyncPeriod,
-			defaultBackupTTL:                  defaultBackupTTL,
-			defaultCSISnapshotTimeout:         defaultCSISnapshotTimeout,
-			storeValidationFrequency:          defaultStoreValidationFrequency,
-			podVolumeOperationTimeout:         defaultPodVolumeOperationTimeout,
-			restoreResourcePriorities:         defaultRestorePriorities,
-			clientQPS:                         defaultClientQPS,
-			clientBurst:                       defaultClientBurst,
-			clientPageSize:                    defaultClientPageSize,
-			profilerAddress:                   defaultProfilerAddress,
-			resourceTerminatingTimeout:        defaultResourceTerminatingTimeout,
-			formatFlag:                        logging.NewFormatFlag(),
-			defaultResticMaintenanceFrequency: restic.DefaultMaintenanceFrequency,
-			defaultVolumesToRestic:            restic.DefaultVolumesToRestic,
-			uploaderType:                      uploader.ResticType,
+			pluginDir:                      "/plugins",
+			metricsAddress:                 defaultMetricsAddress,
+			defaultBackupLocation:          "default",
+			defaultVolumeSnapshotLocations: make(map[string]string),
+			backupSyncPeriod:               defaultBackupSyncPeriod,
+			defaultBackupTTL:               defaultBackupTTL,
+			defaultCSISnapshotTimeout:      defaultCSISnapshotTimeout,
+			storeValidationFrequency:       defaultStoreValidationFrequency,
+			podVolumeOperationTimeout:      defaultPodVolumeOperationTimeout,
+			restoreResourcePriorities:      defaultRestorePriorities,
+			clientQPS:                      defaultClientQPS,
+			clientBurst:                    defaultClientBurst,
+			clientPageSize:                 defaultClientPageSize,
+			profilerAddress:                defaultProfilerAddress,
+			resourceTerminatingTimeout:     defaultResourceTerminatingTimeout,
+			formatFlag:                     logging.NewFormatFlag(),
+			defaultVolumesToRestic:         restic.DefaultVolumesToRestic,
+			uploaderType:                   uploader.ResticType,
 		}
 	)
 
@@ -228,7 +227,7 @@ func NewCommand(f client.Factory) *cobra.Command {
 	command.Flags().StringVar(&config.profilerAddress, "profiler-address", config.profilerAddress, "The address to expose the pprof profiler.")
 	command.Flags().DurationVar(&config.resourceTerminatingTimeout, "terminating-resource-timeout", config.resourceTerminatingTimeout, "How long to wait on persistent volumes and namespaces to terminate during a restore before timing out.")
 	command.Flags().DurationVar(&config.defaultBackupTTL, "default-backup-ttl", config.defaultBackupTTL, "How long to wait by default before backups can be garbage collected.")
-	command.Flags().DurationVar(&config.defaultResticMaintenanceFrequency, "default-restic-prune-frequency", config.defaultResticMaintenanceFrequency, "How often 'restic prune' is run for restic repositories by default.")
+	command.Flags().DurationVar(&config.repoMaintenanceFrequency, "default-restic-prune-frequency", config.repoMaintenanceFrequency, "How often 'prune' is run for backup repositories by default.")
 	command.Flags().DurationVar(&config.garbageCollectionFrequency, "garbage-collection-frequency", config.garbageCollectionFrequency, "How often garbage collection is run for expired backups.")
 	command.Flags().BoolVar(&config.defaultVolumesToRestic, "default-volumes-to-restic", config.defaultVolumesToRestic, "Backup all volumes with restic by default.")
 	command.Flags().StringVar(&config.uploaderType, "uploader-type", config.uploaderType, "Type of uploader to handle the transfer of data of pod volumes")
@@ -260,6 +259,7 @@ type server struct {
 	config                              serverConfig
 	mgr                                 manager.Manager
 	credentialFileStore                 credentials.FileStore
+	credentialSecretStore               credentials.SecretStore
 }
 
 func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*server, error) {
@@ -349,6 +349,8 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 		return nil, err
 	}
 
+	credentialSecretStore, err := credentials.NewNamespacedSecretStore(mgr.GetClient(), f.Namespace())
+
 	s := &server{
 		namespace:                           f.Namespace(),
 		metricsAddress:                      config.metricsAddress,
@@ -368,6 +370,7 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
 		config:                              config,
 		mgr:                                 mgr,
 		credentialFileStore:                 credentialFileStore,
+		credentialSecretStore:               credentialSecretStore,
 	}
 
 	return s, nil
@@ -546,7 +549,7 @@ func (s *server) initRestic() error {
 	s.repoLocker = repository.NewRepoLocker()
 	s.repoEnsurer = repository.NewRepositoryEnsurer(s.sharedInformerFactory.Velero().V1().BackupRepositories(), s.veleroClient.VeleroV1(), s.logger)
 
-	s.repoManager = repository.NewManager(s.namespace, s.mgr.GetClient(), s.repoLocker, s.repoEnsurer, s.credentialFileStore, s.logger)
+	s.repoManager = repository.NewManager(s.namespace, s.mgr.GetClient(), s.repoLocker, s.repoEnsurer, s.credentialFileStore, s.credentialSecretStore, s.logger)
 
 	return nil
 }
@@ -620,6 +623,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.config.podVolumeOperationTimeout,
 			s.config.defaultVolumesToRestic,
 			s.config.clientPageSize,
+			s.config.uploaderType,
 		)
 		cmd.CheckError(err)
 
@@ -781,7 +785,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 	}
 
 	if _, ok := enabledRuntimeControllers[controller.ResticRepo]; ok {
-		if err := controller.NewResticRepoReconciler(s.namespace, s.logger, s.mgr.GetClient(), s.config.defaultResticMaintenanceFrequency, s.repoManager).SetupWithManager(s.mgr); err != nil {
+		if err := controller.NewResticRepoReconciler(s.namespace, s.logger, s.mgr.GetClient(), s.config.repoMaintenanceFrequency, s.repoManager).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", controller.ResticRepo)
 		}
 	}

pkg/controller/backup_deletion_controller.go

@@ -45,6 +45,8 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/vmware-tanzu/velero/pkg/podvolume"
 )
 
 const (
@@ -506,17 +508,5 @@ func getSnapshotsInBackup(ctx context.Context, backup *velerov1api.Backup, kbCli
 		return nil, errors.WithStack(err)
 	}
 
-	var res []repository.SnapshotIdentifier
-	for _, item := range podVolumeBackups.Items {
-		if item.Status.SnapshotID == "" {
-			continue
-		}
-		res = append(res, repository.SnapshotIdentifier{
-			VolumeNamespace:       item.Spec.Pod.Namespace,
-			BackupStorageLocation: backup.Spec.StorageLocation,
-			SnapshotID:            item.Status.SnapshotID,
-		})
-	}
-
-	return res, nil
+	return podvolume.GetSnapshotIdentifier(podVolumeBackups), nil
 }

pkg/controller/backup_deletion_controller_test.go

@@ -771,10 +771,12 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 				{
 					VolumeNamespace: "ns-1",
 					SnapshotID:      "snap-3",
+					RepositoryType:  "restic",
 				},
 				{
 					VolumeNamespace: "ns-1",
 					SnapshotID:      "snap-4",
+					RepositoryType:  "restic",
 				},
 			},
 		},
@@ -822,6 +824,7 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 				{
 					VolumeNamespace: "ns-1",
 					SnapshotID:      "snap-3",
+					RepositoryType:  "restic",
 				},
 			},
 		},

pkg/controller/restic_repository_controller.go

@@ -32,39 +32,34 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	repoconfig "github.com/vmware-tanzu/velero/pkg/repository/config"
-	"github.com/vmware-tanzu/velero/pkg/restic"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
 const (
-	repoSyncPeriod = 5 * time.Minute
+	repoSyncPeriod           = 5 * time.Minute
+	defaultMaintainFrequency = 7 * 24 * time.Hour
 )
 
 type ResticRepoReconciler struct {
 	client.Client
-	namespace                   string
-	logger                      logrus.FieldLogger
-	clock                       clock.Clock
-	defaultMaintenanceFrequency time.Duration
-	repositoryManager           repository.Manager
+	namespace            string
+	logger               logrus.FieldLogger
+	clock                clock.Clock
+	maintenanceFrequency time.Duration
+	repositoryManager    repository.Manager
 }
 
 func NewResticRepoReconciler(namespace string, logger logrus.FieldLogger, client client.Client,
-	defaultMaintenanceFrequency time.Duration, repositoryManager repository.Manager) *ResticRepoReconciler {
+	maintenanceFrequency time.Duration, repositoryManager repository.Manager) *ResticRepoReconciler {
 	c := &ResticRepoReconciler{
 		client,
 		namespace,
 		logger,
 		clock.RealClock{},
-		defaultMaintenanceFrequency,
+		maintenanceFrequency,
 		repositoryManager,
 	}
 
-	if c.defaultMaintenanceFrequency <= 0 {
-		logger.Infof("Invalid default restic maintenance frequency, setting to %v", restic.DefaultMaintenanceFrequency)
-		c.defaultMaintenanceFrequency = restic.DefaultMaintenanceFrequency
-	}
-
 	return c
 }
 
@@ -135,7 +130,7 @@ func (r *ResticRepoReconciler) initializeRepo(ctx context.Context, req *velerov1
 			rr.Status.Phase = velerov1api.BackupRepositoryPhaseNotReady
 
 			if rr.Spec.MaintenanceFrequency.Duration <= 0 {
-				rr.Spec.MaintenanceFrequency = metav1.Duration{Duration: r.defaultMaintenanceFrequency}
+				rr.Spec.MaintenanceFrequency = metav1.Duration{Duration: r.getRepositoryMaintenanceFrequency(req)}
 			}
 		})
 	}
@@ -145,7 +140,7 @@ func (r *ResticRepoReconciler) initializeRepo(ctx context.Context, req *velerov1
 		rr.Spec.ResticIdentifier = repoIdentifier
 
 		if rr.Spec.MaintenanceFrequency.Duration <= 0 {
-			rr.Spec.MaintenanceFrequency = metav1.Duration{Duration: r.defaultMaintenanceFrequency}
+			rr.Spec.MaintenanceFrequency = metav1.Duration{Duration: r.getRepositoryMaintenanceFrequency(req)}
 		}
 	}); err != nil {
 		return err
@@ -161,6 +156,23 @@ func (r *ResticRepoReconciler) initializeRepo(ctx context.Context, req *velerov1
 	})
 }
 
+func (r *ResticRepoReconciler) getRepositoryMaintenanceFrequency(req *velerov1api.BackupRepository) time.Duration {
+	if r.maintenanceFrequency > 0 {
+		r.logger.WithField("frequency", r.maintenanceFrequency).Info("Set user defined maintenance frequency")
+		return r.maintenanceFrequency
+	} else {
+		frequency, err := r.repositoryManager.DefaultMaintenanceFrequency(req)
+		if err != nil || frequency <= 0 {
+			r.logger.WithError(err).WithField("returned frequency", frequency).Warn("Failed to get maitanance frequency, use the default one")
+			frequency = defaultMaintainFrequency
+		} else {
+			r.logger.WithField("frequency", frequency).Info("Set matainenance according to repository suggestion")
+		}
+
+		return frequency
+	}
+}
+
 // ensureRepo checks to see if a repository exists, and attempts to initialize it if
 // it does not exist. An error is returned if the repository can't be connected to
 // or initialized.