Use context with timeout for SQL reconciliation

Author: AMecea

pkg/controller/node/node_controller.go

@@ -47,6 +47,7 @@ import (
 var log = logf.Log.WithName(controllerName)
 
 const controllerName = "controller.mysqlNode"
+const mysqlReconciliationTimeout = 30 * time.Second
 
 // Add creates a new MysqlCluster Controller and adds it to the Manager with default RBAC. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
@@ -136,8 +137,11 @@ type ReconcileMysqlNode struct {
 // Automatically generate RBAC rules to allow the Controller to read and write Deployments
 // +kubebuilder:rbac:groups=core,resources=pods/status,verbs=get;list;watch;create;update;patch;delete
 func (r *ReconcileMysqlNode) Reconcile(request reconcile.Request) (reconcile.Result, error) {
+	ctx, cancel := context.WithTimeout(context.TODO(), mysqlReconciliationTimeout)
+	defer cancel()
+
 	pod := &corev1.Pod{}
-	err := r.Get(context.TODO(), request.NamespacedName, pod)
+	err := r.Get(ctx, request.NamespacedName, pod)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			// Object not found, return.  Created objects are automatically garbage collected.
@@ -151,7 +155,7 @@ func (r *ReconcileMysqlNode) Reconcile(request reconcile.Request) (reconcile.Res
 	log.Info("syncing MySQL Node", "pod", request.NamespacedName.String())
 
 	var cluster *mysqlcluster.MysqlCluster
-	cluster, err = r.getNodeCluster(pod)
+	cluster, err = r.getNodeCluster(ctx, pod)
 	if err != nil {
 		log.Info("cluster is not found")
 		return reconcile.Result{}, err
@@ -163,20 +167,20 @@ func (r *ReconcileMysqlNode) Reconcile(request reconcile.Request) (reconcile.Res
 	}
 
 	var creds *credentials
-	creds, err = r.getCredsSecret(cluster)
+	creds, err = r.getCredsSecret(ctx, cluster)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
 
 	// initialize SQL interface
 	sql := r.getMySQLConnection(cluster, pod, creds)
 
-	err = r.initializeMySQL(sql, cluster, creds)
+	err = r.initializeMySQL(ctx, sql, cluster, creds)
 	if err != nil {
 		updatePodStatusCondition(pod, mysqlcluster.NodeInitializedConditionType,
 			corev1.ConditionFalse, "mysqlInitializationFailed", err.Error())
 
-		if uErr := r.updatePod(pod); uErr != nil {
+		if uErr := r.updatePod(ctx, pod); uErr != nil {
 			return reconcile.Result{}, uErr
 		}
 
@@ -186,16 +190,16 @@ func (r *ReconcileMysqlNode) Reconcile(request reconcile.Request) (reconcile.Res
 	updatePodStatusCondition(pod, mysqlcluster.NodeInitializedConditionType,
 		corev1.ConditionTrue, "mysqlInitializationSucceeded", "success")
 
-	return reconcile.Result{}, r.updatePod(pod)
+	return reconcile.Result{}, r.updatePod(ctx, pod)
 }
 
-func (r *ReconcileMysqlNode) initializeMySQL(sql SQLInterface, cluster *mysqlcluster.MysqlCluster, c *credentials) error {
+func (r *ReconcileMysqlNode) initializeMySQL(ctx context.Context, sql SQLInterface, cluster *mysqlcluster.MysqlCluster, c *credentials) error {
 	// wait for mysql to be ready
-	if err := sql.Wait(); err != nil {
+	if err := sql.Wait(ctx); err != nil {
 		return err
 	}
 
-	enableSuperReadOnly, err := sql.DisableSuperReadOnly()
+	enableSuperReadOnly, err := sql.DisableSuperReadOnly(ctx)
 	if err != nil {
 		return err
 	}
@@ -204,24 +208,24 @@ func (r *ReconcileMysqlNode) initializeMySQL(sql SQLInterface, cluster *mysqlclu
 	// is slave node?
 	if cluster.GetMasterHost() != sql.Host() {
 		log.Info("configure pod as slave", "pod", sql.Host(), "master", cluster.GetMasterHost())
-		if err := sql.SetPurgedGTID(); err != nil {
+		if err := sql.SetPurgedGTID(ctx); err != nil {
 			return err
 		}
 
-		if err := sql.ChangeMasterTo(cluster.GetMasterHost(), c.ReplicationUser, c.ReplicationPassword); err != nil {
+		if err := sql.ChangeMasterTo(ctx, cluster.GetMasterHost(), c.ReplicationUser, c.ReplicationPassword); err != nil {
 			return err
 		}
 	}
 
-	if err := sql.MarkConfigurationDone(); err != nil {
+	if err := sql.MarkConfigurationDone(ctx); err != nil {
 		return err
 	}
 
 	return nil
 }
 
 // getNodeCluster returns the node related MySQL cluster
-func (r *ReconcileMysqlNode) getNodeCluster(pod *corev1.Pod) (*mysqlcluster.MysqlCluster, error) {
+func (r *ReconcileMysqlNode) getNodeCluster(ctx context.Context, pod *corev1.Pod) (*mysqlcluster.MysqlCluster, error) {
 	re := regexp.MustCompile(`^([\w-]+)-mysql-\d*$`)
 	indexStrs := re.FindStringSubmatch(pod.Name)
 	if len(indexStrs) != 2 {
@@ -233,7 +237,7 @@ func (r *ReconcileMysqlNode) getNodeCluster(pod *corev1.Pod) (*mysqlcluster.Mysq
 		Namespace: pod.Namespace,
 	}
 	cluster := mysqlcluster.New(&api.MysqlCluster{})
-	err := r.Get(context.TODO(), clusterKey, cluster.Unwrap())
+	err := r.Get(ctx, clusterKey, cluster.Unwrap())
 	return cluster, err
 }
 
@@ -257,13 +261,13 @@ type credentials struct {
 	ReplicationPassword string
 }
 
-func (r *ReconcileMysqlNode) getCredsSecret(cluster *mysqlcluster.MysqlCluster) (*credentials, error) {
+func (r *ReconcileMysqlNode) getCredsSecret(ctx context.Context, cluster *mysqlcluster.MysqlCluster) (*credentials, error) {
 	secretKey := types.NamespacedName{
 		Name:      cluster.GetNameForResource(mysqlcluster.Secret),
 		Namespace: cluster.Namespace,
 	}
 	secret := &corev1.Secret{}
-	if err := r.Get(context.TODO(), secretKey, secret); err != nil {
+	if err := r.Get(ctx, secretKey, secret); err != nil {
 		return nil, err
 	}
 
@@ -277,8 +281,8 @@ func (r *ReconcileMysqlNode) getCredsSecret(cluster *mysqlcluster.MysqlCluster)
 	return creds, creds.Validate()
 }
 
-func (r *ReconcileMysqlNode) updatePod(pod *corev1.Pod) error {
-	return r.Status().Update(context.TODO(), pod)
+func (r *ReconcileMysqlNode) updatePod(ctx context.Context, pod *corev1.Pod) error {
+	return r.Status().Update(ctx, pod)
 }
 
 func (c *credentials) Validate() error {
@@ -297,6 +301,7 @@ func anyIsEmpty(ss ...string) bool {
 	return zero
 }
 
+// nolint: unparam
 func updatePodStatusCondition(pod *corev1.Pod, condType corev1.PodConditionType,
 	status corev1.ConditionStatus, reason, msg string) {
 	newCondition := corev1.PodCondition{

pkg/controller/node/sql.go

@@ -17,6 +17,7 @@ limitations under the License.
 package node
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"strings"
@@ -29,19 +30,17 @@ import (
 )
 
 const (
-	// mysqlReadyTries represents the number of tries with 1 second sleep between them to check if MySQL is ready
-	mysqlReadyTries = 10
 	// connRetry represents the number of tries to connect to master server
 	connRetry = 10
 )
 
 // SQLInterface expose abstract operations that can be applied on a MySQL node
 type SQLInterface interface {
-	Wait() error
-	DisableSuperReadOnly() (func(), error)
-	ChangeMasterTo(string, string, string) error
-	MarkConfigurationDone() error
-	SetPurgedGTID() error
+	Wait(ctx context.Context) error
+	DisableSuperReadOnly(ctx context.Context) (func(), error)
+	ChangeMasterTo(ctx context.Context, host string, user string, pass string) error
+	MarkConfigurationDone(ctx context.Context) error
+	SetPurgedGTID(ctx context.Context) error
 	Host() string
 }
 
@@ -62,38 +61,35 @@ func newNodeConn(dsn, host string) SQLInterface {
 	}
 }
 
-func (r *nodeSQLRunner) Wait() error {
+// Wait method pings MySQL until it's ready (runs SELECT 1;). The timeout should be set in ctx context.Context
+func (r *nodeSQLRunner) Wait(ctx context.Context) error {
 	log.V(1).Info("wait for mysql to be ready")
 
-	for i := 0; i < mysqlReadyTries; i++ {
-		if err := r.runQuery("SELECT 1"); err == nil {
-			break
+	for {
+		select {
+		case <-ctx.Done():
+			// timeout expired
+			return fmt.Errorf("timeout: mysql is not ready")
+		case <-time.After(time.Second):
+			if err := r.runQuery(ctx, "SELECT 1"); err == nil {
+				return nil
+			}
 		}
-		// wait a second
-		time.Sleep(1 * time.Second)
 	}
-	if err := r.runQuery("SELECT 1"); err != nil {
-		log.V(1).Info("mysql is not ready", "error", err)
-		return err
-	}
-
-	log.V(1).Info("mysql is ready")
-	return nil
-
 }
 
-func (r *nodeSQLRunner) DisableSuperReadOnly() (func(), error) {
+func (r *nodeSQLRunner) DisableSuperReadOnly(ctx context.Context) (func(), error) {
 	enable := func() {
-		err := r.runQuery("SET GLOBAL SUPER_READ_ONLY = 0;")
+		err := r.runQuery(ctx, "SET GLOBAL SUPER_READ_ONLY = 0;")
 		if err != nil {
 			log.Error(err, "failed to set node super read only", "node", r.Host())
 		}
 	}
-	return enable, r.runQuery("SET GLOBAL READ_ONLY = 1; SET GLOBAL SUPER_READ_ONLY = 0;")
+	return enable, r.runQuery(ctx, "SET GLOBAL READ_ONLY = 1; SET GLOBAL SUPER_READ_ONLY = 0;")
 }
 
 // ChangeMasterTo changes the master host and starts slave.
-func (r *nodeSQLRunner) ChangeMasterTo(masterHost, user, pass string) error {
+func (r *nodeSQLRunner) ChangeMasterTo(ctx context.Context, masterHost, user, pass string) error {
 	// slave node
 	query := `
       STOP SLAVE;
@@ -103,14 +99,14 @@ func (r *nodeSQLRunner) ChangeMasterTo(masterHost, user, pass string) error {
 		MASTER_PASSWORD=?,
 		MASTER_CONNECT_RETRY=?;
 	`
-	if err := r.runQuery(query,
+	if err := r.runQuery(ctx, query,
 		masterHost, user, pass, connRetry,
 	); err != nil {
 		return fmt.Errorf("failed to configure slave node, err: %s", err)
 	}
 
 	query = "START SLAVE;"
-	if err := r.runQuery(query); err != nil {
+	if err := r.runQuery(ctx, query); err != nil {
 		log.Info("failed to start slave in the simple mode, trying a second method")
 		// TODO: https://bugs.mysql.com/bug.php?id=83713
 		query2 := `
@@ -120,14 +116,14 @@ func (r *nodeSQLRunner) ChangeMasterTo(masterHost, user, pass string) error {
 		  reset slave;
 		  start slave;
 		`
-		if err := r.runQuery(query2); err != nil {
+		if err := r.runQuery(ctx, query2); err != nil {
 			return fmt.Errorf("failed to start slave node, err: %s", err)
 		}
 	}
 	return nil
 }
 
-func (r *nodeSQLRunner) MarkConfigurationDone() error {
+func (r *nodeSQLRunner) MarkConfigurationDone(ctx context.Context) error {
 	query := `
     CREATE TABLE IF NOT EXISTS %s.%s (
 		ok tinyint(1) NOT NULL
@@ -137,7 +133,7 @@ func (r *nodeSQLRunner) MarkConfigurationDone() error {
     `
 	query = fmt.Sprintf(query, constants.OperatorDbName, "readiness")
 
-	if err := r.runQuery(query); err != nil {
+	if err := r.runQuery(ctx, query); err != nil {
 		return fmt.Errorf("failed to mark configuration done, err: %s", err)
 	}
 	return nil
@@ -148,7 +144,7 @@ func (r *nodeSQLRunner) Host() string {
 }
 
 // runQuery executes a query
-func (r *nodeSQLRunner) runQuery(q string, args ...interface{}) error {
+func (r *nodeSQLRunner) runQuery(ctx context.Context, q string, args ...interface{}) error {
 	db, close, err := r.dbConn()
 	if err != nil {
 		return err
@@ -160,24 +156,23 @@ func (r *nodeSQLRunner) runQuery(q string, args ...interface{}) error {
 	if !r.enableBinLog {
 		q = "SET @@SESSION.SQL_LOG_BIN = 0;\n" + q
 	}
-	if _, err := db.Exec(q, args...); err != nil {
+	if _, err := db.ExecContext(ctx, q, args...); err != nil {
 		return err
 	}
 
 	return nil
 }
 
 // readFromMysql executes the given query and loads the values into give variables
-func (r *nodeSQLRunner) readFromMysql(query string, values ...interface{}) error {
+func (r *nodeSQLRunner) readFromMysql(ctx context.Context, query string, values ...interface{}) error {
 	db, close, err := r.dbConn()
 	if err != nil {
 		return err
 	}
 	defer close()
 
-	// nolint: gosec
 	log.V(1).Info("running query", "query", query)
-	row := db.QueryRow(query)
+	row := db.QueryRowContext(ctx, query)
 	if row == nil {
 		return fmt.Errorf("no row found")
 	}
@@ -206,14 +201,14 @@ func (r *nodeSQLRunner) dbConn() (*sql.DB, func(), error) {
 	return db, close, nil
 }
 
-func (r *nodeSQLRunner) SetPurgedGTID() error {
+func (r *nodeSQLRunner) SetPurgedGTID(ctx context.Context) error {
 	// first check if the GTID should be set, if the table exists or if the GTID was set before (used)
 	// nolint: gosec
 	qq := fmt.Sprintf("SELECT used FROM %[1]s.%[2]s WHERE id=1",
 		constants.OperatorDbName, constants.OperatorGtidsTableName)
 
 	var used bool
-	if err := r.readFromMysql(qq, &used); err != nil {
+	if err := r.readFromMysql(ctx, qq, &used); err != nil {
 		// if it's a: "Table doesn't exist" error then GTID should not be set, it's a master case.
 		if isMySQLError(err, 1146) {
 			log.V(1).Info("GTID purged table does not exists", "host", r.Host())
@@ -239,7 +234,7 @@ func (r *nodeSQLRunner) SetPurgedGTID() error {
       COMMIT;
     `, constants.OperatorDbName, constants.OperatorGtidsTableName)
 
-	if err := r.runQuery(query); err != nil {
+	if err := r.runQuery(ctx, query); err != nil {
 		return err
 	}
 

pkg/controller/node/sql_fake_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package node
 
 import (
+	"context"
 	"fmt"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -27,27 +28,27 @@ type fakeSQLRunner struct{}
 // test if fakeer implements interface
 var _ SQLInterface = &fakeSQLRunner{}
 
-func (f *fakeSQLRunner) Wait() error {
+func (f *fakeSQLRunner) Wait(ctx context.Context) error {
 	return nil
 }
 
-func (f *fakeSQLRunner) DisableSuperReadOnly() (func(), error) {
+func (f *fakeSQLRunner) DisableSuperReadOnly(ctx context.Context) (func(), error) {
 	return func() {}, nil
 }
 
-func (f *fakeSQLRunner) ChangeMasterTo(host, user, pass string) error {
+func (f *fakeSQLRunner) ChangeMasterTo(ctx context.Context, host, user, pass string) error {
 	return nil
 }
 
-func (f *fakeSQLRunner) MarkConfigurationDone() error {
+func (f *fakeSQLRunner) MarkConfigurationDone(ctx context.Context) error {
 	return nil
 }
 
 func (f *fakeSQLRunner) Host() string {
 	return ""
 }
 
-func (f *fakeSQLRunner) SetPurgedGTID() error {
+func (f *fakeSQLRunner) SetPurgedGTID(ctx context.Context) error {
 	return nil
 }