Skip to content

Commit

Permalink
Add maintenance job
Browse files Browse the repository at this point in the history
Signed-off-by: Ming Qiu <mqiu@vmware.com>
  • Loading branch information
qiuming-best committed Feb 23, 2024
1 parent 2a1ae0e commit 375d8ab
Show file tree
Hide file tree
Showing 16 changed files with 1,845 additions and 23 deletions.
1 change: 1 addition & 0 deletions changelogs/unreleased/7451-qiuming-best
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add repository maintenance job
22 changes: 22 additions & 0 deletions pkg/cmd/cli/install/install.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,9 @@ import (
"strings"
"time"

"github.com/sirupsen/logrus"

"github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/uploader"

"github.com/pkg/errors"
Expand All @@ -38,6 +41,7 @@ import (
"github.com/vmware-tanzu/velero/pkg/cmd/util/output"
"github.com/vmware-tanzu/velero/pkg/install"
kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
"github.com/vmware-tanzu/velero/pkg/util/logging"
)

// Options collects all the options for installing Velero into a Kubernetes cluster.
Expand Down Expand Up @@ -84,6 +88,9 @@ type Options struct {
DefaultSnapshotMoveData bool
DisableInformerCache bool
ScheduleSkipImmediately bool
FormatFlag *logging.FormatFlag
LogLevelFlag *logging.LevelFlag
MaintenanceCfg repository.MaintenanceConfig
}

// BindFlags adds command line values to the options struct.
Expand Down Expand Up @@ -128,6 +135,13 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
flags.BoolVar(&o.DefaultSnapshotMoveData, "default-snapshot-move-data", o.DefaultSnapshotMoveData, "Bool flag to configure Velero server to move data by default for all snapshots supporting data movement. Optional.")
flags.BoolVar(&o.DisableInformerCache, "disable-informer-cache", o.DisableInformerCache, "Disable informer cache for Get calls on restore. With this enabled, it will speed up restore in cases where there are backup resources which already exist in the cluster, but for very large clusters this will increase velero memory usage. Default is false (don't disable). Optional.")
flags.BoolVar(&o.ScheduleSkipImmediately, "schedule-skip-immediately", o.ScheduleSkipImmediately, "Skip the first scheduled backup immediately after creating a schedule. Default is false (don't skip).")
flags.Var(o.FormatFlag, "log-format", fmt.Sprintf("The format for log output. Valid values are %s.", strings.Join(o.FormatFlag.AllowedValues(), ", ")))
flags.Var(o.LogLevelFlag, "log-level", fmt.Sprintf("The level at which to log. Valid values are %s.", strings.Join(o.LogLevelFlag.AllowedValues(), ", ")))
flags.IntVar(&o.MaintenanceCfg.KeepLatestMaitenanceJobs, "keep-latest-maintenance-jobs", o.MaintenanceCfg.KeepLatestMaitenanceJobs, "Number of latest maintenance jobs to keep each repository. Optional.")
flags.StringVar(&o.MaintenanceCfg.CPURequest, "maintenance-job-cpu-request", o.MaintenanceCfg.CPURequest, "CPU request for maintenance jobs. Default is no limit.")
flags.StringVar(&o.MaintenanceCfg.MemRequest, "maintenance-job-mem-request", o.MaintenanceCfg.MemRequest, "Memory request for maintenance jobs. Default is no limit.")
flags.StringVar(&o.MaintenanceCfg.CPULimit, "maintenance-job-cpu-limit", o.MaintenanceCfg.CPULimit, "CPU limit for maintenance jobs. Default is no limit.")
flags.StringVar(&o.MaintenanceCfg.MemLimit, "maintenance-job-mem-limit", o.MaintenanceCfg.MemLimit, "Memory limit for maintenance jobs. Default is no limit.")
}

