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).

pkg/cluster/k8sres.go

@@ -464,14 +464,16 @@ func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.Statefu
 		return nil, fmt.Errorf("could not generate volume claim template: %v", err)
 	}
 
+	numberOfInstaces := c.getNumberOfInstances(spec)
+
 	statefulSet := &v1beta1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      c.statefulSetName(),
 			Namespace: c.Namespace,
 			Labels:    c.labelsSet(),
 		},
 		Spec: v1beta1.StatefulSetSpec{
-			Replicas:             &spec.NumberOfInstances,
+			Replicas:             &numberOfInstaces,
 			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 to %d (min: %d, max: %d)", 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)

pkg/util/config/config.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 	"time"
 
+	"fmt"
 	"github.com/zalando-incubator/postgres-operator/pkg/spec"
 )
 
@@ -33,6 +34,8 @@ type Resources struct {
 	PodEnvironmentConfigMap string            `name:"pod_environment_configmap" default:""`
 	NodeEOLLabel            map[string]string `name:"node_eol_label" default:"lifecycle-status:pending-decommission"`
 	NodeReadinessLabel      map[string]string `name:"node_readiness_label" default:"lifecycle-status:ready"`
+	MaxInstances            int32             `name:"max_instances" default:"-1"`
+	MinInstances            int32             `name:"min_instances" default:"-1"`
 }
 
 // Auth describes authentication specific configuration parameters
@@ -108,6 +111,9 @@ func NewFromMap(m map[string]string) *Config {
 			panic(err)
 		}
 	}
+	if err := validate(&cfg); err != nil {
+		panic(err)
+	}
 
 	return &cfg
 }
@@ -123,3 +129,13 @@ func Copy(c *Config) Config {
 
 	return cfg
 }
+
+func validate(cfg *Config) (err error) {
+	if cfg.MinInstances > 0 && cfg.MaxInstances > 0 && cfg.MinInstances > cfg.MaxInstances {
+		err = fmt.Errorf("minimum number of instances %d is set higher than the maximum number %d")
+	}
+	if cfg.Workers == 0 {
+		err = fmt.Errorf("number of works should be higher than 0")
+	}
+	return
+}