Add a function to delete backups logic

Author: MerceaOtniel

deploy/mysqlclusters.yaml

@@ -25,6 +25,10 @@ spec:
               description: Specify under crontab format interval to take backups leave
                 it empty to deactivate the backup process Defaults to ""
               type: string
+            backupScheduleJobsHistoryLimit:
+              description: If set keeps last BackupScheduleJobsHistoryLimit Backups
+              format: int32
+              type: integer
             backupSecretName:
               type: string
             backupUri:

examples/example-cluster.yaml

@@ -3,7 +3,7 @@ kind: MysqlCluster
 metadata:
   name: foo
 spec:
-  replicas: 3
+  replicas: 1
   secretName: the-secret
   # mysqlVersion: 5.7
   # initBucketUri: gs://bucket_name/backup.xtrabackup.gz
@@ -16,6 +16,8 @@ spec:
   # backupSchedule:
   # backupUri: s3://bucket_name/
   # backupSecretName:
+  # backupScheduleJobsHistoryLimit:
+
 
   ## Configs that will be added to my.cnf for cluster
   # mysqlConf:

pkg/apis/mysql/v1alpha1/types.go

@@ -82,6 +82,10 @@ type ClusterSpec struct {
 	BackupUri        string `json:"backupUri,omitempty"`
 	BackupSecretName string `json:"backupSecretName,omitempty"`
 
+	// If set keeps last BackupScheduleJobsHistoryLimit Backups
+	// +optional
+	BackupScheduleJobsHistoryLimit *int `json:"backupScheduleJobsHistoryLimit"`
+
 	// A map[string]string that will be passed to my.cnf file.
 	// +optional
 	MysqlConf MysqlConf `json:"mysqlConf,omitempty"`

pkg/apis/mysql/v1alpha1/zz_generated.deepcopy.go

@@ -111,6 +111,15 @@ func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 			**out = **in
 		}
 	}
+	if in.BackupScheduleJobsHistoryLimit != nil {
+		in, out := &in.BackupScheduleJobsHistoryLimit, &out.BackupScheduleJobsHistoryLimit
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(int)
+			**out = **in
+		}
+	}
 	if in.MysqlConf != nil {
 		in, out := &in.MysqlConf, &out.MysqlConf
 		*out = make(MysqlConf, len(*in))

pkg/controller/clustercontroller/backups.go

@@ -48,6 +48,8 @@ type job struct {
 
 	lock     *sync.Mutex
 	myClient myclientset.Interface
+
+	BackupScheduleJobsHistoryLimit *int
 }
 
 func (c *Controller) registerClusterInBackupCron(cluster *api.MysqlCluster) error {
@@ -86,17 +88,23 @@ func (c *Controller) registerClusterInBackupCron(cluster *api.MysqlCluster) erro
 		myClient:      c.myClient,
 		BackupRunning: new(bool),
 		lock:          new(sync.Mutex),
+		BackupScheduleJobsHistoryLimit: cluster.Spec.BackupScheduleJobsHistoryLimit,
 	}, cluster.Name)
 
 	return nil
 }
 
 func (j job) Run() {
 	backupName := fmt.Sprintf("%s-auto-backup-%s", j.Name, time.Now().Format("2006-01-02t15-04-05"))
-	glog.Infof("Schedul backup job started. Creating backup %s..", backupName)
+	glog.Infof("Scheduled backup job started for %s/%s ", j.Namespace, backupName)
+
+	if j.BackupScheduleJobsHistoryLimit != nil {
+		defer j.backupGC()
+	}
 
 	// Wrap backup creation to ensure that lock is released when backup is
 	// created
+
 	created := func() bool {
 		j.lock.Lock()
 		defer j.lock.Unlock()
@@ -121,7 +129,6 @@ func (j job) Run() {
 					ClusterName: j.Name,
 				},
 			})
-
 			if err == nil {
 				break
 			}
@@ -140,7 +147,6 @@ func (j job) Run() {
 		*j.BackupRunning = true
 		return true
 	}()
-
 	if !created {
 		return
 	}
@@ -170,3 +176,39 @@ func (j job) Run() {
 			backupName, j.Name, err)
 	}
 }
