Merge branch 'master' of https://github.com/zalando/postgres-operator into standby

pkg/cluster/cluster.go

@@ -81,6 +81,7 @@ type Cluster struct {
 	currentProcess   Process
 	processMu        sync.RWMutex // protects the current operation for reporting, no need to hold the master mutex
 	specMu           sync.RWMutex // protects the spec for reporting, no need to hold the master mutex
+
 }
 
 type compareStatefulsetResult struct {
@@ -298,6 +299,13 @@ func (c *Cluster) Create() error {
 		c.logger.Infof("databases have been successfully created")
 	}
 
+	if c.Postgresql.Spec.EnableLogicalBackup {
+		if err := c.createLogicalBackupJob(); err != nil {
+			return fmt.Errorf("could not create a k8s cron job for logical backups: %v", err)
+		}
+		c.logger.Info("a k8s cron job for logical backup has been successfully created")
+	}
+
 	if err := c.listResources(); err != nil {
 		c.logger.Errorf("could not list resources: %v", err)
 	}
@@ -334,7 +342,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *comp
 	if c.Statefulset.Spec.Template.Spec.ServiceAccountName != statefulSet.Spec.Template.Spec.ServiceAccountName {
 		needsReplace = true
 		needsRollUpdate = true
-		reasons = append(reasons, "new statefulset's serviceAccountName service asccount name doesn't match the current one")
+		reasons = append(reasons, "new statefulset's serviceAccountName service account name doesn't match the current one")
 	}
 	if *c.Statefulset.Spec.Template.Spec.TerminationGracePeriodSeconds != *statefulSet.Spec.Template.Spec.TerminationGracePeriodSeconds {
 		needsReplace = true
@@ -454,16 +462,16 @@ func (c *Cluster) compareContainers(description string, setA, setB []v1.Containe
 func compareResources(a *v1.ResourceRequirements, b *v1.ResourceRequirements) bool {
 	equal := true
 	if a != nil {
-		equal = compareResoucesAssumeFirstNotNil(a, b)
+		equal = compareResourcesAssumeFirstNotNil(a, b)
 	}
 	if equal && (b != nil) {
-		equal = compareResoucesAssumeFirstNotNil(b, a)
+		equal = compareResourcesAssumeFirstNotNil(b, a)
 	}
 
 	return equal
 }
 
-func compareResoucesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.ResourceRequirements) bool {
+func compareResourcesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.ResourceRequirements) bool {
 	if b == nil || (len(b.Requests) == 0) {
 		return len(a.Requests) == 0
 	}
@@ -481,8 +489,10 @@ func compareResoucesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.Resource
 
 }
 
-// Update changes Kubernetes objects according to the new specification. Unlike the sync case, the missing object.
-// (i.e. service) is treated as an error.
+// 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
+// for a cluster that had no such job before. In this case a missing job is not an error.
 func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	updateFailed := false
 
@@ -569,6 +579,43 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		}
 	}()
 
