feat: support take backup from specified node

Author: drivebyer

CHANGELOG.md

@@ -8,6 +8,7 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ### Added
  * `MysqlDatabase` `MysqlUser` Add delete policy
  * Add `PtHeartbeatResources` in `.Spec.PodSpec` to allow the user specifying resources for pt-heartbeat.
+ * Add `CandidateNode` field in `MysqlBackup.Spec` to allow the user specifying candidate node for backup.
  * Set `MysqlCluter.Spec.BackupSchedule` to empty string to disable recurrent backups
 ### Changed
  * Set default MySQL server version to `5.7.35`

config/crd/bases/mysql.presslabs.org_mysqlbackups.yaml

@@ -36,6 +36,9 @@ spec:
                 backupURL:
                   description: BackupURL represents the URL to the backup location, this can be partially specifyied. Default is used the one specified in the cluster.
                   type: string
+                candidateNode:
+                  description: CandidateNode is the node host that will be used to take the backup. If not set, the operator will calculate by itself.
+                  type: string
                 clusterName:
                   description: ClustterName represents the cluster for which to take backup
                   type: string

deploy/charts/mysql-operator/crds/mysql.presslabs.org_mysqlbackups.yaml

@@ -37,6 +37,9 @@ spec:
                 backupURL:
                   description: BackupURL represents the URL to the backup location, this can be partially specifyied. Default is used the one specified in the cluster.
                   type: string
+                candidateNode:
+                  description: CandidateNode is the node host that will be used to take the backup. If not set, the operator will calculate by itself.
+                  type: string
                 clusterName:
                   description: ClustterName represents the cluster for which to take backup
                   type: string

pkg/apis/mysql/v1alpha1/mysqlbackup_types.go

@@ -44,6 +44,11 @@ type MysqlBackupSpec struct {
 	// default it's used softDelete.
 	// +optional
 	RemoteDeletePolicy DeletePolicy `json:"remoteDeletePolicy,omitempty"`
+
+	// CandidateNode is the node host that will be used to take the backup.
+	// If not set, the operator will calculate by itself.
+	// +optional
+	CandidateNode string `json:"candidateNode,omitempty"`
 }
 
 // BackupCondition defines condition struct for backup resource
@@ -94,7 +99,6 @@ type MysqlBackupStatus struct {
 
 // MysqlBackup is the Schema for the mysqlbackups API
 // +kubebuilder:object:root=true
-//
 type MysqlBackup struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
@@ -105,7 +109,6 @@ type MysqlBackup struct {
 
 // MysqlBackupList contains a list of MysqlBackup
 // +kubebuilder:object:root=true
-//
 type MysqlBackupList struct {
 	metav1.TypeMeta `json:",inline"`
 	metav1.ListMeta `json:"metadata,omitempty"`

pkg/controller/mysqlbackup/internal/syncer/job.go

@@ -91,6 +91,10 @@ func (s *jobSyncer) SyncFn() error {
 // getBackupCandidate returns the hostname of the first not-lagged and
 // replicating slave node, else returns the master node.
 func (s *jobSyncer) getBackupCandidate() string {
+	if s.backup.Spec.CandidateNode != "" {
+		return s.backup.Spec.CandidateNode
+	}
+
 	for _, node := range s.cluster.Status.Nodes {
 		master := s.cluster.GetNodeCondition(node.Name, api.NodeConditionMaster)
 		replicating := s.cluster.GetNodeCondition(node.Name, api.NodeConditionReplicating)

pkg/controller/mysqlbackup/mysqlbackup_controller.go

@@ -151,6 +151,10 @@ func (r *ReconcileMysqlBackup) Reconcile(ctx context.Context, request reconcile.
 	// set defaults for the backup base on the related cluster
 	backup.SetDefaults(cluster)
 
+	if err := backup.Validate(cluster); err != nil {
+		return reconcile.Result{}, err
+	}
+
 	syncers := []syncer.Interface{
 		backupSyncer.NewDeleteJobSyncer(r.Client, r.scheme, backup, cluster, r.opt, r.recorder),
 		backupSyncer.NewJobSyncer(r.Client, r.scheme, backup, cluster, r.opt),

pkg/controller/mysqlbackup/mysqlbackup_controller_test.go

@@ -316,6 +316,48 @@ var _ = Describe("MysqlBackup controller", func() {
 			Expect(c.Delete(context.TODO(), cluster.Unwrap())).To(Succeed())
 		})
 	})
+
+	When("candidate node is setted to wrong node", func() {
+		BeforeEach(func() {
+			backup.Spec.CandidateNode = cluster.GetPodHostname(3)
+			// create a cluster and a backup
+			Expect(c.Create(context.TODO(), backup.Unwrap())).To(Succeed())
+			Expect(c.Create(context.TODO(), cluster.Unwrap())).To(Succeed())
+
+			// first reconcile request for the backup
+			Eventually(requests, timeout).Should(Receive(Equal(expectedRequest)))
+			testutil.DrainChan(requests)
+		})
+
+		AfterEach(func() {
+			Expect(c.Delete(context.TODO(), backup.Unwrap())).To(Succeed())
+			c.Delete(context.TODO(), cluster.Unwrap())
+		})
+
+		It("should skip creating job", func() {
+			job := &batch.Job{}
+			Expect(c.Get(context.TODO(), jobKey, job)).NotTo(Succeed())
+		})
+
+		When("candidate node update to master", func() {
+			BeforeEach(func() {
+				backup.Spec.CandidateNode = cluster.GetPodHostname(0)
+				// update backup
+				Expect(c.Update(context.TODO(), backup.Unwrap())).To(Succeed())
+
+				// first reconcile request for the backup
+				Eventually(requests, timeout).Should(Receive(Equal(expectedRequest)))
+				testutil.DrainChan(requests)
+			})
+
+			It("should take backup from master", func() {
+				job := &batch.Job{}
+				Expect(c.Get(context.TODO(), jobKey, job)).To(Succeed())
+				Expect(job.Spec.Template.Spec.Containers[0].Args).To(ContainElement(Equal(cluster.GetPodHostname(0))))
+			})
+		})
+
+	})
 })
 
 func refreshFn(c client.Client, backupKey types.NamespacedName) func() *api.MysqlBackup {

pkg/internal/mysqlbackup/validation.go

@@ -0,0 +1,42 @@
+/*
+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.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mysqlbackup
+
+import (
+	"fmt"
+
+	"github.com/bitpoke/mysql-operator/pkg/internal/mysqlcluster"
+)
+
+// Validate checks if the backup spec is validated
+func (c *MysqlBackup) Validate(cluster *mysqlcluster.MysqlCluster) error {
+	// TODO: this validation should be done in an admission web-hook
+
+	if c.Spec.CandidateNode != "" {
+		validate := false
+		for i := 0; i < int(*cluster.Spec.Replicas); i++ {
+			if c.Spec.CandidateNode == cluster.GetPodHostname(i) {
+				validate = true
+				break
+			}
+		}
+		if !validate {
+			return fmt.Errorf("spec.candidateNode is not a valid node")
+		}
+	}
+	return nil
+}