etcd: one option to do defragement before backup.

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.

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 {

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 {

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 {

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)
 	}

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 {

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
 		}