Introduce higher and lower bounds for the number of instances (#178)

* Introduce higher and lower bounds for the number of instances

Reduce the number of instances to the min_instances if it is lower and
to the max_instances if it is higher. -1 for either of those means there
is no lower or upper bound.

In addition, terminate the operator when there is a nonsense in the
configuration (i.e. max_instances < min_instances).

Reviewed by Jan Mußler and Sergey Dudoladov.

pkg/cluster/k8sres.go

@@ -464,6 +464,8 @@ func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.Statefu
 		return nil, fmt.Errorf("could not generate volume claim template: %v", err)
 	}
 
+	numberOfInstances := c.getNumberOfInstances(spec)
+
 	statefulSet := &v1beta1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      c.statefulSetName(),
@@ -471,7 +473,7 @@ func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.Statefu
 			Labels:    c.labelsSet(),
 		},
 		Spec: v1beta1.StatefulSetSpec{
-			Replicas:             &spec.NumberOfInstances,
+			Replicas:             &numberOfInstances,
 			ServiceName:          c.serviceName(Master),
 			Template:             *podTemplate,
 			VolumeClaimTemplates: []v1.PersistentVolumeClaim{*volumeClaimTemplate},
@@ -481,6 +483,25 @@ func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.Statefu
 	return statefulSet, nil
 }
 
+func (c *Cluster) getNumberOfInstances(spec *spec.PostgresSpec) (newcur int32) {
+	min := c.OpConfig.MinInstances
+	max := c.OpConfig.MaxInstances
+	cur := spec.NumberOfInstances
+	newcur = cur
+
+	if max >= 0 && newcur > max {
+		newcur = max
+	}
+	if min >= 0 && newcur < min {
+		newcur = min
+	}
+	if newcur != cur {
+		c.logger.Infof("adjusted number of instances from %d to %d (min: %d, max: %d)", cur, newcur, min, max)
+	}
+
+	return
+}
+
 func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string) (*v1.PersistentVolumeClaim, error) {
 	metadata := metav1.ObjectMeta{
 		Name: constants.DataVolumeName,

pkg/cluster/sync.go

@@ -267,7 +267,7 @@ func (c *Cluster) syncStatefulSet() error {
 		c.logger.Infof("found pods without the statefulset: trigger rolling update")
 
 	} else {
-		// statefulset is alrady there, make sure we use its definition in order to compare with the spec.
+		// statefulset is already there, make sure we use its definition in order to compare with the spec.
 		c.Statefulset = sset
 
 		desiredSS, err := c.generateStatefulSet(&c.Spec)