// NewInstallOptions instantiates a new, default InstallOptions struct.
Expand Down Expand Up @@ -157,6 +171,11 @@ func NewInstallOptions() *Options {
DefaultSnapshotMoveData: false,
DisableInformerCache: false,
ScheduleSkipImmediately: false,
FormatFlag: logging.NewFormatFlag(),
LogLevelFlag: logging.LogLevelFlag(logrus.InfoLevel),
MaintenanceCfg: repository.MaintenanceConfig{
KeepLatestMaitenanceJobs: repository.DefaultKeepLatestMaitenanceJobs,
},
}
}

Expand Down Expand Up @@ -224,6 +243,9 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
DefaultSnapshotMoveData: o.DefaultSnapshotMoveData,
DisableInformerCache: o.DisableInformerCache,
ScheduleSkipImmediately: o.ScheduleSkipImmediately,
FormatFlag: o.FormatFlag,
LogLevelFlag: o.LogLevelFlag,
MaintenanceCfg: o.MaintenanceCfg,
}, nil
}

Expand Down
176 changes: 176 additions & 0 deletions pkg/cmd/server/repomaintenance/repo_maintenance.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,176 @@
package repomaintenance

import (
"context"
"fmt"
"os"
"strings"

"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/pflag"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/vmware-tanzu/velero/internal/credentials"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
velerocli "github.com/vmware-tanzu/velero/pkg/client"
"github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/repository/provider"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
"github.com/vmware-tanzu/velero/pkg/util/logging"
)

type Options struct {
RepoName string
BackupStorageLocation string
RepoType string
KeepLatestMaitenanceJobs int
LogLevelFlag *logging.LevelFlag
FormatFlag *logging.FormatFlag
}

func (o *Options) BindFlags(flags *pflag.FlagSet) {
flags.StringVar(&o.RepoName, "repo-name", "", "namespace of the pod/volume that the snapshot is for")
flags.StringVar(&o.BackupStorageLocation, "backup-storage-location", "", "backup's storage location name")
flags.StringVar(&o.RepoType, "repo-type", velerov1api.BackupRepositoryTypeKopia, "type of the repository where the snapshot is stored")
flags.Var(o.LogLevelFlag, "log-level", fmt.Sprintf("The level at which to log. Valid values are %s.", strings.Join(o.LogLevelFlag.AllowedValues(), ", ")))
flags.Var(o.FormatFlag, "log-format", fmt.Sprintf("The format for log output. Valid values are %s.", strings.Join(o.FormatFlag.AllowedValues(), ", ")))
}

func NewCommand(f velerocli.Factory) *cobra.Command {
o := &Options{
LogLevelFlag: logging.LogLevelFlag(logrus.InfoLevel),
FormatFlag: logging.NewFormatFlag(),
}
cmd := &cobra.Command{
Use: "repo-maintenance",
Hidden: true,
Short: "VELERO SERVER COMMAND ONLY - not intended to be run directly by users",
Run: func(c *cobra.Command, args []string) {
o.Run(f)
},
}

o.BindFlags(cmd.Flags())
return cmd
}

func checkError(err error, log logrus.FieldLogger) error {
if err != nil {
if err != context.Canceled {
log.Errorf("An error occurred: %v", err)
return err
}
}
return nil
}

func (o *Options) Run(f velerocli.Factory) {
logger := logging.DefaultLogger(o.LogLevelFlag.Parse(), o.FormatFlag.Parse())
logger.SetOutput(os.Stdout)
var getError error
errorFile, err := os.Create("/dev/termination-log")
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to create termination log file: %v\n", err)
getError = err
return
}

defer func() {
err := errorFile.Close()
if err != nil {
getError = err
}
if getError != nil {
os.Exit(1) // indicate the command executed failed
}
}()

logger.AddHook(&logging.FileHook{
File: errorFile})

scheme := runtime.NewScheme()
err = velerov1api.AddToScheme(scheme)
if getError = checkError(err, logger); getError != nil {
return
}

err = v1.AddToScheme(scheme)
if getError = checkError(err, logger); getError != nil {
return
}

