add MapContains function

pkg/cluster/cluster.go

@@ -316,7 +316,6 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *comp
 		reasons = append(reasons, "new statefulset's container specification doesn't match the current one")
 	}
 	if len(c.Statefulset.Spec.Template.Spec.Containers) == 0 {
-
 		c.logger.Warningf("statefulset %q has no container", util.NameFromMeta(c.Statefulset.ObjectMeta))
 		return &compareStatefulsetResult{}
 	}

pkg/cluster/pod.go

@@ -12,19 +12,39 @@ import (
 )
 
 func (c *Cluster) listPods() ([]v1.Pod, error) {
-	ns := c.Namespace
 	listOptions := metav1.ListOptions{
 		LabelSelector: c.labelsSet().String(),
 	}
 
-	pods, err := c.KubeClient.Pods(ns).List(listOptions)
+	pods, err := c.KubeClient.Pods(c.Namespace).List(listOptions)
 	if err != nil {
 		return nil, fmt.Errorf("could not get list of pods: %v", err)
 	}
 
 	return pods.Items, nil
 }
 
+func (c *Cluster) getRolePods(role PostgresRole) ([]v1.Pod, error) {
+	listOptions := metav1.ListOptions{
+		LabelSelector: c.roleLabelsSet(role).String(),
+	}
+
+	pods, err := c.KubeClient.Pods(c.Namespace).List(listOptions)
+	if err != nil {
+		return nil, fmt.Errorf("could not get list of pods: %v", err)
+	}
+
+	if len(pods.Items) == 0 {
+		return nil, fmt.Errorf("no pods")
+	}
+
+	if role == Master && len(pods.Items) > 1 {
+		return nil, fmt.Errorf("too many masters")
+	}
+
+	return pods.Items, nil
+}
+
 func (c *Cluster) deletePods() error {
 	c.logger.Debugln("deleting pods")
 	pods, err := c.listPods()

pkg/cluster/resources.go

@@ -119,6 +119,33 @@ func (c *Cluster) createStatefulSet() (*v1beta1.StatefulSet, error) {
 	return statefulSet, nil
 }
 
+func (c *Cluster) scaleDown(newStatefulSet *v1beta1.StatefulSet) error {
+	podName := fmt.Sprintf("%s-0", c.Statefulset.Name) //TODO:
+	masterCandidatePod, err := c.KubeClient.Pods(c.OpConfig.Namespace).Get(podName, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("could not get master candidate pod: %v", err)
+	}
+
+	//TODO: check if the pod is master pod
+
+	// some sanity check
+	if !util.MapContains(masterCandidatePod.Labels, c.OpConfig.ClusterLabels) ||
+		!util.MapContains(masterCandidatePod.Labels, map[string]string{c.OpConfig.ClusterNameLabel:c.Name}) {
+		return fmt.Errorf("pod %q does not belong to cluster", podName)
+	}
+
+	masterPod, err := c.getRolePods(Master)
+	if err != nil {
+		return fmt.Errorf("could not get master pod: %v", err)
+	}
+
+	if err := c.patroni.Failover(&masterPod[0], masterCandidatePod.Name); err != nil {
+		return fmt.Errorf("could not failover: %v", err)
+	}
+
+	return nil
+}
+
 func (c *Cluster) updateStatefulSet(newStatefulSet *v1beta1.StatefulSet) error {
 	c.setProcessName("updating statefulset")
 	if c.Statefulset == nil {
@@ -133,6 +160,11 @@ func (c *Cluster) updateStatefulSet(newStatefulSet *v1beta1.StatefulSet) error {
 		return fmt.Errorf("could not form patch for the statefulset %q: %v", statefulSetName, err)
 	}
 
+	//Scale down
+	if *c.Statefulset.Spec.Replicas > *newStatefulSet.Spec.Replicas {
+		c.scaleDown(newStatefulSet)
+	}
+
 	statefulSet, err := c.KubeClient.StatefulSets(c.Statefulset.Namespace).Patch(
 		c.Statefulset.Name,
 		types.MergePatchType,

pkg/util/patroni/patroni.go

@@ -113,6 +113,7 @@ func (p *Patroni) Failover(master *v1.Pod, candidate string) error {
 	return nil
 }
 
+// Status returns patroni status
 func (p *Patroni) Status(pod *v1.Pod) (*Status, error) {
 	status := &Status{}
 

pkg/util/util.go

@@ -103,3 +103,19 @@ func FindNamedStringSubmatch(r *regexp.Regexp, s string) map[string]string {
 
 	return res
 }
+
+// MapContains returns true if and only if haystack contains all the keys from the needle with matching corresponding values
+func MapContains(haystack, needle map[string]string) bool {
+	if len(haystack) < len(needle) {
+		return false
+	}
+
+	for k, v := range needle {
+		v2, ok := haystack[k]
+		if !ok || v2 != v {
+			return false
+		}
+	}
+
+	return true
+}

pkg/util/util_test.go

@@ -46,6 +46,18 @@ var substractTest = []struct {
 	{[]string{"a", "b", "c", "d"}, []string{"a", "bb", "c", "d"}, []string{"b"}, false},
 }
 
+var mapContaintsTest = []struct {
+	inA map[string]string
+	inB map[string]string
+	out bool
+}{
+	{map[string]string{"1": "a", "2": "b", "3": "c", "4": "c"}, map[string]string{"1": "a", "2": "b", "3": "c"}, true},
+	{map[string]string{"1": "a", "2": "b", "3": "c", "4": "c"}, map[string]string{"1": "a", "2": "b", "3": "d"}, false},
+	{map[string]string{}, map[string]string{}, true},
+	{map[string]string{"3": "c", "4": "c"}, map[string]string{"1": "a", "2": "b", "3": "c"}, false},
+	{map[string]string{"3": "c", "4": "c"}, map[string]string{}, true},
+}
+
 var substringMatch = []struct {
 	inRegex *regexp.Regexp
 	inStr   string
@@ -122,3 +134,12 @@ func TestFindNamedStringSubmatch(t *testing.T) {
 		}
 	}
 }
+
+func TestMapContains(t *testing.T) {
+	for _, tt := range mapContaintsTest {
+		res := MapContains(tt.inA, tt.inB)
+		if res != tt.out {
+			t.Errorf("MapContains expected: %#v, got: %#v", tt.out, res)
+		}
+	}
+}