Merge remote-tracking branch 'altinity/0.25.5' into 0.25.5

Author: sunsingerus

tests/e2e/manifests/chi/test-036-volume-re-provisioning-1.yaml

@@ -3,7 +3,6 @@ kind: "ClickHouseInstallation"
 metadata:
   name: "test-036-volume-re-provisioning"
 spec:
-
   useTemplates:
     - name: clickhouse-version
   configuration:
@@ -16,6 +15,8 @@ spec:
       layout:
         shardsCount: 1
         replicasCount: 2
+    profiles:
+      clickhouse_operator/log_queries: 1
   defaults:
     storageManagement:
       provisioner: Operator

tests/e2e/manifests/chi/test-036-volume-re-provisioning-2.yaml

@@ -0,0 +1,54 @@
+apiVersion: "clickhouse.altinity.com/v1"
+kind: "ClickHouseInstallation"
+metadata:
+  name: "test-036-volume-re-provisioning"
+spec:
+  useTemplates:
+    - name: clickhouse-version
+  configuration:
+    zookeeper:
+      nodes:
+        - host: zookeeper
+          port: 2181
+    clusters:
+    - name: simple
+      layout:
+        shardsCount: 1
+        replicasCount: 2
+    profiles:
+      clickhouse_operator/log_queries: 1
+    settings:
+      storage_configuration/disks/disk2/path: /var/lib/clickhouse2/
+      storage_configuration/policies/default/volumes/default/disk: default
+      storage_configuration/policies/default/volumes/disk2/disk: disk2
+  defaults:
+    storageManagement:
+      provisioner: Operator
+    templates:
+      volumeClaimTemplate: default
+  templates:
+    volumeClaimTemplates:
+      - name: default
+        spec:
+          accessModes:
+            - ReadWriteOnce
+          resources:
+            requests:
+              storage: 1Gi
+      - name: disk2
+        spec:
+          accessModes:
+            - ReadWriteOnce
+          resources:
+            requests:
+              storage: 1Gi
+    podTemplates:
+      - name: default
+        spec:
+          containers:
+            - name: clickhouse-pod
+              volumeMounts:
+                - name: default
+                  mountPath: /var/lib/clickhouse
+                - name: disk2
+                  mountPath: /var/lib/clickhouse2

tests/e2e/test_operator.py

@@ -1508,25 +1508,6 @@ def test_010014_0(self):
                     host=f"chi-{chi_name}-{cluster}-1-0",
                 )
 
-    def check_query_log(log, do_not_log, since = '1970-01-01', flags = None):
-        out = clickhouse.query(chi_name, f"select query, substring(hostname, 1, locate('.', hostname)-1) from cluster('all-sharded', system.query_log) where user = 'clickhouse_operator' and event_time>='{since}'")
-        for q in log:
-            found = 0
-            with Then(f"system.query_log should contain {q} statements", flags = flags):
-                for l in out.splitlines():
-                    if l.lower().startswith(q.lower()):
-                        found = 1
-                        print(l)
-                assert found, error(out)
-        for q in do_not_log:
-            found = 0
-            with Then(f"system.query_log should NOT contain {q} statements", flags = flags):
-                for l in out.splitlines():
-                    if l.lower().startswith(q.lower()):
-                        found = 1
-                        print(l)
-                assert not found, error(out)
-
     def check_schema_propagation(replicas):
         for replica in replicas:
             host = f"chi-{chi_name}-{cluster}-0-{replica}"
@@ -1651,7 +1632,7 @@ def check_schema_propagation(replicas):
 
         check_schema_propagation(replicas)
 
-        check_query_log(['CREATE'], [], query_log_start)
+        util.check_query_log(chi_name, ['CREATE'], [], query_log_start)
 
     with When("Restart (Zoo)Keeper pod"):
         if self.context.keeper_type == "zookeeper":
@@ -1707,7 +1688,7 @@ def check_schema_propagation(replicas):
                 )
                 assert out == "1"
 
-        check_query_log(['SYSTEM DROP REPLICA'], ['DROP TABLE', 'DROP DATABASE'], query_log_start)
+        util.check_query_log(chi_name, ['SYSTEM DROP REPLICA'], ['DROP TABLE', 'DROP DATABASE'], query_log_start)
 
     with When("Add replica one more time"):
         manifest = "manifests/chi/test-014-0-replication-2.yaml"
@@ -1743,7 +1724,7 @@ def check_schema_propagation(replicas):
             note(f"Found {out} replicated tables in {self.context.keeper_type}")
             assert "DB::Exception: No node" in out or out == "0"
 