config, err := f.ClientConfig()
if getError = checkError(err, logger); getError != nil {
return
}

cli, err := client.New(config, client.Options{
Scheme: scheme,
})
if getError = checkError(err, logger); getError != nil {
return
}

credentialFileStore, err := credentials.NewNamespacedFileStore(
cli,
f.Namespace(),
"/tmp/credentials",
filesystem.NewFileSystem(),
)
if getError = checkError(err, logger); getError != nil {
return
}

credentialSecretStore, err := credentials.NewNamespacedSecretStore(cli, f.Namespace())
if getError = checkError(err, logger); getError != nil {
return
}

var repoProvider provider.Provider
if o.RepoType == velerov1api.BackupRepositoryTypeRestic {
repoProvider = provider.NewResticRepositoryProvider(credentialFileStore, filesystem.NewFileSystem(), logger)
} else {
repoProvider = provider.NewUnifiedRepoProvider(
credentials.CredentialGetter{
FromFile: credentialFileStore,
FromSecret: credentialSecretStore,
}, o.RepoType, cli, logger)
}

// backupRepository
repo, err := repository.GetBackupRepository(context.Background(), cli, f.Namespace(),
repository.BackupRepositoryKey{
VolumeNamespace: o.RepoName,
BackupLocation: o.BackupStorageLocation,
RepositoryType: o.RepoType,
}, true)

if getError = checkError(err, logger); getError != nil {
return
}

// bsl
bsl := &velerov1api.BackupStorageLocation{}
err = cli.Get(context.Background(), client.ObjectKey{Namespace: f.Namespace(), Name: repo.Spec.BackupStorageLocation}, bsl)
if getError = checkError(err, logger); getError != nil {
return
}

para := provider.RepoParam{
BackupRepo: repo,
BackupLocation: bsl,
}

err = repoProvider.BoostRepoConnect(context.Background(), para)
if getError = checkError(err, logger); getError != nil {
return
}

