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 <[email protected]>

Author: FxKu

charts/postgres-operator/crds/postgresqls.yaml

@@ -474,6 +474,39 @@ spec:
                     type: string
                   gs_wal_path:
                     type: string
+              streams:
+                type: array
+                nullable: true
+                items:
+                  type: object
+                  required:
+                    - applicationId
+                    - database
+                    - tables
+                  properties:
+                    applicationId:
+                      type: string
+                    batchSize:
+                      type: integer
+                    database:
+                      type: string
+                    filter:
+                      type: object
+                      additionalProperties:
+                        type: string
+                    tables:
+                      type: object
+                      additionalProperties:
+                        type: object
+                        required:
+                          - eventType
+                        properties:
+                          eventType:
+                            type: string
+                          idColumn:
+                            type: string
+                          payloadColumn:
+                            type: string
               teamId:
                 type: string
               tls:

charts/postgres-operator/templates/clusterrole.yaml

@@ -34,6 +34,22 @@ rules:
   - get
   - list
   - watch
+# all verbs allowed for event streams
+{{- if .Values.enableStreams }}
+- apiGroups:
+  - zalando.org
+  resources:
+  - fabriceventstreams
+  verbs:
+  - create
+  - delete
+  - deletecollection
+  - get
+  - list
+  - patch
+  - update
+  - watch
+{{- end }}
 # to create or get/update CRDs when starting up
 - apiGroups:
   - apiextensions.k8s.io

charts/postgres-operator/values.yaml

@@ -368,6 +368,9 @@ configConnectionPooler:
   connection_pooler_default_cpu_limit: "1"
   connection_pooler_default_memory_limit: 100Mi
 
+# Zalando's internal CDC stream feature
+enableStreams: false
+
 rbac:
   # Specifies whether RBAC resources should be created
   create: true

docs/reference/cluster_manifest.md

@@ -536,3 +536,54 @@ Those parameters are grouped under the `tls` top-level key.
   relative to the "/tls/", which is mount path of the tls secret.
   If `caSecretName` is defined, the ca.crt path is relative to "/tlsca/",
   otherwise to the same "/tls/".