-        check_query_log(['SYSTEM DROP REPLICA'], ['DROP TABLE', 'DROP DATABASE'], query_log_start, flags=XFAIL)
+        util.check_query_log(chi_name, ['SYSTEM DROP REPLICA'], ['DROP TABLE', 'DROP DATABASE'], query_log_start, flags=XFAIL)
 
     with When("Delete chi"):
         kubectl.delete_chi("test-014-replication")
@@ -3789,17 +3770,12 @@ def test_010036(self):
     """Check clickhouse operator recreates volumes and schema if volume is broken."""
     create_shell_namespace_clickhouse_template()
 
-    with Given("I create shells"):
-        shell = get_shell()
-        self.context.shell = shell
-        shell_2 = get_shell()
-
-    manifest = f"manifests/chi/test-036-volume-re-provisioning-1.yaml"
+    manifest = "manifests/chi/test-036-volume-re-provisioning-1.yaml"
     chi = yaml_manifest.get_name(util.get_full_path(manifest))
     cluster = "simple"
     util.require_keeper(keeper_type=self.context.keeper_type)
 
-    with Given("chi exists"):
+    with Given("CHI with two replicas is created"):
         kubectl.create_and_check(
             manifest=manifest,
             check={
@@ -3812,33 +3788,31 @@ def test_010036(self):
     wait_for_cluster(chi, cluster, 1, 2)
 
     with And("I create replicated table with some data"):
-        clickhouse.query(chi, "CREATE DATABASE test_036 ON CLUSTER '{cluster}'")
+        clickhouse.query(chi, "CREATE DATABASE IF NOT EXISTS test_036 ON CLUSTER '{cluster}'")
         create_table = """
-            CREATE TABLE test_036.test_local_036 ON CLUSTER '{cluster}' (a UInt32)
+            CREATE TABLE IF NOT EXISTS test_036.test_local_036 ON CLUSTER '{cluster}' (a UInt32)
             Engine = ReplicatedMergeTree('/clickhouse/{installation}/tables/{shard}/{database}/{table}', '{replica}')
             PARTITION BY tuple()
             ORDER BY a
             """.replace("\r", "").replace("\n", "")
         clickhouse.query(chi, create_table)
         clickhouse.query(chi, f"INSERT INTO test_036.test_local_036 select * from numbers(10000)")
 
-        clickhouse.query(chi, "CREATE DATABASE test_036_mem ON CLUSTER '{cluster}' Engine = Memory")
-        clickhouse.query(chi, "CREATE VIEW test_036_mem.test_view ON CLUSTER '{cluster}' AS SELECT * from system.tables")
+        clickhouse.query(chi, "CREATE DATABASE IF NOT EXISTS test_036_mem ON CLUSTER '{cluster}' Engine = Memory")
+        clickhouse.query(chi, "CREATE VIEW IF NOT EXISTS test_036_mem.test_view ON CLUSTER '{cluster}' AS SELECT * from system.tables")
 
-    def delete_pv():
+    def delete_pv(volume):
         with When("Delete PV", description="delete PV on replica 0"):
             # Prepare counters
             pvc_count = kubectl.get_count("pvc", chi=chi)
             pv_count = kubectl.get_count("pv")
             print(f"pvc_count: {pvc_count}")
             print(f"pv_count: {pv_count}")
 
-            pv_name = kubectl.get_pv_name("default-chi-test-036-volume-re-provisioning-simple-0-0-0")
+            pv_name = kubectl.get_pv_name(f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0")
             # retry
-            kubectl.launch(f"delete pv {pv_name} --force &", shell=shell_2)
-            kubectl.launch(
-                f"""patch pv {pv_name} --type='json' --patch='[{{"op":"remove","path":"/metadata/finalizers"}}]'"""
-            )
+            kubectl.launch(f"delete pv {pv_name} --force &")
+            kubectl.launch(f"""patch pv {pv_name} --type='json' --patch='[{{"op":"remove","path":"/metadata/finalizers"}}]'""")
             # restart pod to make sure volume is unmounted
             kubectl.launch("delete pod chi-test-036-volume-re-provisioning-simple-0-0-0")
             # Give it some time to be deleted
@@ -3852,53 +3826,55 @@ def delete_pv():
                 assert new_pvc_count == pvc_count
                 assert new_pv_count < pv_count
 
-        with And("Wait for PVC to detect PV is lost"):
-            # Need to add more retries on real kubernetes
-            kubectl.wait_field(
-                kind="pvc",
-                name="default-chi-test-036-volume-re-provisioning-simple-0-0-0",
-                field=".status.phase",
-                value="Lost",
-            )
+            with And("Wait for PVC to detect PV is lost"):
+                # Need to add more retries on real kubernetes
+                kubectl.wait_field(
+                    kind="pvc",
+                    name=f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0",
+                    field=".status.phase",
+                    value="Lost",
+                )
 
-    def delete_sts_and_pvc():
+    def delete_sts_and_pvc(volume):
         with When("Delete StatefulSet and PVC", description="delete StatefulSet on replica 0"):
             kubectl.launch("delete sts chi-test-036-volume-re-provisioning-simple-0-0")
-            kubectl.launch("delete pvc default-chi-test-036-volume-re-provisioning-simple-0-0-0")
+            kubectl.launch(f"delete pvc {volume}-chi-test-036-volume-re-provisioning-simple-0-0-0")
 
-        with Then("Wait for StatefulSet is deleted"):
-            for i in range(5):
-                if kubectl.get_count("sts", "chi-test-036-volume-re-provisioning-simple-0-0") == 0:
-                    break
-                time.sleep(10)
+            with Then("Wait for StatefulSet is deleted"):
+                for i in range(5):
+                    if kubectl.get_count("sts", "chi-test-036-volume-re-provisioning-simple-0-0") == 0:
+                        break
+                    time.sleep(10)
 
-        with Then("Wait for PVC is deleted"):
-            for i in range(5):
-                if kubectl.get_count("pvc", "default-chi-test-036-volume-re-provisioning-simple-0-0-0") == 0:
-                    break
-                time.sleep(10)
+            with Then("Wait for PVC is deleted"):
+                for i in range(5):
+                    if kubectl.get_count("pvc", f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0") == 0:
+                        break
+                    time.sleep(10)
 
         assert kubectl.get_count("sts", "chi-test-036-volume-re-provisioning-simple-0-0") == 0, "StatefulSet is not deleted"
-        assert kubectl.get_count("pvc", "default-chi-test-036-volume-re-provisioning-simple-0-0-0") == 0, "PVC is not deleted"
+        assert kubectl.get_count("pvc", f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0") == 0, "PVC is not deleted"
 
-    def delete_pvc():
-        with When("Delete PVC", description="delete PVC on replica 0"):
+    def delete_pvc(volume):
+        with Then("Delete PVC", description="delete PVC on replica 0"):
             # Prepare counters
             pvc_count = kubectl.get_count("pvc", chi=chi)
             pv_count = kubectl.get_count("pv")
             print(f"pvc_count: {pvc_count}")
             print(f"pv_count: {pv_count}")
 
-            pvc_name = f"default-chi-test-036-volume-re-provisioning-simple-0-0-0"
+            pvc_name = f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0"
             # retry
-            kubectl.launch(f"delete pvc {pvc_name} --force &", shell=shell_2)
-            kubectl.launch(
-                f"""patch pvc {pvc_name} --type='json' --patch='[{{"op":"remove","path":"/metadata/finalizers"}}]'"""
-            )
+            kubectl.launch(f"""patch pvc {pvc_name} --type='json' --patch='[{{"op":"remove","path":"/metadata/finalizers"}}]'""")
+            kubectl.launch(f"delete pvc {pvc_name} --force &")
+
             # restart pod to make sure volume is unmounted
             kubectl.launch("delete pod chi-test-036-volume-re-provisioning-simple-0-0-0")
-            # Give it some time to be deleted
-            time.sleep(10)
+            with Then("Wait for PVC is deleted"):
+                for i in range(5):
+                    if kubectl.get_count("pvc", f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0") == 0:
+                        break
+                    time.sleep(10)
 
             with Then("PVC should be deleted, PV should be deleted as well"):
                 new_pvc_count = kubectl.get_count("pvc", chi=chi)
@@ -3908,25 +3884,26 @@ def delete_pvc():
                 assert new_pvc_count < pvc_count
                 assert new_pv_count < pv_count
 
-    def check_data_is_recovered(reconcile_task_id):
-        with Then(f"Kick operator to start reconcile cycle to fix lost {reconcile_task_id}"):
+    def recover_volume(volume, reconcile_task_id):
+        with When(f"Kick operator to start reconcile cycle to fix lost {volume} volume"):
             kubectl.force_reconcile(chi, reconcile_task_id)
             wait_for_cluster(chi, cluster, 1, 2)
 
-        with Then("I check PV is in place"):
-            kubectl.wait_field(
-                "pvc",
-                "default-chi-test-036-volume-re-provisioning-simple-0-0-0",
-                ".status.phase",
-                "Bound",
-            )
-            kubectl.wait_object(
-                "pv",
-                kubectl.get_pv_name("default-chi-test-036-volume-re-provisioning-simple-0-0-0"),
-            )
-            size = kubectl.get_pv_size("default-chi-test-036-volume-re-provisioning-simple-0-0-0")
-            assert size == "1Gi", error()
+            with Then("I check PV is in place"):
+                kubectl.wait_field(
+                    "pvc",
+                    f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0",
+                    ".status.phase",
+                    "Bound",
+                )
+                kubectl.wait_object(
+                    "pv",
+                    kubectl.get_pv_name(f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0"),
+                )
+                size = kubectl.get_pv_size(f"{volume}-chi-test-036-volume-re-provisioning-simple-0-0-0")
+                assert size == "1Gi", error()
 
+    def check_data_is_recovered():
         with And("I check data on each replica"):
             for replica in (0,1):
                 with By(f"Check that databases exist on replica {replica}"):
@@ -3951,14 +3928,39 @@ def check_data_is_recovered(reconcile_task_id):
                         )
                 assert r == "1", error()
 
-    delete_sts_and_pvc()
-    check_data_is_recovered("reconcile-after-STS-and-PVC-deleted")
+    delete_sts_and_pvc("default")
+    recover_volume("default", "reconcile-after-STS-and-PVC-deleted")
+    check_data_is_recovered()
 
-    delete_pvc()
-    check_data_is_recovered("reconcile-after-PVC-deleted")
+    query_log_start = clickhouse.query(chi, 'select now()')
+    delete_pvc("default")
+    recover_volume("default", "reconcile-after-PVC-deleted")
+    check_data_is_recovered()
+    util.check_query_log(chi, ['SYSTEM DROP REPLICA'], [], since = query_log_start)
+
+    delete_pv("default")
+    recover_volume("default", "reconcile-after-PV-deleted")
+    check_data_is_recovered()
+
+    with Then("Add a second disk"):
+        kubectl.create_and_check(
+            manifest="manifests/chi/test-036-volume-re-provisioning-2.yaml",
+            check={
+                "apply_templates": {current().context.clickhouse_template},
+                "pod_count": 2,
+                "do_not_delete": 1,
+            },
+        )
+        wait_for_cluster(chi, cluster, 1, 2)
+        with Then("Confirm there are two disks"):
+            out = clickhouse.query(chi, "select count() from system.disks")
+            assert out == "2"
 
-    delete_pv()
-    check_data_is_recovered("reconcile-after-PV-deleted")
+    query_log_start = clickhouse.query(chi, 'select now()')
+    delete_pvc("disk2")
+    recover_volume("disk2", "reconcile-after-disk2-PVC-deleted")
+    check_data_is_recovered()
+    util.check_query_log(chi, [], ['SYSTEM DROP REPLICA'], since = query_log_start)
 
     with Finally("I clean up"):
         delete_test_namespace()

tests/e2e/util.py

@@ -8,6 +8,7 @@
 import e2e.util as util
 
 from testflows.core import fail, Given, Then, But, current, message
+from testflows.asserts import error
 
 
 current_dir = os.path.dirname(os.path.abspath(__file__))
@@ -358,4 +359,23 @@ def require_expandable_storage_class():
         assert len(default_storage_class) > 0
         allow_volume_expansion = kubectl.get_field("storageclass", default_storage_class, ".allowVolumeExpansion")
         if allow_volume_expansion != "true":
-            kubectl.launch(f"patch storageclass {default_storage_class} -p '{{\"allowVolumeExpansion\":true}}'")
+            kubectl.launch(f"patch storageclass {default_storage_class} -p '{{\"allowVolumeExpansion\":true}}'")
+
+def check_query_log(chi_name, log, do_not_log, since = '1970-01-01', flags = None):
+    out = clickhouse.query(chi_name, f"select query, substring(hostname, 1, locate('.', hostname)-1) from cluster('all-sharded', system.query_log) where user = 'clickhouse_operator' and event_time>='{since}'")
+    for q in log:
+        found = 0
+        with Then(f"system.query_log should contain {q} statements", flags = flags):
+            for l in out.splitlines():
+                if l.lower().startswith(q.lower()):
+                    found = 1
+                    print(l)
+            assert found, error(out)
+    for q in do_not_log:
+        found = 0
+        with Then(f"system.query_log should NOT contain {q} statements", flags = flags):
+            for l in out.splitlines():
+                if l.lower().startswith(q.lower()):
+                    found = 1
+                    print(l)
+            assert not found, error(out)