err = repoProvider.PruneRepo(context.Background(), para)
if getError = checkError(err, logger); getError != nil {
return
}
}
27 changes: 26 additions & 1 deletion pkg/cmd/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ import (
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
appsv1 "k8s.io/api/apps/v1"
batchv1api "k8s.io/api/batch/v1"
corev1api "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand All @@ -58,6 +60,7 @@ import (
"github.com/vmware-tanzu/velero/pkg/buildinfo"
"github.com/vmware-tanzu/velero/pkg/client"
"github.com/vmware-tanzu/velero/pkg/cmd"
"github.com/vmware-tanzu/velero/pkg/cmd/server/repomaintenance"
"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
"github.com/vmware-tanzu/velero/pkg/controller"
Expand Down Expand Up @@ -135,6 +138,7 @@ type serverConfig struct {
defaultSnapshotMoveData bool
disableInformerCache bool
scheduleSkipImmediately bool
maintenanceCfg repository.MaintenanceConfig
}

func NewCommand(f client.Factory) *cobra.Command {
Expand Down Expand Up @@ -166,6 +170,9 @@ func NewCommand(f client.Factory) *cobra.Command {
defaultSnapshotMoveData: false,
disableInformerCache: defaultDisableInformerCache,
scheduleSkipImmediately: false,
maintenanceCfg: repository.MaintenanceConfig{
KeepLatestMaitenanceJobs: repository.DefaultKeepLatestMaitenanceJobs,
},
}
)

Expand Down Expand Up @@ -239,7 +246,17 @@ func NewCommand(f client.Factory) *cobra.Command {
command.Flags().BoolVar(&config.defaultSnapshotMoveData, "default-snapshot-move-data", config.defaultSnapshotMoveData, "Move data by default for all snapshots supporting data movement.")
command.Flags().BoolVar(&config.disableInformerCache, "disable-informer-cache", config.disableInformerCache, "Disable informer cache for Get calls on restore. With this enabled, it will speed up restore in cases where there are backup resources which already exist in the cluster, but for very large clusters this will increase velero memory usage. Default is false (don't disable).")
command.Flags().BoolVar(&config.scheduleSkipImmediately, "schedule-skip-immediately", config.scheduleSkipImmediately, "Skip the first scheduled backup immediately after creating a schedule. Default is false (don't skip).")
command.Flags().IntVar(&config.maintenanceCfg.KeepLatestMaitenanceJobs, "keep-latest-maintenance-jobs", config.maintenanceCfg.KeepLatestMaitenanceJobs, "Number of latest maintenance jobs to keep each repository. Optional.")
command.Flags().StringVar(&config.maintenanceCfg.CPURequest, "maintenance-job-cpu-request", config.maintenanceCfg.CPURequest, "CPU request for maintenance job. Default is no limit.")
command.Flags().StringVar(&config.maintenanceCfg.MemRequest, "maintenance-job-mem-request", config.maintenanceCfg.MemRequest, "Memory request for maintenance job. Default is no limit.")
command.Flags().StringVar(&config.maintenanceCfg.CPULimit, "maintenance-job-cpu-limit", config.maintenanceCfg.CPULimit, "CPU limit for maintenance job. Default is no limit.")
command.Flags().StringVar(&config.maintenanceCfg.MemLimit, "maintenance-job-mem-limit", config.maintenanceCfg.MemLimit, "Memory limit for maintenance job. Default is no limit.")

// maintenance job log setting inherited from velero server
config.maintenanceCfg.FormatFlag = config.formatFlag
config.maintenanceCfg.LogLevelFlag = logLevelFlag

command.AddCommand(repomaintenance.NewCommand(f))
return command
}

Expand Down Expand Up @@ -346,6 +363,14 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
cancelFunc()
return nil, err
}
if err := batchv1api.AddToScheme(scheme); err != nil {
cancelFunc()
return nil, err
}
if err := appsv1.AddToScheme(scheme); err != nil {
cancelFunc()
return nil, err
}

Check warning on line 373 in pkg/cmd/server/server.go

View check run for this annotation

Codecov / codecov/patch

pkg/cmd/server/server.go#L366-L373

Added lines #L366 - L373 were not covered by tests

ctrl.SetLogger(logrusr.New(logger))

Expand Down Expand Up @@ -642,7 +667,7 @@ func (s *server) initRepoManager() error {
s.repoLocker = repository.NewRepoLocker()
s.repoEnsurer = repository.NewEnsurer(s.mgr.GetClient(), s.logger, s.config.resourceTimeout)

s.repoManager = repository.NewManager(s.namespace, s.mgr.GetClient(), s.repoLocker, s.repoEnsurer, s.credentialFileStore, s.credentialSecretStore, s.logger)
s.repoManager = repository.NewManager(s.namespace, s.mgr.GetClient(), s.repoLocker, s.repoEnsurer, s.credentialFileStore, s.credentialSecretStore, s.config.maintenanceCfg, s.logger)

Check warning on line 670 in pkg/cmd/server/server.go

View check run for this annotation

Codecov / codecov/patch

pkg/cmd/server/server.go#L670

Added line #L670 was not covered by tests

return nil
}
Expand Down
36 changes: 32 additions & 4 deletions pkg/controller/backup_repository_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,9 @@ import (
)

const (
repoSyncPeriod = 5 * time.Minute
defaultMaintainFrequency = 7 * 24 * time.Hour
repoSyncPeriod = 5 * time.Minute
defaultMaintainFrequency = 7 * 24 * time.Hour
defaultMaintainRetryFrequency = 5 * time.Minute
)

