merge with master

pkg/cluster/cluster.go

@@ -232,6 +232,10 @@ func (c *Cluster) Create() error {
 
 	c.setStatus(acidv1.ClusterStatusCreating)
 
+	if err = c.validateResources(&c.Spec); err != nil {
+		return fmt.Errorf("insufficient resource limits specified: %v", err)
+	}
+
 	for _, role := range []PostgresRole{Master, Replica} {
 
 		if c.Endpoints[role] != nil {
@@ -499,6 +503,44 @@ func compareResourcesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.Resourc
 
 }
 
+func (c *Cluster) validateResources(spec *acidv1.PostgresSpec) error {
+
+	// setting limits too low can cause unnecessary evictions / OOM kills
+	const (
+		cpuMinLimit    = "256m"
+		memoryMinLimit = "256Mi"
+	)
+
+	var (
+		isSmaller bool
+		err       error
+	)
+
+	cpuLimit := spec.Resources.ResourceLimits.CPU
+	if cpuLimit != "" {
+		isSmaller, err = util.IsSmallerQuantity(cpuLimit, cpuMinLimit)
+		if err != nil {
+			return fmt.Errorf("error validating CPU limit: %v", err)
+		}
+		if isSmaller {
+			return fmt.Errorf("defined CPU limit %s is below required minimum %s to properly run postgresql resource", cpuLimit, cpuMinLimit)
+		}
+	}
+
+	memoryLimit := spec.Resources.ResourceLimits.Memory
+	if memoryLimit != "" {
+		isSmaller, err = util.IsSmallerQuantity(memoryLimit, memoryMinLimit)
+		if err != nil {
+			return fmt.Errorf("error validating memory limit: %v", err)
+		}
+		if isSmaller {
+			return fmt.Errorf("defined memory limit %s is below required minimum %s to properly run postgresql resource", memoryLimit, memoryMinLimit)
+		}
+	}
+
+	return nil
+}
+
 // Update changes Kubernetes objects according to the new specification. Unlike the sync case, the missing object
 // (i.e. service) is treated as an error
 // logical backup cron jobs are an exception: a user-initiated Update can enable a logical backup job
@@ -509,6 +551,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
+	oldStatus := c.Status
 	c.setStatus(acidv1.ClusterStatusUpdating)
 	c.setSpec(newSpec)
 
@@ -520,6 +563,22 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		}
 	}()
 
+	if err := c.validateResources(&newSpec.Spec); err != nil {
+		err = fmt.Errorf("insufficient resource limits specified: %v", err)
+
+		// cancel update only when (already too low) pod resources were edited
+		// if cluster was successfully running before the update, continue but log a warning
+		isCPULimitSmaller, err2 := util.IsSmallerQuantity(newSpec.Spec.Resources.ResourceLimits.CPU, oldSpec.Spec.Resources.ResourceLimits.CPU)
+		isMemoryLimitSmaller, err3 := util.IsSmallerQuantity(newSpec.Spec.Resources.ResourceLimits.Memory, oldSpec.Spec.Resources.ResourceLimits.Memory)
+
+		if oldStatus.Running() && !isCPULimitSmaller && !isMemoryLimitSmaller && err2 == nil && err3 == nil {
+			c.logger.Warning(err)
+		} else {
+			updateFailed = true
+			return err
+		}
+	}
+
 	if oldSpec.Spec.PgVersion != newSpec.Spec.PgVersion { // PG versions comparison
 		c.logger.Warningf("postgresql version change(%q -> %q) has no effect", oldSpec.Spec.PgVersion, newSpec.Spec.PgVersion)
 		//we need that hack to generate statefulset with the old version

pkg/cluster/cluster_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util/config"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
 	"github.com/zalando/postgres-operator/pkg/util/teams"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 )
 
 const (
@@ -328,3 +328,57 @@ func TestShouldDeleteSecret(t *testing.T) {
 		}
 	}
 }
