Pod syncers no found error should be skiped, add some tests, fixes #160

Author: AMecea

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

@@ -0,0 +1,57 @@
+/*
+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 mysqlcluster
+
+import "fmt"
+
+// SyncErrCode is the error code
+type SyncErrCode int
+
+const (
+	// PodNotFound represents the error when pod is not found
+	PodNotFound SyncErrCode = iota
+)
+
+// SyncError is the error type for pod syncer
+type SyncError struct {
+	syncer  string
+	code    SyncErrCode
+	details string
+}
+
+func (e *SyncError) Error() string {
+	return fmt.Sprintf("%s(%d: %s)", e.syncer, e.code, e.details)
+}
+
+// NewError returns a syncer error
+func NewError(code SyncErrCode, syncer, details string) error {
+	return &SyncError{
+		syncer:  syncer,
+		code:    code,
+		details: details,
+	}
+}
+
+// IsError check if a error is of given code
+func IsError(err error, code SyncErrCode) bool {
+	switch t := err.(type) {
+	default:
+		return false
+	case *SyncError:
+		return t.code == code
+	}
+}

…luster/internal/syncer/healty_service.go …uster/internal/syncer/healthy_service.gopkg/controller/mysqlcluster/internal/syncer/healty_service.go renamed to pkg/controller/mysqlcluster/internal/syncer/healthy_service.go pkg/controller/mysqlcluster/internal/syncer/healty_service.go renamed to pkg/controller/mysqlcluster/internal/syncer/healthy_service.go

@@ -40,7 +40,7 @@ func NewHealthySVCSyncer(c client.Client, scheme *runtime.Scheme, cluster *mysql
 
 		out.Spec.Type = "ClusterIP"
 		out.Spec.Selector = cluster.GetLabels()
-		out.Spec.Selector["healty"] = "yes"
+		out.Spec.Selector["healthy"] = "yes"
 
 		if len(out.Spec.Ports) != 1 {
 			out.Spec.Ports = make([]core.ServicePort, 1)

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

@@ -65,7 +65,7 @@ func (s *podSyncer) SyncFn(in runtime.Object) error {
 
 	// raise error if pod is not created
 	if out.CreationTimestamp.IsZero() {
-		return fmt.Errorf("pod is not created")
+		return NewError(PodNotFound, "PodSyncer", "pod is not found")
 	}
 
 	master := s.cluster.GetNodeCondition(s.hostname, api.NodeConditionMaster)
@@ -86,18 +86,18 @@ func (s *podSyncer) SyncFn(in runtime.Object) error {
 		role = labelMaster
 	}
 
-	// set healty label
-	healty := labelNotHealty
+	// set healthy label
+	healthy := labelNotHealty
 	if isMaster || !isMaster && isReplicating && !isLagged {
-		healty = labelHealty
+		healthy = labelHealty
 	}
 
 	if len(out.ObjectMeta.Labels) == 0 {
 		out.ObjectMeta.Labels = map[string]string{}
 	}
 
 	out.ObjectMeta.Labels["role"] = role
-	out.ObjectMeta.Labels["healty"] = healty
+	out.ObjectMeta.Labels["healthy"] = healthy
 
 	return nil
 }

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

@@ -107,7 +107,7 @@ var _ = Describe("Pod syncer", func() {
 		Expect(err).ToNot(Succeed())
 	})
 
-	It("should mark pods as healty", func() {
+	It("should mark pods as healthy", func() {
 		cluster.UpdateNodeConditionStatus(cluster.GetPodHostname(1), api.NodeConditionLagged, core.ConditionFalse)
 		cluster.UpdateNodeConditionStatus(cluster.GetPodHostname(1), api.NodeConditionReplicating, core.ConditionTrue)
 		// update cluster

pkg/controller/mysqlcluster/mysqlcluster_controller.go

@@ -188,22 +188,31 @@ func (r *ReconcileMysqlCluster) Reconcile(request reconcile.Request) (reconcile.
 		syncers = append(syncers, clustersyncer.NewPDBSyncer(r.Client, r.scheme, cluster))
 	}
 
-	syncers = append(syncers, r.getPodSyncers(cluster)...)
-
-	// add pods syncers for every node status
+	// run the syncers
 	for _, sync := range syncers {
 		if err = syncer.Sync(context.TODO(), sync, r.recorder); err != nil {
 			return reconcile.Result{}, err
 		}
 	}
 
+	// run the pod syncers
+	log.Info("cluster status", "status", cluster.Status)
+	for _, sync := range r.getPodSyncers(cluster) {
+		if err = syncer.Sync(context.TODO(), sync, r.recorder); err != nil {
+			// if it's pod not found then skip the error
+			if !clustersyncer.IsError(err, clustersyncer.PodNotFound) {
+				return reconcile.Result{}, err
+			}
+		}
+	}
+
 	// Perform any cleanup
 	pvcCleaner := cleaner.NewPvcCleaner(cluster, r.opt)
 	err = pvcCleaner.Run(context.TODO(), r.Client, r.scheme, r.recorder)
-
 	if err != nil {
 		return reconcile.Result{}, err
 	}
+
 	return reconcile.Result{}, nil
 }
 

pkg/controller/mysqlcluster/mysqlcluster_controller_suite_test.go

@@ -28,8 +28,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
 
 	"github.com/presslabs/mysql-operator/pkg/apis"
+	"github.com/presslabs/mysql-operator/pkg/controller/internal/testutil"
 )
 
 var cfg *rest.Config
@@ -43,6 +45,8 @@ func TestMysqlClusterController(t *testing.T) {
 var _ = BeforeSuite(func() {
 	var err error
 
+	logf.SetLogger(testutil.NewTestLogger(GinkgoWriter))
+
 	t = &envtest.Environment{
 		CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "config", "crds")},
 	}

pkg/controller/mysqlcluster/mysqlcluster_controller_test.go

@@ -13,7 +13,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// nolint: errcheck
+// nolint: errcheck, unparam
 package mysqlcluster
 
 import (
@@ -81,6 +81,7 @@ var _ = Describe("MysqlCluster controller", func() {
 		var (
 			expectedRequest reconcile.Request
 			cluster         *mysqlcluster.MysqlCluster
+			clusterKey      types.NamespacedName
 			secret          *corev1.Secret
 			components      clusterComponents
 		)
@@ -107,6 +108,10 @@ var _ = Describe("MysqlCluster controller", func() {
 					SecretName: secret.Name,
 				},
 			})
+			clusterKey = types.NamespacedName{
+				Name:      cluster.Name,
+				Namespace: cluster.Namespace,
+			}
 
 			components = []runtime.Object{
 				&appsv1.StatefulSet{
@@ -147,6 +152,7 @@ var _ = Describe("MysqlCluster controller", func() {
 				},
 			}
 
+			By("create the MySQL cluster")
 			Expect(c.Create(context.TODO(), secret)).To(Succeed())
 			Expect(c.Create(context.TODO(), cluster.Unwrap())).To(Succeed())
 
@@ -169,6 +175,7 @@ var _ = Describe("MysqlCluster controller", func() {
 
 			// We need to make sure that the controller does not create infinite
 			// loops
+			By("wait for no more reconcile requests")
 			Consistently(requests).ShouldNot(Receive(Equal(expectedRequest)))
 		})
 
@@ -247,6 +254,92 @@ var _ = Describe("MysqlCluster controller", func() {
 				Eventually(requests, timeout).Should(Receive(Equal(expectedRequest)))
 				Eventually(getClusterConditions(c, cluster), timeout).Should(haveCondWithStatus(api.ClusterConditionReady, corev1.ConditionTrue))
 			})
+			It("should label pods as healthy and as master accordingly", func() {
+				pod0 := getPod(cluster, 0)
+				Expect(c.Create(context.TODO(), pod0)).To(Succeed())
+				pod1 := getPod(cluster, 1)
+				Expect(c.Create(context.TODO(), pod1)).To(Succeed())
+				pod2 := getPod(cluster, 2)
+				Expect(c.Create(context.TODO(), pod2)).To(Succeed())
+
+				// update cluster conditions
+				By("update cluster status")
+				Expect(c.Get(context.TODO(), clusterKey, cluster.Unwrap())).To(Succeed())
+				cluster.Status.Nodes = []api.NodeStatus{
+					nodeStatusForPod(cluster, pod0, true, false, false),
+					nodeStatusForPod(cluster, pod1, false, false, true),
+					nodeStatusForPod(cluster, pod2, false, false, false),
+				}
+				Expect(c.Status().Update(context.TODO(), cluster.Unwrap())).To(Succeed())
+
+				Eventually(requests, timeout).Should(Receive(Equal(expectedRequest)))
+
+				// assert pods labels
+				// master
+				Expect(c.Get(context.TODO(), objToKey(pod0), pod0)).To(Succeed())
+				Expect(pod0).To(haveLabelWithValue("role", "master"))
+				Expect(pod0).To(haveLabelWithValue("healthy", "yes"))
+
+				// replica
+				Expect(c.Get(context.TODO(), objToKey(pod1), pod1)).To(Succeed())
+				Expect(pod1).To(haveLabelWithValue("role", "replica"))
+				Expect(pod1).To(haveLabelWithValue("healthy", "yes"))
+			})
+			It("should label pods as master eaven if pods does not exists", func() {
+				pod0 := getPod(cluster, 0)
+				Expect(c.Create(context.TODO(), pod0)).To(Succeed())
+				pod1 := getPod(cluster, 1)
+				pod2 := getPod(cluster, 2)
+
+				// update cluster conditions
+				By("update cluster status")
+				Expect(c.Get(context.TODO(), clusterKey, cluster.Unwrap())).To(Succeed())
+				cluster.Status.Nodes = []api.NodeStatus{
+					nodeStatusForPod(cluster, pod0, true, false, false),
+					nodeStatusForPod(cluster, pod1, false, false, false),
+					nodeStatusForPod(cluster, pod2, false, false, false),
+				}
+				Expect(c.Status().Update(context.TODO(), cluster.Unwrap())).To(Succeed())
+
+				Eventually(requests, timeout).Should(Receive(Equal(expectedRequest)))
+
+				// assert pods labels
+				// master
+				Expect(c.Get(context.TODO(), objToKey(pod0), pod0)).To(Succeed())
+				Expect(pod0).To(haveLabelWithValue("role", "master"))
+				Expect(pod0).To(haveLabelWithValue("healthy", "yes"))
+
+				// check pod is not created
+				Expect(c.Get(context.TODO(), objToKey(pod1), pod1)).ToNot(Succeed())
+			})
+			It("should label as unhealthy if lagged", func() {
+				pod0 := getPod(cluster, 0)
+				Expect(c.Create(context.TODO(), pod0)).To(Succeed())
+				pod1 := getPod(cluster, 1)
+				Expect(c.Create(context.TODO(), pod1)).To(Succeed())
+
+				// update cluster conditions
+				By("update cluster status")
+				Expect(c.Get(context.TODO(), clusterKey, cluster.Unwrap())).To(Succeed())
+				cluster.Status.Nodes = []api.NodeStatus{
+					nodeStatusForPod(cluster, pod0, false, true, false),
+					nodeStatusForPod(cluster, pod1, true, false, true),
+				}
+				Expect(c.Status().Update(context.TODO(), cluster.Unwrap())).To(Succeed())
+
+				Eventually(requests, timeout).Should(Receive(Equal(expectedRequest)))
+
+				// assert pods labels
+				// master
+				Expect(c.Get(context.TODO(), objToKey(pod0), pod0)).To(Succeed())
+				Expect(pod0).To(haveLabelWithValue("role", "replica"))
+				Expect(pod0).To(haveLabelWithValue("healthy", "no"))
+
+				// replica
+				Expect(c.Get(context.TODO(), objToKey(pod1), pod1)).To(Succeed())
+				Expect(pod1).To(haveLabelWithValue("role", "master"))
+				Expect(pod1).To(haveLabelWithValue("healthy", "yes"))
+			})
 		})
 	})
 })
@@ -257,6 +350,65 @@ func removeAllCreatedResource(c client.Client, clusterComps []runtime.Object) {
 	}
 }
 
+func objToKey(o runtime.Object) types.NamespacedName {
+	obj, _ := o.(*corev1.Pod)
+	return types.NamespacedName{
+		Name:      obj.Name,
+		Namespace: obj.Namespace,
+	}
+}
+
+func getPod(cluster *mysqlcluster.MysqlCluster, index int) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-%d", cluster.GetNameForResource(mysqlcluster.StatefulSet), index),
+			Namespace: cluster.Namespace,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				corev1.Container{
+					Name:  "dummy",
+					Image: "dummy",
+				},
+			},
+		},
+	}
+}
+
+func nodeStatusForPod(cluster *mysqlcluster.MysqlCluster, pod *corev1.Pod, master, lagged, replicating bool) api.NodeStatus {
+	name := fmt.Sprintf("%s.%s.%s", pod.Name, cluster.GetNameForResource(mysqlcluster.HeadlessSVC), pod.Namespace)
+
+	boolToStatus := func(c bool) corev1.ConditionStatus {
+		if c {
+			return corev1.ConditionTrue
+		}
+		return corev1.ConditionFalse
+	}
+
+	t := time.Now()
+
+	return api.NodeStatus{
+		Name: name,
+		Conditions: []api.NodeCondition{
+			api.NodeCondition{
+				Type:               api.NodeConditionMaster,
+				Status:             boolToStatus(master),
+				LastTransitionTime: metav1.NewTime(t),
+			},
+			api.NodeCondition{
+				Type:               api.NodeConditionLagged,
+				Status:             boolToStatus(lagged),
+				LastTransitionTime: metav1.NewTime(t),
+			},
+			api.NodeCondition{
+				Type:               api.NodeConditionReplicating,
+				Status:             boolToStatus(replicating),
+				LastTransitionTime: metav1.NewTime(t),
+			},
+		},
+	}
+}
+
 // getClusterConditions is a helper func that returns a functions that returns cluster status conditions
 func getClusterConditions(c client.Client, cluster *mysqlcluster.MysqlCluster) func() []api.ClusterCondition {
 	return func() []api.ClusterCondition {
@@ -273,3 +425,11 @@ func haveCondWithStatus(condType api.ClusterConditionType, status corev1.Conditi
 		"Status": Equal(status),
 	}))
 }
+
+func haveLabelWithValue(label, value string) gomegatypes.GomegaMatcher {
+	return PointTo(MatchFields(IgnoreExtras, Fields{
+		"ObjectMeta": MatchFields(IgnoreExtras, Fields{
+			"Labels": HaveKeyWithValue(label, value),
+		}),
+	}))
+}

pkg/controller/orchestrator/orchestrator_reconcile.go

@@ -295,7 +295,7 @@ func (ou *orcUpdater) getRecoveriesToAck(recoveries []orc.TopologyRecovery) []or
 }
 
 func (ou *orcUpdater) acknowledgeRecoveries(toAck []orc.TopologyRecovery) error {
-	comment := fmt.Sprintf("Statefulset '%s' is healty for more then %d seconds",
+	comment := fmt.Sprintf("Statefulset '%s' is healthy for more then %d seconds",
 		ou.cluster.GetNameForResource(mysqlcluster.StatefulSet), recoveryGraceTime,
 	)