type BackupRepoReconciler struct {
Expand Down Expand Up @@ -277,16 +278,42 @@ func ensureRepo(repo *velerov1api.BackupRepository, repoManager repository.Manag
return repoManager.PrepareRepo(repo)
}

func (r *BackupRepoReconciler) runMaintenanceIfDue(ctx context.Context, req *velerov1api.BackupRepository, log logrus.FieldLogger) error {
log.Debug("backupRepositoryController.runMaintenanceIfDue")
func (r *BackupRepoReconciler) shouldRetryMaintenance(req *velerov1api.BackupRepository, log logrus.FieldLogger) bool {
log.Debug("Checking if maintenance should be retried")

if req.Status.Message == "" {
return true
}

job, err := repository.GetLatestMaintenanceJob(r.Client, req.Name)
if err != nil {
log.WithError(err).Error("error getting latest maintenance job")
return false
}

Check warning on line 292 in pkg/controller/backup_repository_controller.go

View check run for this annotation

Codecov / codecov/patch

pkg/controller/backup_repository_controller.go#L288-L292

Added lines #L288 - L292 were not covered by tests

if job != nil && job.Status.Failed > 0 && job.CreationTimestamp.Add(defaultMaintainRetryFrequency).Before(r.clock.Now()) {
log.Debug("Latest maintenance job failed and is older than retry frequency, retrying maintenance")
return true
} else {
log.Debugf("Latest maintenance job %s is not older than retry frequency, not retrying maintenance", job.Name)
}

Check warning on line 299 in pkg/controller/backup_repository_controller.go

View check run for this annotation

Codecov / codecov/patch

pkg/controller/backup_repository_controller.go#L294-L299

Added lines #L294 - L299 were not covered by tests

return false

Check warning on line 301 in pkg/controller/backup_repository_controller.go

View check run for this annotation

Codecov / codecov/patch

pkg/controller/backup_repository_controller.go#L301

Added line #L301 was not covered by tests
}

func (r *BackupRepoReconciler) runMaintenanceIfDue(ctx context.Context, req *velerov1api.BackupRepository, log logrus.FieldLogger) error {
now := r.clock.Now()

if !dueForMaintenance(req, now) {
log.Debug("not due for maintenance")
return nil
}

if !r.shouldRetryMaintenance(req, log) {
log.Debug("not retrying maintenance")
return nil
}

Check warning on line 315 in pkg/controller/backup_repository_controller.go

View check run for this annotation

Codecov / codecov/patch

pkg/controller/backup_repository_controller.go#L313-L315

Added lines #L313 - L315 were not covered by tests

log.Info("Running maintenance on backup repository")

// prune failures should be displayed in the `.status.message` field but
Expand All @@ -300,6 +327,7 @@ func (r *BackupRepoReconciler) runMaintenanceIfDue(ctx context.Context, req *vel
}

return r.patchBackupRepository(ctx, req, func(rr *velerov1api.BackupRepository) {
rr.Status.Message = ""
rr.Status.LastMaintenanceTime = &metav1.Time{Time: now}
})
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/datapath/file_system.go
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,7 @@ func (fs *fileSystemBR) Cancel() {

func (fs *fileSystemBR) boostRepoConnect(ctx context.Context, repositoryType string, credentialGetter *credentials.CredentialGetter) error {
if repositoryType == velerov1api.BackupRepositoryTypeKopia {
if err := repoProvider.NewUnifiedRepoProvider(*credentialGetter, repositoryType, fs.log).BoostRepoConnect(ctx, repoProvider.RepoParam{BackupLocation: fs.backupLocation, BackupRepo: fs.backupRepo}); err != nil {
if err := repoProvider.NewUnifiedRepoProvider(*credentialGetter, repositoryType, fs.client, fs.log).BoostRepoConnect(ctx, repoProvider.RepoParam{BackupLocation: fs.backupLocation, BackupRepo: fs.backupRepo}); err != nil {

Check warning on line 187 in pkg/datapath/file_system.go

View check run for this annotation

Codecov / codecov/patch

pkg/datapath/file_system.go#L187

Added line #L187 was not covered by tests
return err
}
} else {
Expand Down
Loading

0 comments on commit 375d8ab

Please sign in to comment.