Fix upgrade, don't set GTID_PURGED if was an upgrade

Author: AMecea

docs/operator-upgrades.md

@@ -41,6 +41,7 @@ field) to make it smaller that will prevent you from hitting this
 
 
 The operator should do all the work for you but you have to make sure that you have the latest
-`0.2.x` version of the operator.
+`0.2.x` version of the operator. For a smooth upgrade, it's recommended to have clusters with only
+one node or the master node to be node 0. Doing so will prevent the operator to wait for a failover.
 
 This release drop support for `emptyDir` volume source.

hack/upgrades-tests/test_upgrade_v0.3.0.sh

@@ -156,60 +156,18 @@ function wait_cluster_ready {
 
 }
 
-function read_endpoint {
-    name=${1:-my-cluster}
+function run_query {
+    pod=${1}-mysql-${2}
+    query=${3}
 
-    echo "${name}-mysql.${CL_NAMESPACE}"
+    kubectl exec $pod -- mysql --defaults-file=/etc/mysql/client.conf -NB -e "$query"
 }
 
-function start_check_job {
-    name=${1:-my-cluster}
-    root_pass=${2:-not-so-secure}
+function run_query_old {
+    pod=${1}-mysql-${2}
+    query=${3}
 
-    msg "Create check job ($name) ..."
-    cat <<EOF | kubectl apply -f -
-apiVersion: batch/v1
-kind: Job
-metadata:
-  name: check-mysql-cluster
-spec:
-  backoffLimit: 0
-  template:
-    spec:
-      restartPolicy: Never
-      containers:
-      - name: check
-        image: percona:5.7
-        command: ["/bin/bash", "-c"]
-        args:
-          - |
-            set -e
-            while true; do
-            mysql --host=$(read_endpoint ${name}) --user=root --password=${root_pass} -e 'SELECT 1'
-            sleep 5
-            done
-EOF
-}
-
-function check_job_ok {
-    JSONPATH='{range @.status.conditions[*]}{@.type}={@.status};{end}'
-
-    msg "Check job status ..."
-    sleep 10
-    out=$(kubectl get jobs check-mysql-cluster -o jsonpath="$JSONPATH")
-    if [[ "$out" == *"Failed=True"* ]]; then
-        msg "Cluser was down!!" error
-        return 1
-    else
-        msg "Cluster was ok!!" success
-    fi
-}
-
-function apply_new_crds {
-    file=${1:-02x-crds.yaml}
-
-    msg "Apply new crds ($file)..."
-    kubectl apply -f ${file}
+    kubectl exec $pod -- mysql --defaults-file=/etc/mysql/client.cnf -NB -e "$query"
 }
 
 
@@ -259,12 +217,8 @@ function cmd_test {
     # wait for cluster to be ready
     wait_cluster_ready $CL_NAME
 
-    # start a job that checks for mysql to be up
-    # start_check_job $CL_NAME
-
-    # check job
-    # check_job_ok
-    # [ $? -ne 0 ] && exit 1
+    run_query_old $CL_NAME 0 "CREATE DATABASE IF NOT EXISTS test; CREATE TABLE IF NOT EXISTS test.test (name varchar(10) PRIMARY KEY)"
+    run_query_old $CL_NAME 0 "INSERT INTO test.test VALUES ('test1')"
 
     # trigger failover
     kubectl delete pod $CL_NAME-mysql-0
@@ -284,10 +238,12 @@ function cmd_test {
 
     check_cluster_version $CL_NAME 300
 
-    # check the job status to determine if the cluster was down
-    # check_job_ok
-    # [ $? -ne 0 ] && exit 1
-
+    current=$(run_query $CL_NAME 0 "SELECT 'ok' FROM test.test WHERE name='test1'")
+    if [ "$current" == "ok" ]; then
+        msg "Ok sql test" success
+    else
+        msg "Bad sql test" error
+    fi
 }
 
 

pkg/controller/mysqlcluster/internal/upgrades/upgrades.go

@@ -78,11 +78,6 @@ func (u *upgrader) Run(ctx context.Context) error {
 	// in maintenance except node 0.
 	// TODO: or set promotion rules
 	if int(*sts.Spec.Replicas) > 1 {
-
-		if err = u.setNodesInMaintenaceExcept(insts, 0); err != nil {
-			return err
-		}
-
 		one := int32(1)
 		sts.Spec.Replicas = &one
 		if err = u.client.Update(ctx, sts); err != nil {
@@ -221,35 +216,6 @@ func (u *upgrader) getPodOldHostname(node int) string {
 		u.cluster.Namespace)
 }
 
-func (u *upgrader) setNodesInMaintenaceExcept(insts []orc.Instance, node int) error {
-	maintenances, err := u.orcClient.Maintenance()
-	if err != nil {
-		return err
-	}
-
-	for _, inst := range insts {
-		if inst.Key.Hostname == u.getPodOldHostname(node) {
-			// execept given node
-			continue
-		}
-
-		inMaintenance := false
-		for _, m := range maintenances {
-			if m.Key.Hostname == inst.Key.Hostname {
-				inMaintenance = true
-			}
-		}
-
-		if !inMaintenance {
-			if err := u.orcClient.BeginMaintenance(inst.Key, "upgrader", "upgrade"); err != nil {
-				return err
-			}
-		}
-	}
-
-	return nil
-}
-
 func (u *upgrader) forgetFromOrc() error {
 	for node := 0; node < int(*u.cluster.Spec.Replicas); node++ {
 		if err := u.orcClient.Forget(u.getPodOldHostname(node), 3306); err != nil {

pkg/controller/node/node_controller.go

@@ -171,7 +171,10 @@ func (r *ReconcileMysqlNode) Reconcile(request reconcile.Request) (reconcile.Res
 
 	// if it's a old version cluster then don't do anything
 	if shouldUpdateToVersion(cluster, 300) {
-		return reconcile.Result{}, nil
+		// if the cluster is upgraded then set on the cluster an annotations that skips the GTID configuration
+		// TODO: this should be removed in the next versions
+		cluster.Annotations["mysql.presslabs.org/SkipGTIDPurged"] = "true"
+		return reconcile.Result{}, r.Update(ctx, cluster.Unwrap())
 	}
 
 	// get cluster credentials from k8s secret, like replication and operator credentials
@@ -209,6 +212,15 @@ func (r *ReconcileMysqlNode) initializeMySQL(ctx context.Context, sql SQLInterfa
 		return err
 	}
 
+	// check if MySQL was configured before to avoid multiple times reconfiguration
+	if configured, err := sql.IsConfigured(ctx); err != nil {
+		return err
+	} else if configured {
+		// already configured. For example this can be reached if the pod status update fails
+		log.V(1).Info("MySQL is already configure - skip")
+		return nil
+	}
+
 	// disable MySQL SUPER readonly to be able to modify settings in MySQL
 	enableSuperReadOnly, err := sql.DisableSuperReadOnly(ctx)
 	if err != nil {
@@ -219,8 +231,12 @@ func (r *ReconcileMysqlNode) initializeMySQL(ctx context.Context, sql SQLInterfa
 	// is slave node?
 	if cluster.GetMasterHost() != sql.Host() {
 		log.Info("configure pod as slave", "pod", sql.Host(), "master", cluster.GetMasterHost())
-		if err := sql.SetPurgedGTID(ctx); err != nil {
-			return err
+
+		// check if the skip annotation is set on the cluster first
+		if _, ok := cluster.Annotations["mysql.presslabs.org/SkipGTIDPurged"]; !ok {
+			if err := sql.SetPurgedGTID(ctx); err != nil {
+				return err
+			}
 		}
 
 		if err := sql.ChangeMasterTo(ctx, cluster.GetMasterHost(), c.ReplicationUser, c.ReplicationPassword); err != nil {

pkg/controller/node/sql.go

@@ -40,6 +40,7 @@ type SQLInterface interface {
 	DisableSuperReadOnly(ctx context.Context) (func(), error)
 	ChangeMasterTo(ctx context.Context, host string, user string, pass string) error
 	MarkConfigurationDone(ctx context.Context) error
+	IsConfigured(ctx context.Context) (bool, error)
 	SetPurgedGTID(ctx context.Context) error
 	Host() string
 }
@@ -125,14 +126,13 @@ func (r *nodeSQLRunner) ChangeMasterTo(ctx context.Context, masterHost, user, pa
 
 // MarkConfigurationDone write in a MEMORY table value. The readiness probe checks for that value to exist to succeed.
 func (r *nodeSQLRunner) MarkConfigurationDone(ctx context.Context) error {
-	// nolint: gosec
-	query := fmt.Sprintf("REPLACE INTO %s.%s VALUES ('%s', '1');",
-		constants.OperatorDbName, constants.OperatorStatusTableName, "configured")
+	return r.writeStatusValue(ctx, "configured", "1")
+}
 
-	if err := r.runQuery(ctx, query); err != nil {
-		return fmt.Errorf("failed to mark configuration done, err: %s", err)
-	}
-	return nil
+// IsConfigured returns true if MySQL was configured, a key was set in the status table
+func (r *nodeSQLRunner) IsConfigured(ctx context.Context) (bool, error) {
+	val, err := r.readStatusValue(ctx, "configured")
+	return val == "1", err
 }
 
 func (r *nodeSQLRunner) Host() string {
@@ -239,6 +239,7 @@ func (r *nodeSQLRunner) SetPurgedGTID(ctx context.Context) error {
 	return nil
 }
 
+// readStatusValue read from status table the value under the given key
 func (r *nodeSQLRunner) readStatusValue(ctx context.Context, key string) (string, error) {
 	// nolint: gosec
 	qq := fmt.Sprintf("SELECT value FROM %s.%s WHERE name='%s'",
@@ -254,6 +255,19 @@ func (r *nodeSQLRunner) readStatusValue(ctx context.Context, key string) (string
 	return value, nil
 }
 
+// writeStatusValue updates the value at the provided key
+func (r *nodeSQLRunner) writeStatusValue(ctx context.Context, key, value string) error {
+	// nolint: gosec
+	query := fmt.Sprintf("REPLACE INTO %s.%s VALUES ('%s', '%s');",
+		constants.OperatorDbName, constants.OperatorStatusTableName, key, value)
+
+	if err := r.runQuery(ctx, query); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // isMySQLError checks if a mysql error is of the given code.
 // more information about mysql error codes can be found here:
 // https://dev.mysql.com/doc/refman/8.0/en/server-error-reference.html

pkg/controller/node/sql_fake_test.go

@@ -52,6 +52,10 @@ func (f *fakeSQLRunner) SetPurgedGTID(ctx context.Context) error {
 	return nil
 }
 
+func (f *fakeSQLRunner) IsConfigured(ctx context.Context) (bool, error) {
+	return false, nil
+}
+
 var _ = Describe("SQL functions", func() {
 	It("should find not found error", func() {
 		err := fmt.Errorf("Error 1146: Table 'a.a' doesn't exist")