+
+## Change data capture streams
+
+This sections enables change data capture (CDC) streams via Postgres' 
+[logical decoding](https://www.postgresql.org/docs/14/logicaldecoding.html)
+feature and `pgoutput` plugin. While the Postgres operator takes responsibility
+for providing the setup to publish change events, it relies on external tools
+to consume them. At Zalando, we are using a workflow based on
+[Debezium Connector](https://debezium.io/documentation/reference/stable/connectors/postgresql.html)
+which can feed streams into Zalando’s distributed event broker [Nakadi](https://nakadi.io/)
+among others.
+
+The Postgres Operator creates custom resources for Zalando's internal CDC
+operator which will be used to set up the consumer part. Each stream object
+can have the following properties:
+
+* **applicationId**
+  The application name to which the database and CDC belongs to. For each
+  set of streams with a distinct `applicationId` a separate stream CR as well
+  as a separate logical replication slot will be created. This means there can
+  be different streams in the same database and streams with the same
+  `applicationId` are bundled in one stream CR. The stream CR will be called
+  like the Postgres cluster plus "-<applicationId>" suffix. Required.
+
+* **database**
+  Name of the database from where events will be published via Postgres'
+  logical decoding feature. The operator will take care of updating the
+  database configuration (setting `wal_level: logical`, creating logical
+  replication slots, using output plugin `pgoutput` and creating a dedicated
+  replication user). Required.
+
+* **tables**
+  Defines a map of table names and their properties (`eventType`, `idColumn`
+  and `payloadColumn`). The CDC operator is following the [outbox pattern](https://debezium.io/blog/2019/02/19/reliable-microservices-data-exchange-with-the-outbox-pattern/).
+  The application is responsible for putting events into a (JSON/B or VARCHAR)
+  payload column of the outbox table in the structure of the specified target
+  event type. The operator will create a [PUBLICATION](https://www.postgresql.org/docs/14/logical-replication-publication.html)
+  in Postgres for all tables specified for one `database` and `applicationId`.
+  The CDC operator will consume from it shortly after transactions are
+  committed to the outbox table. The `idColumn` will be used in telemetry for
+  the CDC operator. The names for `idColumn` and `payloadColumn` can be
+  configured. Defaults are `id` and `payload`. The target `eventType` has to
+  be defined. Required.
+
+* **filter**
+  Streamed events can be filtered by a jsonpath expression for each table.
+  Optional.
+
+* **batchSize**
+  Defines the size of batches in which events are consumed. Optional.
+  Defaults to 1.

hack/update-codegen.sh

@@ -17,7 +17,7 @@ trap "cleanup" EXIT SIGINT
 
 bash "${CODEGEN_PKG}/generate-groups.sh" all \
   "${OPERATOR_PACKAGE_ROOT}/pkg/generated" "${OPERATOR_PACKAGE_ROOT}/pkg/apis" \
-  "acid.zalan.do:v1" \
+  "acid.zalan.do:v1 zalando.org:v1" \
   --go-header-file "${SCRIPT_ROOT}"/hack/custom-boilerplate.go.txt
 
 cp -r "${OPERATOR_PACKAGE_ROOT}"/pkg/* "${TARGET_CODE_DIR}"

manifests/complete-postgres-manifest.yaml

@@ -3,6 +3,7 @@ kind: postgresql
 metadata:
   name: acid-test-cluster
 #  labels:
+#    application: test-app
 #    environment: demo
 #  annotations:
 #    "acid.zalan.do/controller": "second-operator"
@@ -17,7 +18,7 @@ spec:
     - superuser
     - createdb
     foo_user: []
-#    flyway: [] 
+#    flyway: []
 #  usersWithSecretRotation:
 #  - foo_user
 #  usersWithInPlaceSecretRotation:
@@ -203,3 +204,21 @@ spec:
 #              operator: In
 #              values:
 #                - enabled
+
+# Enables change data capture streams for defined database tables
+#  streams:
+#  - applicationId: test-app
+#    database: foo
+#    tables:
+#      data.state_pending_outbox:
+#        eventType: test-app.status-pending
+#      data.state_approved_outbox:
+#        eventType: test-app.status-approved
+#      data.orders_outbox:
+#        eventType: test-app.order-completed
+#        idColumn: o_id
+#        payloadColumn: o_payload
+#    # Optional. Filter ignores events before a certain txnId and lsn. Can be used to skip bad events
+#    filter:
+#      data.orders_outbox: "[?(@.source.txId > 500 && @.source.lsn > 123456)]"
+#    batchSize: 1000

manifests/operator-service-account-rbac.yaml

@@ -35,6 +35,20 @@ rules:
   - get
   - list
   - watch
+# all verbs allowed for event streams (Zalando-internal feature)
+# - apiGroups:
+#   - zalando.org
+#   resources:
+#   - fabriceventstreams
+#   verbs:
+#   - create
+#   - delete
+#   - deletecollection
+#   - get
+#   - list
+#   - patch
+#   - update
+#   - watch
 # to create or get/update CRDs when starting up
 - apiGroups:
   - apiextensions.k8s.io

manifests/postgresql.crd.yaml

@@ -472,6 +472,39 @@ spec:
                     type: string
                   gs_wal_path:
                     type: string
+              streams:
+                type: array
+                nullable: true
+                items:
+                  type: object
+                  required:
+                    - applicationId
+                    - database
+                    - tables
+                  properties:
+                    applicationId:
+                      type: string
+                    batchSize:
+                      type: integer
+                    database:
+                      type: string
+                    filter:
+                      type: object
+                      additionalProperties:
+                        type: string
+                    tables:
+                      type: object
+                      additionalProperties:
+                        type: object
+                        required:
+                          - eventType
+                        properties:
+                          eventType:
+                            type: string
+                          idColumn:
+                            type: string
+                          payloadColumn:
+                            type: string
               teamId:
                 type: string
               tls:

pkg/apis/acid.zalan.do/v1/crds.go

@@ -716,6 +716,54 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
 							},
 						},
 					},
+					"streams": {
+						Type: "array",
+						Items: &apiextv1.JSONSchemaPropsOrArray{
+							Schema: &apiextv1.JSONSchemaProps{
+								Type:     "object",
+								Required: []string{"applicationId", "database", "tables"},
+								Properties: map[string]apiextv1.JSONSchemaProps{
+									"applicationId": {
+										Type: "string",
+									},
+									"batchSize": {
+										Type: "integer",
+									},
+									"database": {
+										Type: "string",
+									},
+									"filter": {
+										Type: "object",
+										AdditionalProperties: &apiextv1.JSONSchemaPropsOrBool{
+											Schema: &apiextv1.JSONSchemaProps{
+												Type: "string",
+											},
+										},
+									},
+									"tables": {
+										Type: "object",
+										AdditionalProperties: &apiextv1.JSONSchemaPropsOrBool{
+											Schema: &apiextv1.JSONSchemaProps{
+												Type:     "object",
+												Required: []string{"eventType"},
+												Properties: map[string]apiextv1.JSONSchemaProps{
+													"eventType": {
+														Type: "string",
+													},
+													"idColumn": {
+														Type: "string",
+													},
+													"payloadColumn": {
+														Type: "string",
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
 					"teamId": {
 						Type: "string",
 					},

pkg/apis/acid.zalan.do/v1/postgresql_type.go

@@ -77,6 +77,7 @@ type PostgresSpec struct {
 	ServiceAnnotations    map[string]string           `json:"serviceAnnotations,omitempty"`
 	TLS                   *TLSDescription             `json:"tls,omitempty"`
 	AdditionalVolumes     []AdditionalVolume          `json:"additionalVolumes,omitempty"`
+	Streams               []Stream                    `json:"streams,omitempty"`
 
 	// deprecated json tags
 	InitContainersOld       []v1.Container `json:"init_containers,omitempty"`
@@ -231,3 +232,17 @@ type ConnectionPooler struct {
 
 	Resources `json:"resources,omitempty"`
 }
+
+type Stream struct {
+	ApplicationId string                 `json:"applicationId"`
+	Database      string                 `json:"database"`
+	Tables        map[string]StreamTable `json:"tables"`
+	Filter        map[string]string      `json:"filter,omitempty"`
+	BatchSize     uint32                 `json:"batchSize,omitempty"`
+}
+
+type StreamTable struct {
+	EventType     string `json:"eventType"`
+	IdColumn      string `json:"idColumn,omitempty" defaults:"id"`
+	PayloadColumn string `json:"payloadColumn,omitempty" defaults:"payload"`
+}

pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go

@@ -0,0 +1,6 @@
+package zalando
+
+const (
+	// GroupName is the group name for the operator CRDs
+	GroupName = "zalando.org"
+)