+
+func TestPodAnnotations(t *testing.T) {
+	testName := "TestPodAnnotations"
+	tests := []struct {
+		subTest  string
+		operator map[string]string
+		database map[string]string
+		merged   map[string]string
+	}{
+		{
+			subTest:  "No Annotations",
+			operator: make(map[string]string),
+			database: make(map[string]string),
+			merged:   make(map[string]string),
+		},
+		{
+			subTest:  "Operator Config Annotations",
+			operator: map[string]string{"foo": "bar"},
+			database: make(map[string]string),
+			merged:   map[string]string{"foo": "bar"},
+		},
+		{
+			subTest:  "Database Config Annotations",
+			operator: make(map[string]string),
+			database: map[string]string{"foo": "bar"},
+			merged:   map[string]string{"foo": "bar"},
+		},
+		{
+			subTest:  "Database Config overrides Operator Config Annotations",
+			operator: map[string]string{"foo": "bar", "global": "foo"},
+			database: map[string]string{"foo": "baz", "local": "foo"},
+			merged:   map[string]string{"foo": "baz", "global": "foo", "local": "foo"},
+		},
+	}
+
+	for _, tt := range tests {
+		cl.OpConfig.CustomPodAnnotations = tt.operator
+		cl.Postgresql.Spec.PodAnnotations = tt.database
+
+		annotations := cl.generatePodAnnotations(&cl.Postgresql.Spec)
+		for k, v := range annotations {
+			if observed, expected := v, tt.merged[k]; observed != expected {
+				t.Errorf("%v expects annotation value %v for key %v, but found %v",
+					testName+"/"+tt.subTest, expected, observed, k)
+			}
+		}
+		for k, v := range tt.merged {
+			if observed, expected := annotations[k], v; observed != expected {
+				t.Errorf("%v expects annotation value %v for key %v, but found %v",
+					testName+"/"+tt.subTest, expected, observed, k)
+			}
+		}
+	}
+}

pkg/cluster/k8sres.go

@@ -430,6 +430,7 @@ func mountShmVolumeNeeded(opConfig config.Config, pgSpec *acidv1.PostgresSpec) *
 func generatePodTemplate(
 	namespace string,
 	labels labels.Set,
+	annotations map[string]string,
 	spiloContainer *v1.Container,
 	initContainers []v1.Container,
 	sidecarContainers []v1.Container,
@@ -485,13 +486,17 @@ func generatePodTemplate(
 
 	template := v1.PodTemplateSpec{
 		ObjectMeta: metav1.ObjectMeta{
-			Labels:    labels,
-			Namespace: namespace,
+			Labels:      labels,
+			Namespace:   namespace,
+			Annotations: annotations,
 		},
 		Spec: podSpec,
 	}
 	if kubeIAMRole != "" {
-		template.Annotations = map[string]string{constants.KubeIAmAnnotation: kubeIAMRole}
+		if template.Annotations == nil {
+			template.Annotations = make(map[string]string)
+		}
+		template.Annotations[constants.KubeIAmAnnotation] = kubeIAMRole
 	}
 
 	return &template, nil
@@ -715,6 +720,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 
 	var (
 		err                 error
+		initContainers      []v1.Container
 		sidecarContainers   []v1.Container
 		podTemplate         *v1.PodTemplateSpec
 		volumeClaimTemplate *v1.PersistentVolumeClaim
@@ -735,7 +741,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 			limit = c.OpConfig.DefaultMemoryLimit
 		}
 
-		isSmaller, err := util.RequestIsSmallerThanLimit(request, limit)
+		isSmaller, err := util.IsSmallerQuantity(request, limit)
 		if err != nil {
 			return nil, err
 		}
@@ -762,7 +768,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 				limit = c.OpConfig.DefaultMemoryLimit
 			}
 
-			isSmaller, err := util.RequestIsSmallerThanLimit(sidecarRequest, sidecarLimit)
+			isSmaller, err := util.IsSmallerQuantity(sidecarRequest, sidecarLimit)
 			if err != nil {
 				return nil, err
 			}
@@ -781,6 +787,13 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		return nil, fmt.Errorf("could not generate resource requirements: %v", err)
 	}
 
+	if spec.InitContainers != nil && len(spec.InitContainers) > 0 {
+		if c.OpConfig.EnableInitContainers != nil && !(*c.OpConfig.EnableInitContainers) {
+			c.logger.Warningf("initContainers specified but disabled in configuration - next statefulset creation would fail")
+		}
+		initContainers = spec.InitContainers
+	}
+
 	customPodEnvVarsList := make([]v1.EnvVar, 0)
 
 	if c.OpConfig.PodEnvironmentConfigMap != "" {
@@ -867,9 +880,14 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 	}
 
 	// generate sidecar containers
