| 551 | } |
| 552 | |
| 553 | func (c *Cluster) syncStatefulSet() error { |
| 554 | var ( |
| 555 | restartWait uint32 |
| 556 | configPatched bool |
| 557 | restartPrimaryFirst bool |
| 558 | ) |
| 559 | podsToRecreate := make([]v1.Pod, 0) |
| 560 | isSafeToRecreatePods := true |
| 561 | postponeReasons := make([]string, 0) |
| 562 | switchoverCandidates := make([]spec.NamespacedName, 0) |
| 563 | |
| 564 | pods, err := c.listPods() |
| 565 | if err != nil { |
| 566 | c.logger.Warnf("could not list pods of the statefulset: %v", err) |
| 567 | } |
| 568 | |
| 569 | // NB: Be careful to consider the codepath that acts on podsRollingUpdateRequired before returning early. |
| 570 | sset, err := c.KubeClient.StatefulSets(c.Namespace).Get(context.TODO(), c.statefulSetName(), metav1.GetOptions{}) |
| 571 | if err != nil && !k8sutil.ResourceNotFound(err) { |
| 572 | return fmt.Errorf("error during reading of statefulset: %v", err) |
| 573 | } |
| 574 | |
| 575 | if err != nil { |
| 576 | // statefulset does not exist, try to re-create it |
| 577 | c.logger.Infof("cluster's statefulset does not exist") |
| 578 | |
| 579 | sset, err = c.createStatefulSet() |
| 580 | if err != nil { |
| 581 | return fmt.Errorf("could not create missing statefulset: %v", err) |
| 582 | } |
| 583 | |
| 584 | if err = c.waitStatefulsetPodsReady(); err != nil { |
| 585 | return fmt.Errorf("cluster is not ready: %v", err) |
| 586 | } |
| 587 | |
| 588 | if len(pods) > 0 { |
| 589 | for _, pod := range pods { |
| 590 | if err = c.markRollingUpdateFlagForPod(&pod, "pod from previous statefulset"); err != nil { |
| 591 | c.logger.Warnf("marking old pod for rolling update failed: %v", err) |
| 592 | } |
| 593 | podsToRecreate = append(podsToRecreate, pod) |
| 594 | } |
| 595 | } |
| 596 | c.logger.Infof("created missing statefulset %q", util.NameFromMeta(sset.ObjectMeta)) |
| 597 | |
| 598 | } else { |
| 599 | desiredSts, err := c.generateStatefulSet(&c.Spec) |
| 600 | if err != nil { |
| 601 | return fmt.Errorf("could not generate statefulset: %v", err) |
| 602 | } |
| 603 | c.logger.Debug("syncing statefulsets") |
| 604 | // check if there are still pods with a rolling update flag |
| 605 | for _, pod := range pods { |
| 606 | if c.getRollingUpdateFlagFromPod(&pod) { |
| 607 | podsToRecreate = append(podsToRecreate, pod) |
| 608 | } else { |
| 609 | role := PostgresRole(pod.Labels[c.OpConfig.PodRoleLabel]) |
| 610 | if role == Master { |