add resource annotation and ignore recovery type (#2817)

* add resource annotation and ignore recovery type
* Update docs/reference/cluster_manifest.md

---------

Co-authored-by: Ida Novindasari <idanovinda@gmail.com>

charts/postgres-operator/crds/postgresqls.yaml

@@ -514,6 +514,9 @@ spec:
                       type: string
                     batchSize:
                       type: integer
+                    cpu:
+                      type: string
+                      pattern: '^(\d+m|\d+(\.\d{1,3})?)$'
                     database:
                       type: string
                     enableRecovery:
@@ -522,6 +525,9 @@ spec:
                       type: object
                       additionalProperties:
                         type: string
+                    memory:
+                      type: string
+                      pattern: '^(\d+(e\d+)?|\d+(\.\d+)?(e\d+)?[EPTGMK]i?)$'
                     tables:
                       type: object
                       additionalProperties:
@@ -533,6 +539,8 @@ spec:
                             type: string
                           idColumn:
                             type: string
+                          ignoreRecovery:
+                            type: boolean
                           payloadColumn:
                             type: string
                           recoveryEventType:

docs/reference/cluster_manifest.md

@@ -652,11 +652,11 @@ 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
+  set of streams with a distinct `applicationId` a separate stream resource as
-  as a separate logical replication slot will be created. This means there can
+  well as a separate logical replication slot will be created. This means there
-  be different streams in the same database and streams with the same
+  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
+  `applicationId` are bundled in one stream resource. The stream resource will
-  like the Postgres cluster plus "-<applicationId>" suffix. Required.
+  be called like the Postgres cluster plus "-<applicationId>" suffix. Required.
 
 * **database**
   Name of the database from where events will be published via Postgres'
@@ -667,7 +667,8 @@ can have the following properties:
 
 * **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/).
+  and `payloadColumn`). Required.
+  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/16/logical-replication-publication.html)
@@ -676,12 +677,27 @@ can have the following properties:
   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.
+  be defined. One can also specify a `recoveryEventType` that will be used
+  for a dead letter queue. By enabling `ignoreRecovery`, you can choose to
+  ignore failing events.
 
 * **filter**
   Streamed events can be filtered by a jsonpath expression for each table.
   Optional.
 
+* **enableRecovery**
+  Flag to enable a dead letter queue recovery for all streams tables.
+  Alternatively, recovery can also be enable for single outbox tables by only
+  specifying a `recoveryEventType` and no `enableRecovery` flag. When set to
+  false or missing, events will be retried until consuming succeeded. You can
+  use a `filter` expression to get rid of poison pills. Optional.
+
 * **batchSize**
   Defines the size of batches in which events are consumed. Optional.
   Defaults to 1.
+
+* **cpu**
+  CPU requests to be set as an annotation on the stream resource. Optional.
+
+* **memory**
+  memory requests to be set as an annotation on the stream resource. Optional.

manifests/postgresql.crd.yaml

@@ -512,6 +512,9 @@ spec:
                       type: string
                     batchSize:
                       type: integer
+                    cpu:
+                      type: string
+                      pattern: '^(\d+m|\d+(\.\d{1,3})?)$'
                     database:
                       type: string
                     enableRecovery:
@@ -520,6 +523,9 @@ spec:
                       type: object
                       additionalProperties:
                         type: string
+                    memory:
+                      type: string
+                      pattern: '^(\d+(e\d+)?|\d+(\.\d+)?(e\d+)?[EPTGMK]i?)$'
                     tables:
                       type: object
                       additionalProperties:
@@ -531,6 +537,8 @@ spec:
                             type: string
                           idColumn:
                             type: string
+                          ignoreRecovery:
+                            type: boolean
                           payloadColumn:
                             type: string
                           recoveryEventType:

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

@@ -258,6 +258,8 @@ type Stream struct {
 	Tables         map[string]StreamTable `json:"tables"`
 	Filter         map[string]*string     `json:"filter,omitempty"`
 	BatchSize      *uint32                `json:"batchSize,omitempty"`
+	CPU            *string                `json:"cpu,omitempty"`
+	Memory         *string                `json:"memory,omitempty"`
 	EnableRecovery *bool                  `json:"enableRecovery,omitempty"`
 }
 
