Enable slot and publication deletion when stream application is removed (#2684)

* refactor syncing publication section
* update createOrUpdateStream function to allow resource deletion when removed from manifest
* add minimal FES CRD to enable FES resources creation for E2E test
* fix bug of removing manifest slots in syncStream
* e2e test: fixing typo with major upgrade test
* e2e test: should create and delete FES resource
* e2e test: should not delete manual created resources
* e2e test: enable cluster role for FES with patching instead of deploying in manifest

Author: idanovinda

e2e/tests/k8s_api.py

@@ -20,6 +20,7 @@ def __init__(self):
 
         self.config = config.load_kube_config()
         self.k8s_client = client.ApiClient()
+        self.rbac_api = client.RbacAuthorizationV1Api()
 
         self.core_v1 = client.CoreV1Api()
         self.apps_v1 = client.AppsV1Api()

e2e/tests/test_e2e.py

@@ -129,7 +129,8 @@ def setUpClass(cls):
                          "infrastructure-roles.yaml",
                          "infrastructure-roles-new.yaml",
                          "custom-team-membership.yaml",
-                         "e2e-storage-class.yaml"]:
+                         "e2e-storage-class.yaml",
+                         "fes.crd.yaml"]:
             result = k8s.create_with_kubectl("manifests/" + filename)
             print("stdout: {}, stderr: {}".format(result.stdout, result.stderr))
 
@@ -199,6 +200,7 @@ def test_additional_owner_roles(self):
         self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "postgres", owner_query)), 3,
             "Not all additional users found in database", 10, 5)
 
+
     @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
     def test_additional_pod_capabilities(self):
         '''
@@ -1203,7 +1205,7 @@ def check_version_14():
             version = p["server_version"][0:2]
             return version
 
-        self.evantuallyEqual(check_version_14, "14", "Version was not upgrade to 14")
+        self.eventuallyEqual(check_version_14, "14", "Version was not upgrade to 14")
 
     @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
     def test_persistent_volume_claim_retention_policy(self):
@@ -1989,6 +1991,123 @@ def test_standby_cluster(self):
                 "acid.zalan.do", "v1", "default", "postgresqls", "acid-standby-cluster")
             time.sleep(5)
 
+    @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
+    def test_stream_resources(self):
+        '''
+           Create and delete fabric event streaming resources.
+        '''
+        k8s = self.k8s
+        self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"},
+            "Operator does not get in sync")
+        leader = k8s.get_cluster_leader_pod()
+
+        # patch ClusterRole with CRUD privileges on FES resources
+        cluster_role = k8s.api.rbac_api.read_cluster_role("postgres-operator")
+        fes_cluster_role_rule = client.V1PolicyRule(
+            api_groups=["zalando.org"],
+            resources=["fabriceventstreams"],
+            verbs=["create", "delete", "deletecollection", "get", "list", "patch", "update", "watch"]
+        )
+        cluster_role.rules.append(fes_cluster_role_rule)
+        k8s.api.rbac_api.patch_cluster_role("postgres-operator", cluster_role)
+
+        # create a table in one of the database of acid-minimal-cluster
+        create_stream_table = """
+            CREATE TABLE test_table (id int, payload jsonb);
+        """
+        self.query_database(leader.metadata.name, "foo", create_stream_table)
+
+        # update the manifest with the streams section
+        patch_streaming_config = {
+            "spec": {
+                 "patroni": {
+                    "slots": {
+                        "manual_slot": {
+                            "type": "physical"
+                        }
+                    }
+                 },
+                "streams": [
+                    {
+                        "applicationId": "test-app",
+                        "batchSize": 100,
+                        "database": "foo",
+                        "enableRecovery": True,
+                        "tables": {
+                            "test_table": {
+                                "eventType": "test-event",
+                                "idColumn": "id",
+                                "payloadColumn": "payload",
+                                "recoveryEventType": "test-event-dlq"
+                            }
+                        }
+                    }
+                ]
+            }
+        }
+        k8s.api.custom_objects_api.patch_namespaced_custom_object(
+            'acid.zalan.do', 'v1', 'default', 'postgresqls', 'acid-minimal-cluster', patch_streaming_config)
+        self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
+
+        # check if publication, slot, and fes resource are created
+        get_publication_query = """
+            SELECT * FROM pg_publication WHERE pubname = 'fes_foo_test_app';
+        """
+        get_slot_query = """
+            SELECT * FROM pg_replication_slots WHERE slot_name = 'fes_foo_test_app';
+        """
+        self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "foo", get_publication_query)), 1,
+            "Publication is not created", 10, 5)
+        self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "foo", get_slot_query)), 1,
+            "Replication slot is not created", 10, 5)
+        self.eventuallyEqual(lambda: len(k8s.api.custom_objects_api.list_namespaced_custom_object(
+                "zalando.org", "v1", "default", "fabriceventstreams", label_selector="cluster-name=acid-minimal-cluster")["items"]), 1,
+                "Could not find Fabric Event Stream resource", 10, 5)
+
+        # grant create and ownership of test_table to foo_user, reset search path to default
+        grant_permission_foo_user = """
+            GRANT CREATE ON DATABASE foo TO foo_user;
+            ALTER TABLE test_table OWNER TO foo_user;
+            ALTER ROLE foo_user RESET search_path;
+        """
+        self.query_database(leader.metadata.name, "foo", grant_permission_foo_user)
+        # non-postgres user creates a publication
+        create_nonstream_publication = """
+            CREATE PUBLICATION mypublication FOR TABLE test_table;
+        """
+        self.query_database_with_user(leader.metadata.name, "foo", create_nonstream_publication, "foo_user")
+
+        # remove the streams section from the manifest
+        patch_streaming_config_removal = {
+            "spec": {
+                "streams": []
+            }
+        }
+        k8s.api.custom_objects_api.patch_namespaced_custom_object(
+            'acid.zalan.do', 'v1', 'default', 'postgresqls', 'acid-minimal-cluster', patch_streaming_config_removal)
+        self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, "Operator does not get in sync")
+
+        # check if publication, slot, and fes resource are removed
+        self.eventuallyEqual(lambda: len(k8s.api.custom_objects_api.list_namespaced_custom_object(
+                "zalando.org", "v1", "default", "fabriceventstreams", label_selector="cluster-name=acid-minimal-cluster")["items"]), 0,
+                'Could not delete Fabric Event Stream resource', 10, 5)
+        self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "foo", get_publication_query)), 0,
+            "Publication is not deleted", 10, 5)
+        self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "foo", get_slot_query)), 0,
+            "Replication slot is not deleted", 10, 5)
+
+        # check the manual_slot and mypublication should not get deleted
+        get_manual_slot_query = """
+            SELECT * FROM pg_replication_slots WHERE slot_name = 'manual_slot';
+        """
+        get_nonstream_publication_query = """
+            SELECT * FROM pg_publication WHERE pubname = 'mypublication';
+        """
+        self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "postgres", get_manual_slot_query)), 1,
+            "Slot defined in patroni config is deleted", 10, 5)
+        self.eventuallyEqual(lambda: len(self.query_database(leader.metadata.name, "foo", get_nonstream_publication_query)), 1,
+            "Publication defined not in stream section is deleted", 10, 5)
+
     @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
     def test_taint_based_eviction(self):
         '''
