From 37d69934394b960ea235aab963fac2650ff91f54 Mon Sep 17 00:00:00 2001
From: Felix Kunde <felix-kunde@gmx.de>
Date: Thu, 27 Jun 2024 14:30:52 +0200
Subject: [PATCH] remove stream resources after drop from Postgres manifest
 (#2563)

* remove stream resources after drop from Postgres manifest
---
 pkg/cluster/cluster.go      |  2 +-
 pkg/cluster/streams.go      |  3 ++-
 pkg/cluster/streams_test.go | 17 +++++++++++++++++
 pkg/util/k8sutil/k8sutil.go | 26 ++++++++++++++++++++++++--
 4 files changed, 44 insertions(+), 4 deletions(-)

diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index 23004ef9b..e3acdb835 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -1082,7 +1082,7 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
 	}
 
 	// streams
-	if len(newSpec.Spec.Streams) > 0 {
+	if len(newSpec.Spec.Streams) > 0 || len(oldSpec.Spec.Streams) != len(newSpec.Spec.Streams) {
 		if err := c.syncStreams(); err != nil {
 			c.logger.Errorf("could not sync streams: %v", err)
 			updateFailed = true
diff --git a/pkg/cluster/streams.go b/pkg/cluster/streams.go
index a135c5767..ec4221b4b 100644
--- a/pkg/cluster/streams.go
+++ b/pkg/cluster/streams.go
@@ -327,7 +327,8 @@ func (c *Cluster) syncStreams() error {
 	if len(slotsToSync) > 0 {
 		requiredPatroniConfig.Slots = slotsToSync
 	} else {
-		return nil
+		// try to delete existing stream resources
+		return c.deleteStreams()
 	}
 
 	c.logger.Debug("syncing logical replication slots")
diff --git a/pkg/cluster/streams_test.go b/pkg/cluster/streams_test.go
index f71178823..63c38311b 100644
--- a/pkg/cluster/streams_test.go
+++ b/pkg/cluster/streams_test.go
@@ -455,4 +455,21 @@ func TestUpdateFabricEventStream(t *testing.T) {
 	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)
 	}
+
+	mockClient := k8sutil.NewMockKubernetesClient()
+	cluster.KubeClient.CustomResourceDefinitionsGetter = mockClient.CustomResourceDefinitionsGetter
+
+	// remove streams from manifest
+	pgPatched.Spec.Streams = nil
+	pgUpdated, err := cluster.KubeClient.Postgresqls(namespace).Update(
+		context.TODO(), pgPatched, metav1.UpdateOptions{})
+	assert.NoError(t, err)
+
+	cluster.Postgresql.Spec = pgUpdated.Spec
+	cluster.syncStreams()
+
+	streamList, err := cluster.KubeClient.FabricEventStreams(namespace).List(context.TODO(), listOptions)
+	if len(streamList.Items) > 0 || err != nil {
+		t.Errorf("stream resource has not been removed or unexpected error %v", err)
+	}
 }
diff --git a/pkg/util/k8sutil/k8sutil.go b/pkg/util/k8sutil/k8sutil.go
index 3db1122a9..7ae402fe3 100644
--- a/pkg/util/k8sutil/k8sutil.go
+++ b/pkg/util/k8sutil/k8sutil.go
@@ -16,8 +16,9 @@ import (
 	"github.com/zalando/postgres-operator/pkg/spec"
 	apiappsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
-	apiextv1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1"
+	apiextv1client "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -59,7 +60,7 @@ type KubernetesClient struct {
 	appsv1.DeploymentsGetter
 	rbacv1.RoleBindingsGetter
 	policyv1.PodDisruptionBudgetsGetter
-	apiextv1.CustomResourceDefinitionsGetter
+	apiextv1client.CustomResourceDefinitionsGetter
 	clientbatchv1.CronJobsGetter
 	acidv1.OperatorConfigurationsGetter
 	acidv1.PostgresTeamsGetter
@@ -71,6 +72,13 @@ type KubernetesClient struct {
 	Zalandov1ClientSet *zalandoclient.Clientset
 }
 
+type mockCustomResourceDefinition struct {
+	apiextv1client.CustomResourceDefinitionInterface
+}
+
+type MockCustomResourceDefinitionsGetter struct {
+}
+
 type mockSecret struct {
 	corev1.SecretInterface
 }
@@ -240,6 +248,18 @@ func (client *KubernetesClient) SetFinalizer(clusterName spec.NamespacedName, pg
 	return updatedPg, nil
 }
 
+func (c *mockCustomResourceDefinition) Get(ctx context.Context, name string, options metav1.GetOptions) (*apiextv1.CustomResourceDefinition, error) {
+	return &apiextv1.CustomResourceDefinition{}, nil
+}
+
+func (c *mockCustomResourceDefinition) Create(ctx context.Context, crd *apiextv1.CustomResourceDefinition, options metav1.CreateOptions) (*apiextv1.CustomResourceDefinition, error) {
+	return &apiextv1.CustomResourceDefinition{}, nil
+}
+
+func (mock *MockCustomResourceDefinitionsGetter) CustomResourceDefinitions() apiextv1client.CustomResourceDefinitionInterface {
+	return &mockCustomResourceDefinition{}
+}
+
 func (c *mockSecret) Get(ctx context.Context, name string, options metav1.GetOptions) (*v1.Secret, error) {
 	oldFormatSecret := &v1.Secret{}
 	oldFormatSecret.Name = "testcluster"
@@ -444,6 +464,8 @@ func NewMockKubernetesClient() KubernetesClient {
 		ConfigMapsGetter:  &MockConfigMapsGetter{},
 		DeploymentsGetter: &MockDeploymentGetter{},
 		ServicesGetter:    &MockServiceGetter{},
+
+		CustomResourceDefinitionsGetter: &MockCustomResourceDefinitionsGetter{},
 	}
 }