K8SPS-301: reduce error logs on pod deletion (#659)

* K8SPS-301: reduce error logs on pod deletion

https://perconadev.atlassian.net/browse/K8SPS-301

* fix test

* fix merge

* fix regression

* get ready pods for orchestrator

---------

Co-authored-by: Viacheslav Sarzhan <[email protected]>

Author: pooknull

pkg/controller/ps/controller.go

@@ -188,7 +188,7 @@ func (r *PerconaServerMySQLReconciler) deleteMySQLPods(ctx context.Context, cr *
 	}
 
 	if cr.Spec.MySQL.IsAsync() {
-		orcPod, err := getOrcPod(ctx, r.Client, cr, 0)
+		orcPod, err := getReadyOrcPod(ctx, r.Client, cr)
 		if err != nil {
 			return nil
 		}
@@ -204,10 +204,9 @@ func (r *PerconaServerMySQLReconciler) deleteMySQLPods(ctx context.Context, cr *
 			return errors.Wrap(err, "get operator password")
 		}
 
-		firstPodFQDN := fmt.Sprintf("%s.%s.%s", firstPod.Name, mysql.ServiceName(cr), cr.Namespace)
-		firstPodUri := fmt.Sprintf("%s:%s@%s", apiv1alpha1.UserOperator, operatorPass, firstPodFQDN)
+		firstPodUri := fmt.Sprintf("%s:%s@%s", apiv1alpha1.UserOperator, operatorPass, mysql.PodFQDN(cr, &firstPod))
 
-		um := database.NewReplicationManager(&firstPod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, firstPodFQDN)
+		um := database.NewReplicationManager(&firstPod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, mysql.PodFQDN(cr, &firstPod))
 
 		mysh, err := mysqlsh.NewWithExec(r.ClientCmd, &firstPod, firstPodUri)
 		if err != nil {
@@ -221,11 +220,11 @@ func (r *PerconaServerMySQLReconciler) deleteMySQLPods(ctx context.Context, cr *
 
 		log.Info("Got primary", "primary", clusterStatus.DefaultReplicaSet.Primary)
 
-		if !strings.HasPrefix(clusterStatus.DefaultReplicaSet.Primary, firstPodFQDN) {
+		if !strings.HasPrefix(clusterStatus.DefaultReplicaSet.Primary, mysql.PodFQDN(cr, &firstPod)) {
 			log.Info("Primary is not pod-0", "primary", clusterStatus.DefaultReplicaSet.Primary)
 
 			log.Info("Ensuring pod-0 is the primary")
-			err := mysh.SetPrimaryInstanceWithExec(ctx, cr.InnoDBClusterName(), firstPodFQDN)
+			err := mysh.SetPrimaryInstanceWithExec(ctx, cr.InnoDBClusterName(), mysql.PodFQDN(cr, &firstPod))
 			if err != nil {
 				return errors.Wrap(err, "set primary instance")
 			}
@@ -634,7 +633,7 @@ func (r *PerconaServerMySQLReconciler) reconcileOrchestrator(ctx context.Context
 		return nil
 	}
 
-	orcPod, err := getOrcPod(ctx, r.Client, cr, 0)
+	orcPod, err := getReadyOrcPod(ctx, r.Client, cr)
 	if err != nil {
 		return nil
 	}
@@ -785,7 +784,7 @@ func (r *PerconaServerMySQLReconciler) reconcileReplication(ctx context.Context,
 		return nil
 	}
 
-	pod, err := getOrcPod(ctx, r.Client, cr, 0)
+	pod, err := getReadyOrcPod(ctx, r.Client, cr)
 	if err != nil {
 		return nil
 	}
@@ -833,17 +832,17 @@ func (r *PerconaServerMySQLReconciler) reconcileGroupReplication(ctx context.Con
 		return nil
 	}
 
-	firstPod, err := getMySQLPod(ctx, r.Client, cr, 0)
+	pod, err := getReadyMySQLPod(ctx, r.Client, cr)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "get ready mysql pod")
 	}
 
 	operatorPass, err := k8s.UserPassword(ctx, r.Client, cr, apiv1alpha1.UserOperator)
 	if err != nil {
 		return errors.Wrap(err, "get operator password")
 	}
 
-	db := database.NewReplicationManager(firstPod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, mysql.ServiceName(cr))
+	db := database.NewReplicationManager(pod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, mysql.ServiceName(cr))
 	cond := meta.FindStatusCondition(cr.Status.Conditions, apiv1alpha1.ConditionInnoDBClusterBootstrapped)
 	if cond == nil || cond.Status == metav1.ConditionFalse {
 		if exists, err := db.CheckIfDatabaseExists(ctx, "mysql_innodb_cluster_metadata"); err != nil || !exists {
@@ -863,7 +862,7 @@ func (r *PerconaServerMySQLReconciler) reconcileGroupReplication(ctx context.Con
 	}
 
 	if exists, err := db.CheckIfDatabaseExists(ctx, "mysql_innodb_cluster_metadata"); err != nil || !exists {
-		return errors.New("InnoDB cluster is already bootstrapped, but failed to check its status")
+		return errors.Wrap(err, "InnoDB cluster is already bootstrapped, but failed to check its status")
 	}
 
 	return nil
@@ -978,19 +977,17 @@ func (r *PerconaServerMySQLReconciler) reconcileMySQLRouter(ctx context.Context,
 			return nil
 		}
 
-		firstPod, err := getMySQLPod(ctx, r.Client, cr, 0)
+		pod, err := getReadyMySQLPod(ctx, r.Client, cr)
 		if err != nil {
-			return err
+			return errors.Wrap(err, "get ready mysql pod")
 		}
 
 		operatorPass, err := k8s.UserPassword(ctx, r.Client, cr, apiv1alpha1.UserOperator)
 		if err != nil {
 			return errors.Wrap(err, "get operator password")
 		}
 
-		firstPodUri := mysql.PodName(cr, 0) + "." + mysql.ServiceName(cr) + "." + cr.Namespace
-
-		db := database.NewReplicationManager(firstPod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, firstPodUri)
+		db := database.NewReplicationManager(pod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, mysql.PodFQDN(cr, pod))
 		if exist, err := db.CheckIfDatabaseExists(ctx, "mysql_innodb_cluster_metadata"); err != nil || !exist {
 			log.V(1).Info("Waiting for InnoDB Cluster", "cluster", cr.Name)
 			return nil
@@ -1026,7 +1023,7 @@ func (r *PerconaServerMySQLReconciler) cleanupOutdated(ctx context.Context, cr *
 }
 
 func (r *PerconaServerMySQLReconciler) getPrimaryFromOrchestrator(ctx context.Context, cr *apiv1alpha1.PerconaServerMySQL) (*orchestrator.Instance, error) {
-	pod, err := getOrcPod(ctx, r.Client, cr, 0)
+	pod, err := getReadyOrcPod(ctx, r.Client, cr)
 	if err != nil {
 		return nil, err
 	}
@@ -1048,14 +1045,12 @@ func (r *PerconaServerMySQLReconciler) getPrimaryFromGR(ctx context.Context, cr
 		return "", errors.Wrap(err, "get operator password")
 	}
 
-	fqdn := mysql.FQDN(cr, 0)
-
-	firstPod, err := getMySQLPod(ctx, r.Client, cr, 0)
+	pod, err := getReadyMySQLPod(ctx, r.Client, cr)
 	if err != nil {
-		return "", err
+		return "", errors.Wrap(err, "get ready mysql pod")
 	}
 
-	um := database.NewReplicationManager(firstPod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, fqdn)
+	um := database.NewReplicationManager(pod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, mysql.PodFQDN(cr, pod))
 
 	return um.GetGroupReplicationPrimary(ctx)
 }
@@ -1079,7 +1074,7 @@ func (r *PerconaServerMySQLReconciler) getPrimaryHost(ctx context.Context, cr *a
 func (r *PerconaServerMySQLReconciler) stopAsyncReplication(ctx context.Context, cr *apiv1alpha1.PerconaServerMySQL, primary *orchestrator.Instance) error {
 	log := logf.FromContext(ctx).WithName("stopAsyncReplication")
 
-	orcPod, err := getOrcPod(ctx, r.Client, cr, 0)
+	orcPod, err := getReadyOrcPod(ctx, r.Client, cr)
 	if err != nil {
 		return err
 	}
@@ -1134,7 +1129,7 @@ func (r *PerconaServerMySQLReconciler) stopAsyncReplication(ctx context.Context,
 func (r *PerconaServerMySQLReconciler) startAsyncReplication(ctx context.Context, cr *apiv1alpha1.PerconaServerMySQL, replicaPass string, primary *orchestrator.Instance) error {
 	log := logf.FromContext(ctx).WithName("startAsyncReplication")
 
-	orcPod, err := getOrcPod(ctx, r.Client, cr, 0)
+	orcPod, err := getReadyOrcPod(ctx, r.Client, cr)
 	if err != nil {
 		return nil
 	}
@@ -1177,6 +1172,20 @@ func (r *PerconaServerMySQLReconciler) startAsyncReplication(ctx context.Context
 	return errors.Wrap(g.Wait(), "start replication on replicas")
 }
 
+func getReadyMySQLPod(ctx context.Context, cl client.Reader, cr *apiv1alpha1.PerconaServerMySQL) (*corev1.Pod, error) {
+	pods, err := k8s.PodsByLabels(ctx, cl, mysql.MatchLabels(cr))
+	if err != nil {
+		return nil, errors.Wrap(err, "get pods")
+	}
+
+	for i, pod := range pods {
+		if k8s.IsPodReady(pod) {
+			return &pods[i], nil
+		}
+	}
+	return nil, errors.New("no ready pods")
+}
+
 func getMySQLPod(ctx context.Context, cl client.Reader, cr *apiv1alpha1.PerconaServerMySQL, idx int) (*corev1.Pod, error) {
 	pod := &corev1.Pod{}
 
@@ -1188,15 +1197,18 @@ func getMySQLPod(ctx context.Context, cl client.Reader, cr *apiv1alpha1.PerconaS
 	return pod, nil
 }
 
-func getOrcPod(ctx context.Context, cl client.Reader, cr *apiv1alpha1.PerconaServerMySQL, idx int) (*corev1.Pod, error) {
-	pod := &corev1.Pod{}
-
-	nn := types.NamespacedName{Namespace: cr.Namespace, Name: orchestrator.PodName(cr, idx)}
-	if err := cl.Get(ctx, nn, pod); err != nil {
-		return nil, err
+func getReadyOrcPod(ctx context.Context, cl client.Reader, cr *apiv1alpha1.PerconaServerMySQL) (*corev1.Pod, error) {
+	pods, err := k8s.PodsByLabels(ctx, cl, orchestrator.MatchLabels(cr))
+	if err != nil {
+		return nil, errors.Wrap(err, "get pods")
 	}
 
-	return pod, nil
+	for i, pod := range pods {
+		if k8s.IsPodReady(pod) {
+			return &pods[i], nil
+		}
+	}
+	return nil, errors.New("no ready pods")
 }
 
 func getPodIndexFromHostname(hostname string) (int, error) {

pkg/controller/ps/status.go

@@ -217,13 +217,12 @@ func (r *PerconaServerMySQLReconciler) isGRReady(ctx context.Context, cr *apiv1a
 		return false, errors.Wrap(err, "get operator password")
 	}
 
-	firstPod, err := getMySQLPod(ctx, r.Client, cr, 0)
+	pod, err := getReadyMySQLPod(ctx, r.Client, cr)
 	if err != nil {
-		return false, err
+		return false, errors.Wrap(err, "get ready mysql pod")
 	}
 
-	firstPodUri := mysql.PodName(cr, 0) + "." + mysql.ServiceName(cr) + "." + cr.Namespace
-	db := database.NewReplicationManager(firstPod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, firstPodUri)
+	db := database.NewReplicationManager(pod, r.ClientCmd, apiv1alpha1.UserOperator, operatorPass, mysql.PodFQDN(cr, pod))
 
 	dbExists, err := db.CheckIfDatabaseExists(ctx, "mysql_innodb_cluster_metadata")
 	if err != nil {

pkg/controller/ps/status_test.go

@@ -1138,6 +1138,7 @@ func makeFakeReadyPods(cr *apiv1alpha1.PerconaServerMySQL, amount int, podType s
 			},
 		},
 		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
 			Conditions: []corev1.PodCondition{
 				{
 					Type:   corev1.ContainersReady,

pkg/k8s/utils.go

@@ -94,6 +94,9 @@ func IsPodWithNameReady(ctx context.Context, cl client.Client, nn types.Namespac
 }
 
 func IsPodReady(pod corev1.Pod) bool {
+	if pod.Status.Phase != corev1.PodRunning || pod.DeletionTimestamp != nil {
+		return false
+	}
 	for _, cond := range pod.Status.Conditions {
 		if cond.Type == corev1.ContainersReady && cond.Status == corev1.ConditionTrue {
 			return true

pkg/mysql/mysql.go

@@ -109,6 +109,10 @@ func FQDN(cr *apiv1alpha1.PerconaServerMySQL, idx int) string {
 	return fmt.Sprintf("%s.%s.%s", PodName(cr, idx), ServiceName(cr), cr.Namespace)
 }
 
+func PodFQDN(cr *apiv1alpha1.PerconaServerMySQL, pod *corev1.Pod) string {
+	return fmt.Sprintf("%s.%s.%s", pod.Name, ServiceName(cr), cr.Namespace)
+}
+
 func MatchLabels(cr *apiv1alpha1.PerconaServerMySQL) map[string]string {
 	return util.SSMapMerge(cr.MySQLSpec().Labels,
 		map[string]string{naming.LabelComponent: ComponentName},