-	if sidecarContainers, err = generateSidecarContainers(sideCars, volumeMounts, defaultResources,
-		c.OpConfig.SuperUsername, c.credentialSecretName(c.OpConfig.SuperUsername), c.logger); err != nil {
-		return nil, fmt.Errorf("could not generate sidecar containers: %v", err)
+	if sideCars != nil && len(sideCars) > 0 {
+		if c.OpConfig.EnableSidecars != nil && !(*c.OpConfig.EnableSidecars) {
+			c.logger.Warningf("sidecars specified but disabled in configuration - next statefulset creation would fail")
+		}
+		if sidecarContainers, err = generateSidecarContainers(sideCars, volumeMounts, defaultResources,
+			c.OpConfig.SuperUsername, c.credentialSecretName(c.OpConfig.SuperUsername), c.logger); err != nil {
+			return nil, fmt.Errorf("could not generate sidecar containers: %v", err)
+		}
 	}
 
 	tolerationSpec := tolerations(&spec.Tolerations, c.OpConfig.PodToleration)
@@ -881,12 +899,15 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 		effectiveFSGroup = spec.SpiloFSGroup
 	}
 
+	annotations := c.generatePodAnnotations(spec)
+
 	// generate pod template for the statefulset, based on the spilo container and sidecars
 	if podTemplate, err = generatePodTemplate(
 		c.Namespace,
 		c.labelsSet(true),
+		annotations,
 		spiloContainer,
-		spec.InitContainers,
+		initContainers,
 		sidecarContainers,
 		&tolerationSpec,
 		effectiveFSGroup,
@@ -949,6 +970,24 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
 	return statefulSet, nil
 }
 
