HCD-241: Add a dedicated DSE 6.9 upgrade test

dtest.py

@@ -233,7 +233,7 @@ def test_failure_due_to_timeout(err, *args):
         return False
 
 
-@flaky(rerun_filter=test_failure_due_to_timeout)
+#@flaky(rerun_filter=test_failure_due_to_timeout)
 class Tester(object):
 
     def __getattribute__(self, name):

dtest_setup.py

@@ -61,7 +61,10 @@ def default_ignore_log_patterns():
             # queries which probe for the default superuser role to be directed to the wrong node, _if_ they are sent
             # while peers are performing UnsafeJoins. This behaviour is safe, but it can cause flaky tests if the
             # ERROR unexpectedly appears in the log
-            r'Received a read request from .* for a token .* that is not owned by the current replica .* key=cassandra .*'
+            r'Received a read request from .* for a token .* that is not owned by the current replica .* key=cassandra .*',
+            # the two below error patterns are needed by DSE upgrade tests
+            r'that is incompatible with the version this node can handle',
+            r'Unable to merge schema from',
             ]
 
 

requirements.txt

@@ -9,7 +9,7 @@
 #
 # In case you want to test a patch with your own CCM branch, further to changing below CCM repo and branch name, you need to add -e flag at the beginning
 # Example: -e git+https://github.com/userb/ccm.git@cassandra-17182#egg=ccm
-git+https://github.com/datastax/cassandra-ccm.git@converged-cassandra#egg=ccm
+-e git+https://github.com/datastax/cassandra-ccm.git@HCD-267#egg=ccm
 click==8.0.4
 decorator==5.1.1
 docopt==0.6.2

upgrade_tests/upgrade_from_dse_test.py

