mirror of
https://github.com/vmware-tanzu/velero.git
synced 2026-09-05 15:47:15 +00:00
update sync controller for backup locations
Signed-off-by: Steve Kriss <steve@heptio.com>
This commit is contained in:
Steve Kriss
@@ -17,7 +17,6 @@ limitations under the License.
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
@@ -27,133 +26,172 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
|
||||
api "github.com/heptio/ark/pkg/apis/ark/v1"
|
||||
arkv1api "github.com/heptio/ark/pkg/apis/ark/v1"
|
||||
"github.com/heptio/ark/pkg/cloudprovider"
|
||||
arkv1client "github.com/heptio/ark/pkg/generated/clientset/versioned/typed/ark/v1"
|
||||
informers "github.com/heptio/ark/pkg/generated/informers/externalversions/ark/v1"
|
||||
listers "github.com/heptio/ark/pkg/generated/listers/ark/v1"
|
||||
"github.com/heptio/ark/pkg/plugin"
|
||||
"github.com/heptio/ark/pkg/util/kube"
|
||||
"github.com/heptio/ark/pkg/util/stringslice"
|
||||
)
|
||||
|
||||
type backupSyncController struct {
|
||||
client arkv1client.BackupsGetter
|
||||
cloudBackupLister cloudprovider.BackupLister
|
||||
bucket string
|
||||
syncPeriod time.Duration
|
||||
namespace string
|
||||
backupLister listers.BackupLister
|
||||
backupInformerSynced cache.InformerSynced
|
||||
logger logrus.FieldLogger
|
||||
*genericController
|
||||
|
||||
client arkv1client.BackupsGetter
|
||||
backupLister listers.BackupLister
|
||||
backupStorageLocationLister listers.BackupStorageLocationLister
|
||||
namespace string
|
||||
newPluginManager func(logrus.FieldLogger) plugin.Manager
|
||||
listCloudBackups func(logrus.FieldLogger, cloudprovider.ObjectStore, string) ([]*arkv1api.Backup, error)
|
||||
}
|
||||
|
||||
func NewBackupSyncController(
|
||||
client arkv1client.BackupsGetter,
|
||||
cloudBackupLister cloudprovider.BackupLister,
|
||||
bucket string,
|
||||
backupInformer informers.BackupInformer,
|
||||
backupStorageLocationInformer informers.BackupStorageLocationInformer,
|
||||
syncPeriod time.Duration,
|
||||
namespace string,
|
||||
backupInformer informers.BackupInformer,
|
||||
pluginRegistry plugin.Registry,
|
||||
logger logrus.FieldLogger,
|
||||
logLevel logrus.Level,
|
||||
) Interface {
|
||||
if syncPeriod < time.Minute {
|
||||
logger.Infof("Provided backup sync period %v is too short. Setting to 1 minute", syncPeriod)
|
||||
syncPeriod = time.Minute
|
||||
}
|
||||
return &backupSyncController{
|
||||
client: client,
|
||||
cloudBackupLister: cloudBackupLister,
|
||||
bucket: bucket,
|
||||
syncPeriod: syncPeriod,
|
||||
namespace: namespace,
|
||||
backupLister: backupInformer.Lister(),
|
||||
backupInformerSynced: backupInformer.Informer().HasSynced,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Run is a blocking function that continually runs the object storage -> Ark API
|
||||
// sync process according to the controller's syncPeriod. It will return when it
|
||||
// receives on the ctx.Done() channel.
|
||||
func (c *backupSyncController) Run(ctx context.Context, workers int) error {
|
||||
c.logger.Info("Running backup sync controller")
|
||||
c.logger.Info("Waiting for caches to sync")
|
||||
if !cache.WaitForCacheSync(ctx.Done(), c.backupInformerSynced) {
|
||||
return errors.New("timed out waiting for caches to sync")
|
||||
c := &backupSyncController{
|
||||
genericController: newGenericController("backup-sync", logger),
|
||||
client: client,
|
||||
namespace: namespace,
|
||||
backupLister: backupInformer.Lister(),
|
||||
backupStorageLocationLister: backupStorageLocationInformer.Lister(),
|
||||
|
||||
newPluginManager: func(logger logrus.FieldLogger) plugin.Manager {
|
||||
return plugin.NewManager(logger, logLevel, pluginRegistry)
|
||||
},
|
||||
listCloudBackups: cloudprovider.ListBackups,
|
||||
}
|
||||
c.logger.Info("Caches are synced")
|
||||
wait.Until(c.run, c.syncPeriod, ctx.Done())
|
||||
return nil
|
||||
|
||||
c.resyncFunc = c.run
|
||||
c.resyncPeriod = syncPeriod
|
||||
c.cacheSyncWaiters = []cache.InformerSynced{
|
||||
backupInformer.Informer().HasSynced,
|
||||
backupStorageLocationInformer.Informer().HasSynced,
|
||||
}
|
||||
|
||||
return c
|
||||
}
|
||||
|
||||
const gcFinalizer = "gc.ark.heptio.com"
|
||||
|
||||
func (c *backupSyncController) run() {
|
||||
c.logger.Info("Syncing backups from object storage")
|
||||
backups, err := c.cloudBackupLister.ListBackups(c.bucket)
|
||||
c.logger.Info("Syncing backups from backup storage into cluster")
|
||||
|
||||
locations, err := c.backupStorageLocationLister.BackupStorageLocations(c.namespace).List(labels.Everything())
|
||||
if err != nil {
|
||||
c.logger.WithError(err).Error("error listing backups")
|
||||
c.logger.WithError(errors.WithStack(err)).Error("Error getting backup storage locations from lister")
|
||||
return
|
||||
}
|
||||
c.logger.WithField("backupCount", len(backups)).Info("Got backups from object storage")
|
||||
|
||||
cloudBackupNames := sets.NewString()
|
||||
for _, cloudBackup := range backups {
|
||||
logContext := c.logger.WithField("backup", kube.NamespaceAndName(cloudBackup))
|
||||
logContext.Info("Syncing backup")
|
||||
pluginManager := c.newPluginManager(c.logger)
|
||||
|
||||
cloudBackupNames.Insert(cloudBackup.Name)
|
||||
for _, location := range locations {
|
||||
log := c.logger.WithField("backupLocation", location.Name)
|
||||
log.Info("Syncing backups from backup location")
|
||||
|
||||
// If we're syncing backups made by pre-0.8.0 versions, the server removes all finalizers
|
||||
// faster than the sync finishes. Just process them as we find them.
|
||||
cloudBackup.Finalizers = stringslice.Except(cloudBackup.Finalizers, gcFinalizer)
|
||||
|
||||
cloudBackup.Namespace = c.namespace
|
||||
cloudBackup.ResourceVersion = ""
|
||||
|
||||
// Backup only if backup does not exist in Kubernetes or if we are not able to get the backup for any reason.
|
||||
_, err := c.client.Backups(cloudBackup.Namespace).Get(cloudBackup.Name, metav1.GetOptions{})
|
||||
objectStore, err := getObjectStoreForLocation(location, pluginManager)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("Error getting object store for location")
|
||||
continue
|
||||
}
|
||||
|
||||
backupsInBackupStore, err := c.listCloudBackups(log, objectStore, location.Spec.ObjectStorage.Bucket)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("Error listing backups in object store")
|
||||
continue
|
||||
}
|
||||
|
||||
log.WithField("backupCount", len(backupsInBackupStore)).Info("Got backups from object store")
|
||||
|
||||
cloudBackupNames := sets.NewString()
|
||||
for _, cloudBackup := range backupsInBackupStore {
|
||||
log = log.WithField("backup", kube.NamespaceAndName(cloudBackup))
|
||||
log.Debug("Checking cloud backup to see if it needs to be synced into the cluster")
|
||||
|
||||
cloudBackupNames.Insert(cloudBackup.Name)
|
||||
|
||||
// use the controller's namespace when getting the backup because that's where we
|
||||
// are syncing backups to, regardless of the namespace of the cloud backup.
|
||||
_, err := c.client.Backups(c.namespace).Get(cloudBackup.Name, metav1.GetOptions{})
|
||||
if err == nil {
|
||||
log.Debug("Backup already exists in cluster")
|
||||
continue
|
||||
}
|
||||
if !kuberrs.IsNotFound(err) {
|
||||
logContext.WithError(errors.WithStack(err)).Error("Error getting backup from client, proceeding with backup sync")
|
||||
log.WithError(errors.WithStack(err)).Error("Error getting backup from client, proceeding with sync into cluster")
|
||||
}
|
||||
|
||||
if _, err := c.client.Backups(cloudBackup.Namespace).Create(cloudBackup); err != nil && !kuberrs.IsAlreadyExists(err) {
|
||||
logContext.WithError(errors.WithStack(err)).Error("Error syncing backup from object storage")
|
||||
// remove the pre-v0.8.0 gcFinalizer if it exists
|
||||
// TODO(1.0): remove this
|
||||
cloudBackup.Finalizers = stringslice.Except(cloudBackup.Finalizers, gcFinalizer)
|
||||
cloudBackup.Namespace = c.namespace
|
||||
cloudBackup.ResourceVersion = ""
|
||||
|
||||
// update the StorageLocation field and label since the name of the location
|
||||
// may be different in this cluster than in the cluster that created the
|
||||
// backup.
|
||||
cloudBackup.Spec.StorageLocation = location.Name
|
||||
if cloudBackup.Labels == nil {
|
||||
cloudBackup.Labels = make(map[string]string)
|
||||
}
|
||||
cloudBackup.Labels[arkv1api.StorageLocationLabel] = cloudBackup.Spec.StorageLocation
|
||||
|
||||
_, err = c.client.Backups(cloudBackup.Namespace).Create(cloudBackup)
|
||||
switch {
|
||||
case err != nil && kuberrs.IsAlreadyExists(err):
|
||||
log.Debug("Backup already exists in cluster")
|
||||
case err != nil && !kuberrs.IsAlreadyExists(err):
|
||||
log.WithError(errors.WithStack(err)).Error("Error syncing backup into cluster")
|
||||
default:
|
||||
log.Debug("Synced backup into cluster")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
c.deleteUnused(cloudBackupNames)
|
||||
return
|
||||
c.deleteOrphanedBackups(location.Name, cloudBackupNames, log)
|
||||
}
|
||||
}
|
||||
|
||||
// deleteUnused deletes backup objects from Kubernetes if they are complete
|
||||
// and there is no corresponding backup in the object storage.
|
||||
func (c *backupSyncController) deleteUnused(cloudBackupNames sets.String) {
|
||||
// Backups objects in Kubernetes
|
||||
backups, err := c.backupLister.Backups(c.namespace).List(labels.Everything())
|
||||
// deleteOrphanedBackups deletes backup objects from Kubernetes that have the specified location
|
||||
// and a phase of Completed, but no corresponding backup in object storage.
|
||||
func (c *backupSyncController) deleteOrphanedBackups(locationName string, cloudBackupNames sets.String, log logrus.FieldLogger) {
|
||||
locationSelector := labels.Set(map[string]string{
|
||||
arkv1api.StorageLocationLabel: locationName,
|
||||
}).AsSelector()
|
||||
|
||||
backups, err := c.backupLister.Backups(c.namespace).List(locationSelector)
|
||||
if err != nil {
|
||||
c.logger.WithError(errors.WithStack(err)).Error("Error listing backup from Kubernetes")
|
||||
log.WithError(errors.WithStack(err)).Error("Error listing backups from cluster")
|
||||
return
|
||||
}
|
||||
if len(backups) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// For each completed backup object in Kubernetes, delete it if it
|
||||
// does not have a corresponding backup in object storage
|
||||
for _, backup := range backups {
|
||||
if backup.Status.Phase == api.BackupPhaseCompleted && !cloudBackupNames.Has(backup.Name) {
|
||||
if err := c.client.Backups(backup.Namespace).Delete(backup.Name, nil); err != nil {
|
||||
c.logger.WithError(errors.WithStack(err)).Error("Error deleting unused backup from Kubernetes")
|
||||
} else {
|
||||
c.logger.Debugf("Deleted backup: %s", backup.Name)
|
||||
}
|
||||
log = log.WithField("backup", backup.Name)
|
||||
if backup.Status.Phase != arkv1api.BackupPhaseCompleted || cloudBackupNames.Has(backup.Name) {
|
||||
continue
|
||||
}
|
||||
|
||||
if err := c.client.Backups(backup.Namespace).Delete(backup.Name, nil); err != nil {
|
||||
log.WithError(errors.WithStack(err)).Error("Error deleting orphaned backup from cluster")
|
||||
} else {
|
||||
log.Debug("Deleted orphaned backup from cluster")
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user