+func (c *Cluster) generatePodAnnotations(spec *acidv1.PostgresSpec) map[string]string {
+	annotations := make(map[string]string)
+	for k, v := range c.OpConfig.CustomPodAnnotations {
+		annotations[k] = v
+	}
+	if spec != nil || spec.PodAnnotations != nil {
+		for k, v := range spec.PodAnnotations {
+			annotations[k] = v
+		}
+	}
+
+	if len(annotations) == 0 {
+		return nil
+	}
+
+	return annotations
+}
+
 func generateScalyrSidecarSpec(clusterName, APIKey, serverURL, dockerImage string,
 	containerResources *acidv1.Resources, logger *logrus.Entry) *acidv1.Sidecar {
 	if APIKey == "" || dockerImage == "" {
@@ -1393,7 +1432,7 @@ func (c *Cluster) generatePodDisruptionBudget() *policybeta1.PodDisruptionBudget
 	pdbEnabled := c.OpConfig.EnablePodDisruptionBudget
 
 	// if PodDisruptionBudget is disabled or if there are no DB pods, set the budget to 0.
-	if (pdbEnabled != nil && !*pdbEnabled) || c.Spec.NumberOfInstances <= 0 {
+	if (pdbEnabled != nil && !(*pdbEnabled)) || c.Spec.NumberOfInstances <= 0 {
 		minAvailable = intstr.FromInt(0)
 	}
 
@@ -1469,10 +1508,13 @@ func (c *Cluster) generateLogicalBackupJob() (*batchv1beta1.CronJob, error) {
 			},
 		}}
 
+	annotations := c.generatePodAnnotations(&c.Spec)
+
 	// re-use the method that generates DB pod templates
 	if podTemplate, err = generatePodTemplate(
 		c.Namespace,
 		labels,
+		annotations,
 		logicalBackupContainer,
 		[]v1.Container{},
 		[]v1.Container{},
@@ -1486,8 +1528,8 @@ func (c *Cluster) generateLogicalBackupJob() (*batchv1beta1.CronJob, error) {
 		util.False(),
 		false,
 		"",
-		"",
-		""); err != nil {
+		c.OpConfig.AdditionalSecretMount,
+		c.OpConfig.AdditionalSecretMountPath); err != nil {
 		return nil, fmt.Errorf("could not generate pod template for logical backup pod: %v", err)
 	}
 
@@ -1535,6 +1577,10 @@ func (c *Cluster) generateLogicalBackupPodEnvVars() []v1.EnvVar {
 			Name:  "SCOPE",
 			Value: c.Name,
 		},
+		{
+			Name:  "CLUSTER_NAME_LABEL",
+			Value: c.OpConfig.ClusterNameLabel,
+		},
 		{
 			Name: "POD_NAMESPACE",
 			ValueFrom: &v1.EnvVarSource{
@@ -1549,6 +1595,14 @@ func (c *Cluster) generateLogicalBackupPodEnvVars() []v1.EnvVar {
 			Name:  "LOGICAL_BACKUP_S3_BUCKET",
 			Value: c.OpConfig.LogicalBackup.LogicalBackupS3Bucket,
 		},
+		{
+			Name:  "LOGICAL_BACKUP_S3_ENDPOINT",
+			Value: c.OpConfig.LogicalBackup.LogicalBackupS3Endpoint,
+		},
+		{
+			Name:  "LOGICAL_BACKUP_S3_SSE",
+			Value: c.OpConfig.LogicalBackup.LogicalBackupS3SSE,
+		},
 		{
 			Name:  "LOGICAL_BACKUP_S3_BUCKET_SCOPE_SUFFIX",
 			Value: getBucketScopeSuffix(string(c.Postgresql.GetUID())),
@@ -1587,8 +1641,15 @@ func (c *Cluster) generateLogicalBackupPodEnvVars() []v1.EnvVar {
 		},
 	}
 
-	c.logger.Debugf("Generated logical backup env vars %v", envVars)
+	if c.OpConfig.LogicalBackup.LogicalBackupS3AccessKeyID != "" {
+		envVars = append(envVars, v1.EnvVar{Name: "AWS_ACCESS_KEY_ID", Value: c.OpConfig.LogicalBackup.LogicalBackupS3AccessKeyID})
+	}
 
+	if c.OpConfig.LogicalBackup.LogicalBackupS3SecretAccessKey != "" {
+		envVars = append(envVars, v1.EnvVar{Name: "AWS_SECRET_ACCESS_KEY", Value: c.OpConfig.LogicalBackup.LogicalBackupS3SecretAccessKey})
+	}
+
+	c.logger.Debugf("Generated logical backup env vars %v", envVars)
 	return envVars
 }
 

pkg/cluster/k8sres_test.go

@@ -3,7 +3,7 @@ package cluster
 import (
 	"reflect"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 
 	"testing"
 

pkg/cluster/resources.go

@@ -13,7 +13,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	"github.com/zalando/postgres-operator/pkg/util"
-	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
 	"github.com/zalando/postgres-operator/pkg/util/retryutil"
 )
@@ -66,6 +65,17 @@ func (c *Cluster) listResources() error {
 
 func (c *Cluster) createStatefulSet() (*appsv1.StatefulSet, error) {
 	c.setProcessName("creating statefulset")
+	// check if it's allowed that spec contains initContainers
+	if c.Spec.InitContainers != nil && len(c.Spec.InitContainers) > 0 &&
+		c.OpConfig.EnableInitContainers != nil && !(*c.OpConfig.EnableInitContainers) {
+		return nil, fmt.Errorf("initContainers specified but disabled in configuration")
+	}
+	// check if it's allowed that spec contains sidecars
+	if c.Spec.Sidecars != nil && len(c.Spec.Sidecars) > 0 &&
+		c.OpConfig.EnableSidecars != nil && !(*c.OpConfig.EnableSidecars) {
+		return nil, fmt.Errorf("sidecar containers specified but disabled in configuration")
+	}
+
 	statefulSetSpec, err := c.generateStatefulSet(&c.Spec)
 	if err != nil {
 		return nil, fmt.Errorf("could not generate statefulset: %v", err)
@@ -278,7 +288,8 @@ func (c *Cluster) replaceStatefulSet(newStatefulSet *appsv1.StatefulSet) error {
 	oldStatefulset := c.Statefulset
 
 	options := metav1.DeleteOptions{PropagationPolicy: &deletePropagationPolicy}
-	if err := c.KubeClient.StatefulSets(oldStatefulset.Namespace).Delete(oldStatefulset.Name, &options); err != nil {
+	err := c.KubeClient.StatefulSets(oldStatefulset.Namespace).Delete(oldStatefulset.Name, &options)
+	if err != nil {
 		return fmt.Errorf("could not delete statefulset %q: %v", statefulSetName, err)
 	}
 	// make sure we clear the stored statefulset status if the subsequent create fails.
@@ -286,11 +297,16 @@ func (c *Cluster) replaceStatefulSet(newStatefulSet *appsv1.StatefulSet) error {
 	// wait until the statefulset is truly deleted
 	c.logger.Debugf("waiting for the statefulset to be deleted")
 
-	err := retryutil.Retry(constants.StatefulsetDeletionInterval, constants.StatefulsetDeletionTimeout,
+	err = retryutil.Retry(c.OpConfig.ResourceCheckInterval, c.OpConfig.ResourceCheckTimeout,
 		func() (bool, error) {
-			_, err := c.KubeClient.StatefulSets(oldStatefulset.Namespace).Get(oldStatefulset.Name, metav1.GetOptions{})
-
-			return err != nil, nil
+			_, err2 := c.KubeClient.StatefulSets(oldStatefulset.Namespace).Get(oldStatefulset.Name, metav1.GetOptions{})
+			if err2 == nil {
+				return false, nil
+			}
+			if k8sutil.ResourceNotFound(err2) {
+				return true, nil
+			}
+			return false, err2
 		})
 	if err != nil {
 		return fmt.Errorf("could not delete statefulset: %v", err)
@@ -380,13 +396,27 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
 			return fmt.Errorf("could not delete service %q: %v", serviceName, err)
 		}
 
-		c.Endpoints[role] = nil
-		svc, err := c.KubeClient.Services(serviceName.Namespace).Create(newService)
+		// wait until the service is truly deleted
+		c.logger.Debugf("waiting for service to be deleted")
+
+		err = retryutil.Retry(c.OpConfig.ResourceCheckInterval, c.OpConfig.ResourceCheckTimeout,
+			func() (bool, error) {
+				_, err2 := c.KubeClient.Services(serviceName.Namespace).Get(serviceName.Name, metav1.GetOptions{})
+				if err2 == nil {
+					return false, nil
+				}
+				if k8sutil.ResourceNotFound(err2) {
+					return true, nil
+				}
+				return false, err2
+			})
 		if err != nil {
-			return fmt.Errorf("could not create service %q: %v", serviceName, err)
+			return fmt.Errorf("could not delete service %q: %v", serviceName, err)
 		}
 
-		c.Services[role] = svc
+		// make sure we clear the stored service and endpoint status if the subsequent create fails.
+		c.Services[role] = nil
+		c.Endpoints[role] = nil
 		if role == Master {
 			// create the new endpoint using the addresses obtained from the previous one
 			endpointSpec := c.generateEndpoint(role, currentEndpoint.Subsets)
@@ -398,6 +428,13 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
 			c.Endpoints[role] = ep
 		}
 
+		svc, err := c.KubeClient.Services(serviceName.Namespace).Create(newService)
+		if err != nil {
+			return fmt.Errorf("could not create service %q: %v", serviceName, err)
+		}
+
+		c.Services[role] = svc
+
 		return nil
 	}
 

pkg/cluster/sync.go

@@ -24,6 +24,7 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
+	oldStatus := c.Status
 	c.setSpec(newSpec)
 
 	defer func() {
@@ -35,6 +36,16 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 		}
 	}()
 
+	if err = c.validateResources(&c.Spec); err != nil {
+		err = fmt.Errorf("insufficient resource limits specified: %v", err)
+		if oldStatus.Running() {
+			c.logger.Warning(err)
+			err = nil
+		} else {
+			return err
+		}
+	}
+
 	if err = c.initUsers(); err != nil {
 		err = fmt.Errorf("could not init users: %v", err)
 		return err

pkg/cluster/types.go

@@ -5,7 +5,7 @@ import (
 
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	appsv1 "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	policybeta1 "k8s.io/api/policy/v1beta1"
 	"k8s.io/apimachinery/pkg/types"
 )