| from distutils.version import LooseVersion |
| |
| from cassandra import ConsistencyLevel |
| from ccmlib.common import is_win |
| |
| from dtest import Tester, debug, create_ks, create_cf |
| from tools.assertions import assert_length_equal |
| from tools.data import insert_c1c2 |
| from tools.decorators import since |
| from tools.jmxutils import (JolokiaAgent, make_mbean, |
| remove_perf_disable_shared_mem) |
| |
| |
| @since("2.2", max_version="4") |
| class TestDeprecatedRepairAPI(Tester): |
| """ |
| @jira_ticket CASSANDRA-9570 |
| |
| Test if deprecated repair JMX API runs with expected parameters |
| """ |
| |
| def force_repair_async_1_test(self): |
| """ |
| test forceRepairAsync(String keyspace, boolean isSequential, |
| Collection<String> dataCenters, |
| Collection<String> hosts, |
| boolean primaryRange, boolean fullRepair, String... columnFamilies) |
| """ |
| opt = self._deprecated_repair_jmx("forceRepairAsync(java.lang.String,boolean,java.util.Collection,java.util.Collection,boolean,boolean,[Ljava.lang.String;)", |
| ['ks', True, [], [], False, False, ["cf"]]) |
| self.assertEqual(opt["parallelism"], "parallel" if is_win() else "sequential", opt) |
| self.assertEqual(opt["primary_range"], "false", opt) |
| self.assertEqual(opt["incremental"], "true", opt) |
| self.assertEqual(opt["job_threads"], "1", opt) |
| self.assertEqual(opt["data_centers"], "[]", opt) |
| self.assertEqual(opt["hosts"], "[]", opt) |
| self.assertEqual(opt["column_families"], "[cf]", opt) |
| |
| def force_repair_async_2_test(self): |
| """ |
| test forceRepairAsync(String keyspace, int parallelismDegree, |
| Collection<String> dataCenters, |
| Collection<String> hosts, |
| boolean primaryRange, boolean fullRepair, String... columnFamilies) |
| """ |
| opt = self._deprecated_repair_jmx("forceRepairAsync(java.lang.String,int,java.util.Collection,java.util.Collection,boolean,boolean,[Ljava.lang.String;)", |
| ['ks', 1, [], [], True, True, []]) |
| self.assertEqual(opt["parallelism"], "parallel", opt) |
| self.assertEqual(opt["primary_range"], "true", opt) |
| self.assertEqual(opt["incremental"], "false", opt) |
| self.assertEqual(opt["job_threads"], "1", opt) |
| self.assertEqual(opt["data_centers"], "[]", opt) |
| self.assertEqual(opt["hosts"], "[]", opt) |
| self.assertEqual(opt["column_families"], "[]", opt) |
| |
| def force_repair_async_3_test(self): |
| """ |
| test forceRepairAsync(String keyspace, boolean isSequential, |
| boolean isLocal, boolean primaryRange, |
| boolean fullRepair, String... columnFamilies) |
| """ |
| opt = self._deprecated_repair_jmx("forceRepairAsync(java.lang.String,boolean,boolean,boolean,boolean,[Ljava.lang.String;)", |
| ['ks', False, False, False, False, ["cf"]]) |
| self.assertEqual(opt["parallelism"], "parallel", opt) |
| self.assertEqual(opt["primary_range"], "false", opt) |
| self.assertEqual(opt["incremental"], "true", opt) |
| self.assertEqual(opt["job_threads"], "1", opt) |
| self.assertEqual(opt["data_centers"], "[]", opt) |
| self.assertEqual(opt["hosts"], "[]", opt) |
| self.assertEqual(opt["column_families"], "[cf]", opt) |
| |
| def force_repair_range_async_1_test(self): |
| """ |
| test forceRepairRangeAsync(String beginToken, String endToken, |
| String keyspaceName, boolean isSequential, |
| Collection<String> dataCenters, |
| Collection<String> hosts, boolean fullRepair, |
| String... columnFamilies) |
| """ |
| opt = self._deprecated_repair_jmx("forceRepairRangeAsync(java.lang.String,java.lang.String,java.lang.String,boolean,java.util.Collection,java.util.Collection,boolean,[Ljava.lang.String;)", |
| ["0", "1000", "ks", True, ["dc1"], [], False, ["cf"]]) |
| self.assertEqual(opt["parallelism"], "parallel" if is_win() else "sequential", opt) |
| self.assertEqual(opt["primary_range"], "false", opt) |
| self.assertEqual(opt["incremental"], "true", opt) |
| self.assertEqual(opt["job_threads"], "1", opt) |
| self.assertEqual(opt["data_centers"], "[dc1]", opt) |
| self.assertEqual(opt["hosts"], "[]", opt) |
| self.assertEqual(opt["ranges"], "1", opt) |
| self.assertEqual(opt["column_families"], "[cf]", opt) |
| |
| def force_repair_range_async_2_test(self): |
| """ |
| test forceRepairRangeAsync(String beginToken, String endToken, |
| String keyspaceName, int parallelismDegree, |
| Collection<String> dataCenters, |
| Collection<String> hosts, |
| boolean fullRepair, String... columnFamilies) |
| """ |
| opt = self._deprecated_repair_jmx("forceRepairRangeAsync(java.lang.String,java.lang.String,java.lang.String,int,java.util.Collection,java.util.Collection,boolean,[Ljava.lang.String;)", |
| ["0", "1000", "ks", 2, [], [], True, ["cf"]]) |
| self.assertEqual(opt["parallelism"], "parallel" if is_win() else "dc_parallel", opt) |
| self.assertEqual(opt["primary_range"], "false", opt) |
| self.assertEqual(opt["incremental"], "false", opt) |
| self.assertEqual(opt["job_threads"], "1", opt) |
| self.assertEqual(opt["data_centers"], "[]", opt) |
| self.assertEqual(opt["hosts"], "[]", opt) |
| self.assertEqual(opt["ranges"], "1", opt) |
| self.assertEqual(opt["column_families"], "[cf]", opt) |
| |
| def force_repair_range_async_3_test(self): |
| """ |
| test forceRepairRangeAsync(String beginToken, String endToken, |
| String keyspaceName, boolean isSequential, |
| boolean isLocal, boolean fullRepair, |
| String... columnFamilies) |
| """ |
| opt = self._deprecated_repair_jmx("forceRepairRangeAsync(java.lang.String,java.lang.String,java.lang.String,boolean,boolean,boolean,[Ljava.lang.String;)", |
| ["0", "1000", "ks", True, True, True, ["cf"]]) |
| self.assertEqual(opt["parallelism"], "parallel" if is_win() else "sequential", opt) |
| self.assertEqual(opt["primary_range"], "false", opt) |
| self.assertEqual(opt["incremental"], "false", opt) |
| self.assertEqual(opt["job_threads"], "1", opt) |
| self.assertEqual(opt["data_centers"], "[dc1]", opt) |
| self.assertEqual(opt["hosts"], "[]", opt) |
| self.assertEqual(opt["ranges"], "1", opt) |
| self.assertEqual(opt["column_families"], "[cf]", opt) |
| |
| def _deprecated_repair_jmx(self, method, arguments): |
| """ |
| * Launch a two node, two DC cluster |
| * Create a keyspace and table |
| * Insert some data |
| * Call the deprecated repair JMX API based on the arguments passed into this method |
| * Check the node log to see if the correct repair was performed based on the jmx args |
| """ |
| cluster = self.cluster |
| |
| debug("Starting cluster..") |
| cluster.populate([1, 1]) |
| node1, node2 = cluster.nodelist() |
| remove_perf_disable_shared_mem(node1) |
| cluster.start() |
| supports_pull_repair = cluster.version() >= LooseVersion('3.10') |
| |
| session = self.patient_cql_connection(node1) |
| create_ks(session, 'ks', 2) |
| create_cf(session, 'cf', read_repair=0.0, columns={'c1': 'text', 'c2': 'text'}) |
| |
| insert_c1c2(session, n=1000, consistency=ConsistencyLevel.ALL) |
| |
| # Run repair |
| mbean = make_mbean('db', 'StorageService') |
| with JolokiaAgent(node1) as jmx: |
| # assert repair runs and returns valid cmd number |
| self.assertEqual(jmx.execute_method(mbean, method, arguments), 1) |
| # wait for log to start |
| node1.watch_log_for("Starting repair command") |
| # get repair parameters from the log |
| l = node1.grep_log(("Starting repair command #1" + (" \([^\)]+\)" if cluster.version() >= LooseVersion("3.10") else "") + |
| ", repairing keyspace ks with repair options \(parallelism: (?P<parallelism>\w+), primary range: (?P<pr>\w+), " |
| "incremental: (?P<incremental>\w+), job threads: (?P<jobs>\d+), ColumnFamilies: (?P<cfs>.+), dataCenters: (?P<dc>.+), " |
| "hosts: (?P<hosts>.+), # of ranges: (?P<ranges>\d+)(, pull repair: (?P<pullrepair>true|false))?\)")) |
| |
| assert_length_equal(l, 1) |
| line, m = l[0] |
| |
| if supports_pull_repair: |
| self.assertEqual(m.group("pullrepair"), "false", "Pull repair cannot be enabled through the deprecated API so the pull repair option should always be false.") |
| |
| return {"parallelism": m.group("parallelism"), |
| "primary_range": m.group("pr"), |
| "incremental": m.group("incremental"), |
| "job_threads": m.group("jobs"), |
| "column_families": m.group("cfs"), |
| "data_centers": m.group("dc"), |
| "hosts": m.group("hosts"), |
| "ranges": m.group("ranges")} |
