create CDC event stream CRD (#1570)

* provide event stream API
* check manifest settings for logical decoding before creating streams
* operator updates Postgres config and creates replication user
* name FES like the Postgres cluster
* add delete case and fix updating streams + update unit test
* check if fes CRD exists before syncing
* existing slot must use the same plugin
* make id and payload columns configurable
* sync streams only when they are defined in manifest
* introduce applicationId for separate stream CRDs
* add FES to RBAC in chart
* disable streams in chart
* switch to pgoutput plugin and let operator create publications
* reflect code review and additional refactoring

Co-authored-by: Paŭlo Ebermann <paul.ebermann@zalando.de>

pkg/cluster/cluster.go

@@ -361,6 +361,12 @@ func (c *Cluster) Create() error {
 	// something fails, report warning
 	c.createConnectionPooler(c.installLookupFunction)
 
+	if len(c.Spec.Streams) > 0 {
+		if err = c.syncStreams(); err != nil {
+			c.logger.Errorf("could not create streams: %v", err)
+		}
+	}
+
 	return nil
 }
 
@@ -861,6 +867,13 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 		updateFailed = true
 	}
 
+	if len(c.Spec.Streams) > 0 {
+		if err := c.syncStreams(); err != nil {
+			c.logger.Errorf("could not sync streams: %v", err)
+			updateFailed = true
+		}
+	}
+
 	if !updateFailed {
 		// Major version upgrade must only fire after success of earlier operations and should stay last
 		if err := c.majorVersionUpgrade(); err != nil {
@@ -896,6 +909,10 @@ func (c *Cluster) Delete() {
 	defer c.mu.Unlock()
 	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Delete", "Started deletion of new cluster resources")
 
+	if err := c.deleteStreams(); err != nil {
+		c.logger.Warningf("could not delete event streams: %v", err)
+	}
+
 	// 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 {
@@ -1050,6 +1067,23 @@ func (c *Cluster) initSystemUsers() {
 			c.systemUsers[constants.ConnectionPoolerUserKeyName] = connectionPoolerUser
 		}
 	}
+
+	// replication users for event streams are another exception
+	// the operator will create one replication user for all streams
+	if len(c.Spec.Streams) > 0 {
+		username := constants.EventStreamSourceSlotPrefix + constants.UserRoleNameSuffix
+		streamUser := spec.PgUser{
+			Origin:    spec.RoleConnectionPooler,
+			Name:      username,
+			Namespace: c.Namespace,
+			Flags:     []string{constants.RoleFlagLogin, constants.RoleFlagReplication},
+			Password:  util.RandomPassword(constants.PasswordLength),
+		}
+
+		if _, exists := c.pgUsers[username]; !exists {
+			c.pgUsers[username] = streamUser
+		}
+	}
 }
 
 func (c *Cluster) initPreparedDatabaseRoles() error {

pkg/cluster/database.go

@@ -46,6 +46,13 @@ const (
 	createExtensionSQL      = `CREATE EXTENSION IF NOT EXISTS "%s" SCHEMA "%s"`
 	alterExtensionSQL       = `ALTER EXTENSION "%s" SET SCHEMA "%s"`
 
+	getPublicationsSQL = `SELECT p.pubname, string_agg(pt.schemaname || '.' || pt.tablename, ', ' ORDER BY pt.schemaname, pt.tablename)
+	        FROM pg_publication p
+			JOIN pg_publication_tables pt ON pt.pubname = p.pubname
+			GROUP BY p.pubname;`
+	createPublicationSQL = `CREATE PUBLICATION "%s" FOR TABLE %s WITH (publish = 'insert, update');`
+	alterPublicationSQL  = `ALTER PUBLICATION "%s" SET TABLE %s;`
+
 	globalDefaultPrivilegesSQL = `SET ROLE TO "%s";
 			ALTER DEFAULT PRIVILEGES GRANT USAGE ON SCHEMAS TO "%s","%s";
 			ALTER DEFAULT PRIVILEGES GRANT SELECT ON TABLES TO "%s";
@@ -572,6 +579,68 @@ func (c *Cluster) execCreateOrAlterExtension(extName, schemaName, statement, doi
 	return nil
 }
 
+// getPublications returns the list of current database publications with tables
+// The caller is responsible for opening and closing the database connection
+func (c *Cluster) getPublications() (publications map[string]string, err error) {
+	var (
+		rows *sql.Rows
+	)
+
+	if rows, err = c.pgDb.Query(getPublicationsSQL); err != nil {
+		return nil, fmt.Errorf("could not query database publications: %v", err)
+	}
+
+	defer func() {
+		if err2 := rows.Close(); err2 != nil {
+			if err != nil {
+				err = fmt.Errorf("error when closing query cursor: %v, previous error: %v", err2, err)
+			} else {
+				err = fmt.Errorf("error when closing query cursor: %v", err2)
+			}
+		}
+	}()
+
+	dbPublications := make(map[string]string)
+
+	for rows.Next() {
+		var (
+			dbPublication       string
+			dbPublicationTables string
+		)
+
+		if err = rows.Scan(&dbPublication, &dbPublicationTables); err != nil {
+			return nil, fmt.Errorf("error when processing row: %v", err)
+		}
+		dbPublications[dbPublication] = dbPublicationTables
+	}
+
+	return dbPublications, err
+}
+
+// executeCreatePublication creates new publication for given tables
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeCreatePublication(pubName, tableList string) error {
+	return c.execCreateOrAlterPublication(pubName, tableList, createPublicationSQL,
+		"creating publication", "create publication")
+}
+
+// executeAlterExtension changes the table list of the given publication.
+// The caller is responsible for opening and closing the database connection.
+func (c *Cluster) executeAlterPublication(pubName, tableList string) error {
+	return c.execCreateOrAlterPublication(pubName, tableList, alterPublicationSQL,
+		"changing publication", "alter publication tables")
+}
+
+func (c *Cluster) execCreateOrAlterPublication(pubName, tableList, statement, doing, operation string) error {
+
+	c.logger.Debugf("%s %q with table list %q", doing, pubName, tableList)
+	if _, err := c.pgDb.Exec(fmt.Sprintf(statement, pubName, tableList)); err != nil {
+		return fmt.Errorf("could not execute %s: %v", operation, err)
+	}
+
+	return nil
+}
+
 // Creates a connection pool credentials lookup function in every database to
 // perform remote authentication.
 func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {

pkg/cluster/streams.go

@@ -0,0 +1,363 @@
+package cluster
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+
+	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
+	zalandov1 "github.com/zalando/postgres-operator/pkg/apis/zalando.org/v1"
+	"github.com/zalando/postgres-operator/pkg/util"
+	"github.com/zalando/postgres-operator/pkg/util/constants"
+	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (c *Cluster) createStreams(appId string) error {
+	c.setProcessName("creating streams")
+
+	fes := c.generateFabricEventStream(appId)
+	if _, err := c.KubeClient.FabricEventStreams(c.Namespace).Create(context.TODO(), fes, metav1.CreateOptions{}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *Cluster) updateStreams(newEventStreams *zalandov1.FabricEventStream) error {
+	c.setProcessName("updating event streams")
+
+	if _, err := c.KubeClient.FabricEventStreams(newEventStreams.Namespace).Update(context.TODO(), newEventStreams, metav1.UpdateOptions{}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *Cluster) deleteStreams() error {
+	c.setProcessName("deleting event streams")
+
+	// check if stream CRD is installed before trying a delete
+	_, err := c.KubeClient.CustomResourceDefinitions().Get(context.TODO(), constants.EventStreamCRDName, metav1.GetOptions{})
+	if k8sutil.ResourceNotFound(err) {
+		return nil
+	}
+
+	err = c.KubeClient.FabricEventStreams(c.Namespace).Delete(context.TODO(), c.Name, metav1.DeleteOptions{})
+	if err != nil {
+		return fmt.Errorf("could not delete event stream custom resource: %v", err)
+	}
+
+	return nil
+}
+
+func gatherApplicationIds(streams []acidv1.Stream) []string {
+	appIds := make([]string, 0)
+	for _, stream := range streams {
+		if !util.SliceContains(appIds, stream.ApplicationId) {
+			appIds = append(appIds, stream.ApplicationId)
+		}
+	}
+
+	return appIds
+}
+
+func (c *Cluster) syncPostgresConfig(requiredPatroniConfig acidv1.Patroni) error {
+	errorMsg := "no pods found to update config"
+
+	// if streams are defined wal_level must be switched to logical
+	requiredPgParameters := map[string]string{"wal_level": "logical"}
+
+	// apply config changes in pods
+	pods, err := c.listPods()
+	if err != nil {
+		errorMsg = fmt.Sprintf("could not list pods of the statefulset: %v", err)
+	}
+	for i, pod := range pods {
+		podName := util.NameFromMeta(pods[i].ObjectMeta)
+		effectivePatroniConfig, effectivePgParameters, err := c.patroni.GetConfig(&pod)
+		if err != nil {
+			errorMsg = fmt.Sprintf("could not get Postgres config from pod %s: %v", podName, err)
+			continue
+		}
+
+		_, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, effectivePatroniConfig, requiredPatroniConfig, effectivePgParameters, requiredPgParameters)
+		if err != nil {
+			errorMsg = fmt.Sprintf("could not set PostgreSQL configuration options for pod %s: %v", podName, err)
+			continue
+		}
+
+		// Patroni's config endpoint is just a "proxy" to DCS. It is enough to patch it only once and it doesn't matter which pod is used
+		return nil
+	}
+
+	return fmt.Errorf(errorMsg)
+}
+
+func (c *Cluster) syncPublication(publication, dbName string, tables map[string]acidv1.StreamTable) error {
+	createPublications := make(map[string]string)
+	alterPublications := make(map[string]string)
+
+	defer func() {
+		if err := c.closeDbConn(); err != nil {
+			c.logger.Errorf("could not close database connection: %v", err)
+		}
+	}()
+
+	// check for existing publications
+	if err := c.initDbConnWithName(dbName); err != nil {
+		return fmt.Errorf("could not init database connection")
+	}
+
+	currentPublications, err := c.getPublications()
+	if err != nil {
+		return fmt.Errorf("could not get current publications: %v", err)
+	}
+
+	tableNames := make([]string, len(tables))
+	i := 0
+	for t := range tables {
+		tableName, schemaName := getTableSchema(t)
+		tableNames[i] = fmt.Sprintf("%s.%s", schemaName, tableName)
+		i++
+	}
+	sort.Strings(tableNames)
+	tableList := strings.Join(tableNames, ", ")
+
+	currentTables, exists := currentPublications[publication]
+	if !exists {
+		createPublications[publication] = tableList
+	} else if currentTables != tableList {
+		alterPublications[publication] = tableList
+	}
+
+	if len(createPublications)+len(alterPublications) == 0 {
+		return nil
+	}
+
+	for publicationName, tables := range createPublications {
+		if err = c.executeCreatePublication(publicationName, tables); err != nil {
+			return fmt.Errorf("creation of publication %q failed: %v", publicationName, err)
+		}
+	}
+	for publicationName, tables := range alterPublications {
+		if err = c.executeAlterPublication(publicationName, tables); err != nil {
+			return fmt.Errorf("update of publication %q failed: %v", publicationName, err)
+		}
+	}
+
+	return nil
+}
+
+func (c *Cluster) generateFabricEventStream(appId string) *zalandov1.FabricEventStream {
+	eventStreams := make([]zalandov1.EventStream, 0)
+
+	for _, stream := range c.Spec.Streams {
+		if stream.ApplicationId != appId {
+			continue
+		}
+		for tableName, table := range stream.Tables {
+			streamSource := c.getEventStreamSource(stream, tableName, table.IdColumn)
+			streamFlow := getEventStreamFlow(stream, table.PayloadColumn)
+			streamSink := getEventStreamSink(stream, table.EventType)
+
+			eventStreams = append(eventStreams, zalandov1.EventStream{
+				EventStreamFlow:   streamFlow,
+				EventStreamSink:   streamSink,
+				EventStreamSource: streamSource})
+		}
+	}
+
+	return &zalandov1.FabricEventStream{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: constants.EventStreamCRDApiVersion,
+			Kind:       constants.EventStreamCRDKind,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        fmt.Sprintf("%s-%s", c.Name, appId),
+			Namespace:   c.Namespace,
+			Annotations: c.AnnotationsToPropagate(c.annotationsSet(nil)),
+			// make cluster StatefulSet the owner (like with connection pooler objects)
+			OwnerReferences: c.ownerReferences(),
+		},
+		Spec: zalandov1.FabricEventStreamSpec{
+			ApplicationId: appId,
+			EventStreams:  eventStreams,
+		},
+	}
+}
+
+func (c *Cluster) getEventStreamSource(stream acidv1.Stream, tableName, idColumn string) zalandov1.EventStreamSource {
+	table, schema := getTableSchema(tableName)
+	streamFilter := stream.Filter[tableName]
+	return zalandov1.EventStreamSource{
+		Type:             constants.EventStreamSourcePGType,
+		Schema:           schema,
+		EventStreamTable: getOutboxTable(table, idColumn),
+		Filter:           streamFilter,
+		Connection: c.getStreamConnection(
+			stream.Database,
+			constants.EventStreamSourceSlotPrefix+constants.UserRoleNameSuffix,
+			stream.ApplicationId),
+	}
+}
+
+func getEventStreamFlow(stream acidv1.Stream, payloadColumn string) zalandov1.EventStreamFlow {
+	return zalandov1.EventStreamFlow{
+		Type:          constants.EventStreamFlowPgGenericType,
+		PayloadColumn: payloadColumn,
+	}
+}
+
+func getEventStreamSink(stream acidv1.Stream, eventType string) zalandov1.EventStreamSink {
+	return zalandov1.EventStreamSink{
+		Type:         constants.EventStreamSinkNakadiType,
+		EventType:    eventType,
+		MaxBatchSize: stream.BatchSize,
+	}
+}
+
+func getTableSchema(fullTableName string) (tableName, schemaName string) {
+	schemaName = "public"
+	tableName = fullTableName
+	if strings.Contains(fullTableName, ".") {
+		schemaName = strings.Split(fullTableName, ".")[0]
+		tableName = strings.Split(fullTableName, ".")[1]
+	}
+
+	return tableName, schemaName
+}
+
+func getOutboxTable(tableName, idColumn string) zalandov1.EventStreamTable {
+	return zalandov1.EventStreamTable{
+		Name:     tableName,
+		IDColumn: idColumn,
+	}
+}
+
+func getSlotName(dbName, appId string) string {
+	return fmt.Sprintf("%s_%s_%s", constants.EventStreamSourceSlotPrefix, dbName, strings.Replace(appId, "-", "_", -1))
+}
+
+func (c *Cluster) getStreamConnection(database, user, appId string) zalandov1.Connection {
+	return zalandov1.Connection{
+		Url:        fmt.Sprintf("jdbc:postgresql://%s.%s/%s?user=%s&ssl=true&sslmode=require", c.Name, c.Namespace, database, user),
+		SlotName:   getSlotName(database, appId),
+		PluginType: constants.EventStreamSourcePluginType,
+		DBAuth: zalandov1.DBAuth{
+			Type:        constants.EventStreamSourceAuthType,
+			Name:        c.credentialSecretNameForCluster(user, c.Name),
+			UserKey:     "username",
+			PasswordKey: "password",
+		},
+	}
+}
+
+func (c *Cluster) syncStreams() error {
+
+	c.setProcessName("syncing streams")
+
+	_, err := c.KubeClient.CustomResourceDefinitions().Get(context.TODO(), constants.EventStreamCRDName, metav1.GetOptions{})
+	if k8sutil.ResourceNotFound(err) {
+		c.logger.Debugf("event stream CRD not installed, skipping")
+		return nil
+	}
+
+	slots := make(map[string]map[string]string)
+	publications := make(map[string]map[string]acidv1.StreamTable)
+
+	requiredPatroniConfig := c.Spec.Patroni
+	if len(requiredPatroniConfig.Slots) > 0 {
+		slots = requiredPatroniConfig.Slots
+	}
+
+	// gather list of required slots and publications
+	for _, stream := range c.Spec.Streams {
+		slot := map[string]string{
+			"database": stream.Database,
+			"plugin":   constants.EventStreamSourcePluginType,
+			"type":     "logical",
+		}
+		slotName := getSlotName(stream.Database, stream.ApplicationId)
+		if _, exists := slots[slotName]; !exists {
+			slots[slotName] = slot
+			publications[slotName] = stream.Tables
+		} else {
+			streamTables := publications[slotName]
+			for tableName, table := range stream.Tables {
+				if _, exists := streamTables[tableName]; !exists {
+					streamTables[tableName] = table
+				}
+			}
+			publications[slotName] = streamTables
+		}
+	}
+
+	// no slots = no streams defined
+	if len(slots) > 0 {
+		requiredPatroniConfig.Slots = slots
+	} else {
+		return nil
+	}
+
+	// add extra logical slots to Patroni config
+	c.logger.Debug("syncing Postgres config for logical decoding")
+	err = c.syncPostgresConfig(requiredPatroniConfig)
+	if err != nil {
+		return fmt.Errorf("failed to snyc Postgres config for event streaming: %v", err)
+	}
+
+	// next, create publications to each created slot
+	c.logger.Debug("syncing database publications")
+	for publication, tables := range publications {
+		// but first check for existing publications
+		dbName := slots[publication]["database"]
+		err = c.syncPublication(publication, dbName, tables)
+		if err != nil {
+			c.logger.Warningf("could not sync publication %q in database %q: %v", publication, dbName, err)
+		}
+	}
+
+	err = c.createOrUpdateStreams()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *Cluster) createOrUpdateStreams() error {
+
+	appIds := gatherApplicationIds(c.Spec.Streams)
+	for _, appId := range appIds {
+		fesName := fmt.Sprintf("%s-%s", c.Name, appId)
+		effectiveStreams, err := c.KubeClient.FabricEventStreams(c.Namespace).Get(context.TODO(), fesName, metav1.GetOptions{})
+		if err != nil {
+			if !k8sutil.ResourceNotFound(err) {
+				return fmt.Errorf("failed reading event stream %s: %v", fesName, err)
+			}
+
+			c.logger.Infof("event streams do not exist, create it")
+			err = c.createStreams(appId)
+			if err != nil {
+				return fmt.Errorf("failed creating event stream %s: %v", fesName, err)
+			}
+			c.logger.Infof("event stream %q has been successfully created", fesName)
+		} else {
+			desiredStreams := c.generateFabricEventStream(appId)
+			if !reflect.DeepEqual(effectiveStreams.Spec, desiredStreams.Spec) {
+				c.logger.Debug("updating event streams")
+				desiredStreams.ObjectMeta.ResourceVersion = effectiveStreams.ObjectMeta.ResourceVersion
+				err = c.updateStreams(desiredStreams)
+				if err != nil {
+					return fmt.Errorf("failed updating event stream %s: %v", fesName, err)
+				}
+				c.logger.Infof("event stream %q has been successfully updated", fesName)
+			}
+		}
+	}
+
+	return nil
+}

pkg/cluster/streams_test.go

@@ -0,0 +1,244 @@
+package cluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strings"
+
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
+	v1 "github.com/zalando/postgres-operator/pkg/apis/zalando.org/v1"
+	fakezalandov1 "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/fake"
+	"github.com/zalando/postgres-operator/pkg/util"
+	"github.com/zalando/postgres-operator/pkg/util/config"
+	"github.com/zalando/postgres-operator/pkg/util/constants"
+	"github.com/zalando/postgres-operator/pkg/util/k8sutil"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func newFakeK8sStreamClient() (k8sutil.KubernetesClient, *fake.Clientset) {
+	zalandoClientSet := fakezalandov1.NewSimpleClientset()
+	clientSet := fake.NewSimpleClientset()
+
+	return k8sutil.KubernetesClient{
+		FabricEventStreamsGetter: zalandoClientSet.ZalandoV1(),
+		PostgresqlsGetter:        zalandoClientSet.AcidV1(),
+		PodsGetter:               clientSet.CoreV1(),
+		StatefulSetsGetter:       clientSet.AppsV1(),
+	}, clientSet
+}
+
+var (
+	clusterName string = "acid-test-cluster"
+	namespace   string = "default"
+	appId       string = "test-app"
+	dbName      string = "foo"
+	fesUser     string = constants.EventStreamSourceSlotPrefix + constants.UserRoleNameSuffix
+	fesName     string = fmt.Sprintf("%s-%s", clusterName, appId)
+	slotName    string = fmt.Sprintf("%s_%s_%s", constants.EventStreamSourceSlotPrefix, dbName, strings.Replace(appId, "-", "_", -1))
+
+	pg = acidv1.Postgresql{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Postgresql",
+			APIVersion: "acid.zalan.do/v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      clusterName,
+			Namespace: namespace,
+		},
+		Spec: acidv1.PostgresSpec{
+			Databases: map[string]string{
+				dbName: dbName + constants.UserRoleNameSuffix,
+			},
+			Streams: []acidv1.Stream{
+				{
+					ApplicationId: appId,
+					Database:      "foo",
+					Tables: map[string]acidv1.StreamTable{
+						"data.bar": acidv1.StreamTable{
+							EventType:     "stream_type_a",
+							IdColumn:      "b_id",
+							PayloadColumn: "b_payload",
+						},
+					},
+					Filter: map[string]string{
+						"data.bar": "[?(@.source.txId > 500 && @.source.lsn > 123456)]",
+					},
+					BatchSize: uint32(100),
+				},
+			},
+			Volume: acidv1.Volume{
+				Size: "1Gi",
+			},
+		},
+	}
+
+	fes = &v1.FabricEventStream{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: constants.EventStreamCRDApiVersion,
+			Kind:       constants.EventStreamCRDKind,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fesName,
+			Namespace: namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				metav1.OwnerReference{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "acid-test-cluster",
+					Controller: util.True(),
+				},
+			},
+		},
+		Spec: v1.FabricEventStreamSpec{
+			ApplicationId: appId,
+			EventStreams: []v1.EventStream{
+				{
+					EventStreamFlow: v1.EventStreamFlow{
+						PayloadColumn: "b_payload",
+						Type:          constants.EventStreamFlowPgGenericType,
+					},
+					EventStreamSink: v1.EventStreamSink{
+						EventType:    "stream_type_a",
+						MaxBatchSize: uint32(100),
+						Type:         constants.EventStreamSinkNakadiType,
+					},
+					EventStreamSource: v1.EventStreamSource{
+						Filter: "[?(@.source.txId > 500 && @.source.lsn > 123456)]",
+						Connection: v1.Connection{
+							DBAuth: v1.DBAuth{
+								Name:        fmt.Sprintf("fes-user.%s.credentials.postgresql.acid.zalan.do", clusterName),
+								PasswordKey: "password",
+								Type:        constants.EventStreamSourceAuthType,
+								UserKey:     "username",
+							},
+							Url:        fmt.Sprintf("jdbc:postgresql://%s.%s/foo?user=%s&ssl=true&sslmode=require", clusterName, namespace, fesUser),
+							SlotName:   slotName,
+							PluginType: constants.EventStreamSourcePluginType,
+						},
+						Schema: "data",
+						EventStreamTable: v1.EventStreamTable{
+							IDColumn: "b_id",
+							Name:     "bar",
+						},
+						Type: constants.EventStreamSourcePGType,
+					},
+				},
+			},
+		},
+	}
+)
+
+func TestGenerateFabricEventStream(t *testing.T) {
+	client, _ := newFakeK8sStreamClient()
+
+	var cluster = New(
+		Config{
+			OpConfig: config.Config{
+				Auth: config.Auth{
+					SecretNameTemplate: "{username}.{cluster}.credentials.{tprkind}.{tprgroup}",
+				},
+				PodManagementPolicy: "ordered_ready",
+				Resources: config.Resources{
+					ClusterLabels:        map[string]string{"application": "spilo"},
+					ClusterNameLabel:     "cluster-name",
+					DefaultCPURequest:    "300m",
+					DefaultCPULimit:      "300m",
+					DefaultMemoryRequest: "300Mi",
+					DefaultMemoryLimit:   "300Mi",
+					PodRoleLabel:         "spilo-role",
+				},
+			},
+		}, client, pg, logger, eventRecorder)
+
+	cluster.Name = clusterName
+	cluster.Namespace = namespace
+
+	_, err := cluster.createStatefulSet()
+	assert.NoError(t, err)
+
+	err = cluster.createOrUpdateStreams()
+	assert.NoError(t, err)
+
+	result := cluster.generateFabricEventStream(appId)
+
+	if !reflect.DeepEqual(result, fes) {
+		t.Errorf("Malformed FabricEventStream, expected %#v, got %#v", fes, result)
+	}
+
+	streamCRD, err := cluster.KubeClient.FabricEventStreams(namespace).Get(context.TODO(), fesName, metav1.GetOptions{})
+	assert.NoError(t, err)
+
+	if !reflect.DeepEqual(streamCRD, fes) {
+		t.Errorf("Malformed FabricEventStream, expected %#v, got %#v", fes, streamCRD)
+	}
+}
+
+func TestUpdateFabricEventStream(t *testing.T) {
+	client, _ := newFakeK8sStreamClient()
+
+	var cluster = New(
+		Config{
+			OpConfig: config.Config{
+				PodManagementPolicy: "ordered_ready",
+				Resources: config.Resources{
+					ClusterLabels:        map[string]string{"application": "spilo"},
+					ClusterNameLabel:     "cluster-name",
+					DefaultCPURequest:    "300m",
+					DefaultCPULimit:      "300m",
+					DefaultMemoryRequest: "300Mi",
+					DefaultMemoryLimit:   "300Mi",
+					PodRoleLabel:         "spilo-role",
+				},
+			},
+		}, client, pg, logger, eventRecorder)
+
+	_, err := cluster.KubeClient.Postgresqls(namespace).Create(
+		context.TODO(), &pg, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	err = cluster.createOrUpdateStreams()
+	assert.NoError(t, err)
+
+	var pgSpec acidv1.PostgresSpec
+	pgSpec.Streams = []acidv1.Stream{
+		{
+			ApplicationId: appId,
+			Database:      dbName,
+			Tables: map[string]acidv1.StreamTable{
+				"data.bar": acidv1.StreamTable{
+					EventType:     "stream_type_b",
+					IdColumn:      "b_id",
+					PayloadColumn: "b_payload",
+				},
+			},
+			BatchSize: uint32(250),
+		},
+	}
+	patch, err := json.Marshal(struct {
+		PostgresSpec interface{} `json:"spec"`
+	}{&pgSpec})
+	assert.NoError(t, err)
+
+	pgPatched, err := cluster.KubeClient.Postgresqls(namespace).Patch(
+		context.TODO(), cluster.Name, types.MergePatchType, patch, metav1.PatchOptions{}, "spec")
+	assert.NoError(t, err)
+
+	cluster.Postgresql.Spec = pgPatched.Spec
+	err = cluster.createOrUpdateStreams()
+	assert.NoError(t, err)
+
+	streamCRD, err := cluster.KubeClient.FabricEventStreams(namespace).Get(context.TODO(), fesName, metav1.GetOptions{})
+	assert.NoError(t, err)
+
+	result := cluster.generateFabricEventStream(appId)
+	if !reflect.DeepEqual(result, streamCRD) {
+		t.Errorf("Malformed FabricEventStream, expected %#v, got %#v", streamCRD, result)
+	}
+}

pkg/cluster/sync.go

@@ -81,7 +81,7 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 		return err
 	}
 
-	c.logger.Debugf("syncing statefulsets")
+	c.logger.Debug("syncing statefulsets")
 	if err = c.syncStatefulSet(); err != nil {
 		if !k8sutil.ResourceAlreadyExists(err) {
 			err = fmt.Errorf("could not sync statefulsets: %v", err)
@@ -107,17 +107,17 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 
 	// create database objects unless we are running without pods or disabled that feature explicitly
 	if !(c.databaseAccessDisabled() || c.getNumberOfInstances(&newSpec.Spec) <= 0 || c.Spec.StandbyCluster != nil) {
-		c.logger.Debugf("syncing roles")
+		c.logger.Debug("syncing roles")
 		if err = c.syncRoles(); err != nil {
 			err = fmt.Errorf("could not sync roles: %v", err)
 			return err
 		}
-		c.logger.Debugf("syncing databases")
+		c.logger.Debug("syncing databases")
 		if err = c.syncDatabases(); err != nil {
 			err = fmt.Errorf("could not sync databases: %v", err)
 			return err
 		}
-		c.logger.Debugf("syncing prepared databases with schemas")
+		c.logger.Debug("syncing prepared databases with schemas")
 		if err = c.syncPreparedDatabases(); err != nil {
 			err = fmt.Errorf("could not sync prepared database: %v", err)
 			return err
@@ -129,6 +129,14 @@ func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
 		return fmt.Errorf("could not sync connection pooler: %v", err)
 	}
 
+	if len(c.Spec.Streams) > 0 {
+		c.logger.Debug("syncing streams")
+		if err = c.syncStreams(); err != nil {
+			err = fmt.Errorf("could not sync streams: %v", err)
+			return err
+		}
+	}
+
 	// Major version upgrade must only run after success of all earlier operations, must remain last item in sync
 	if err := c.majorVersionUpgrade(); err != nil {
 		c.logger.Errorf("major version upgrade failed: %v", err)
@@ -401,7 +409,7 @@ func (c *Cluster) syncStatefulSet() error {
 	}
 
 	// get Postgres config, compare with manifest and update via Patroni PATCH endpoint if it differs
-	// Patroni's config endpoint is just a "proxy" to DCS. It is enough to patch it only once and it doesn't matter which pod is used.
+	// Patroni's config endpoint is just a "proxy" to DCS. It is enough to patch it only once and it doesn't matter which pod is used
 	for i, pod := range pods {
 		patroniConfig, pgParameters, err := c.getPatroniConfig(&pod)
 		if err != nil {
@@ -414,11 +422,13 @@ func (c *Cluster) syncStatefulSet() error {
 		// empty config probably means cluster is not fully initialized yet, e.g. restoring from backup
 		// do not attempt a restart
 		if !reflect.DeepEqual(patroniConfig, acidv1.Patroni{}) || len(pgParameters) > 0 {
-			restartMasterFirst, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, patroniConfig, pgParameters)
+			// compare config returned from Patroni with what is specified in the manifest
+			restartMasterFirst, err = c.checkAndSetGlobalPostgreSQLConfiguration(&pod, patroniConfig, c.Spec.Patroni, pgParameters, c.Spec.Parameters)
 			if err != nil {
 				c.logger.Warningf("could not set PostgreSQL configuration options for pod %s: %v", pods[i].Name, err)
 				continue
 			}
+
 			// it could take up to LoopWait to apply the config
 			time.Sleep(time.Duration(restartWait)*time.Second + time.Second*2)
 			break
@@ -526,40 +536,39 @@ func (c *Cluster) AnnotationsToPropagate(annotations map[string]string) map[stri
 
 // checkAndSetGlobalPostgreSQLConfiguration checks whether cluster-wide API parameters
 // (like max_connections) have changed and if necessary sets it via the Patroni API
-func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, patroniConfig acidv1.Patroni, effectivePgParameters map[string]string) (bool, error) {
+func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, effectivePatroniConfig, desiredPatroniConfig acidv1.Patroni, effectivePgParameters, desiredPgParameters map[string]string) (bool, error) {
 	configToSet := make(map[string]interface{})
 	parametersToSet := make(map[string]string)
 	restartMaster := make([]bool, 0)
 	requiresMasterRestart := false
 
-	// compare options from config with c.Spec.Patroni from manifest
-	desiredPatroniConfig := c.Spec.Patroni
-	if desiredPatroniConfig.LoopWait > 0 && desiredPatroniConfig.LoopWait != patroniConfig.LoopWait {
+	// compare effective and desired Patroni config options
+	if desiredPatroniConfig.LoopWait > 0 && desiredPatroniConfig.LoopWait != effectivePatroniConfig.LoopWait {
 		configToSet["loop_wait"] = desiredPatroniConfig.LoopWait
 	}
-	if desiredPatroniConfig.MaximumLagOnFailover > 0 && desiredPatroniConfig.MaximumLagOnFailover != patroniConfig.MaximumLagOnFailover {
+	if desiredPatroniConfig.MaximumLagOnFailover > 0 && desiredPatroniConfig.MaximumLagOnFailover != effectivePatroniConfig.MaximumLagOnFailover {
 		configToSet["maximum_lag_on_failover"] = desiredPatroniConfig.MaximumLagOnFailover
 	}
-	if desiredPatroniConfig.PgHba != nil && !reflect.DeepEqual(desiredPatroniConfig.PgHba, patroniConfig.PgHba) {
+	if desiredPatroniConfig.PgHba != nil && !reflect.DeepEqual(desiredPatroniConfig.PgHba, effectivePatroniConfig.PgHba) {
 		configToSet["pg_hba"] = desiredPatroniConfig.PgHba
 	}
-	if desiredPatroniConfig.RetryTimeout > 0 && desiredPatroniConfig.RetryTimeout != patroniConfig.RetryTimeout {
+	if desiredPatroniConfig.RetryTimeout > 0 && desiredPatroniConfig.RetryTimeout != effectivePatroniConfig.RetryTimeout {
 		configToSet["retry_timeout"] = desiredPatroniConfig.RetryTimeout
 	}
-	if desiredPatroniConfig.SynchronousMode != patroniConfig.SynchronousMode {
+	if desiredPatroniConfig.SynchronousMode != effectivePatroniConfig.SynchronousMode {
 		configToSet["synchronous_mode"] = desiredPatroniConfig.SynchronousMode
 	}
-	if desiredPatroniConfig.SynchronousModeStrict != patroniConfig.SynchronousModeStrict {
+	if desiredPatroniConfig.SynchronousModeStrict != effectivePatroniConfig.SynchronousModeStrict {
 		configToSet["synchronous_mode_strict"] = desiredPatroniConfig.SynchronousModeStrict
 	}
-	if desiredPatroniConfig.TTL > 0 && desiredPatroniConfig.TTL != patroniConfig.TTL {
+	if desiredPatroniConfig.TTL > 0 && desiredPatroniConfig.TTL != effectivePatroniConfig.TTL {
 		configToSet["ttl"] = desiredPatroniConfig.TTL
 	}
 
 	// check if specified slots exist in config and if they differ
 	slotsToSet := make(map[string]map[string]string)
 	for slotName, desiredSlot := range desiredPatroniConfig.Slots {
-		if effectiveSlot, exists := patroniConfig.Slots[slotName]; exists {
+		if effectiveSlot, exists := effectivePatroniConfig.Slots[slotName]; exists {
 			if reflect.DeepEqual(desiredSlot, effectiveSlot) {
 				continue
 			}
@@ -570,8 +579,7 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, patroniC
 		configToSet["slots"] = slotsToSet
 	}
 
-	// compare parameters under postgresql section with c.Spec.Postgresql.Parameters from manifest
-	desiredPgParameters := c.Spec.Parameters
+	// compare effective and desired parameters under postgresql section in Patroni config
 	for desiredOption, desiredValue := range desiredPgParameters {
 		effectiveValue := effectivePgParameters[desiredOption]
 		if isBootstrapOnlyParameter(desiredOption) && (effectiveValue != desiredValue) {
@@ -614,7 +622,7 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration(pod *v1.Pod, patroniC
 	c.logger.Debugf("patching Postgres config via Patroni API on pod %s with following options: %s",
 		podName, configToSetJson)
 	if err = c.patroni.SetConfig(pod, configToSet); err != nil {
-		return requiresMasterRestart, fmt.Errorf("could not patch postgres parameters with a pod %s: %v", podName, err)
+		return requiresMasterRestart, fmt.Errorf("could not patch postgres parameters within pod %s: %v", podName, err)
 	}
 
 	return requiresMasterRestart, nil

pkg/cluster/sync_test.go

@@ -256,7 +256,7 @@ func TestCheckAndSetGlobalPostgreSQLConfiguration(t *testing.T) {
 	}
 
 	for _, tt := range tests {
-		requireMasterRestart, err := cluster.checkAndSetGlobalPostgreSQLConfiguration(mockPod, tt.patroni, tt.pgParams)
+		requireMasterRestart, err := cluster.checkAndSetGlobalPostgreSQLConfiguration(mockPod, tt.patroni, cluster.Spec.Patroni, tt.pgParams, cluster.Spec.Parameters)
 		assert.NoError(t, err)
 		if requireMasterRestart != tt.restartMaster {
 			t.Errorf("%s - %s: unexpect master restart strategy, got %v, expected %v", testName, tt.subtest, requireMasterRestart, tt.restartMaster)