+	// logical backup job
+	func() {
+
+		// create if it did not exist
+		if !oldSpec.Spec.EnableLogicalBackup && newSpec.Spec.EnableLogicalBackup {
+			c.logger.Debugf("creating backup cron job")
+			if err := c.createLogicalBackupJob(); err != nil {
+				c.logger.Errorf("could not create a k8s cron job for logical backups: %v", err)
+				updateFailed = true
+				return
+			}
+		}
+
+		// delete if no longer needed
+		if oldSpec.Spec.EnableLogicalBackup && !newSpec.Spec.EnableLogicalBackup {
+			c.logger.Debugf("deleting backup cron job")
+			if err := c.deleteLogicalBackupJob(); err != nil {
+				c.logger.Errorf("could not delete a k8s cron job for logical backups: %v", err)
+				updateFailed = true
+				return
+			}
+
+		}
+
+		// apply schedule changes
+		// this is the only parameter of logical backups a user can overwrite in the cluster manifest
+		if (oldSpec.Spec.EnableLogicalBackup && newSpec.Spec.EnableLogicalBackup) &&
+			(newSpec.Spec.LogicalBackupSchedule != oldSpec.Spec.LogicalBackupSchedule) {
+			c.logger.Debugf("updating schedule of the backup cron job")
+			if err := c.syncLogicalBackupJob(); err != nil {
+				c.logger.Errorf("could not sync logical backup jobs: %v", err)
+				updateFailed = true
+			}
+		}
+
+	}()
+
 	// Roles and Databases
 	if !(c.databaseAccessDisabled() || c.getNumberOfInstances(&c.Spec) <= 0) {
 		c.logger.Debugf("syncing roles")
@@ -597,6 +644,12 @@ func (c *Cluster) Delete() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
+	// delete the backup job before the stateful set of the cluster to prevent connections to non-existing pods
+	// deleting the cron job also removes pods and batch jobs it created
+	if err := c.deleteLogicalBackupJob(); err != nil {
+		c.logger.Warningf("could not remove the logical backup k8s cron job; %v", err)
+	}
+
 	if err := c.deleteStatefulSet(); err != nil {
 		c.logger.Warningf("could not delete statefulset: %v", err)
 	}
@@ -629,6 +682,7 @@ func (c *Cluster) Delete() {
 	if err := c.deletePatroniClusterObjects(); err != nil {
 		c.logger.Warningf("could not remove leftover patroni objects; %v", err)
 	}
+
 }
 
 //NeedsRepair returns true if the cluster should be included in the repair scan (based on its in-memory status).
@@ -821,7 +875,7 @@ func (c *Cluster) initInfrastructureRoles() error {
 	return nil
 }
 
-// resolves naming conflicts between existing and new roles by chosing either of them.
+// resolves naming conflicts between existing and new roles by choosing either of them.
 func (c *Cluster) resolveNameConflict(currentRole, newRole *spec.PgUser) spec.PgUser {
 	var result spec.PgUser
 	if newRole.Origin >= currentRole.Origin {
@@ -915,7 +969,7 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
 	// signal the role label waiting goroutine to close the shop and go home
 	close(stopCh)
 	// wait until the goroutine terminates, since unregisterPodSubscriber
-	// must be called before the outer return; otherwsise we risk subscribing to the same pod twice.
+	// must be called before the outer return; otherwise we risk subscribing to the same pod twice.
 	wg.Wait()
 	// close the label waiting channel no sooner than the waiting goroutine terminates.
 	close(podLabelErr)
@@ -950,8 +1004,8 @@ func (c *Cluster) deletePatroniClusterObjects() error {
 	if !c.patroniUsesKubernetes() {
 		c.logger.Infof("not cleaning up Etcd Patroni objects on cluster delete")
 	}
-	c.logger.Debugf("removing leftover Patroni objects (endpoints or configmaps)")
-	for _, deleter := range []simpleActionWithResult{c.deletePatroniClusterEndpoints, c.deletePatroniClusterConfigMaps} {
+	c.logger.Debugf("removing leftover Patroni objects (endpoints, services and configmaps)")
+	for _, deleter := range []simpleActionWithResult{c.deletePatroniClusterEndpoints, c.deletePatroniClusterServices, c.deletePatroniClusterConfigMaps} {
 		if err := deleter(); err != nil {
 			return err
 		}
@@ -983,6 +1037,19 @@ func (c *Cluster) deleteClusterObject(
 	return nil
 }
 
+func (c *Cluster) deletePatroniClusterServices() error {
+	get := func(name string) (spec.NamespacedName, error) {
+		svc, err := c.KubeClient.Services(c.Namespace).Get(name, metav1.GetOptions{})
+		return util.NameFromMeta(svc.ObjectMeta), err
+	}
+
+	deleteServiceFn := func(name string) error {
+		return c.KubeClient.Services(c.Namespace).Delete(name, c.deleteOptions)
+	}
+
+	return c.deleteClusterObject(get, deleteServiceFn, "service")
+}
+
 func (c *Cluster) deletePatroniClusterEndpoints() error {
 	get := func(name string) (spec.NamespacedName, error) {
 		ep, err := c.KubeClient.Endpoints(c.Namespace).Get(name, metav1.GetOptions{})

pkg/cluster/database.go

@@ -34,7 +34,7 @@ func (c *Cluster) pgConnectionString() string {
 	password := c.systemUsers[constants.SuperuserKeyName].Password
 
 	return fmt.Sprintf("host='%s' dbname=postgres sslmode=require user='%s' password='%s' connect_timeout='%d'",
-		fmt.Sprintf("%s.%s.svc.cluster.local", c.Name, c.Namespace),
+		fmt.Sprintf("%s.%s.svc.%s", c.Name, c.Namespace, c.OpConfig.ClusterDomain),
 		c.systemUsers[constants.SuperuserKeyName].Name,
 		strings.Replace(password, "$", "\\$", -1),
 		constants.PostgresConnectTimeout/time.Second)

pkg/cluster/k8sres.go

@@ -20,6 +20,8 @@ import (
 	"github.com/zalando/postgres-operator/pkg/util"
 	"github.com/zalando/postgres-operator/pkg/util/config"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
+	batchv1 "k8s.io/api/batch/v1"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	"k8s.io/apimachinery/pkg/labels"
 )
 
@@ -147,7 +149,7 @@ func fillResourceList(spec acidv1.ResourceDescription, defaults acidv1.ResourceD
 	return requests, nil
 }
 
-func generateSpiloJSONConfiguration(pg *acidv1.PostgresqlParam, patroni *acidv1.Patroni, pamRoleName string, logger *logrus.Entry) string {
+func generateSpiloJSONConfiguration(pg *acidv1.PostgresqlParam, patroni *acidv1.Patroni, pamRoleName string, logger *logrus.Entry) (string, error) {
 	config := spiloConfiguration{}
 
 	config.Bootstrap = pgBootstrap{}
@@ -247,12 +249,9 @@ PatroniInitDBParams:
 			Options:  []string{constants.RoleFlagCreateDB, constants.RoleFlagNoLogin},
 		},
 	}
-	result, err := json.Marshal(config)
-	if err != nil {
-		logger.Errorf("cannot convert spilo configuration into JSON: %v", err)
-		return ""
-	}
-	return string(result)
+
+	res, err := json.Marshal(config)
+	return string(res), err
 }
 
 func getLocalAndBoostrapPostgreSQLParameters(parameters map[string]string) (local, bootstrap map[string]string) {
@@ -330,7 +329,7 @@ func tolerations(tolerationsSpec *[]v1.Toleration, podToleration map[string]stri
 	return []v1.Toleration{}
 }
 
-// isBootstrapOnlyParameter checks asgainst special Patroni bootstrap parameters.
+// isBootstrapOnlyParameter checks against special Patroni bootstrap parameters.
 // Those parameters must go to the bootstrap/dcs/postgresql/parameters section.
 // See http://patroni.readthedocs.io/en/latest/dynamic_configuration.html.
 func isBootstrapOnlyParameter(param string) bool {
@@ -352,7 +351,7 @@ func generateVolumeMounts() []v1.VolumeMount {
 	}
 }
 
-func generateSpiloContainer(
+func generateContainer(
 	name string,
 	dockerImage *string,
 	resourceRequirements *v1.ResourceRequirements,
@@ -433,6 +432,7 @@ func generatePodTemplate(
 	initContainers []v1.Container,
 	sidecarContainers []v1.Container,
 	tolerationsSpec *[]v1.Toleration,
+	spiloFSGroup *int64,
 	nodeAffinity *v1.Affinity,
 	terminateGracePeriod int64,
 	podServiceAccountName string,
@@ -446,6 +446,11 @@ func generatePodTemplate(
 	terminateGracePeriodSeconds := terminateGracePeriod
 	containers := []v1.Container{*spiloContainer}
 	containers = append(containers, sidecarContainers...)
+	securityContext := v1.PodSecurityContext{}
+
+	if spiloFSGroup != nil {
+		securityContext.FSGroup = spiloFSGroup
+	}
 
 	podSpec := v1.PodSpec{
 		ServiceAccountName:            podServiceAccountName,
@@ -453,6 +458,7 @@ func generatePodTemplate(
 		Containers:                    containers,
 		InitContainers:                initContainers,
 		Tolerations:                   *tolerationsSpec,
+		SecurityContext:               &securityContext,
 	}
 
 	if shmVolume {
@@ -785,7 +791,10 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
 		return nil, fmt.Errorf("s3_wal_path is empty for standby cluster")
 	}
 
-	spiloConfiguration := generateSpiloJSONConfiguration(&spec.PostgresqlParam, &spec.Patroni, c.OpConfig.PamRoleName, c.logger)
+	spiloConfiguration, err := generateSpiloJSONConfiguration(&spec.PostgresqlParam, &spec.Patroni, c.OpConfig.PamRoleName, c.logger)
+	if err != nil {
+		return nil, fmt.Errorf("could not generate Spilo JSON configuration: %v", err)
+	}
 
 	// generate environment variables for the spilo container
 	spiloEnvVars := deduplicateEnvVars(
@@ -799,7 +808,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
 
 	// generate the spilo container
 	c.logger.Debugf("Generating Spilo container, environment variables: %v", spiloEnvVars)
-	spiloContainer := generateSpiloContainer(c.containerName(),
+	spiloContainer := generateContainer(c.containerName(),
 		&effectiveDockerImage,
 		resourceRequirements,
 		spiloEnvVars,
@@ -836,6 +845,12 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
 	tolerationSpec := tolerations(&spec.Tolerations, c.OpConfig.PodToleration)
 	effectivePodPriorityClassName := util.Coalesce(spec.PodPriorityClassName, c.OpConfig.PodPriorityClassName)
 
+	// determine the FSGroup for the spilo pod
+	effectiveFSGroup := c.OpConfig.Resources.SpiloFSGroup
+	if spec.SpiloFSGroup != nil {
+		effectiveFSGroup = spec.SpiloFSGroup
+	}
+
 	// generate pod template for the statefulset, based on the spilo container and sidecars
 	if podTemplate, err = generatePodTemplate(
 		c.Namespace,
@@ -844,6 +859,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
 		spec.InitContainers,
 		sidecarContainers,
 		&tolerationSpec,
+		effectiveFSGroup,
 		nodeAffinity(c.OpConfig.NodeReadinessLabel),
 		int64(c.OpConfig.PodTerminateGracePeriod.Seconds()),
 		c.OpConfig.PodServiceAccountName,
@@ -1021,6 +1037,7 @@ func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string
 		return nil, fmt.Errorf("could not parse volume size: %v", err)
 	}
 
+	volumeMode := v1.PersistentVolumeFilesystem
 	volumeClaim := &v1.PersistentVolumeClaim{
 		ObjectMeta: metadata,
 		Spec: v1.PersistentVolumeClaimSpec{
@@ -1031,6 +1048,7 @@ func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string
 				},
 			},
 			StorageClassName: storageClassName,
+			VolumeMode:       &volumeMode,
 		},
 	}
 
@@ -1310,3 +1328,168 @@ func (c *Cluster) getClusterServiceConnectionParameters(clusterName string) (hos
 	port = "5432"
 	return
 }
+
+func (c *Cluster) generateLogicalBackupJob() (*batchv1beta1.CronJob, error) {
+
+	var (
+		err                  error
+		podTemplate          *v1.PodTemplateSpec
+		resourceRequirements *v1.ResourceRequirements
+	)
+
+	// NB: a cron job creates standard batch jobs according to schedule; these batch jobs manage pods and clean-up
+
+	c.logger.Debug("Generating logical backup pod template")
+
+	// allocate for the backup pod the same amount of resources as for normal DB pods
+	defaultResources := c.makeDefaultResources()
+	resourceRequirements, err = generateResourceRequirements(c.Spec.Resources, defaultResources)
+	if err != nil {
+		return nil, fmt.Errorf("could not generate resource requirements for logical backup pods: %v", err)
+	}
+
+	envVars := c.generateLogicalBackupPodEnvVars()
+	logicalBackupContainer := generateContainer(
+		"logical-backup",
+		&c.OpConfig.LogicalBackup.LogicalBackupDockerImage,
+		resourceRequirements,
+		envVars,
+		[]v1.VolumeMount{},
+		c.OpConfig.SpiloPrivileged, // use same value as for normal DB pods
+	)
+
+	labels := map[string]string{
+		"version":     c.Name,
+		"application": "spilo-logical-backup",
+	}
+	podAffinityTerm := v1.PodAffinityTerm{
+		LabelSelector: &metav1.LabelSelector{
+			MatchLabels: labels,
+		},
+		TopologyKey: "kubernetes.io/hostname",
+	}
+	podAffinity := v1.Affinity{
+		PodAffinity: &v1.PodAffinity{
+			PreferredDuringSchedulingIgnoredDuringExecution: []v1.WeightedPodAffinityTerm{{
+				Weight:          1,
+				PodAffinityTerm: podAffinityTerm,
+			},
+			},
+		}}
+
+	// re-use the method that generates DB pod templates
+	if podTemplate, err = generatePodTemplate(
+		c.Namespace,
+		c.labelsSet(true),
+		logicalBackupContainer,
+		[]v1.Container{},
+		[]v1.Container{},
+		&[]v1.Toleration{},
+		nil,
+		nodeAffinity(c.OpConfig.NodeReadinessLabel),
+		int64(c.OpConfig.PodTerminateGracePeriod.Seconds()),
+		c.OpConfig.PodServiceAccountName,
+		c.OpConfig.KubeIAMRole,
+		"",
+		false,
+		false,
+		""); err != nil {
+		return nil, fmt.Errorf("could not generate pod template for logical backup pod: %v", err)
+	}
+
+	// overwrite specific params of logical backups pods
+	podTemplate.Spec.Affinity = &podAffinity
+	podTemplate.Spec.RestartPolicy = "Never" // affects containers within a pod
+
+	// configure a batch job
+
+	jobSpec := batchv1.JobSpec{
+		Template: *podTemplate,
+	}
+
+	// configure a cron job
+
+	jobTemplateSpec := batchv1beta1.JobTemplateSpec{
+		Spec: jobSpec,
+	}
+
+	schedule := c.Postgresql.Spec.LogicalBackupSchedule
+	if schedule == "" {
+		schedule = c.OpConfig.LogicalBackupSchedule
+	}
+
+	cronJob := &batchv1beta1.CronJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      c.getLogicalBackupJobName(),
+			Namespace: c.Namespace,
+			Labels:    c.labelsSet(true),
+		},
+		Spec: batchv1beta1.CronJobSpec{
+			Schedule:          schedule,
+			JobTemplate:       jobTemplateSpec,
+			ConcurrencyPolicy: batchv1beta1.ForbidConcurrent,
+		},
+	}
+
+	return cronJob, nil
+}
+
+func (c *Cluster) generateLogicalBackupPodEnvVars() []v1.EnvVar {
+
+	envVars := []v1.EnvVar{
+		{
+			Name:  "SCOPE",
+			Value: c.Name,
+		},
+		// Bucket env vars
+		{
+			Name:  "LOGICAL_BACKUP_S3_BUCKET",
+			Value: c.OpConfig.LogicalBackup.LogicalBackupS3Bucket,
+		},
+		{
+			Name:  "LOGICAL_BACKUP_S3_BUCKET_SCOPE_SUFFIX",
+			Value: getBucketScopeSuffix(string(c.Postgresql.GetUID())),
+		},
+		// Postgres env vars
+		{
+			Name:  "PG_VERSION",
+			Value: c.Spec.PgVersion,
+		},
+		{
+			Name:  "PGPORT",
+			Value: "5432",
+		},
+		{
+			Name:  "PGUSER",
+			Value: c.OpConfig.SuperUsername,
+		},
+		{
+			Name:  "PGDATABASE",
+			Value: c.OpConfig.SuperUsername,
+		},
+		{
+			Name:  "PGSSLMODE",
+			Value: "require",
+		},
+		{
+			Name: "PGPASSWORD",
+			ValueFrom: &v1.EnvVarSource{
+				SecretKeyRef: &v1.SecretKeySelector{
+					LocalObjectReference: v1.LocalObjectReference{
+						Name: c.credentialSecretName(c.OpConfig.SuperUsername),
+					},
+					Key: "password",
+				},
+			},
+		},
+	}
+
+	c.logger.Debugf("Generated logical backup env vars %v", envVars)
+
+	return envVars
+}
+
+// getLogicalBackupJobName returns the name; the job itself may not exists
+func (c *Cluster) getLogicalBackupJobName() (jobName string) {
+	return "logical-backup-" + c.clusterName().Name
+}

pkg/cluster/k8sres_test.go

@@ -3,11 +3,12 @@ package cluster
 import (
 	"k8s.io/api/core/v1"
 
+	"testing"
+
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	"github.com/zalando/postgres-operator/pkg/util/config"
 	"github.com/zalando/postgres-operator/pkg/util/constants"
 	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
-	"testing"
 )
 
 func True() *bool {
@@ -20,6 +21,69 @@ func False() *bool {
 	return &b
 }
 
+func TestGenerateSpiloJSONConfiguration(t *testing.T) {
+	var cluster = New(
+		Config{
+			OpConfig: config.Config{
+				ProtectedRoles: []string{"admin"},
+				Auth: config.Auth{
+					SuperUsername:       superUserName,
+					ReplicationUsername: replicationUserName,
+				},
+			},
+		}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
+
+	testName := "TestGenerateSpiloConfig"
+	tests := []struct {
+		subtest  string
+		pgParam  *acidv1.PostgresqlParam
+		patroni  *acidv1.Patroni
+		role     string
+		opConfig config.Config
+		result   string
+	}{
+		{
+			subtest:  "Patroni default configuration",
+			pgParam:  &acidv1.PostgresqlParam{PgVersion: "9.6"},
+			patroni:  &acidv1.Patroni{},
+			role:     "zalandos",
+			opConfig: config.Config{},
+			result:   `{"postgresql":{"bin_dir":"/usr/lib/postgresql/9.6/bin"},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"}],"users":{"zalandos":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{}}}`,
+		},
+		{
+			subtest: "Patroni configured",
+			pgParam: &acidv1.PostgresqlParam{PgVersion: "11"},
+			patroni: &acidv1.Patroni{
+				InitDB: map[string]string{
+					"encoding":       "UTF8",
+					"locale":         "en_US.UTF-8",
+					"data-checksums": "true",
+				},
+				PgHba:                []string{"hostssl all all 0.0.0.0/0 md5", "host    all all 0.0.0.0/0 md5"},
+				TTL:                  30,
+				LoopWait:             10,
+				RetryTimeout:         10,
+				MaximumLagOnFailover: 33554432,
+				Slots:                map[string]map[string]string{"permanent_logical_1": {"type": "logical", "database": "foo", "plugin": "pgoutput"}},
+			},
+			role:     "zalandos",
+			opConfig: config.Config{},
+			result:   `{"postgresql":{"bin_dir":"/usr/lib/postgresql/11/bin","pg_hba":["hostssl all all 0.0.0.0/0 md5","host    all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"zalandos":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"slots":{"permanent_logical_1":{"database":"foo","plugin":"pgoutput","type":"logical"}}}}}`,
+		},
+	}
+	for _, tt := range tests {
+		cluster.OpConfig = tt.opConfig
+		result, err := generateSpiloJSONConfiguration(tt.pgParam, tt.patroni, tt.role, logger)
+		if err != nil {
+			t.Errorf("Unexpected error: %v", err)
+		}
+		if tt.result != result {
+			t.Errorf("%s %s: Spilo Config is %v, expected %v for role %#v and param %#v",
+				testName, tt.subtest, result, tt.result, tt.role, tt.pgParam)
+		}
+	}
+}
+
 func TestCreateLoadBalancerLogic(t *testing.T) {
 	var cluster = New(
 		Config{

pkg/cluster/resources.go

@@ -6,7 +6,8 @@ import (
 	"strings"
 
 	"k8s.io/api/apps/v1beta1"
-	"k8s.io/api/core/v1"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
+	v1 "k8s.io/api/core/v1"
 	policybeta1 "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -360,7 +361,7 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
 	// TODO: check if it possible to change the service type with a patch in future versions of Kubernetes
 	if newService.Spec.Type != c.Services[role].Spec.Type {
 		// service type has changed, need to replace the service completely.
-		// we cannot use just pach the current service, since it may contain attributes incompatible with the new type.
+		// we cannot use just patch the current service, since it may contain attributes incompatible with the new type.
 		var (
 			currentEndpoint *v1.Endpoints
 			err             error
@@ -368,7 +369,7 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
 
 		if role == Master {
 			// for the master service we need to re-create the endpoint as well. Get the up-to-date version of
-			// the addresses stored in it before the service is deleted (deletion of the service removes the endpooint)
+			// the addresses stored in it before the service is deleted (deletion of the service removes the endpoint)
 			currentEndpoint, err = c.KubeClient.Endpoints(c.Namespace).Get(c.endpointName(role), metav1.GetOptions{})
 			if err != nil {
 				return fmt.Errorf("could not get current cluster %s endpoints: %v", role, err)
@@ -609,6 +610,51 @@ func (c *Cluster) createRoles() (err error) {
 	return c.syncRoles()
 }
 
+func (c *Cluster) createLogicalBackupJob() (err error) {
+
+	c.setProcessName("creating a k8s cron job for logical backups")
+
+	logicalBackupJobSpec, err := c.generateLogicalBackupJob()
+	if err != nil {
+		return fmt.Errorf("could not generate k8s cron job spec: %v", err)
+	}
+	c.logger.Debugf("Generated cronJobSpec: %v", logicalBackupJobSpec)
+
+	_, err = c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Create(logicalBackupJobSpec)
+	if err != nil {
+		return fmt.Errorf("could not create k8s cron job: %v", err)
+	}
+
+	return nil
+}
+
+func (c *Cluster) patchLogicalBackupJob(newJob *batchv1beta1.CronJob) error {
+	c.setProcessName("patching logical backup job")
+
+	patchData, err := specPatch(newJob.Spec)
+	if err != nil {
+		return fmt.Errorf("could not form patch for the logical backup job: %v", err)
+	}
+
+	// update the backup job spec
+	_, err = c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Patch(
+		c.getLogicalBackupJobName(),
+		types.MergePatchType,
+		patchData, "")
+	if err != nil {
+		return fmt.Errorf("could not patch logical backup job: %v", err)
+	}
+
+	return nil
+}
+
+func (c *Cluster) deleteLogicalBackupJob() error {
+
+	c.logger.Info("removing the logical backup job")
+
+	return c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Delete(c.getLogicalBackupJobName(), c.deleteOptions)
+}
+
 // GetServiceMaster returns cluster's kubernetes master Service
 func (c *Cluster) GetServiceMaster() *v1.Service {
 	return c.Services[Master]

pkg/cluster/sync.go

@@ -3,7 +3,8 @@ package cluster
 import (
 	"fmt"
 
-	"k8s.io/api/core/v1"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
+	v1 "k8s.io/api/core/v1"
 	policybeta1 "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -92,6 +93,16 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 		return err
 	}
 
+	// create a logical backup job unless we are running without pods or disable that feature explicitly
+	if c.Spec.EnableLogicalBackup && c.getNumberOfInstances(&c.Spec) > 0 {
+
+		c.logger.Debug("syncing logical backup job")
+		if err = c.syncLogicalBackupJob(); err != nil {
+			err = fmt.Errorf("could not sync the logical backup job: %v", err)
+			return err
+		}
+	}
+
 	return err
 }
 
@@ -519,3 +530,56 @@ func (c *Cluster) syncDatabases() error {
 
 	return nil
 }
+
+func (c *Cluster) syncLogicalBackupJob() error {
+	var (
+		job        *batchv1beta1.CronJob
+		desiredJob *batchv1beta1.CronJob
+		err        error
+	)
+	c.setProcessName("syncing the logical backup job")
+
+	// sync the job if it exists
+
+	jobName := c.getLogicalBackupJobName()
+	if job, err = c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Get(jobName, metav1.GetOptions{}); err == nil {
+
+		desiredJob, err = c.generateLogicalBackupJob()
+		if err != nil {
+			return fmt.Errorf("could not generate the desired logical backup job state: %v", err)
+		}
+		if match, reason := k8sutil.SameLogicalBackupJob(job, desiredJob); !match {
+			c.logger.Infof("logical job %q is not in the desired state and needs to be updated",
+				c.getLogicalBackupJobName(),
+			)
+			if reason != "" {
+				c.logger.Infof("reason: %s", reason)
+			}
+			if err = c.patchLogicalBackupJob(desiredJob); err != nil {
+				return fmt.Errorf("could not update logical backup job to match desired state: %v", err)
+			}
+			c.logger.Info("the logical backup job is synced")
+		}
+		return nil
+	}
+	if !k8sutil.ResourceNotFound(err) {
+		return fmt.Errorf("could not get logical backp job: %v", err)
+	}
+
+	// no existing logical backup job, create new one
+	c.logger.Info("could not find the cluster's logical backup job")
+
+	if err = c.createLogicalBackupJob(); err == nil {
+		c.logger.Infof("created missing logical backup job %q", jobName)
+	} else {
+		if !k8sutil.ResourceAlreadyExists(err) {
+			return fmt.Errorf("could not create missing logical backup job: %v", err)
+		}
+		c.logger.Infof("logical backup job %q already exists", jobName)
+		if _, err = c.KubeClient.CronJobsGetter.CronJobs(c.Namespace).Get(jobName, metav1.GetOptions{}); err != nil {
+			return fmt.Errorf("could not fetch existing logical backup job: %v", err)
+		}
+	}
+
+	return nil
+}

pkg/cluster/util.go

@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"k8s.io/api/apps/v1beta1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	policybeta1 "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"