| import re |
| import time |
| import pytest |
| import logging |
| |
| from ccmlib.node import TimeoutError |
| from dtest import Tester |
| |
| since = pytest.mark.since |
| logger = logging.getLogger(__name__) |
| |
| |
| class TestSchemaAgreementUpgrade(Tester): |
| """ |
| Verifies that upgrades do not produce migration storms. |
| |
| For upgrades from 3.0 to 3.x we have encountered that the beloved 'cdc' column |
| caused rolling upgrades to produce migration storms, because the schema digest calculated |
| by 3.11 differs from that in 3.0, because 3.11 included the 'cdc' column in the digest |
| calculation. |
| |
| This test verifies that (after some grace time, i.e. nodes in status NORMAL) neither the |
| upgraded node nor the other nodes perform schema pulls (migration tasks) resulting in a |
| "migration storm". |
| |
| A few schema migrations however happen before all nodes are in status NORMAL. This is not ideal, |
| but legit. In an ideal world, those should not happen - but we would have to do the schema |
| changes atomically - and currently there is just nothing like a concurrent schema change. |
| |
| The tests run for a quite some time, about 5 minutes per test. |
| But there's some debug output for your entertainment - enjoy. |
| |
| The node.start/stop calls pass wait_other_notice=False intentionally, because there seem to |
| be issues in 3.0 to thoroughly detect and _log_ the status change - i.e. the wait_other_notice |
| check doesn't seem to be reliable - at least not in this test. |
| """ |
| |
| # The number of seconds we wait for schema migration log entries to verify |
| migration_check_time = 30 |
| |
| @pytest.fixture(autouse=True) |
| def fixture_add_additional_log_patterns(self, fixture_dtest_setup): |
| fixture_dtest_setup.ignore_log_patterns = ( |
| # This one occurs if we do a non-rolling upgrade, the node |
| # it's trying to send the migration to hasn't started yet, |
| # and when it does, it gets replayed and everything is fine. |
| r'Can\'t send migration request: node.*is down', |
| ) |
| |
| def _prepare(self, version, num_nodes=3): |
| cluster = self.cluster |
| |
| # Forcing cluster version on purpose |
| cluster.set_install_dir(version=version) |
| self.fixture_dtest_setup.reinitialize_cluster_for_different_version() |
| cluster.populate(num_nodes).start() |
| |
| return cluster |
| |
| def _set_verify_log_mark(self, nodes): |
| for node in nodes: |
| node.verify_log_mark = node.mark_log(filename='debug.log') |
| |
| def _expect_no_schema_migrations(self, nodes): |
| """ |
| Inspects the debug.log files from the given nodes whether any log about schema migration is present. |
| """ |
| |
| # Verify that there are _no_ log messages like: |
| expressions = [" - [pP]ulling schema from endpoint", |
| " - [Ss]ubmitting migration task", |
| " - [Pp]ulled schema from endpoint"] |
| logger.debug("Inspecting log files of {}...".format([n.name for n in nodes])) |
| all_matchings = "" |
| for node in nodes: |
| try: |
| matchings = node.watch_log_for(expressions, from_mark=node.verify_log_mark, timeout=0, filename='debug.log') |
| all_matchings = all_matchings + "\n{}: {}".format(node.name, matchings) |
| except TimeoutError: |
| # good |
| logger.debug(" {}: log files don't show schema migration messages (good)".format(node.name)) |
| if all_matchings != "": |
| msg = "Expected no schema migration log entries, but got:{}".format(all_matchings) |
| logger.debug(msg) # debug message for the validation test case (3.0 vs 3.11.1) |
| pytest.fail(msg) |
| |
| def _wait_for_status_normal(self, node, mark): |
| # Wait until the node is in state NORMAL (otherwise we can expect |
| # schema migrations for the first upgraded node). |
| node.watch_log_for("Node /{} state jump to NORMAL".format(node.address()), |
| from_mark=mark, timeout=300, filename='debug.log') |
| |
| def _bounce_node(self, node): |
| logger.debug("Bouncing {}...".format(node.name)) |
| logger.debug(" Stopping...") |
| node.stop(wait_other_notice=False) # intentionally set to wait_other_notice=False |
| mark = node.mark_log(filename='debug.log') |
| logger.debug(" Starting...") |
| node.start(wait_other_notice=False) # intentionally set to wait_other_notice=False |
| logger.debug(" Waiting for status NORMAL...") |
| self._wait_for_status_normal(node, mark) |
| |
| def _min_version(self, nodes): |
| """ |
| Retrieve the minimum major version (x.y) the given nodes run. |
| """ |
| |
| min_version = 99.9 |
| for node in nodes: |
| short_version = node.get_base_cassandra_version() |
| logger.debug("{} is on {} ({})".format(node.name, short_version, node.get_cassandra_version())) |
| if short_version < min_version: |
| min_version = short_version |
| return min_version |
| |
| def _upgrade_schema_agreement_test(self, upgrade_path): |
| """ |
| Test the upgrade though the specified versions and verify that there is schema agreement. |
| In theory, upgrading a 2-node cluster and verifying that might be sufficient, but the intention |
| of this test is also to catch potential edge cases. Therefore it uses a 3-node cluster and |
| bounces the upgraded nodes for an additional verification. |
| |
| The most effective way to check for schema migration is to inspect the debug log file for |
| related messages. However, since schema migrations are normal during upgrades and may happen |
| in a lazy fashion (e.g. potentially delayed by migration-interval), it does not work reliably - |
| i.e. the test would depend heavily on the test environment (hardware). |
| |
| Therefore, at most points, we can only efficiently check the schema versions as shown via |
| 'nodetool describecluster'. |
| |
| If these tests become flaky, tuning the (initial) migration-delay should help. |
| |
| :param upgrade_path: two-dimensional array containing the versions and whether each version |
| logs a '"Gossiping my * schema version' instead of the usual 'Gossiping my |
| schema version' message, which is only true for 3.11.2 (but not 4.0). |
| """ |
| |
| # prepare the cluster with initial version from the upgrade path |
| logger.debug('Starting upgrade test with {}'.format(upgrade_path[0][1])) |
| cluster = self._prepare(version=upgrade_path[0][1]) |
| |
| nodes = self.cluster.nodelist() |
| |
| # perform _rolling_ upgrades from one version to another |
| for (gossip_log_with_product_version, version) in upgrade_path[1:]: |
| logger.debug("") |
| logger.debug("Upgrading cluster to {}".format(version)) |
| cluster.set_install_dir(version=version) |
| self.fixture_dtest_setup.reinitialize_cluster_for_different_version() |
| |
| for node in nodes: |
| other_nodes = [n for n in nodes if n != node] |
| |
| logger.debug("") |
| logger.debug("Stopping {} for upgrade...".format(node.name)) |
| # needed to "patch" the config file (especially since 4.0) and get the correct version number |
| node.set_install_dir(version=version) |
| node.stop(wait_other_notice=False) # intentionally set to wait_other_notice=False |
| |
| # remember the logfile-mark when the node was upgraded |
| upgrade_log_mark = node.mark_log(filename='debug.log') |
| logger.debug("Starting upgraded {}...".format(node.name)) |
| node.start(wait_other_notice=False) # intentionally set to wait_other_notice=False |
| |
| # wait until the upgraded node is in status NORMAL |
| self._wait_for_status_normal(node, upgrade_log_mark) |
| |
| # If it's a 3.11.2 node, check that the correct schema version is announced |
| min_version = self._min_version(nodes) |
| logger.debug("Minimum version: {}".format(min_version)) |
| if gossip_log_with_product_version: |
| # 3.11.2 nodes (and only 3.11.2) indicate whether they announce |
| # a "3.0 compatible" or "real" "3.11" schema version. |
| watch_part = "Gossiping my {} schema version".format("3.0 compatible" if min_version == 3.0 else "3.11") |
| logger.debug("Inspecting log for '{}'...".format(watch_part)) |
| matchings = node.watch_log_for(watch_part, from_mark=upgrade_log_mark, timeout=120, filename='debug.log') |
| logger.debug(" Found: {}".format(matchings)) |
| |
| # Only log the schema information for debug purposes here. Primarily want to catch the |
| # schema migration race. |
| for n in nodes: |
| out, _, _ = n.nodetool("describecluster") |
| logger.debug("nodetool describecluster of {}:".format(n.name)) |
| logger.debug(out) |
| |
| # We expect no schema migrations at this point. |
| self._set_verify_log_mark(other_nodes) |
| logger.debug(" Sleep for {} seconds...".format(self.migration_check_time)) |
| time.sleep(self.migration_check_time) |
| self._expect_no_schema_migrations(other_nodes) |
| |
| # Try to trigger the schema migration race by bouncing the upgraded node. |
| # Bouncing a nodes causes a new gossip-digest-'generation', which in turn causes |
| # the whole endpoint state to propagate - including the schema version, which, in theory, |
| # should trigger the race. |
| # It is expected, that the _other_ nodes do not try to pull the schema. |
| logger.debug("") |
| logger.debug("Try to trigger schema migration race by bouncing the upgraded node") |
| self._bounce_node(node) |
| self._set_verify_log_mark(other_nodes) |
| logger.debug(" Sleep for {} seconds...".format(self.migration_check_time)) |
| time.sleep(self.migration_check_time) |
| self._expect_no_schema_migrations(other_nodes) |
| |
| # Even if it was impossible to trigger a schema version race, compare the schema versions. |
| # Kind of a last resort. Sometimes it's difficult to trigger the race. But we know that we |
| # only want to have one schema version. |
| for n in nodes: |
| out, _, _ = n.nodetool("describecluster") |
| logger.debug("nodetool describecluster of {}:".format(n.name)) |
| logger.debug(out) |
| versions = out.split('Schema versions:')[1].strip() |
| num_schemas = len(re.findall(r'\[.*?\]', versions)) |
| self.assertEqual(num_schemas, 1, "Multiple schema versions detected on {}: {}".format(n.name, out)) |
| |
| def upgrade_schema_agreement_30_3112_test(self): |
| """ |
| Test the upgrade from 3.0.latest to 3.11.2 and |
| verify schema agreement and no further migrations. |
| """ |
| self._upgrade_schema_agreement_test(upgrade_path=[[False, 'alias:apache/cassandra-3.0'], |
| [True, 'alias:apache/cassandra-3.11']]) |
| |
| def upgrade_schema_agreement_3111_3112_test(self): |
| """ |
| Test the upgrade from 3.11.1 to 3.11.2 and |
| verify schema agreement and no further migrations. |
| """ |
| self._upgrade_schema_agreement_test(upgrade_path=[[False, 'alias:apache/cassandra-3.11.1'], |
| [True, 'alias:apache/cassandra-3.11']]) |
| |
| def upgrade_schema_agreement_30latest_3111_test(self): |
| """ |
| Cross-check that the dtest still works (it expects the assertion on the log files). |
| |
| If this test becomes flaky, check whether the time to verify (self.migration_check_time) the log files is |
| still good enough. However, as long as the vast majority of runs of this tests succeeds, the functionality |
| itself works. |
| """ |
| with self.assertRaises(AssertionError, msg="Expected no schema migration log entries for the last {} seconds".format(self.migration_check_time)): |
| self._upgrade_schema_agreement_test(upgrade_path=[[False, 'alias:apache/cassandra-3.0'], |
| [False, 'alias:apache/cassandra-3.11.1']]) |
| |
| # REVIEWER NOTICE: |
| # The following tests should not make it into the dtest repo. The above tests using cassandra-3.11 cannot work |
| # until the corresponding is committed. |
| |
| def upgrade_schema_agreement_ok30_test(self): |
| """ |
| Test the upgrade from 3.0 to 3.11.2 and verify that there is no migration storm. |
| |
| TO BE CLEAR: THIS TEST IS INTENDED TO FAIL AND NOT INTENDED TO BE COMMITTED! |
| """ |
| self._upgrade_schema_agreement_test(upgrade_path=[[False, 'alias:apache/cassandra-3.0'], |
| [True, 'alias:snazy/schema-migration-upgrade-bug-trunk']]) |
| |
| def upgrade_schema_agreement_ok3112_test(self): |
| """ |
| Test the upgrade from 3.0 to 3.11.2 and verify that there is no migration storm. |
| |
| TO BE CLEAR: THIS TEST IS INTENDED TO FAIL AND NOT INTENDED TO BE COMMITTED! |
| """ |
| self._upgrade_schema_agreement_test(upgrade_path=[[False, 'alias:apache/cassandra-3.11'], |
| [True, 'alias:snazy/schema-migration-upgrade-bug-3.11']]) |
