Node controller - for configuring MySQL nodes

pkg/controller/add_nodecontroller.go

@@ -0,0 +1,24 @@
+/*
+Copyright 2019 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 controller
+
+import "github.com/presslabs/mysql-operator/pkg/controller/node"
+
+func init() {
+	// AddToManagerFuncs is a list of functions to create controllers and add them to a manager.
+	AddToManagerFuncs = append(AddToManagerFuncs, node.Add)
+}

pkg/controller/mysqlcluster/internal/syncer/statefullset.go

@@ -41,6 +41,7 @@ import (
 const (
 	confVolumeName    = "conf"
 	confMapVolumeName = "config-map"
+	initDBVolumeName  = "init-scripts"
 	dataVolumeName    = "data"
 )
 
@@ -133,11 +134,12 @@ func (s *sfsSyncer) ensurePodSpec() core.PodSpec {
 		PriorityClassName:  s.cluster.Spec.PodSpec.PriorityClassName,
 		Tolerations:        s.cluster.Spec.PodSpec.Tolerations,
 		ServiceAccountName: s.cluster.Spec.PodSpec.ServiceAccountName,
-		ReadinessGates: []core.PodReadinessGate{
-			{
-				ConditionType: mysqlcluster.NodeInitializedConditionType,
-			},
-		},
+		// TODO: uncomment this when limiting operator for k8s version > 1.13
+		// ReadinessGates: []core.PodReadinessGate{
+		// 	{
+		// 		ConditionType: mysqlcluster.NodeInitializedConditionType,
+		// 	},
+		// },
 	}
 }
 
@@ -408,6 +410,10 @@ func (s *sfsSyncer) ensureVolumes() []core.Volume {
 			EmptyDir: &core.EmptyDirVolumeSource{},
 		}),
 
+		ensureVolume(initDBVolumeName, core.VolumeSource{
+			EmptyDir: &core.EmptyDirVolumeSource{},
+		}),
+
 		ensureVolume(confMapVolumeName, core.VolumeSource{
 			ConfigMap: &core.ConfigMapVolumeSource{
 				LocalObjectReference: core.LocalObjectReference{
@@ -492,6 +498,10 @@ func (s *sfsSyncer) getVolumeMountsFor(name string) []core.VolumeMount {
 				Name:      confMapVolumeName,
 				MountPath: ConfMapVolumeMountPath,
 			},
+			{
+				Name:      initDBVolumeName,
+				MountPath: constants.InitDBVolumeMountPath,
+			},
 		}
 
 	case containerCloneName, containerMysqlName, containerSidecarName:
@@ -504,6 +514,10 @@ func (s *sfsSyncer) getVolumeMountsFor(name string) []core.VolumeMount {
 				Name:      dataVolumeName,
 				MountPath: DataVolumeMountPath,
 			},
+			{
+				Name:      initDBVolumeName,
+				MountPath: constants.InitDBVolumeMountPath,
+			},
 		}
 
 	case containerHeartBeatName, containerKillerName:

pkg/controller/node/node_controller.go

@@ -71,11 +71,12 @@ func isOwnedByMySQL(meta metav1.Object) bool {
 		return false
 	}
 
-	for _, o := range meta.GetOwnerReferences() {
-		if o.Kind == "MysqlCluster" {
-			return true
-		}
+	// TODO: add more checks here
+	labels := meta.GetLabels()
+	if val, ok := labels["app.kubernetes.io/managed-by"]; ok {
+		return val == "mysql.presslabs.org"
 	}
+
 	return false
 }
 
@@ -142,19 +143,20 @@ func (r *ReconcileMysqlNode) Reconcile(request reconcile.Request) (reconcile.Res
 
 	cluster, err := r.getNodeCluster(pod)
 	if err != nil {
-		return reconcile.Result{}, nil
+		return reconcile.Result{}, err
 	}
 
 	var sql *nodeSQLRunner
 	sql, err = r.getMySQLConnection(cluster, pod)
 	if err != nil {
-		return reconcile.Result{}, nil
+		return reconcile.Result{}, err
 	}
 
 	err = r.initializeMySQL(sql, cluster)
 	if err != nil {
 		updatePodStatusCondition(pod, mysqlcluster.NodeInitializedConditionType,
 			corev1.ConditionFalse, "mysqlInitializationFailed", err.Error())
+
 		if uErr := r.updatePod(pod); uErr != nil {
 			return reconcile.Result{}, uErr
 		}
@@ -174,16 +176,24 @@ func (r *ReconcileMysqlNode) initializeMySQL(sql *nodeSQLRunner, cluster *mysqlc
 		return err
 	}
 
+	if err := sql.DisableSuperReadOnly(); err != nil {
+		return err
+	}
+
 	// is slave node
 	if cluster.GetMasterHost() != sql.Host() {
 		log.Info("configure pod as slave", "pod", sql.Host(), "master", cluster.GetMasterHost())
-		if err := sql.ConfigureSlaveNode(cluster.GetMasterHost()); err != nil {
+		if err := sql.SetGtidPurged(); err != nil {
 			return err
 		}
-		// TODO: check if node should be set read-only or not (super)
-		if err := sql.SetReadOnly(); err != nil {
+
+		if err := sql.ChangeMasterTo(cluster.GetMasterHost()); err != nil {
 			return err
 		}
+		// TODO: check if node should be set read-only or not (super)
+		// if err := sql.SetReadOnly(); err != nil {
+		// 	return err
+		// }
 	}
 
 	if err := sql.MarkConfigurationDone(); err != nil {
@@ -227,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)
+	err = r.Get(context.TODO(), clusterKey, cluster.Unwrap())
 	return cluster, err
 }
 
@@ -257,7 +267,7 @@ func (r *ReconcileMysqlNode) getMySQLConnection(cluster *mysqlcluster.MysqlClust
 		user, pass, host, constants.MysqlPort,
 	)
 
-	return newNodeConn(dsn, pod.Spec.Hostname, repU, repP), nil
+	return newNodeConn(dsn, host, repU, repP), nil
 }
 
 func (r *ReconcileMysqlNode) updatePod(pod *corev1.Pod) error {