Merge branch 'master' into kube-api-client-limits

.gitignore

@@ -104,3 +104,5 @@ e2e/tls
 mocks
 
 ui/.npm/
+
+.DS_Store

charts/postgres-operator-ui/templates/deployment.yaml

@@ -9,7 +9,7 @@ metadata:
   name: {{ template "postgres-operator-ui.fullname" . }}
   namespace: {{ .Release.Namespace }}
 spec:
-  replicas: 1
+  replicas: {{ .Values.replicaCount }}
   selector:
     matchLabels:
       app.kubernetes.io/name: {{ template "postgres-operator-ui.name" . }}
@@ -102,4 +102,4 @@ spec:
 {{ toYaml .Values.tolerations | indent 8 }}
     {{- if .Values.priorityClassName }}
       priorityClassName: {{ .Values.priorityClassName }}
-    {{- end }}
+    {{- end }}

charts/postgres-operator/templates/clusterrole.yaml

@@ -141,7 +141,7 @@ rules:
   - get
   - list
   - patch
-{{- if toString .Values.configKubernetes.storage_resize_mode | eq "pvc" }}
+{{- if or (toString .Values.configKubernetes.storage_resize_mode | eq "pvc") (toString .Values.configKubernetes.storage_resize_mode | eq "mixed") }}
   - update
 {{- end }}
  # to read existing PVs. Creation should be done via dynamic provisioning

e2e/tests/test_e2e.py

@@ -2204,6 +2204,8 @@ def test_stream_resources(self):
                         {
                             "applicationId": "test-app",
                             "batchSize": 100,
+                            "cpu": "100m",
+                            "memory": "200Mi",
                             "database": "foo",
                             "enableRecovery": True,
                             "tables": {
@@ -2225,7 +2227,7 @@ def test_stream_resources(self):
                                     "eventType": "test-event",
                                     "idColumn": "id",
                                     "payloadColumn": "payload",
-                                    "recoveryEventType": "test-event-dlq"
+                                    "ignoreRecovery": True
                                 }
                             }
                         }

kubectl-pg/go.mod

@@ -51,13 +51,13 @@ require (
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/crypto v0.26.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
 	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
 	golang.org/x/net v0.25.0 // indirect
 	golang.org/x/oauth2 v0.18.0 // indirect
-	golang.org/x/sys v0.23.0 // indirect
-	golang.org/x/term v0.23.0 // indirect
-	golang.org/x/text v0.17.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/term v0.27.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect

kubectl-pg/go.sum

@@ -137,8 +137,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw=
-golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 h1:hNQpMuAJe5CtcUqCXaWga3FHu+kQvCqcsoVaQgSV60o=
 golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -166,18 +166,18 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM=
-golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU=
-golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=
+golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
+golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
-golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc=
-golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

pkg/cluster/streams.go

@@ -179,29 +179,19 @@ 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{}
+	var err, err2 error
 
 	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
-				}
-			}
+
+		err = setResourceAnnotation(&resourceAnnotations, stream.CPU, constants.EventStreamCpuAnnotationKey)
+		err2 = setResourceAnnotation(&resourceAnnotations, stream.Memory, constants.EventStreamMemoryAnnotationKey)
+		if err != nil || err2 != nil {
+			c.logger.Warningf("could not set resource annotation for event stream: %v", err)
 		}
+
 		for tableName, table := range stream.Tables {
 			streamSource := c.getEventStreamSource(stream, tableName, table.IdColumn)
 			streamFlow := getEventStreamFlow(table.PayloadColumn)
@@ -236,6 +226,27 @@ func (c *Cluster) generateFabricEventStream(appId string) *zalandov1.FabricEvent
 	}
 }
 