+
+func (j *job) backupGC() {
+
+	var err error
+
+	backups, err := j.myClient.Mysql().MysqlBackups(j.Namespace).List(metav1.ListOptions{
+		LabelSelector: fmt.Sprintf("recurrent=true"),
+	})
+
+	var clusterBackups []api.MysqlBackup
+
+	for i := 0; i < len(backups.Items); i++ {
+		if backups.Items[i].Spec.ClusterName == j.Name {
+			clusterBackups = append(clusterBackups, backups.Items[i])
+		}
+	}
+
+	if err != nil {
+		glog.Infof("Failed to obtain backup list for %s/%s, error:%s", j.Namespace, j.Name, err)
+		return
+	}
+	var size = len(clusterBackups)
+
+	if size > *j.BackupScheduleJobsHistoryLimit {
+		for i := 0; i < size-(*j.BackupScheduleJobsHistoryLimit); i++ {
+
+			err := j.myClient.Mysql().MysqlBackups(j.Namespace).Delete(clusterBackups[i].Name, &(metav1.DeleteOptions{}))
+
+			if err != nil {
+				glog.Warningf("Failed to remove backup %s/%s", clusterBackups[i].Name, err)
+			}
+		}
+		clusterBackups = clusterBackups[(size - (*j.BackupScheduleJobsHistoryLimit)):]
+	}
+
+}

pkg/controller/clustercontroller/backups_test.go

@@ -1,12 +1,9 @@
 /*
 Copyright 2018 Pressinfra SRL
-
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
-
     http://www.apache.org/licenses/LICENSE-2.0
-
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -56,6 +53,8 @@ var _ = Describe("Test cluster reconciliation queue", func() {
 		controller = newController(stop, client, myClient, rec)
 		backupPollingTime = 10 * time.Millisecond
 		backupWatchTimeout = time.Second
+		two := 2
+		cluster.Spec.BackupScheduleJobsHistoryLimit = &two
 
 	})
 
@@ -67,6 +66,7 @@ var _ = Describe("Test cluster reconciliation queue", func() {
 		Context("cluster with schedule backups", func() {
 			It("try to register multiple times", func() {
 				cluster.Spec.BackupSchedule = "0 * * * *"
+
 				_, err := myClient.MysqlV1alpha1().MysqlClusters(tutil.Namespace).Create(cluster)
 				Expect(err).To(Succeed())
 
@@ -113,15 +113,16 @@ var _ = Describe("Test cluster reconciliation queue", func() {
 					myClient:      myClient,
 					BackupRunning: new(bool),
 					lock:          new(sync.Mutex),
+					BackupScheduleJobsHistoryLimit: cluster.Spec.BackupScheduleJobsHistoryLimit,
 				}
 
 				go j.Run()
-
 				Eventually(func() bool {
 					j.lock.Lock() // avoid a data race
 					defer j.lock.Unlock()
 					return *j.BackupRunning
 				}).Should(Equal(true))
+
 				Eventually(func() []api.MysqlBackup {
 					bs, _ := myClient.Mysql().MysqlBackups(j.Namespace).List(metav1.ListOptions{})
 					return bs.Items
@@ -139,6 +140,28 @@ var _ = Describe("Test cluster reconciliation queue", func() {
 					return *j.BackupRunning
 				}).Should(Equal(false))
 			})
+
+			It("start jobs to schedule backups", func() {
+
+				j := job{
+					Name:          cluster.Name,
+					Namespace:     cluster.Namespace,
+					myClient:      myClient,
+					BackupRunning: new(bool),
+					lock:          new(sync.Mutex),
+					BackupScheduleJobsHistoryLimit: cluster.Spec.BackupScheduleJobsHistoryLimit,
+				}
+
+				for i := 0; i < *(j.BackupScheduleJobsHistoryLimit)+1; i++ {
+					j.Run()
+				}
+
+				Eventually(func() []api.MysqlBackup {
+					bs, _ := myClient.Mysql().MysqlBackups(j.Namespace).List(metav1.ListOptions{})
+					return bs.Items
+				}).Should(HaveLen(*j.BackupScheduleJobsHistoryLimit))
+
+			})
 		})
 	})
 })

pkg/openapi/openapi_generated.go

@@ -317,6 +317,13 @@ func schema_pkg_apis_mysql_v1alpha1_ClusterSpec(ref common.ReferenceCallback) co
 							Format: "",
 						},
 					},
+					"backupScheduleJobsHistoryLimit": {
+						SchemaProps: spec.SchemaProps{
+							Description: "If set keeps last BackupScheduleJobsHistoryLimit Backups",
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
 					"mysqlConf": {
 						SchemaProps: spec.SchemaProps{
 							Description: "A map[string]string that will be passed to my.cnf file.",