@@ -265,6 +267,7 @@ type Stream struct {
 type StreamTable struct {
 	EventType         string  `json:"eventType"`
 	RecoveryEventType string  `json:"recoveryEventType,omitempty"`
+	IgnoreRecovery    *bool   `json:"ignoreRecovery,omitempty"`
 	IdColumn          *string `json:"idColumn,omitempty"`
 	PayloadColumn     *string `json:"payloadColumn,omitempty"`
 }

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

@@ -1336,6 +1336,16 @@ func (in *Stream) DeepCopyInto(out *Stream) {
 		*out = new(uint32)
 		**out = **in
 	}
+	if in.CPU != nil {
+		in, out := &in.CPU, &out.CPU
+		*out = new(string)
+		**out = **in
+	}
+	if in.Memory != nil {
+		in, out := &in.Memory, &out.Memory
+		*out = new(string)
+		**out = **in
+	}
 	if in.EnableRecovery != nil {
 		in, out := &in.EnableRecovery, &out.EnableRecovery
 		*out = new(bool)
@@ -1357,6 +1367,11 @@ func (in *Stream) DeepCopy() *Stream {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *StreamTable) DeepCopyInto(out *StreamTable) {
 	*out = *in
+	if in.IgnoreRecovery != nil {
+		in, out := &in.IgnoreRecovery, &out.IgnoreRecovery
+		*out = new(bool)
+		**out = **in
+	}
 	if in.IdColumn != nil {
 		in, out := &in.IdColumn, &out.IdColumn
 		*out = new(string)

pkg/cluster/streams.go

@@ -178,16 +178,35 @@ func (c *Cluster) syncPublication(dbName string, databaseSlotsList map[string]za
 
 func (c *Cluster) generateFabricEventStream(appId string) *zalandov1.FabricEventStream {
 	eventStreams := make([]zalandov1.EventStream, 0)
+	resourceAnnotations := map[string]string{}
 
 	for _, stream := range c.Spec.Streams {
 		if stream.ApplicationId != appId {
 			continue
 		}
+		if stream.CPU != nil {
+			cpu, exists := resourceAnnotations[constants.EventStreamCpuAnnotationKey]
+			if exists {
+				isSmaller, _ := util.IsSmallerQuantity(cpu, *stream.CPU)
+				if isSmaller {
+					resourceAnnotations[constants.EventStreamCpuAnnotationKey] = *stream.CPU
+				}
+			}
+		}
+		if stream.Memory != nil {
+			memory, exists := resourceAnnotations[constants.EventStreamMemoryAnnotationKey]
+			if exists {
+				isSmaller, _ := util.IsSmallerQuantity(memory, *stream.Memory)
+				if isSmaller {
+					resourceAnnotations[constants.EventStreamMemoryAnnotationKey] = *stream.Memory
+				}
+			}
+		}
 		for tableName, table := range stream.Tables {
 			streamSource := c.getEventStreamSource(stream, tableName, table.IdColumn)
 			streamFlow := getEventStreamFlow(table.PayloadColumn)
 			streamSink := getEventStreamSink(stream, table.EventType)
-			streamRecovery := getEventStreamRecovery(stream, table.RecoveryEventType, table.EventType)
+			streamRecovery := getEventStreamRecovery(stream, table.RecoveryEventType, table.EventType, table.IgnoreRecovery)
 
 			eventStreams = append(eventStreams, zalandov1.EventStream{
 				EventStreamFlow:     streamFlow,
@@ -207,7 +226,7 @@ func (c *Cluster) generateFabricEventStream(appId string) *zalandov1.FabricEvent
 			Name:            fmt.Sprintf("%s-%s", c.Name, strings.ToLower(util.RandomPassword(5))),
 			Namespace:       c.Namespace,
 			Labels:          c.labelsSet(true),
-			Annotations:     c.AnnotationsToPropagate(c.annotationsSet(nil)),
+			Annotations:     c.AnnotationsToPropagate(c.annotationsSet(resourceAnnotations)),
 			OwnerReferences: c.ownerReferences(),
 		},
 		Spec: zalandov1.FabricEventStreamSpec{
@@ -247,7 +266,7 @@ func getEventStreamSink(stream acidv1.Stream, eventType string) zalandov1.EventS
 	}
 }
 
-func getEventStreamRecovery(stream acidv1.Stream, recoveryEventType, eventType string) zalandov1.EventStreamRecovery {
+func getEventStreamRecovery(stream acidv1.Stream, recoveryEventType, eventType string, ignoreRecovery *bool) zalandov1.EventStreamRecovery {
 	if (stream.EnableRecovery != nil && !*stream.EnableRecovery) ||
 		(stream.EnableRecovery == nil && recoveryEventType == "") {
 		return zalandov1.EventStreamRecovery{
@@ -255,6 +274,12 @@ func getEventStreamRecovery(stream acidv1.Stream, recoveryEventType, eventType s
 		}
 	}
 
+	if ignoreRecovery != nil && *ignoreRecovery {
+		return zalandov1.EventStreamRecovery{
+			Type: constants.EventStreamRecoveryIgnoreType,
+		}
+	}
+
 	if stream.EnableRecovery != nil && *stream.EnableRecovery && recoveryEventType == "" {
 		recoveryEventType = fmt.Sprintf("%s-%s", eventType, constants.EventStreamRecoverySuffix)
 	}

pkg/cluster/streams_test.go

@@ -65,12 +65,18 @@ var (
 							EventType:         "stream-type-b",
 							RecoveryEventType: "stream-type-b-dlq",
 						},
+						"data.foofoobar": {
+							EventType:      "stream-type-c",
+							IgnoreRecovery: util.True(),
+						},
 					},
 					EnableRecovery: util.True(),
 					Filter: map[string]*string{
 						"data.bar": k8sutil.StringToPointer("[?(@.source.txId > 500 && @.source.lsn > 123456)]"),
 					},
 					BatchSize: k8sutil.UInt32ToPointer(uint32(100)),
+					CPU:       k8sutil.StringToPointer("250m"),
+					Memory:    k8sutil.StringToPointer("500Mi"),
 				},
 			},
 			TeamID: "acid",
@@ -88,6 +94,10 @@ var (
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      fmt.Sprintf("%s-12345", clusterName),
 			Namespace: namespace,
+			Annotations: map[string]string{
+				constants.EventStreamCpuAnnotationKey:    "250m",
+				constants.EventStreamMemoryAnnotationKey: "500Mi",
+			},
 			Labels: map[string]string{
 				"application":  "spilo",
 				"cluster-name": clusterName,
@@ -180,6 +190,37 @@ var (
 						Type: constants.EventStreamSourcePGType,
 					},
 				},
+				{
+					EventStreamFlow: zalandov1.EventStreamFlow{
+						Type: constants.EventStreamFlowPgGenericType,
+					},
+					EventStreamRecovery: zalandov1.EventStreamRecovery{
+						Type: constants.EventStreamRecoveryIgnoreType,
+					},
+					EventStreamSink: zalandov1.EventStreamSink{
+						EventType:    "stream-type-c",
+						MaxBatchSize: k8sutil.UInt32ToPointer(uint32(100)),
+						Type:         constants.EventStreamSinkNakadiType,
+					},
+					EventStreamSource: zalandov1.EventStreamSource{
+						Connection: zalandov1.Connection{
+							DBAuth: zalandov1.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: zalandov1.EventStreamTable{
+							Name: "foofoobar",
+						},
+						Type: constants.EventStreamSourcePGType,
+					},
+				},
 			},
 		},
 	}
@@ -528,8 +569,8 @@ func TestSyncStreams(t *testing.T) {
 
 func TestSameStreams(t *testing.T) {
 	testName := "TestSameStreams"
-	annotationsA := map[string]string{"owned-by": "acid"}
+	annotationsA := map[string]string{constants.EventStreamMemoryAnnotationKey: "500Mi"}
-	annotationsB := map[string]string{"owned-by": "foo"}
+	annotationsB := map[string]string{constants.EventStreamMemoryAnnotationKey: "1Gi"}
 
 	stream1 := zalandov1.EventStream{
 		EventStreamFlow:     zalandov1.EventStreamFlow{},
@@ -621,6 +662,13 @@ func TestSameStreams(t *testing.T) {
 			match:    false,
 			reason:   "event stream specs differ",
 		},
+		{
+			subTest:  "event stream annotations differ",
+			streamsA: newFabricEventStream([]zalandov1.EventStream{stream2}, nil),
+			streamsB: newFabricEventStream([]zalandov1.EventStream{stream3}, annotationsA),
+			match:    false,
+			reason:   "event stream specs differ",
+		},
 		{
 			subTest:  "event stream annotations differ",
 			streamsA: newFabricEventStream([]zalandov1.EventStream{stream2}, annotationsA),

pkg/util/constants/streams.go

@@ -11,7 +11,10 @@ const (
 	EventStreamSourceAuthType      = "DatabaseAuthenticationSecret"
 	EventStreamFlowPgGenericType   = "PostgresWalToGenericNakadiEvent"
 	EventStreamSinkNakadiType      = "Nakadi"
-	EventStreamRecoveryNoneType  = "None"
 	EventStreamRecoveryDLQType     = "DeadLetter"
+	EventStreamRecoveryIgnoreType  = "Ignore"
+	EventStreamRecoveryNoneType    = "None"
 	EventStreamRecoverySuffix      = "dead-letter-queue"
+	EventStreamCpuAnnotationKey    = "fes.zalando.org/FES_CPU"
+	EventStreamMemoryAnnotationKey = "fes.zalando.org/FES_MEMORY"
 )