diff --git a/pkg/apis/etcd/v1beta2/backup_types.go b/pkg/apis/etcd/v1beta2/backup_types.go index 3f6ff2367..dbc112e53 100644 --- a/pkg/apis/etcd/v1beta2/backup_types.go +++ b/pkg/apis/etcd/v1beta2/backup_types.go @@ -109,6 +109,8 @@ type BackupPolicy struct { // MaxBackups is to specify how many backups we want to keep // 0 is magic number to indicate un-limited backups MaxBackups int `json:"maxBackups,omitempty"` + // DefragmentBeforeBackup is to specify whether we do one defragement before backup. + DefragmentBeforeBackup bool `json:"defragmentBeforeBackup,omitempty"` } // BackupStatus represents the status of the EtcdBackup Custom Resource. diff --git a/pkg/backup/backup_manager.go b/pkg/backup/backup_manager.go index a82ae32eb..fa4fd7126 100644 --- a/pkg/backup/backup_manager.go +++ b/pkg/backup/backup_manager.go @@ -54,7 +54,7 @@ func NewBackupManagerFromWriter(kubecli kubernetes.Interface, bw writer.Writer, // SaveSnap uses backup writer to save etcd snapshot to a specified S3 path // and returns backup etcd server's kv store revision and its version. -func (bm *BackupManager) SaveSnap(ctx context.Context, s3Path string, isPeriodic bool) (int64, string, *metav1.Time, error) { +func (bm *BackupManager) SaveSnap(ctx context.Context, s3Path string, isPeriodic bool, degragementBeforeBackup bool) (int64, string, *metav1.Time, error) { now := time.Now().UTC() etcdcli, rev, err := bm.etcdClientWithMaxRevision(ctx) if err != nil { @@ -67,6 +67,13 @@ func (bm *BackupManager) SaveSnap(ctx context.Context, s3Path string, isPeriodic return 0, "", nil, fmt.Errorf("failed to retrieve etcd version from the status call: %v", err) } + if degragementBeforeBackup { + logrus.Info("running defragment for cluster.") + if err := bm.defrag(ctx, etcdcli); err != nil { + logrus.Error("defrag failed: %v", err) + } + } + rc, err := etcdcli.Snapshot(ctx) if err != nil { return 0, "", nil, fmt.Errorf("failed to receive snapshot (%v)", err) @@ -82,6 +89,28 @@ func (bm *BackupManager) SaveSnap(ctx context.Context, s3Path string, isPeriodic return rev, resp.Version, &metav1.Time{Time: now}, nil } +func (bm *BackupManager) defrag(ctx context.Context, etcdcli *clientv3.Client) error { + mems, err := etcdcli.MemberList(ctx) + if err != nil { + return fmt.Errorf("defrag: failed to get cluster endpoints (%v)", err) + } + + eps := []string{} + for _, me := range mems.Members { + eps = append(eps, me.ClientURLs...) + } + + for _, ep := range eps { + _, err := etcdcli.Defragment(ctx, ep) + if err != nil { + // If this fails we want to continue but just log the error + fmt.Errorf("defrag: failed to defrag member %s (%v)", ep, err) + } + } + + return nil +} + // EnsureMaxBackup to ensure the number of snapshot is under maxcount // if the number of snapshot exceeded than maxcount, delete oldest snapshot func (bm *BackupManager) EnsureMaxBackup(ctx context.Context, basePath string, maxCount int) error { diff --git a/pkg/controller/backup-operator/abs_backup.go b/pkg/controller/backup-operator/abs_backup.go index c1c9c786d..0061e056a 100644 --- a/pkg/controller/backup-operator/abs_backup.go +++ b/pkg/controller/backup-operator/abs_backup.go @@ -29,7 +29,7 @@ import ( // handleABS saves etcd cluster's backup to specificed ABS path. func handleABS(ctx context.Context, kubecli kubernetes.Interface, s *api.ABSBackupSource, endpoints []string, clientTLSSecret, - namespace string, isPeriodic bool, maxBackup int) (*api.BackupStatus, error) { + namespace string, isPeriodic bool, maxBackup int, degragementBeforeBackup bool) (*api.BackupStatus, error) { // TODO: controls NewClientFromSecret with ctx. This depends on upstream kubernetes to support API calls with ctx. cli, err := absfactory.NewClientFromSecret(kubecli, namespace, s.ABSSecret) if err != nil { diff --git a/pkg/controller/backup-operator/gcs_backup.go b/pkg/controller/backup-operator/gcs_backup.go index 1f1f5a921..ac2c9a21f 100644 --- a/pkg/controller/backup-operator/gcs_backup.go +++ b/pkg/controller/backup-operator/gcs_backup.go @@ -29,7 +29,7 @@ import ( // handleGCS saves etcd cluster's backup to specificed GCS path. func handleGCS(ctx context.Context, kubecli kubernetes.Interface, s *api.GCSBackupSource, endpoints []string, clientTLSSecret, - namespace string, isPeriodic bool, maxBackup int) (*api.BackupStatus, error) { + namespace string, isPeriodic bool, maxBackup int, degragementBeforeBackup bool) (*api.BackupStatus, error) { // TODO: controls NewClientFromSecret with ctx. This depends on upstream kubernetes to support API calls with ctx. cli, err := gcsfactory.NewClientFromSecret(ctx, kubecli, namespace, s.GCPSecret) if err != nil { diff --git a/pkg/controller/backup-operator/oss_backup.go b/pkg/controller/backup-operator/oss_backup.go index ed3ebfaa1..3c5929731 100644 --- a/pkg/controller/backup-operator/oss_backup.go +++ b/pkg/controller/backup-operator/oss_backup.go @@ -28,7 +28,7 @@ import ( // handleOSS saves etcd cluster's backup to specificed OSS path. func handleOSS(ctx context.Context, kubecli kubernetes.Interface, s *api.OSSBackupSource, endpoints []string, clientTLSSecret, - namespace string, isPeriodic bool, maxBackup int) (*api.BackupStatus, error) { + namespace string, isPeriodic bool, maxBackup int, degragementBeforeBackup bool) (*api.BackupStatus, error) { if s.Endpoint == "" { s.Endpoint = "http://oss-cn-hangzhou.aliyuncs.com" } @@ -45,7 +45,7 @@ func handleOSS(ctx context.Context, kubecli kubernetes.Interface, s *api.OSSBack bm := backup.NewBackupManagerFromWriter(kubecli, writer.NewOSSWriter(cli.OSS), tlsConfig, endpoints, namespace) - rev, etcdVersion, now, err := bm.SaveSnap(ctx, s.Path, isPeriodic) + rev, etcdVersion, now, err := bm.SaveSnap(ctx, s.Path, isPeriodic, degragementBeforeBackup) if err != nil { return nil, fmt.Errorf("failed to save snapshot (%v)", err) } diff --git a/pkg/controller/backup-operator/s3_backup.go b/pkg/controller/backup-operator/s3_backup.go index 075ddfb65..dbb95711c 100644 --- a/pkg/controller/backup-operator/s3_backup.go +++ b/pkg/controller/backup-operator/s3_backup.go @@ -30,7 +30,7 @@ import ( // TODO: replace this with generic backend interface for other options (PV, Azure) // handleS3 saves etcd cluster's backup to specificed S3 path. func handleS3(ctx context.Context, kubecli kubernetes.Interface, s *api.S3BackupSource, endpoints []string, clientTLSSecret, - namespace string, isPeriodic bool, maxBackup int) (*api.BackupStatus, error) { + namespace string, isPeriodic bool, maxBackup int, degragementBeforeBackup bool) (*api.BackupStatus, error) { // TODO: controls NewClientFromSecret with ctx. This depends on upstream kubernetes to support API calls with ctx. cli, err := s3factory.NewClientFromSecret(kubecli, namespace, s.Endpoint, s.AWSSecret, s.ForcePathStyle) if err != nil { diff --git a/pkg/controller/backup-operator/sync.go b/pkg/controller/backup-operator/sync.go index e5ae56567..7a3963981 100644 --- a/pkg/controller/backup-operator/sync.go +++ b/pkg/controller/backup-operator/sync.go @@ -269,28 +269,28 @@ func (b *Backup) handleBackup(parentContext *context.Context, spec *api.BackupSp switch spec.StorageType { case api.BackupStorageTypeS3: bs, err := handleS3(ctx, b.kubecli, spec.S3, spec.EtcdEndpoints, spec.ClientTLSSecret, - b.namespace, isPeriodic, backupMaxCount) + b.namespace, isPeriodic, backupMaxCount, spec.BackupPolicy.DefragmentBeforeBackup) if err != nil { return nil, err } return bs, nil case api.BackupStorageTypeABS: bs, err := handleABS(ctx, b.kubecli, spec.ABS, spec.EtcdEndpoints, spec.ClientTLSSecret, - b.namespace, isPeriodic, backupMaxCount) + b.namespace, isPeriodic, backupMaxCount, spec.BackupPolicy.DefragmentBeforeBackup) if err != nil { return nil, err } return bs, nil case api.BackupStorageTypeGCS: bs, err := handleGCS(ctx, b.kubecli, spec.GCS, spec.EtcdEndpoints, spec.ClientTLSSecret, - b.namespace, isPeriodic, backupMaxCount) + b.namespace, isPeriodic, backupMaxCount, spec.BackupPolicy.DefragmentBeforeBackup) if err != nil { return nil, err } return bs, nil case api.BackupStorageTypeOSS: bs, err := handleOSS(ctx, b.kubecli, spec.OSS, spec.EtcdEndpoints, spec.ClientTLSSecret, - b.namespace, isPeriodic, backupMaxCount) + b.namespace, isPeriodic, backupMaxCount, spec.BackupPolicy.DefragmentBeforeBackup) if err != nil { return nil, err }