@@ -2115,7 +2234,7 @@ def test_zz_cluster_deletion(self):
             self.eventuallyEqual(lambda: k8s.count_statefulsets_with_label(cluster_label), 0, "Statefulset not deleted")
             self.eventuallyEqual(lambda: k8s.count_deployments_with_label(cluster_label), 0, "Deployments not deleted")
             self.eventuallyEqual(lambda: k8s.count_pdbs_with_label(cluster_label), 0, "Pod disruption budget not deleted")
-            self.eventuallyEqual(lambda: k8s.count_secrets_with_label(cluster_label), 7, "Secrets were deleted although disabled in config")
+            self.eventuallyEqual(lambda: k8s.count_secrets_with_label(cluster_label), 8, "Secrets were deleted although disabled in config")
             self.eventuallyEqual(lambda: k8s.count_pvcs_with_label(cluster_label), 3, "PVCs were deleted although disabled in config")
 
         except timeout_decorator.TimeoutError:

manifests/fes.crd.yaml

@@ -0,0 +1,23 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: fabriceventstreams.zalando.org
+spec:
+  group: zalando.org
+  names:
+    kind: FabricEventStream
+    listKind: FabricEventStreamList
+    plural: fabriceventstreams
+    singular: fabriceventstream
+    shortNames:
+    - fes
+    categories:
+    - all
+  scope: Namespaced
+  versions:
+  - name: v1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object

pkg/apis/zalando.org/v1/fabriceventstream.go

@@ -1,6 +1,7 @@
 package v1
 
 import (
+	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -89,3 +90,8 @@ type DBAuth struct {
 	UserKey     string `json:"userKey,omitempty"`
 	PasswordKey string `json:"passwordKey,omitempty"`
 }
+
+type Slot struct {
+	Slot        map[string]string             `json:"slot"`
+	Publication map[string]acidv1.StreamTable `json:"publication"`
+}

pkg/cluster/database.go

@@ -49,9 +49,12 @@ const (
 	getPublicationsSQL = `SELECT p.pubname, string_agg(pt.schemaname || '.' || pt.tablename, ', ' ORDER BY pt.schemaname, pt.tablename)
 	        FROM pg_publication p
 			LEFT JOIN pg_publication_tables pt ON pt.pubname = p.pubname
+			WHERE p.pubowner = 'postgres'::regrole
+			AND p.pubname LIKE 'fes_%'
 			GROUP BY p.pubname;`
 	createPublicationSQL = `CREATE PUBLICATION "%s" FOR TABLE %s WITH (publish = 'insert, update');`
 	alterPublicationSQL  = `ALTER PUBLICATION "%s" SET TABLE %s;`
+	dropPublicationSQL   = `DROP PUBLICATION "%s";`
 
 	globalDefaultPrivilegesSQL = `SET ROLE TO "%s";
 			ALTER DEFAULT PRIVILEGES GRANT USAGE ON SCHEMAS TO "%s","%s";
@@ -628,6 +631,14 @@ func (c *Cluster) getPublications() (publications map[string]string, err error)
 	return dbPublications, err
 }
 
+func (c *Cluster) executeDropPublication(pubName string) error {
+	c.logger.Infof("dropping publication %q", pubName)
+	if _, err := c.pgDb.Exec(fmt.Sprintf(dropPublicationSQL, pubName)); err != nil {
+		return fmt.Errorf("could not execute drop publication: %v", err)
+	}
+	return nil
+}
+
 // executeCreatePublication creates new publication for given tables
 // The caller is responsible for opening and closing the database connection.
 func (c *Cluster) executeCreatePublication(pubName, tableList string) error {