| from __future__ import with_statement |
| |
| import ConfigParser |
| import copy |
| import errno |
| import glob |
| import logging |
| import os |
| import pprint |
| import re |
| import shutil |
| import signal |
| import subprocess |
| import sys |
| import tempfile |
| import thread |
| import threading |
| import time |
| import traceback |
| import types |
| import unittest.case |
| from collections import OrderedDict |
| from subprocess import CalledProcessError |
| from unittest import TestCase |
| |
| import cassandra |
| import ccmlib.repository |
| from cassandra import ConsistencyLevel |
| from cassandra.auth import PlainTextAuthProvider |
| from cassandra.cluster import Cluster as PyCluster |
| from cassandra.cluster import NoHostAvailable |
| from cassandra.cluster import ExecutionProfile, EXEC_PROFILE_DEFAULT |
| from cassandra.policies import RetryPolicy, WhiteListRoundRobinPolicy |
| from ccmlib.cluster import Cluster |
| from ccmlib.cluster_factory import ClusterFactory |
| from ccmlib.common import get_version_from_build, is_win |
| from nose.exc import SkipTest |
| from nose.tools import assert_greater_equal |
| from six import print_ |
| |
| from plugins.dtestconfig import _CONFIG as CONFIG |
| # We don't want test files to know about the plugins module, so we import |
| # constants here and re-export them. |
| from plugins.dtestconfig import GlobalConfigObject |
| from tools.context import log_filter |
| from tools.funcutils import merge_dicts |
| |
| LOG_SAVED_DIR = "logs" |
| try: |
| os.mkdir(LOG_SAVED_DIR) |
| except OSError: |
| pass |
| |
| LAST_LOG = os.path.join(LOG_SAVED_DIR, "last") |
| |
| LAST_TEST_DIR = 'last_test_dir' |
| |
| DEFAULT_DIR = './' |
| config = ConfigParser.RawConfigParser() |
| if len(config.read(os.path.expanduser('~/.cassandra-dtest'))) > 0: |
| if config.has_option('main', 'default_dir'): |
| DEFAULT_DIR = os.path.expanduser(config.get('main', 'default_dir')) |
| CASSANDRA_DIR = os.environ.get('CASSANDRA_DIR', DEFAULT_DIR) |
| |
| NO_SKIP = os.environ.get('SKIP', '').lower() in ('no', 'false') |
| DEBUG = os.environ.get('DEBUG', '').lower() in ('yes', 'true') |
| TRACE = os.environ.get('TRACE', '').lower() in ('yes', 'true') |
| KEEP_LOGS = os.environ.get('KEEP_LOGS', '').lower() in ('yes', 'true') |
| KEEP_TEST_DIR = os.environ.get('KEEP_TEST_DIR', '').lower() in ('yes', 'true') |
| PRINT_DEBUG = os.environ.get('PRINT_DEBUG', '').lower() in ('yes', 'true') |
| OFFHEAP_MEMTABLES = os.environ.get('OFFHEAP_MEMTABLES', '').lower() in ('yes', 'true') |
| NUM_TOKENS = os.environ.get('NUM_TOKENS', '256') |
| RECORD_COVERAGE = os.environ.get('RECORD_COVERAGE', '').lower() in ('yes', 'true') |
| IGNORE_REQUIRE = os.environ.get('IGNORE_REQUIRE', '').lower() in ('yes', 'true') |
| DATADIR_COUNT = os.environ.get('DATADIR_COUNT', '3') |
| ENABLE_ACTIVE_LOG_WATCHING = os.environ.get('ENABLE_ACTIVE_LOG_WATCHING', '').lower() in ('yes', 'true') |
| RUN_STATIC_UPGRADE_MATRIX = os.environ.get('RUN_STATIC_UPGRADE_MATRIX', '').lower() in ('yes', 'true') |
| |
| # devault values for configuration from configuration plugin |
| _default_config = GlobalConfigObject( |
| vnodes=True, |
| ) |
| |
| if CONFIG is None: |
| CONFIG = _default_config |
| |
| DISABLE_VNODES = not CONFIG.vnodes |
| |
| |
| if os.environ.get('DISABLE_VNODES', '').lower() in ('yes', 'true'): |
| print 'DISABLE_VNODES environment variable deprecated. Use `./run_dtests.py --vnodes false` instead.' |
| |
| |
| CURRENT_TEST = "" |
| |
| logging.basicConfig(filename=os.path.join(LOG_SAVED_DIR, "dtest.log"), |
| filemode='w', |
| format='%(asctime)s,%(msecs)d %(name)s %(current_test)s %(levelname)s %(message)s', |
| datefmt='%H:%M:%S', |
| level=logging.DEBUG) |
| |
| LOG = logging.getLogger('dtest') |
| # set python-driver log level to INFO by default for dtest |
| logging.getLogger('cassandra').setLevel(logging.INFO) |
| |
| |
| def get_sha(repo_dir): |
| try: |
| output = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=repo_dir).strip() |
| prefix = 'github:apache/' |
| local_repo_location = os.environ.get('LOCAL_GIT_REPO') |
| if local_repo_location is not None: |
| prefix = 'local:{}:'.format(local_repo_location) # local: slugs take the form 'local:/some/path/to/cassandra/:branch_name_or_sha' |
| return "{}{}".format(prefix, output) |
| except CalledProcessError as e: |
| if re.search('Not a git repository', e.message) is not None: |
| # we tried to get a sha, but repo_dir isn't a git repo. No big deal, must just be working from a non-git install. |
| return None |
| else: |
| # git call failed for some unknown reason |
| raise |
| |
| |
| # There are times when we want to know the C* version we're testing against |
| # before we call Tester.setUp. In the general case, we can't know that -- the |
| # test method could use any version it wants for self.cluster. However, we can |
| # get the version from build.xml in the C* repository specified by |
| # CASSANDRA_VERSION or CASSANDRA_DIR. This should use the same resolution |
| # strategy as the actual checkout code in Tester.setUp; if it does not, that is |
| # a bug. |
| _cassandra_version_slug = os.environ.get('CASSANDRA_VERSION') |
| # Prefer CASSANDRA_VERSION if it's set in the environment. If not, use CASSANDRA_DIR |
| if _cassandra_version_slug: |
| # fetch but don't build the specified C* version |
| ccm_repo_cache_dir, _ = ccmlib.repository.setup(_cassandra_version_slug) |
| CASSANDRA_VERSION_FROM_BUILD = get_version_from_build(ccm_repo_cache_dir) |
| CASSANDRA_GITREF = get_sha(ccm_repo_cache_dir) # will be set None when not a git repo |
| else: |
| CASSANDRA_VERSION_FROM_BUILD = get_version_from_build(CASSANDRA_DIR) |
| CASSANDRA_GITREF = get_sha(CASSANDRA_DIR) |
| |
| |
| # Determine the location of the libjemalloc jar so that we can specify it |
| # through environment variables when start Cassandra. This reduces startup |
| # time, making the dtests run faster. |
| def find_libjemalloc(): |
| if is_win(): |
| # let the normal bat script handle finding libjemalloc |
| return "" |
| |
| this_dir = os.path.dirname(os.path.realpath(__file__)) |
| script = os.path.join(this_dir, "findlibjemalloc.sh") |
| try: |
| p = subprocess.Popen([script], stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| if stderr or not stdout: |
| return "-" # tells C* not to look for libjemalloc |
| else: |
| return stdout |
| except Exception as exc: |
| print "Failed to run script to prelocate libjemalloc ({}): {}".format(script, exc) |
| return "" |
| |
| |
| CASSANDRA_LIBJEMALLOC = find_libjemalloc() |
| # copy the initial environment variables so we can reset them later: |
| initial_environment = copy.deepcopy(os.environ) |
| |
| |
| class DtestTimeoutError(Exception): |
| pass |
| |
| |
| def reset_environment_vars(): |
| os.environ.clear() |
| os.environ.update(initial_environment) |
| |
| |
| def warning(msg): |
| LOG.warning(msg, extra={"current_test": CURRENT_TEST}) |
| if PRINT_DEBUG: |
| print "WARN: " + msg |
| |
| |
| def debug(msg): |
| LOG.debug(msg, extra={"current_test": CURRENT_TEST}) |
| if PRINT_DEBUG: |
| print msg |
| |
| |
| debug("Python driver version in use: {}".format(cassandra.__version__)) |
| |
| |
| def retry_till_success(fun, *args, **kwargs): |
| timeout = kwargs.pop('timeout', 60) |
| bypassed_exception = kwargs.pop('bypassed_exception', Exception) |
| |
| deadline = time.time() + timeout |
| while True: |
| try: |
| return fun(*args, **kwargs) |
| except bypassed_exception: |
| if time.time() > deadline: |
| raise |
| else: |
| # brief pause before next attempt |
| time.sleep(0.25) |
| |
| |
| class FlakyRetryPolicy(RetryPolicy): |
| """ |
| A retry policy that retries 5 times by default, but can be configured to |
| retry more times. |
| """ |
| |
| def __init__(self, max_retries=5): |
| self.max_retries = max_retries |
| |
| def on_read_timeout(self, *args, **kwargs): |
| if kwargs['retry_num'] < self.max_retries: |
| debug("Retrying read after timeout. Attempt #" + str(kwargs['retry_num'])) |
| return (self.RETRY, None) |
| else: |
| return (self.RETHROW, None) |
| |
| def on_write_timeout(self, *args, **kwargs): |
| if kwargs['retry_num'] < self.max_retries: |
| debug("Retrying write after timeout. Attempt #" + str(kwargs['retry_num'])) |
| return (self.RETRY, None) |
| else: |
| return (self.RETHROW, None) |
| |
| def on_unavailable(self, *args, **kwargs): |
| if kwargs['retry_num'] < self.max_retries: |
| debug("Retrying request after UE. Attempt #" + str(kwargs['retry_num'])) |
| return (self.RETRY, None) |
| else: |
| return (self.RETHROW, None) |
| |
| |
| class Runner(threading.Thread): |
| |
| def __init__(self, func): |
| threading.Thread.__init__(self) |
| self.__func = func |
| self.__error = None |
| self.__stopped = False |
| self.daemon = True |
| |
| def run(self): |
| i = 0 |
| while True: |
| if self.__stopped: |
| return |
| try: |
| self.__func(i) |
| except Exception as e: |
| self.__error = e |
| return |
| i = i + 1 |
| |
| def stop(self): |
| self.__stopped = True |
| self.join() |
| if self.__error is not None: |
| raise self.__error |
| |
| def check(self): |
| if self.__error is not None: |
| raise self.__error |
| |
| |
| def make_execution_profile(retry_policy=FlakyRetryPolicy(), consistency_level=ConsistencyLevel.ONE, **kwargs): |
| return ExecutionProfile(retry_policy=retry_policy, |
| consistency_level=consistency_level, |
| **kwargs) |
| |
| |
| class Tester(TestCase): |
| |
| maxDiff = None |
| allow_log_errors = False # scan the log of each node for errors after every test. |
| cluster_options = None |
| |
| def set_node_to_current_version(self, node): |
| version = os.environ.get('CASSANDRA_VERSION') |
| cdir = CASSANDRA_DIR |
| |
| if version: |
| node.set_install_dir(version=version) |
| else: |
| node.set_install_dir(install_dir=cdir) |
| |
| def init_config(self): |
| init_default_config(self.cluster, self.cluster_options) |
| |
| def setUp(self): |
| self.set_current_tst_name() |
| kill_windows_cassandra_procs() |
| maybe_cleanup_cluster_from_last_test_file() |
| |
| self.test_path = get_test_path() |
| self.cluster = create_ccm_cluster(self.test_path, name='test') |
| |
| self.maybe_begin_active_log_watch() |
| maybe_setup_jacoco(self.test_path) |
| |
| self.init_config() |
| write_last_test_file(self.test_path, self.cluster) |
| |
| set_log_levels(self.cluster) |
| self.connections = [] |
| self.runners = [] |
| |
| # this is intentionally spelled 'tst' instead of 'test' to avoid |
| # making unittest think it's a test method |
| def set_current_tst_name(self): |
| global CURRENT_TEST |
| CURRENT_TEST = self.id() + self._testMethodName |
| |
| def maybe_begin_active_log_watch(self): |
| if ENABLE_ACTIVE_LOG_WATCHING: |
| if not self.allow_log_errors: |
| self.begin_active_log_watch() |
| |
| def begin_active_log_watch(self): |
| """ |
| Calls into ccm to start actively watching logs. |
| |
| In the event that errors are seen in logs, ccm will call back to _log_error_handler. |
| |
| When the cluster is no longer in use, stop_active_log_watch should be called to end log watching. |
| (otherwise a 'daemon' thread will (needlessly) run until the process exits). |
| """ |
| # log watching happens in another thread, but we want it to halt the main |
| # thread's execution, which we have to do by registering a signal handler |
| signal.signal(signal.SIGINT, self._catch_interrupt) |
| self._log_watch_thread = self.cluster.actively_watch_logs_for_error(self._log_error_handler, interval=0.25) |
| |
| def _log_error_handler(self, errordata): |
| """ |
| Callback handler used in conjunction with begin_active_log_watch. |
| When called, prepares exception instance, then will indirectly |
| cause _catch_interrupt to be called, which can raise the exception in the main |
| program thread. |
| |
| @param errordata is a dictonary mapping node name to failure list. |
| """ |
| # in some cases self.allow_log_errors may get set after proactive log checking has been enabled |
| # so we need to double-check first thing before proceeding |
| if self.allow_log_errors: |
| return |
| |
| reportable_errordata = OrderedDict() |
| |
| for nodename, errors in errordata.items(): |
| filtered_errors = list(self.__filter_errors(['\n'.join(msg) for msg in errors])) |
| if len(filtered_errors) is not 0: |
| reportable_errordata[nodename] = filtered_errors |
| |
| # no errors worthy of halting the test |
| if not reportable_errordata: |
| return |
| |
| message = "Errors seen in logs for: {nodes}".format(nodes=", ".join(reportable_errordata.keys())) |
| for nodename, errors in reportable_errordata.items(): |
| for error in errors: |
| message += "\n{nodename}: {error}".format(nodename=nodename, error=error) |
| |
| try: |
| debug('Errors were just seen in logs, ending test (if not ending already)!') |
| print_("Error details: \n{message}".format(message=message)) |
| self.test_is_ending # will raise AttributeError if not present |
| except AttributeError: |
| self.test_is_ending = True |
| self.exit_with_exception = AssertionError("Log error encountered during active log scanning, see stdout") |
| # thread.interrupt_main will SIGINT in the main thread, which we can |
| # catch to raise an exception with useful information |
| thread.interrupt_main() |
| |
| """ |
| Finds files matching the glob pattern specified as argument on |
| the given keyspace in all nodes |
| """ |
| |
| def glob_data_dirs(self, path, ks="ks"): |
| result = [] |
| for node in self.cluster.nodelist(): |
| for data_dir in node.data_directories(): |
| ks_dir = os.path.join(data_dir, ks, path) |
| result.extend(glob.glob(ks_dir)) |
| return result |
| |
| def _catch_interrupt(self, signal, frame): |
| """ |
| Signal handler for registering on SIGINT. |
| |
| If called will look for a stored exception and raise it to abort test. |
| If a stored exception is not present, this handler has likely caught a |
| user interrupt via CTRL-C, and will raise a KeyboardInterrupt. |
| """ |
| try: |
| # check if we have a persisted exception to fail with |
| raise self.exit_with_exception |
| except AttributeError: |
| # looks like this was just a plain CTRL-C event |
| raise KeyboardInterrupt() |
| |
| def copy_logs(self, cluster, directory=None, name=None): |
| """Copy the current cluster's log files somewhere, by default to LOG_SAVED_DIR with a name of 'last'""" |
| if directory is None: |
| directory = LOG_SAVED_DIR |
| if name is None: |
| name = LAST_LOG |
| else: |
| name = os.path.join(directory, name) |
| if not os.path.exists(directory): |
| os.mkdir(directory) |
| logs = [(node.name, node.logfilename(), node.debuglogfilename(), node.gclogfilename(), node.compactionlogfilename()) |
| for node in self.cluster.nodes.values()] |
| if len(logs) is not 0: |
| basedir = str(int(time.time() * 1000)) + '_' + self.id() |
| logdir = os.path.join(directory, basedir) |
| os.mkdir(logdir) |
| for n, log, debuglog, gclog, compactionlog in logs: |
| if os.path.exists(log): |
| self.assertGreaterEqual(os.path.getsize(log), 0) |
| shutil.copyfile(log, os.path.join(logdir, n + ".log")) |
| if os.path.exists(debuglog): |
| self.assertGreaterEqual(os.path.getsize(debuglog), 0) |
| shutil.copyfile(debuglog, os.path.join(logdir, n + "_debug.log")) |
| if os.path.exists(gclog): |
| self.assertGreaterEqual(os.path.getsize(gclog), 0) |
| shutil.copyfile(gclog, os.path.join(logdir, n + "_gc.log")) |
| if os.path.exists(compactionlog): |
| self.assertGreaterEqual(os.path.getsize(compactionlog), 0) |
| shutil.copyfile(compactionlog, os.path.join(logdir, n + "_compaction.log")) |
| if os.path.exists(name): |
| os.unlink(name) |
| if not is_win(): |
| os.symlink(basedir, name) |
| |
| def cql_connection(self, node, keyspace=None, user=None, |
| password=None, compression=True, protocol_version=None, port=None, ssl_opts=None, **kwargs): |
| |
| return self._create_session(node, keyspace, user, password, compression, |
| protocol_version, port=port, ssl_opts=ssl_opts, **kwargs) |
| |
| def exclusive_cql_connection(self, node, keyspace=None, user=None, |
| password=None, compression=True, protocol_version=None, port=None, ssl_opts=None, **kwargs): |
| |
| node_ip = get_ip_from_node(node) |
| wlrr = WhiteListRoundRobinPolicy([node_ip]) |
| |
| return self._create_session(node, keyspace, user, password, compression, |
| protocol_version, port=port, ssl_opts=ssl_opts, load_balancing_policy=wlrr, **kwargs) |
| |
| def _create_session(self, node, keyspace, user, password, compression, protocol_version, |
| port=None, ssl_opts=None, execution_profiles=None, **kwargs): |
| node_ip = get_ip_from_node(node) |
| if not port: |
| port = get_port_from_node(node) |
| |
| if protocol_version is None: |
| protocol_version = get_eager_protocol_version(node.cluster.version()) |
| |
| if user is not None: |
| auth_provider = get_auth_provider(user=user, password=password) |
| else: |
| auth_provider = None |
| |
| profiles = {EXEC_PROFILE_DEFAULT: make_execution_profile(**kwargs) |
| } if not execution_profiles else execution_profiles |
| |
| cluster = PyCluster([node_ip], |
| auth_provider=auth_provider, |
| compression=compression, |
| protocol_version=protocol_version, |
| port=port, |
| ssl_options=ssl_opts, |
| connect_timeout=10, |
| allow_beta_protocol_version=True, |
| execution_profiles=profiles) |
| session = cluster.connect(wait_for_all_pools=True) |
| |
| if keyspace is not None: |
| session.set_keyspace(keyspace) |
| |
| self.connections.append(session) |
| return session |
| |
| def patient_cql_connection(self, node, keyspace=None, |
| user=None, password=None, timeout=30, compression=True, |
| protocol_version=None, port=None, ssl_opts=None, **kwargs): |
| """ |
| Returns a connection after it stops throwing NoHostAvailables due to not being ready. |
| |
| If the timeout is exceeded, the exception is raised. |
| """ |
| if is_win(): |
| timeout *= 2 |
| |
| expected_log_lines = ('Control connection failed to connect, shutting down Cluster:', '[control connection] Error connecting to ') |
| with log_filter('cassandra.cluster', expected_log_lines): |
| session = retry_till_success( |
| self.cql_connection, |
| node, |
| keyspace=keyspace, |
| user=user, |
| password=password, |
| timeout=timeout, |
| compression=compression, |
| protocol_version=protocol_version, |
| port=port, |
| ssl_opts=ssl_opts, |
| bypassed_exception=NoHostAvailable, |
| **kwargs |
| ) |
| |
| return session |
| |
| def patient_exclusive_cql_connection(self, node, keyspace=None, |
| user=None, password=None, timeout=30, compression=True, |
| protocol_version=None, port=None, ssl_opts=None, **kwargs): |
| """ |
| Returns a connection after it stops throwing NoHostAvailables due to not being ready. |
| |
| If the timeout is exceeded, the exception is raised. |
| """ |
| if is_win(): |
| timeout *= 2 |
| |
| return retry_till_success( |
| self.exclusive_cql_connection, |
| node, |
| keyspace=keyspace, |
| user=user, |
| password=password, |
| timeout=timeout, |
| compression=compression, |
| protocol_version=protocol_version, |
| port=port, |
| ssl_opts=ssl_opts, |
| bypassed_exception=NoHostAvailable, |
| **kwargs |
| ) |
| |
| @classmethod |
| def tearDownClass(cls): |
| reset_environment_vars() |
| if os.path.exists(LAST_TEST_DIR): |
| with open(LAST_TEST_DIR) as f: |
| test_path = f.readline().strip('\n') |
| name = f.readline() |
| try: |
| cluster = ClusterFactory.load(test_path, name) |
| # Avoid waiting too long for node to be marked down |
| if KEEP_TEST_DIR: |
| cluster.stop(gently=RECORD_COVERAGE) |
| else: |
| cluster.remove() |
| os.rmdir(test_path) |
| except IOError: |
| # after a restart, /tmp will be emptied so we'll get an IOError when loading the old cluster here |
| pass |
| try: |
| os.remove(LAST_TEST_DIR) |
| except IOError: |
| # Ignore - see comment above |
| pass |
| |
| def tearDown(self): |
| # test_is_ending prevents active log watching from being able to interrupt the test |
| # which we don't want to happen once tearDown begins |
| self.test_is_ending = True |
| |
| reset_environment_vars() |
| |
| for con in self.connections: |
| con.cluster.shutdown() |
| |
| for runner in self.runners: |
| try: |
| runner.stop() |
| except: |
| pass |
| |
| failed = did_fail() |
| try: |
| if not self.allow_log_errors and self.check_logs_for_errors(): |
| failed = True |
| raise AssertionError('Unexpected error in log, see stdout') |
| finally: |
| try: |
| # save the logs for inspection |
| if failed or KEEP_LOGS: |
| self.copy_logs(self.cluster) |
| except Exception as e: |
| print "Error saving log:", str(e) |
| finally: |
| log_watch_thread = getattr(self, '_log_watch_thread', None) |
| cleanup_cluster(self.cluster, self.test_path, log_watch_thread) |
| |
| def check_logs_for_errors(self): |
| for node in self.cluster.nodelist(): |
| errors = list(self.__filter_errors( |
| ['\n'.join(msg) for msg in node.grep_log_for_errors()])) |
| if len(errors) is not 0: |
| for error in errors: |
| print_("Unexpected error in {node_name} log, error: \n{error}".format(node_name=node.name, error=error)) |
| return True |
| |
| def go(self, func): |
| runner = Runner(func) |
| self.runners.append(runner) |
| runner.start() |
| return runner |
| |
| def skip(self, msg): |
| if not NO_SKIP: |
| raise SkipTest(msg) |
| |
| def __filter_errors(self, errors): |
| """Filter errors, removing those that match self.ignore_log_patterns""" |
| if not hasattr(self, 'ignore_log_patterns'): |
| self.ignore_log_patterns = [] |
| for e in errors: |
| for pattern in self.ignore_log_patterns: |
| if re.search(pattern, e): |
| break |
| else: |
| yield e |
| |
| # Disable docstrings printing in nosetest output |
| def shortDescription(self): |
| return None |
| |
| def get_jfr_jvm_args(self): |
| """ |
| @return The JVM arguments required for attaching flight recorder to a Java process. |
| """ |
| return ["-XX:+UnlockCommercialFeatures", "-XX:+FlightRecorder"] |
| |
| def start_jfr_recording(self, nodes): |
| """ |
| Start Java flight recorder provided the cluster was started with the correct jvm arguments. |
| """ |
| for node in nodes: |
| p = subprocess.Popen(['jcmd', str(node.pid), 'JFR.start'], |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| debug(stdout) |
| debug(stderr) |
| |
| def dump_jfr_recording(self, nodes): |
| """ |
| Save Java flight recorder results to file for analyzing with mission control. |
| """ |
| for node in nodes: |
| p = subprocess.Popen(['jcmd', str(node.pid), 'JFR.dump', |
| 'recording=1', 'filename=recording_{}.jfr'.format(node.address())], |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| debug(stdout) |
| debug(stderr) |
| |
| |
| def get_eager_protocol_version(cassandra_version): |
| """ |
| Returns the highest protocol version accepted |
| by the given C* version |
| """ |
| if cassandra_version >= '2.2': |
| protocol_version = 4 |
| elif cassandra_version >= '2.1': |
| protocol_version = 3 |
| elif cassandra_version >= '2.0': |
| protocol_version = 2 |
| else: |
| protocol_version = 1 |
| return protocol_version |
| |
| |
| # We default to UTF8Type because it's simpler to use in tests |
| def create_cf(session, name, key_type="varchar", speculative_retry=None, read_repair=None, compression=None, |
| gc_grace=None, columns=None, validation="UTF8Type", compact_storage=False): |
| |
| additional_columns = "" |
| if columns is not None: |
| for k, v in columns.items(): |
| additional_columns = "{}, {} {}".format(additional_columns, k, v) |
| |
| if additional_columns == "": |
| query = 'CREATE COLUMNFAMILY %s (key %s, c varchar, v varchar, PRIMARY KEY(key, c)) WITH comment=\'test cf\'' % (name, key_type) |
| else: |
| query = 'CREATE COLUMNFAMILY %s (key %s PRIMARY KEY%s) WITH comment=\'test cf\'' % (name, key_type, additional_columns) |
| |
| if compression is not None: |
| query = '%s AND compression = { \'sstable_compression\': \'%sCompressor\' }' % (query, compression) |
| else: |
| # if a compression option is omitted, C* will default to lz4 compression |
| query += ' AND compression = {}' |
| |
| if read_repair is not None: |
| query = '%s AND read_repair_chance=%f AND dclocal_read_repair_chance=%f' % (query, read_repair, read_repair) |
| if gc_grace is not None: |
| query = '%s AND gc_grace_seconds=%d' % (query, gc_grace) |
| if speculative_retry is not None: |
| query = '%s AND speculative_retry=\'%s\'' % (query, speculative_retry) |
| |
| if compact_storage: |
| query += ' AND COMPACT STORAGE' |
| |
| session.execute(query) |
| time.sleep(0.2) |
| |
| |
| def create_ks(session, name, rf): |
| query = 'CREATE KEYSPACE %s WITH replication={%s}' |
| if isinstance(rf, types.IntType): |
| # we assume simpleStrategy |
| session.execute(query % (name, "'class':'SimpleStrategy', 'replication_factor':%d" % rf)) |
| else: |
| assert_greater_equal(len(rf), 0, "At least one datacenter/rf pair is needed") |
| # we assume networkTopologyStrategy |
| options = (', ').join(['\'%s\':%d' % (d, r) for d, r in rf.iteritems()]) |
| session.execute(query % (name, "'class':'NetworkTopologyStrategy', %s" % options)) |
| session.execute('USE {}'.format(name)) |
| |
| |
| def get_auth_provider(user, password): |
| return PlainTextAuthProvider(username=user, password=password) |
| |
| |
| def make_auth(user, password): |
| def private_auth(node_ip): |
| return {'username': user, 'password': password} |
| return private_auth |
| |
| |
| def get_port_from_node(node): |
| """ |
| Return the port that this node is listening on. |
| We only use this to connect the native driver, |
| so we only care about the binary port. |
| """ |
| try: |
| return node.network_interfaces['binary'][1] |
| except Exception: |
| raise RuntimeError("No network interface defined on this node object. {}".format(node.network_interfaces)) |
| |
| |
| def get_ip_from_node(node): |
| if node.network_interfaces['binary']: |
| node_ip = node.network_interfaces['binary'][0] |
| else: |
| node_ip = node.network_interfaces['thrift'][0] |
| return node_ip |
| |
| |
| def kill_windows_cassandra_procs(): |
| # On Windows, forcefully terminate any leftover previously running cassandra processes. This is a temporary |
| # workaround until we can determine the cause of intermittent hung-open tests and file-handles. |
| if is_win(): |
| try: |
| import psutil |
| for proc in psutil.process_iter(): |
| try: |
| pinfo = proc.as_dict(attrs=['pid', 'name', 'cmdline']) |
| except psutil.NoSuchProcess: |
| pass |
| else: |
| if (pinfo['name'] == 'java.exe' and '-Dcassandra' in pinfo['cmdline']): |
| print 'Found running cassandra process with pid: ' + str(pinfo['pid']) + '. Killing.' |
| psutil.Process(pinfo['pid']).kill() |
| except ImportError: |
| debug("WARN: psutil not installed. Cannot detect and kill " |
| "running cassandra processes - you may see cascading dtest failures.") |
| |
| |
| def get_test_path(): |
| test_path = tempfile.mkdtemp(prefix='dtest-') |
| |
| # ccm on cygwin needs absolute path to directory - it crosses from cygwin space into |
| # regular Windows space on wmic calls which will otherwise break pathing |
| if sys.platform == "cygwin": |
| process = subprocess.Popen(["cygpath", "-m", test_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) |
| test_path = process.communicate()[0].rstrip() |
| |
| return test_path |
| |
| |
| # nose will discover this as a test, so we manually make it not a test |
| get_test_path.__test__ = False |
| |
| |
| def create_ccm_cluster(test_path, name): |
| debug("cluster ccm directory: " + test_path) |
| version = os.environ.get('CASSANDRA_VERSION') |
| cdir = CASSANDRA_DIR |
| |
| if version: |
| cluster = Cluster(test_path, name, cassandra_version=version) |
| else: |
| cluster = Cluster(test_path, name, cassandra_dir=cdir) |
| |
| if DISABLE_VNODES: |
| cluster.set_configuration_options(values={'num_tokens': None}) |
| else: |
| cluster.set_configuration_options(values={'initial_token': None, 'num_tokens': NUM_TOKENS}) |
| |
| if OFFHEAP_MEMTABLES: |
| cluster.set_configuration_options(values={'memtable_allocation_type': 'offheap_objects'}) |
| |
| cluster.set_datadir_count(DATADIR_COUNT) |
| cluster.set_environment_variable('CASSANDRA_LIBJEMALLOC', CASSANDRA_LIBJEMALLOC) |
| |
| return cluster |
| |
| |
| def cleanup_cluster(cluster, test_path, log_watch_thread=None): |
| with log_filter('cassandra'): # quiet noise from driver when nodes start going down |
| if KEEP_TEST_DIR: |
| cluster.stop(gently=RECORD_COVERAGE) |
| else: |
| # when recording coverage the jvm has to exit normally |
| # or the coverage information is not written by the jacoco agent |
| # otherwise we can just kill the process |
| if RECORD_COVERAGE: |
| cluster.stop(gently=True) |
| |
| # Cleanup everything: |
| try: |
| if log_watch_thread: |
| stop_active_log_watch(log_watch_thread) |
| finally: |
| debug("removing ccm cluster {name} at: {path}".format(name=cluster.name, path=test_path)) |
| cluster.remove() |
| |
| debug("clearing ssl stores from [{0}] directory".format(test_path)) |
| for filename in ('keystore.jks', 'truststore.jks', 'ccm_node.cer'): |
| try: |
| os.remove(os.path.join(test_path, filename)) |
| except OSError as e: |
| # once we port to py3, which has better reporting for exceptions raised while |
| # handling other excpetions, we should just assert e.errno == errno.ENOENT |
| if e.errno != errno.ENOENT: # ENOENT = no such file or directory |
| raise |
| |
| os.rmdir(test_path) |
| cleanup_last_test_dir() |
| |
| |
| def cleanup_last_test_dir(): |
| if os.path.exists(LAST_TEST_DIR): |
| os.remove(LAST_TEST_DIR) |
| |
| |
| def stop_active_log_watch(log_watch_thread): |
| """ |
| Joins the log watching thread, which will then exit. |
| Should be called after each test, ideally after nodes are stopped but before cluster files are removed. |
| |
| Can be called multiple times without error. |
| If not called, log watching thread will remain running until the parent process exits. |
| """ |
| log_watch_thread.join(timeout=60) |
| |
| |
| def maybe_cleanup_cluster_from_last_test_file(): |
| # cleaning up if a previous execution didn't trigger tearDown (which |
| # can happen if it is interrupted by KeyboardInterrupt) |
| if os.path.exists(LAST_TEST_DIR): |
| with open(LAST_TEST_DIR) as f: |
| test_path = f.readline().strip('\n') |
| name = f.readline() |
| try: |
| cluster = ClusterFactory.load(test_path, name) |
| # Avoid waiting too long for node to be marked down |
| cleanup_cluster(cluster, test_path) |
| except IOError: |
| # after a restart, /tmp will be emptied so we'll get an IOError when loading the old cluster here |
| pass |
| |
| |
| def init_default_config(cluster, cluster_options): |
| # the failure detector can be quite slow in such tests with quick start/stop |
| phi_values = {'phi_convict_threshold': 5} |
| |
| timeout = 10000 |
| if cluster_options is not None: |
| values = merge_dicts(cluster_options, phi_values) |
| else: |
| values = merge_dicts(phi_values, { |
| 'read_request_timeout_in_ms': timeout, |
| 'range_request_timeout_in_ms': timeout, |
| 'write_request_timeout_in_ms': timeout, |
| 'truncate_request_timeout_in_ms': timeout, |
| 'request_timeout_in_ms': timeout |
| }) |
| |
| # No more thrift in 4.0, and start_rpc doesn't exists anymore |
| if cluster.version() >= '4' and 'start_rpc' in values: |
| del values['start_rpc'] |
| |
| cluster.set_configuration_options(values) |
| debug("Done setting configuration options:\n" + pprint.pformat(cluster._config_options, indent=4)) |
| |
| |
| def write_last_test_file(test_path, cluster): |
| with open(LAST_TEST_DIR, 'w') as f: |
| f.write(test_path + '\n') |
| f.write(cluster.name) |
| |
| |
| def set_log_levels(cluster): |
| if DEBUG: |
| cluster.set_log_level("DEBUG") |
| if TRACE: |
| cluster.set_log_level("TRACE") |
| |
| if os.environ.get('DEBUG', 'no').lower() not in ('no', 'false', 'yes', 'true'): |
| classes_to_debug = os.environ.get('DEBUG').split(":") |
| cluster.set_log_level('DEBUG', None if len(classes_to_debug) == 0 else classes_to_debug) |
| |
| if os.environ.get('TRACE', 'no').lower() not in ('no', 'false', 'yes', 'true'): |
| classes_to_trace = os.environ.get('TRACE').split(":") |
| cluster.set_log_level('TRACE', None if len(classes_to_trace) == 0 else classes_to_trace) |
| |
| |
| def maybe_setup_jacoco(test_path, cluster_name='test'): |
| """Setup JaCoCo code coverage support""" |
| |
| if not RECORD_COVERAGE: |
| return |
| |
| # use explicit agent and execfile locations |
| # or look for a cassandra build if they are not specified |
| cdir = CASSANDRA_DIR |
| |
| agent_location = os.environ.get('JACOCO_AGENT_JAR', os.path.join(cdir, 'build/lib/jars/jacocoagent.jar')) |
| jacoco_execfile = os.environ.get('JACOCO_EXECFILE', os.path.join(cdir, 'build/jacoco/jacoco.exec')) |
| |
| if os.path.isfile(agent_location): |
| debug("Jacoco agent found at {}".format(agent_location)) |
| with open(os.path.join( |
| test_path, cluster_name, 'cassandra.in.sh'), 'w') as f: |
| |
| f.write('JVM_OPTS="$JVM_OPTS -javaagent:{jar_path}=destfile={exec_file}"' |
| .format(jar_path=agent_location, exec_file=jacoco_execfile)) |
| |
| if os.path.isfile(jacoco_execfile): |
| debug("Jacoco execfile found at {}, execution data will be appended".format(jacoco_execfile)) |
| else: |
| debug("Jacoco execfile will be created at {}".format(jacoco_execfile)) |
| else: |
| debug("Jacoco agent not found or is not file. Execution will not be recorded.") |
| |
| |
| def did_fail(): |
| if sys.exc_info() == (None, None, None): |
| return False |
| |
| exc_class, _, _ = sys.exc_info() |
| return not issubclass(exc_class, unittest.case.SkipTest) |
| |
| |
| class ReusableClusterTester(Tester): |
| """ |
| A Tester designed for reusing the same cluster across multiple |
| test methods. This makes test suites with many small tests run |
| much, much faster. However, there are a couple of downsides: |
| |
| First, test setup and teardown must be diligent about cleaning |
| up any data or schema elements that may interfere with other |
| tests. |
| |
| Second, errors triggered by one test method may cascade |
| into other test failures. In an attempt to limit this, the |
| cluster will be restarted if a test fails or an exception is |
| caught. However, there may still be undetected problems in |
| Cassandra that cause cascading failures. |
| """ |
| |
| test_path = None |
| cluster = None |
| cluster_options = None |
| |
| @classmethod |
| def setUpClass(cls): |
| kill_windows_cassandra_procs() |
| maybe_cleanup_cluster_from_last_test_file() |
| cls.initialize_cluster() |
| |
| def setUp(self): |
| self.set_current_tst_name() |
| self.connections = [] |
| |
| # TODO enable active log watching |
| # This needs to happen in setUp() and not setUpClass() so that individual |
| # test methods can set allow_log_errors and so that error handling |
| # only fails a single test method instead of the entire class. |
| # The problem with this is that ccm doesn't yet support stopping the |
| # active log watcher -- it runs until the cluster is destroyed. Since |
| # we reuse the same cluster, this doesn't work for us. |
| |
| def tearDown(self): |
| # test_is_ending prevents active log watching from being able to interrupt the test |
| self.test_is_ending = True |
| |
| failed = did_fail() |
| try: |
| if not self.allow_log_errors and self.check_logs_for_errors(): |
| failed = True |
| raise AssertionError('Unexpected error in log, see stdout') |
| finally: |
| try: |
| # save the logs for inspection |
| if failed or KEEP_LOGS: |
| self.copy_logs(self.cluster) |
| except Exception as e: |
| print "Error saving log:", str(e) |
| finally: |
| reset_environment_vars() |
| if failed: |
| cleanup_cluster(self.cluster, self.test_path) |
| kill_windows_cassandra_procs() |
| self.initialize_cluster() |
| |
| @classmethod |
| def initialize_cluster(cls): |
| """ |
| This method is responsible for initializing and configuring a ccm |
| cluster for the next set of tests. This can be called for two |
| different reasons: |
| * A class of tests is starting |
| * A test method failed/errored, so the cluster has been wiped |
| |
| Subclasses that require custom initialization should generally |
| do so by overriding post_initialize_cluster(). |
| """ |
| cls.test_path = get_test_path() |
| cls.cluster = create_ccm_cluster(cls.test_path, name='test') |
| cls.init_config() |
| |
| maybe_setup_jacoco(cls.test_path) |
| cls.init_config() |
| write_last_test_file(cls.test_path, cls.cluster) |
| set_log_levels(cls.cluster) |
| |
| cls.post_initialize_cluster() |
| |
| @classmethod |
| def post_initialize_cluster(cls): |
| """ |
| This method is called after the ccm cluster has been created |
| and default config options have been applied. Any custom |
| initialization for a test class should generally be done |
| here in order to correctly handle cluster restarts after |
| test method failures. |
| """ |
| pass |
| |
| @classmethod |
| def init_config(cls): |
| init_default_config(cls.cluster, cls.cluster_options) |
| |
| |
| class MultiError(Exception): |
| """ |
| Extends Exception to provide reporting multiple exceptions at once. |
| """ |
| |
| def __init__(self, exceptions, tracebacks): |
| # an exception and the corresponding traceback should be found at the same |
| # position in their respective lists, otherwise __str__ will be incorrect |
| self.exceptions = exceptions |
| self.tracebacks = tracebacks |
| |
| def __str__(self): |
| output = "\n****************************** BEGIN MultiError ******************************\n" |
| |
| for (exc, tb) in zip(self.exceptions, self.tracebacks): |
| output += str(exc) |
| output += str(tb) + "\n" |
| |
| output += "****************************** END MultiError ******************************" |
| |
| return output |
| |
| |
| def run_scenarios(scenarios, handler, deferred_exceptions=tuple()): |
| """ |
| Runs multiple scenarios from within a single test method. |
| |
| "Scenarios" are mini-tests where a common procedure can be reused with several different configurations. |
| They are intended for situations where complex/expensive setup isn't required and some shared state is acceptable (or trivial to reset). |
| |
| Arguments: scenarios should be an iterable, handler should be a callable, and deferred_exceptions should be a tuple of exceptions which |
| are safe to delay until the scenarios are all run. For each item in scenarios, handler(item) will be called in turn. |
| |
| Exceptions which occur will be bundled up and raised as a single MultiError exception, either when: a) all scenarios have run, |
| or b) on the first exception encountered which is not whitelisted in deferred_exceptions. |
| """ |
| errors = [] |
| tracebacks = [] |
| |
| for i, scenario in enumerate(scenarios, 1): |
| debug("running scenario {}/{}: {}".format(i, len(scenarios), scenario)) |
| |
| try: |
| handler(scenario) |
| except deferred_exceptions as e: |
| tracebacks.append(traceback.format_exc(sys.exc_info())) |
| errors.append(type(e)('encountered {} {} running scenario:\n {}\n'.format(e.__class__.__name__, e.message, scenario))) |
| debug("scenario {}/{} encountered a deferrable exception, continuing".format(i, len(scenarios))) |
| except Exception as e: |
| # catch-all for any exceptions not intended to be deferred |
| tracebacks.append(traceback.format_exc(sys.exc_info())) |
| errors.append(type(e)('encountered {} {} running scenario:\n {}\n'.format(e.__class__.__name__, e.message, scenario))) |
| debug("scenario {}/{} encountered a non-deferrable exception, aborting".format(i, len(scenarios))) |
| raise MultiError(errors, tracebacks) |
| |
| if errors: |
| raise MultiError(errors, tracebacks) |
