Merge branch 'master' into pending_rolling_updates

2026-10-02 11:16:07 +02:00 · 2018-05-03 11:27:16 +02:00
parent 0618723a61 4c8dfd7e20
commit 43a1db2128
13 changed files with 295 additions and 92 deletions
@@ -42,6 +42,7 @@ type Config struct {
 	OpConfig            config.Config
 	RestConfig          *rest.Config
 	InfrastructureRoles map[string]spec.PgUser // inherited from the controller
+	PodServiceAccount   *v1.ServiceAccount
 }

 type kubeResources struct {
@@ -196,6 +197,39 @@ func (c *Cluster) initUsers() error {
 	return nil
 }

+/*
+  Ensures the service account required by StatefulSets to create pods exists in a namespace before a PG cluster is created there so that a user does not have to deploy the account manually.
+
+  The operator does not sync these accounts after creation.
+*/
+func (c *Cluster) createPodServiceAccounts() error {
+
+	podServiceAccountName := c.Config.OpConfig.PodServiceAccountName
+	_, err := c.KubeClient.ServiceAccounts(c.Namespace).Get(podServiceAccountName, metav1.GetOptions{})
+
+	if err != nil {
+
+		c.setProcessName(fmt.Sprintf("creating pod service account in the namespace %v", c.Namespace))
+
+		c.logger.Infof("the pod service account %q cannot be retrieved in the namespace %q. Trying to deploy the account.", podServiceAccountName, c.Namespace)
+
+		// get a separate copy of service account
+		// to prevent a race condition when setting a namespace for many clusters
+		sa := *c.PodServiceAccount
+		_, err = c.KubeClient.ServiceAccounts(c.Namespace).Create(&sa)
+		if err != nil {
+			return fmt.Errorf("cannot deploy the pod service account %q defined in the config map to the %q namespace: %v", podServiceAccountName, c.Namespace, err)
+		}
+
+		c.logger.Infof("successfully deployed the pod service account %q to the %q namespace", podServiceAccountName, c.Namespace)
+
+	} else {
+		c.logger.Infof("successfully found the service account %q used to create pods to the namespace %q", podServiceAccountName, c.Namespace)
+	}
+
+	return nil
+}
+
 // Create creates the new kubernetes objects associated with the cluster.
 func (c *Cluster) Create() error {
 	c.mu.Lock()
@@ -259,6 +293,11 @@ func (c *Cluster) Create() error {
 	}
 	c.logger.Infof("pod disruption budget %q has been successfully created", util.NameFromMeta(pdb.ObjectMeta))

+	if err = c.createPodServiceAccounts(); err != nil {
+		return fmt.Errorf("could not create pod service account %v : %v", c.OpConfig.PodServiceAccountName, err)
+	}
+	c.logger.Infof("pod service accounts have been successfully synced")
+
 	if c.Statefulset != nil {
 		return fmt.Errorf("statefulset already exists in the cluster")
 	}
@@ -822,6 +861,7 @@ func (c *Cluster) GetStatus() *spec.ClusterStatus {
 // ManualFailover does manual failover to a candidate pod
 func (c *Cluster) ManualFailover(curMaster *v1.Pod, candidate spec.NamespacedName) error {
 	c.logger.Debugf("failing over from %q to %q", curMaster.Name, candidate)
+
 	podLabelErr := make(chan error)
 	stopCh := make(chan struct{})
 	defer close(podLabelErr)
@@ -832,11 +872,12 @@ func (c *Cluster) ManualFailover(curMaster *v1.Pod, candidate spec.NamespacedNam

 		role := Master

-		_, err := c.waitForPodLabel(ch, &role)
-
 		select {
 		case <-stopCh:
-		case podLabelErr <- err:
+		case podLabelErr <- func() error {
+			_, err := c.waitForPodLabel(ch, stopCh, &role)
+			return err
+		}():
 		}
 	}()

@@ -360,10 +360,16 @@ func (c *Cluster) generatePodTemplate(
 	}
 	if c.OpConfig.WALES3Bucket != "" {
 		envVars = append(envVars, v1.EnvVar{Name: "WAL_S3_BUCKET", Value: c.OpConfig.WALES3Bucket})
-		envVars = append(envVars, v1.EnvVar{Name: "WAL_BUCKET_SCOPE_SUFFIX", Value: getWALBucketScopeSuffix(string(uid))})
+		envVars = append(envVars, v1.EnvVar{Name: "WAL_BUCKET_SCOPE_SUFFIX", Value: getBucketScopeSuffix(string(uid))})
 		envVars = append(envVars, v1.EnvVar{Name: "WAL_BUCKET_SCOPE_PREFIX", Value: ""})
 	}

+	if c.OpConfig.LogS3Bucket != "" {
+		envVars = append(envVars, v1.EnvVar{Name: "LOG_S3_BUCKET", Value: c.OpConfig.LogS3Bucket})
+		envVars = append(envVars, v1.EnvVar{Name: "LOG_BUCKET_SCOPE_SUFFIX", Value: getBucketScopeSuffix(string(uid))})
+		envVars = append(envVars, v1.EnvVar{Name: "LOG_BUCKET_SCOPE_PREFIX", Value: ""})
+	}
+
 	if c.patroniUsesKubernetes() {
 		envVars = append(envVars, v1.EnvVar{Name: "DCS_ENABLE_KUBERNETES_API", Value: "true"})
 	} else {
@@ -435,7 +441,7 @@ func (c *Cluster) generatePodTemplate(
 	terminateGracePeriodSeconds := int64(c.OpConfig.PodTerminateGracePeriod.Seconds())

 	podSpec := v1.PodSpec{
-		ServiceAccountName:            c.OpConfig.ServiceAccountName,
+		ServiceAccountName:            c.OpConfig.PodServiceAccountName,
 		TerminationGracePeriodSeconds: &terminateGracePeriodSeconds,
 		Containers:                    []v1.Container{container},
 		Tolerations:                   c.tolerations(tolerationsSpec),
@@ -504,7 +510,7 @@ func (c *Cluster) generatePodTemplate(
 	return &template
 }

-func getWALBucketScopeSuffix(uid string) string {
+func getBucketScopeSuffix(uid string) string {
 	if uid != "" {
 		return fmt.Sprintf("/%s", uid)
 	}
@@ -701,7 +707,7 @@ func (c *Cluster) shouldCreateLoadBalancerForService(role PostgresRole, spec *sp
 		// `enable_load_balancer`` governs LB for a master service
 		// there is no equivalent deprecated operator option for the replica LB
 		if c.OpConfig.EnableLoadBalancer != nil {
-			c.logger.Debugf("The operator configmap sets the deprecated `enable_load_balancer` param. Consider using the `enable_master_load_balancer` or `enable_replica_load_balancer` instead.", c.Name)
+			c.logger.Debugf("The operator configmap sets the deprecated `enable_load_balancer` param. Consider using the `enable_master_load_balancer` or `enable_replica_load_balancer` instead.")
 			return *c.OpConfig.EnableLoadBalancer
 		}

@@ -819,7 +825,7 @@ func (c *Cluster) generateCloneEnvironment(description *spec.CloneDescription) [
 		result = append(result, v1.EnvVar{Name: "CLONE_METHOD", Value: "CLONE_WITH_WALE"})
 		result = append(result, v1.EnvVar{Name: "CLONE_WAL_S3_BUCKET", Value: c.OpConfig.WALES3Bucket})
 		result = append(result, v1.EnvVar{Name: "CLONE_TARGET_TIME", Value: description.EndTimestamp})
-		result = append(result, v1.EnvVar{Name: "CLONE_WAL_BUCKET_SCOPE_SUFFIX", Value: getWALBucketScopeSuffix(description.Uid)})
+		result = append(result, v1.EnvVar{Name: "CLONE_WAL_BUCKET_SCOPE_SUFFIX", Value: getBucketScopeSuffix(description.Uid)})
 		result = append(result, v1.EnvVar{Name: "CLONE_WAL_BUCKET_SCOPE_PREFIX", Value: ""})
 	}

@@ -149,13 +149,19 @@ func (c *Cluster) movePodFromEndOfLifeNode(pod *v1.Pod) (*v1.Pod, error) {
 }

 func (c *Cluster) masterCandidate(oldNodeName string) (*v1.Pod, error) {
+
+	// Wait until at least one replica pod will come up
+	if err := c.waitForAnyReplicaLabelReady(); err != nil {
+		c.logger.Warningf("could not find at least one ready replica: %v", err)
+	}
+
 	replicas, err := c.getRolePods(Replica)
 	if err != nil {
 		return nil, fmt.Errorf("could not get replica pods: %v", err)
 	}

 	if len(replicas) == 0 {
-		c.logger.Warningf("single master pod for cluster %q, migration will cause disruption of the service")
+		c.logger.Warningf("no available master candidates, migration will cause longer downtime of the master instance")
 		return nil, nil
 	}

@@ -168,12 +174,16 @@ func (c *Cluster) masterCandidate(oldNodeName string) (*v1.Pod, error) {
 			}
 		}
 	}
-	c.logger.Debug("no available master candidates on live nodes")
+	c.logger.Warningf("no available master candidates on live nodes")
 	return &replicas[rand.Intn(len(replicas))], nil
 }

 // MigrateMasterPod migrates master pod via failover to a replica
 func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
+	var (
+		masterCandidatePod *v1.Pod
+	)
+
 	oldMaster, err := c.KubeClient.Pods(podName.Namespace).Get(podName.Name, metav1.GetOptions{})

 	if err != nil {
@@ -193,10 +203,13 @@ func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
 		c.logger.Warningf("pod %q is not a master", podName)
 		return nil
 	}
-
-	masterCandidatePod, err := c.masterCandidate(oldMaster.Spec.NodeName)
-	if err != nil {
-		return fmt.Errorf("could not get new master candidate: %v", err)
+	if *c.Statefulset.Spec.Replicas == 1 {
+		c.logger.Warningf("single master pod for cluster %q, migration will cause longer downtime of the master instance", c.clusterName())
+	} else {
+		masterCandidatePod, err = c.masterCandidate(oldMaster.Spec.NodeName)
+		if err != nil {
+			return fmt.Errorf("could not get new master candidate: %v", err)
+		}
 	}

 	// there are two cases for each postgres cluster that has its master pod on the node to migrate from:
@@ -250,6 +263,7 @@ func (c *Cluster) MigrateReplicaPod(podName spec.NamespacedName, fromNodeName st
 func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
 	ch := c.registerPodSubscriber(podName)
 	defer c.unregisterPodSubscriber(podName)
+	stopChan := make(chan struct{})

 	if err := c.KubeClient.Pods(podName.Namespace).Delete(podName.Name, c.deleteOptions); err != nil {
 		return nil, fmt.Errorf("could not delete pod: %v", err)
@@ -258,7 +272,7 @@ func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
 	if err := c.waitForPodDeletion(ch); err != nil {
 		return nil, err
 	}
-	if pod, err := c.waitForPodLabel(ch, nil); err != nil {
+	if pod, err := c.waitForPodLabel(ch, stopChan, nil); err != nil {
 		return nil, err
 	} else {
 		c.logger.Infof("pod %q has been recreated", podName)
@@ -108,11 +108,10 @@ func (c *Cluster) syncService(role PostgresRole) error {

 	svc, err := c.KubeClient.Services(c.Namespace).Get(c.serviceName(role), metav1.GetOptions{})
 	if err == nil {
-
+		c.Services[role] = svc
 		desiredSvc := c.generateService(role, &c.Spec)
 		match, reason := k8sutil.SameService(svc, desiredSvc)
 		if match {
-			c.Services[role] = svc
 			return nil
 		}
 		c.logServiceChanges(role, svc, desiredSvc, false, reason)
@@ -216,18 +216,20 @@ func (c *Cluster) getTeamMembers() ([]string, error) {

 	token, err := c.oauthTokenGetter.getOAuthToken()
 	if err != nil {
-		return []string{}, fmt.Errorf("could not get oauth token: %v", err)
+		c.logger.Warnf("could not get oauth token to authenticate to team service API, returning empty list of team members: %v", err)
+		return []string{}, nil
 	}

 	teamInfo, err := c.teamsAPIClient.TeamInfo(c.Spec.TeamID, token)
 	if err != nil {
-		return nil, fmt.Errorf("could not get team info: %v", err)
+		c.logger.Warnf("could not get team info, returning empty list of team members: %v", err)
+		return []string{}, nil
 	}

 	return teamInfo.Members, nil
 }

-func (c *Cluster) waitForPodLabel(podEvents chan spec.PodEvent, role *PostgresRole) (*v1.Pod, error) {
+func (c *Cluster) waitForPodLabel(podEvents chan spec.PodEvent, stopChan chan struct{}, role *PostgresRole) (*v1.Pod, error) {
 	timeout := time.After(c.OpConfig.PodLabelWaitTimeout)
 	for {
 		select {
@@ -243,6 +245,8 @@ func (c *Cluster) waitForPodLabel(podEvents chan spec.PodEvent, role *PostgresRo
 			}
 		case <-timeout:
 			return nil, fmt.Errorf("pod label wait timeout")
+		case <-stopChan:
+			return nil, fmt.Errorf("pod label wait cancelled")
 		}
 	}
 }
@@ -280,7 +284,10 @@ func (c *Cluster) waitStatefulsetReady() error {
 		})
 }

-func (c *Cluster) waitPodLabelsReady() error {
+func (c *Cluster) _waitPodLabelsReady(anyReplica bool) error {
+	var (
+		podsNumber int
+	)
 	ls := c.labelsSet(false)
 	namespace := c.Namespace

@@ -297,35 +304,56 @@ func (c *Cluster) waitPodLabelsReady() error {
 			c.OpConfig.PodRoleLabel: string(Replica),
 		}).String(),
 	}
-	pods, err := c.KubeClient.Pods(namespace).List(listOptions)
-	if err != nil {
-		return err
+	podsNumber = 1
+	if !anyReplica {
+		pods, err := c.KubeClient.Pods(namespace).List(listOptions)
+		if err != nil {
+			return err
+		}
+		podsNumber = len(pods.Items)
+		c.logger.Debugf("Waiting for %d pods to become ready", podsNumber)
+	} else {
+		c.logger.Debugf("Waiting for any replica pod to become ready")
 	}
-	podsNumber := len(pods.Items)

-	err = retryutil.Retry(c.OpConfig.ResourceCheckInterval, c.OpConfig.ResourceCheckTimeout,
+	err := retryutil.Retry(c.OpConfig.ResourceCheckInterval, c.OpConfig.ResourceCheckTimeout,
 		func() (bool, error) {
-			masterPods, err2 := c.KubeClient.Pods(namespace).List(masterListOption)
-			if err2 != nil {
-				return false, err2
+			masterCount := 0
+			if !anyReplica {
+				masterPods, err2 := c.KubeClient.Pods(namespace).List(masterListOption)
+				if err2 != nil {
+					return false, err2
+				}
+				if len(masterPods.Items) > 1 {
+					return false, fmt.Errorf("too many masters (%d pods with the master label found)",
+						len(masterPods.Items))
+				}
+				masterCount = len(masterPods.Items)
 			}
 			replicaPods, err2 := c.KubeClient.Pods(namespace).List(replicaListOption)
 			if err2 != nil {
 				return false, err2
 			}
-			if len(masterPods.Items) > 1 {
-				return false, fmt.Errorf("too many masters")
-			}
-			if len(replicaPods.Items) == podsNumber {
+			replicaCount := len(replicaPods.Items)
+			if anyReplica && replicaCount > 0 {
+				c.logger.Debugf("Found %d running replica pods", replicaCount)
 				return true, nil
 			}

-			return len(masterPods.Items)+len(replicaPods.Items) == podsNumber, nil
+			return masterCount+replicaCount >= podsNumber, nil
 		})

 	return err
 }

+func (c *Cluster) waitForAnyReplicaLabelReady() error {
+	return c._waitPodLabelsReady(true)
+}
+
+func (c *Cluster) waitForAllPodsLabelReady() error {
+	return c._waitPodLabelsReady(false)
+}
+
 func (c *Cluster) waitStatefulsetPodsReady() error {
 	c.setProcessName("waiting for the pods of the statefulset")
 	// TODO: wait for the first Pod only
@@ -334,7 +362,7 @@ func (c *Cluster) waitStatefulsetPodsReady() error {
 	}

 	// TODO: wait only for master
-	if err := c.waitPodLabelsReady(); err != nil {
+	if err := c.waitForAllPodsLabelReady(); err != nil {
 		return fmt.Errorf("pod labels error: %v", err)
 	}