public_git
/
postgres-operator
mirror of https://github.com/zalando/postgres-operator.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

patch event stream CRD instead of update (#2535)

This commit is contained in:
Felix Kunde 2024-02-14 18:20:02 +01:00 committed by GitHub
parent 569fc57fc8
commit 3d9d0258f0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 53 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/cluster/streams.go
View File

@ -2,6 +2,7 @@ package cluster
import (
"context"
"encoding/json"
"fmt"
"reflect"
"sort"
@ -13,6 +14,7 @@ import (
"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"
)
func (c *Cluster) createStreams(appId string) (*zalandov1.FabricEventStream, error) {
@ -29,8 +31,12 @@ func (c *Cluster) createStreams(appId string) (*zalandov1.FabricEventStream, err
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 {
patch, err := json.Marshal(newEventStreams)
if err != nil {
return fmt.Errorf("could not marshal new event stream CRD %q: %v", newEventStreams.Name, err)
}
if _, err := c.KubeClient.FabricEventStreams(newEventStreams.Namespace).Patch(
context.TODO(), newEventStreams.Name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil {
return err
}

46
pkg/cluster/streams_test.go
View File

@ -294,6 +294,21 @@ func TestSameStreams(t *testing.T) {
},
}
stream3 := zalandov1.EventStream{
EventStreamFlow: zalandov1.EventStreamFlow{},
EventStreamRecovery: zalandov1.EventStreamRecovery{
Type: constants.EventStreamRecoveryNoneType,
},
EventStreamSink: zalandov1.EventStreamSink{
EventType: "stream-type-b",
},
EventStreamSource: zalandov1.EventStreamSource{
EventStreamTable: zalandov1.EventStreamTable{
Name: "bar",
},
},
}
tests := []struct {
subTest string
streamsA []zalandov1.EventStream
@ -336,6 +351,13 @@ func TestSameStreams(t *testing.T) {
match: false,
reason: "number of defined streams is different",
},
{
subTest: "event stream recovery specs differ",
streamsA: []zalandov1.EventStream{stream2},
streamsB: []zalandov1.EventStream{stream3},
match: false,
reason: "event stream specs differ",
},
}
for _, tt := range tests {
@ -409,6 +431,28 @@ func TestUpdateFabricEventStream(t *testing.T) {
result := cluster.generateFabricEventStream(appId)
if match, _ := sameStreams(streams.Items[0].Spec.EventStreams, result.Spec.EventStreams); !match {
t.Errorf("Malformed FabricEventStream, expected %#v, got %#v", streams.Items[0], result)
t.Errorf("Malformed FabricEventStream after updating manifest, expected %#v, got %#v", streams.Items[0], result)
}
// disable recovery
for _, stream := range pg.Spec.Streams {
if stream.ApplicationId == appId {
stream.EnableRecovery = util.False()
}
}
patchData, err = specPatch(pg.Spec)
assert.NoError(t, err)
pgPatched, err = cluster.KubeClient.Postgresqls(namespace).Patch(
context.TODO(), cluster.Name, types.MergePatchType, patchData, metav1.PatchOptions{}, "spec")
assert.NoError(t, err)
cluster.Postgresql.Spec = pgPatched.Spec
err = cluster.createOrUpdateStreams()
assert.NoError(t, err)
result = cluster.generateFabricEventStream(appId)
if match, _ := sameStreams(streams.Items[0].Spec.EventStreams, result.Spec.EventStreams); !match {
t.Errorf("Malformed FabricEventStream after disabling event recovery, expected %#v, got %#v", streams.Items[0], result)
}
}