@@ -0,0 +1,157 @@
+import shutil
+from distutils.version import LooseVersion
+
+import pytest
+import logging
+
+from cassandra import ConsistencyLevel
+from cassandra.query import (SimpleStatement, dict_factory)
+
+from dtest import Tester
+
+from tools.data import rows_to_list
+from tools.misc import generate_ssl_stores
+from upgrade_tests.upgrade_manifest import get_cluster_class, current_dse_6_9, CC4, indev_dse_6_9, indev_dse_6_8, \
+    indev_dse_5_1, current_dse_6_8, current_dse_5_1
+
+since = pytest.mark.since
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.upgrade_test
+@since('4.0', max_version='4.1.x')
+class TestUpgradeFromDSEWithConcurrentWritesAndRestarts(Tester):
+
+    @pytest.fixture(autouse=True)
+    def fixture_add_additional_log_patterns(self, fixture_dtest_setup):
+        fixture_dtest_setup.ignore_log_patterns = fixture_dtest_setup.ignore_log_patterns + [
+            r'Collectd is only supported on Linux',
+            r'that is incompatible with the version this node can handle',
+            r'Unable to merge schema from',
+            # such an error happens once in a while for upgrade, it does not seem to have a functional impact though
+            r'InvalidCrc',
+        ]
+
+    def test_upgrade_from_6_9_with_internode_encryption(self):
+        self.upgrade_while_writing(version_descriptor=current_dse_6_9, internode_encryption=True)
+
+    def test_upgrade_from_6_9(self):
+        self.upgrade_while_writing(version_descriptor=current_dse_6_9, internode_encryption=False)
+
+    # TODO fails def test_upgrade_from_6_8_with_internode_encryption(self):
+    #     self.upgrade_while_writing(version_descriptor=indev_dse_6_8, internode_encryption=True)
+
+    def test_upgrade_from_6_8(self):
+        self.upgrade_while_writing(version_descriptor=current_dse_6_8, internode_encryption=False)
+
+    # TODO fails def test_upgrade_from_5_1_with_internode_encryption(self):
+    #     self.upgrade_while_writing(version_descriptor=indev_dse_5_1, internode_encryption=True)
+
+    def test_upgrade_from_5_1(self):
+        self.upgrade_while_writing(version_descriptor=current_dse_5_1, internode_encryption=False)
+
+    def upgrade_while_writing(self, version_descriptor, internode_encryption=False):
+        cluster = self._setup_nodes(version_descriptor, internode_encryption=internode_encryption)
+        cluster.start()
+
+        if self.cluster.version() >= LooseVersion('6.8'):
+            for node in cluster.nodelist():
+                node.nodetool('nodesyncservice disable')
+
+        nodes = cluster.nodelist()
+
+        out, err, _ = nodes[0].run_cqlsh("CREATE KEYSPACE ks WITH replication = {'class':'SimpleStrategy', 'replication_factor':3}")
+        out, err, _ = nodes[0].run_cqlsh("CREATE TABLE ks.cf (id int primary key)")
+
+        for i in range(0, 100):
+            out, err, _ = cluster.nodelist()[0].run_cqlsh("INSERT INTO ks.cf (id) VALUES ({})".format(i))
+
+        # upgrade nodes ensuring writes with CL=ALL work
+        upgraded_node_1 = self._upgrade_node(cluster, nodes[0], internode_encryption=internode_encryption)
+        self._assert_insert_select_works(upgraded_node_1, 100, 200)
+
+        upgraded_node_2 = self._upgrade_node(cluster, nodes[1], internode_encryption=internode_encryption)
+        self._assert_insert_select_works(upgraded_node_2, 200, 300)
+
+        upgraded_node_3 = self._upgrade_node(cluster, nodes[2], internode_encryption=internode_encryption)
+        self._assert_insert_select_works(upgraded_node_3, 300, 400)
+
+        # rolling restart post upgrade
+        upgraded_node_1.stop(wait_other_notice=True)
+        self._assert_insert_select_works(upgraded_node_2, 400, 500, cl=ConsistencyLevel.QUORUM)
+        upgraded_node_1.start(jvm_args=["-Dcassandra.skip_schema_check=true"], wait_other_notice=160,
+                              wait_for_binary_proto=True)
+        self._assert_insert_select_works(upgraded_node_1, 400, 500)
+
+        upgraded_node_2.stop(wait_other_notice=True)
+        self._assert_insert_select_works(upgraded_node_3, 500, 600, cl=ConsistencyLevel.QUORUM)
+        upgraded_node_2.start(jvm_args=["-Dcassandra.skip_schema_check=true"], wait_other_notice=160,
+                              wait_for_binary_proto=True)
+        self._assert_insert_select_works(upgraded_node_2, 500, 600)
+
+        upgraded_node_3.stop(wait_other_notice=True)
+        self._assert_insert_select_works(upgraded_node_1, 600, 700, cl=ConsistencyLevel.QUORUM)
+        upgraded_node_3.start(jvm_args=["-Dcassandra.skip_schema_check=true"], wait_other_notice=160,
+                              wait_for_binary_proto=True)
+        self._assert_insert_select_works(upgraded_node_3, 600, 700)
+
+    def _setup_nodes(self, version_descriptor, internode_encryption=False):
+        cluster = self.cluster
+
+        meta_family_class = get_cluster_class(version_descriptor.family)
+        self.cluster.__class__ = meta_family_class
+        self.cluster._cassandra_version = self.cluster.dse_username = self.cluster.dse_password = self.cluster.opscenter = None
+
+        cluster.set_install_dir(version=version_descriptor.version)
+
+        if cluster.version() >= LooseVersion('6.8'):
+            cluster.set_configuration_options(values={'endpoint_snitch': 'org.apache.cassandra.locator.SimpleSnitch', 'nodesync': {'enabled': False}})
+        else:
+            cluster.set_configuration_options(values={'endpoint_snitch': 'org.apache.cassandra.locator.SimpleSnitch'})
+
+        if internode_encryption:
+            generate_ssl_stores(self.fixture_dtest_setup.test_path)
+            self.cluster.enable_internode_ssl(self.fixture_dtest_setup.test_path, enable_legacy_ssl_storage_port=True)
+
+        self.fixture_dtest_setup.reinitialize_cluster_for_different_version()
+        self.install_nodetool_legacy_parsing()
+
+        cluster = cluster.populate(3)
+        return cluster
+
+    def _upgrade_node(self, cluster, node, internode_encryption=False):
+        logger.info('stopping dse node ' + str(node.address()))
+        node.stop(wait_other_notice=True)
+        logger.info('stopped dse node ' + str(node.address()))
+
+        meta_family_class = get_cluster_class(CC4)
+        cluster.__class__ = meta_family_class
+        cluster._cassandra_version = self.cluster.dse_username = self.cluster.dse_password = self.cluster.opscenter = None
+
+        if node.get_cassandra_version() >= LooseVersion('4.0'):
+            node.set_configuration_options(values={'nodesync': None}, delete_empty=True, delete_always=True)
+            cluster.set_configuration_options(values={'nodesync': None}, delete_empty=True, delete_always=True)
+
+        old_conf = node.get_conf_dir()
+        node.__class__ = cluster.getNodeClass()
+        shutil.copytree(old_conf, node.get_conf_dir())
+
+        node.set_install_dir(install_dir=self.dtest_config.cassandra_dir)
+
+        self.install_legacy_parsing(node)
+
+        if internode_encryption:
+            node.set_configuration_options(values={'server_encryption_options': {'enabled': True}})
+
+        logger.info('starting hcd node' + str(node.address()))
+        node.start(jvm_args=["-Dcassandra.skip_schema_check=true"], wait_other_notice=160, wait_for_binary_proto=True)
+        logger.info('started hcd node' + str(node.address()))
+        return node
+
+    def _assert_insert_select_works(self, node, start, end, cl=ConsistencyLevel.ALL):
+        session = self.patient_exclusive_cql_connection(node)
+        for i in range(start, end):
+            session.execute("INSERT INTO ks.cf (id) VALUES ({})".format(i))
+        rows = rows_to_list(
+            session.execute(SimpleStatement("SELECT * FROM ks.cf", consistency_level=cl)))
+        assert len(rows) == end, "Expected to read {} rows from upgraded node but got {}".format(end, len(rows))