+func setResourceAnnotation(annotations *map[string]string, resource *string, key string) error {
+	var (
+		isSmaller bool
+		err       error
+	)
+	if resource != nil {
+		currentValue, exists := (*annotations)[key]
+		if exists {
+			isSmaller, err = util.IsSmallerQuantity(currentValue, *resource)
+			if err != nil {
+				return fmt.Errorf("could not compare resource in %q annotation: %v", key, err)
+			}
+		}
+		if isSmaller || !exists {
+			(*annotations)[key] = *resource
+		}
+	}
+
+	return nil
+}
+
 func (c *Cluster) getEventStreamSource(stream acidv1.Stream, tableName string, idColumn *string) zalandov1.EventStreamSource {
 	table, schema := getTableSchema(tableName)
 	streamFilter := stream.Filter[tableName]
@@ -521,10 +532,19 @@ func (c *Cluster) syncStream(appId string) error {
 
 func (c *Cluster) compareStreams(curEventStreams, newEventStreams *zalandov1.FabricEventStream) (match bool, reason string) {
 	reasons := make([]string, 0)
+	desiredAnnotations := make(map[string]string)
 	match = true
 
 	// stream operator can add extra annotations so incl. current annotations in desired annotations
-	desiredAnnotations := c.annotationsSet(curEventStreams.Annotations)
+	for curKey, curValue := range curEventStreams.Annotations {
+		if _, exists := desiredAnnotations[curKey]; !exists {
+			desiredAnnotations[curKey] = curValue
+		}
+	}
+	// add/or override annotations if cpu and memory values were changed
+	for newKey, newValue := range newEventStreams.Annotations {
+		desiredAnnotations[newKey] = newValue
+	}
 	if changed, reason := c.compareAnnotations(curEventStreams.ObjectMeta.Annotations, desiredAnnotations); changed {
 		match = false
 		reasons = append(reasons, fmt.Sprintf("new streams annotations do not match: %s", reason))

pkg/cluster/streams_test.go

@@ -640,49 +640,49 @@ func TestSameStreams(t *testing.T) {
 			streamsA: newFabricEventStream([]zalandov1.EventStream{stream1}, nil),
 			streamsB: newFabricEventStream([]zalandov1.EventStream{stream1, stream2}, nil),
 			match:    false,
-			reason:   "number of defined streams is different",
+			reason:   "new streams EventStreams array does not match : number of defined streams is different",
 		},
 		{
 			subTest:  "different number of streams",
 			streamsA: newFabricEventStream([]zalandov1.EventStream{stream1}, nil),
 			streamsB: newFabricEventStream([]zalandov1.EventStream{stream1, stream2}, nil),
 			match:    false,
-			reason:   "number of defined streams is different",
+			reason:   "new streams EventStreams array does not match : number of defined streams is different",
 		},
 		{
 			subTest:  "event stream specs differ",
 			streamsA: newFabricEventStream([]zalandov1.EventStream{stream1, stream2}, nil),
 			streamsB: fes,
 			match:    false,
-			reason:   "number of defined streams is different",
+			reason:   "new streams annotations do not match:  Added \"fes.zalando.org/FES_CPU\" with value \"250m\". Added \"fes.zalando.org/FES_MEMORY\" with value \"500Mi\"., new streams labels do not match the current ones, new streams EventStreams array does not match : number of defined streams is different",
 		},
 		{
 			subTest:  "event stream recovery specs differ",
 			streamsA: newFabricEventStream([]zalandov1.EventStream{stream2}, nil),
 			streamsB: newFabricEventStream([]zalandov1.EventStream{stream3}, nil),
 			match:    false,
-			reason:   "event stream specs differ",
+			reason:   "new streams EventStreams array does not match : event stream specs differ",
 		},
 		{
-			subTest:  "event stream annotations differ",
+			subTest:  "event stream with new annotations",
 			streamsA: newFabricEventStream([]zalandov1.EventStream{stream2}, nil),
-			streamsB: newFabricEventStream([]zalandov1.EventStream{stream3}, annotationsA),
+			streamsB: newFabricEventStream([]zalandov1.EventStream{stream2}, annotationsA),
 			match:    false,
-			reason:   "event stream specs differ",
+			reason:   "new streams annotations do not match:  Added \"fes.zalando.org/FES_MEMORY\" with value \"500Mi\".",
 		},
 		{
 			subTest:  "event stream annotations differ",
-			streamsA: newFabricEventStream([]zalandov1.EventStream{stream2}, annotationsA),
+			streamsA: newFabricEventStream([]zalandov1.EventStream{stream3}, annotationsA),
 			streamsB: newFabricEventStream([]zalandov1.EventStream{stream3}, annotationsB),
 			match:    false,
-			reason:   "event stream specs differ",
+			reason:   "new streams annotations do not match:  \"fes.zalando.org/FES_MEMORY\" changed from \"500Mi\" to \"1Gi\".",
 		},
 	}
 
 	for _, tt := range tests {
 		streamsMatch, matchReason := cluster.compareStreams(tt.streamsA, tt.streamsB)
-		if streamsMatch != tt.match {
-			t.Errorf("%s %s: unexpected match result when comparing streams: got %s, epxected %s",
+		if streamsMatch != tt.match || matchReason != tt.reason {
+			t.Errorf("%s %s: unexpected match result when comparing streams: got %s, expected %s",
 				testName, tt.subTest, matchReason, tt.reason)
 		}
 	}

pkg/cluster/volumes.go

@@ -151,7 +151,7 @@ func (c *Cluster) populateVolumeMetaData() error {
 	volumeIds := []string{}
 	var volumeID string
 	for _, pv := range pvs {
-		volumeID, err = c.VolumeResizer.ExtractVolumeID(pv.Spec.AWSElasticBlockStore.VolumeID)
+		volumeID, err = c.VolumeResizer.GetProviderVolumeID(pv)
 		if err != nil {
 			continue
 		}

pkg/cluster/volumes_test.go

@@ -216,6 +216,12 @@ func TestMigrateEBS(t *testing.T) {
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-1")).Return("ebs-volume-1", nil)
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-2")).Return("ebs-volume-2", nil)
 
+	resizer.EXPECT().GetProviderVolumeID(gomock.Any()).
+		DoAndReturn(func(pv *v1.PersistentVolume) (string, error) {
+			return resizer.ExtractVolumeID(pv.Spec.AWSElasticBlockStore.VolumeID)
+		}).
+		Times(2)
+
 	resizer.EXPECT().DescribeVolumes(gomock.Eq([]string{"ebs-volume-1", "ebs-volume-2"})).Return(
 		[]volumes.VolumeProperties{
 			{VolumeID: "ebs-volume-1", VolumeType: "gp2", Size: 100},
@@ -322,6 +328,12 @@ func TestMigrateGp3Support(t *testing.T) {
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-2")).Return("ebs-volume-2", nil)
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-3")).Return("ebs-volume-3", nil)
 
+	resizer.EXPECT().GetProviderVolumeID(gomock.Any()).
+		DoAndReturn(func(pv *v1.PersistentVolume) (string, error) {
+			return resizer.ExtractVolumeID(pv.Spec.AWSElasticBlockStore.VolumeID)
+		}).
+		Times(3)
+
 	resizer.EXPECT().DescribeVolumes(gomock.Eq([]string{"ebs-volume-1", "ebs-volume-2", "ebs-volume-3"})).Return(
 		[]volumes.VolumeProperties{
 			{VolumeID: "ebs-volume-1", VolumeType: "gp3", Size: 100, Iops: 3000},
@@ -377,6 +389,12 @@ func TestManualGp2Gp3Support(t *testing.T) {
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-1")).Return("ebs-volume-1", nil)
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-2")).Return("ebs-volume-2", nil)
 
+	resizer.EXPECT().GetProviderVolumeID(gomock.Any()).
+		DoAndReturn(func(pv *v1.PersistentVolume) (string, error) {
+			return resizer.ExtractVolumeID(pv.Spec.AWSElasticBlockStore.VolumeID)
+		}).
+		Times(2)
+
 	resizer.EXPECT().DescribeVolumes(gomock.Eq([]string{"ebs-volume-1", "ebs-volume-2"})).Return(
 		[]volumes.VolumeProperties{
 			{VolumeID: "ebs-volume-1", VolumeType: "gp2", Size: 150, Iops: 3000},
@@ -436,6 +454,12 @@ func TestDontTouchType(t *testing.T) {
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-1")).Return("ebs-volume-1", nil)
 	resizer.EXPECT().ExtractVolumeID(gomock.Eq("aws://eu-central-1b/ebs-volume-2")).Return("ebs-volume-2", nil)
 
+	resizer.EXPECT().GetProviderVolumeID(gomock.Any()).
+		DoAndReturn(func(pv *v1.PersistentVolume) (string, error) {
+			return resizer.ExtractVolumeID(pv.Spec.AWSElasticBlockStore.VolumeID)
+		}).
+		Times(2)
+
 	resizer.EXPECT().DescribeVolumes(gomock.Eq([]string{"ebs-volume-1", "ebs-volume-2"})).Return(
 		[]volumes.VolumeProperties{
 			{VolumeID: "ebs-volume-1", VolumeType: "gp2", Size: 150, Iops: 3000},

pkg/util/constants/aws.go

@@ -7,6 +7,7 @@ const (
 	// EBS related constants
 	EBSVolumeIDStart = "/vol-"
 	EBSProvisioner   = "kubernetes.io/aws-ebs"
+	EBSDriver        = "ebs.csi.aws.com"
 	//https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_VolumeModification.html
 	EBSVolumeStateModifying     = "modifying"
 	EBSVolumeStateOptimizing    = "optimizing"

pkg/util/volumes/ebs.go

@@ -36,7 +36,8 @@ func (r *EBSVolumeResizer) IsConnectedToProvider() bool {
 
 // VolumeBelongsToProvider checks if the given persistent volume is backed by EBS.
 func (r *EBSVolumeResizer) VolumeBelongsToProvider(pv *v1.PersistentVolume) bool {
-	return pv.Spec.AWSElasticBlockStore != nil && pv.Annotations[constants.VolumeStorateProvisionerAnnotation] == constants.EBSProvisioner
+	return (pv.Spec.AWSElasticBlockStore != nil && pv.Annotations[constants.VolumeStorateProvisionerAnnotation] == constants.EBSProvisioner) ||
+		(pv.Spec.CSI != nil && pv.Spec.CSI.Driver == constants.EBSDriver)
 }
 
 // ExtractVolumeID extracts volumeID from "aws://eu-central-1a/vol-075ddfc4a127d0bd4"
@@ -54,7 +55,12 @@ func (r *EBSVolumeResizer) ExtractVolumeID(volumeID string) (string, error) {
 
 // GetProviderVolumeID converts aws://eu-central-1b/vol-00f93d4827217c629 to vol-00f93d4827217c629 for EBS volumes
 func (r *EBSVolumeResizer) GetProviderVolumeID(pv *v1.PersistentVolume) (string, error) {
-	volumeID := pv.Spec.AWSElasticBlockStore.VolumeID
+	var volumeID string = ""
+	if pv.Spec.CSI != nil {
+		volumeID = pv.Spec.CSI.VolumeHandle
+	} else if pv.Spec.AWSElasticBlockStore != nil {
+		volumeID = pv.Spec.AWSElasticBlockStore.VolumeID
+	}
 	if volumeID == "" {
 		return "", fmt.Errorf("got empty volume id for volume %v", pv)
 	}