thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-25 17:12:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

better

Browse Source
This commit is contained in:
Nikita Mikhailov 2020-12-23 21:59:27 +03:00
parent 9a069b2cc0
commit a35ef5ace3
3 changed files with 75 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/integration/test_cluster_copier/test.py
View File

@ -94,8 +94,6 @@ class Task1:
instance.query("INSERT INTO hits_all SELECT * FROM system.numbers LIMIT 1002",
settings={"insert_distributed_sync": 1})
print instance.query("SELECT DISTINCT 'all' AS partition FROM hits_all ORDER BY partition DESC")
def check(self):
assert TSV(self.cluster.instances['s0_0_0'].query("SELECT count() FROM hits_all")) == TSV("1002\n")
assert TSV(self.cluster.instances['s1_0_0'].query("SELECT count() FROM hits_all")) == TSV("1002\n")
@ -312,7 +310,13 @@ def execute_task(task, cmd_options):
# Tests
@pytest.mark.parametrize(('use_sample_offset'),[False,True])
@pytest.mark.parametrize(
('use_sample_offset'),
[
False,
True
]
)
def test_copy_simple(started_cluster, use_sample_offset):
if use_sample_offset:
execute_task(Task1(started_cluster), ['--experimental-use-sample-offset', '1'])
@ -320,7 +324,13 @@ def test_copy_simple(started_cluster, use_sample_offset):
execute_task(Task1(started_cluster), [])
@pytest.mark.parametrize(('use_sample_offset'),[False,True])
@pytest.mark.parametrize(
('use_sample_offset'),
[
False,
True
]
)
def test_copy_with_recovering(started_cluster, use_sample_offset):
if use_sample_offset:
execute_task(Task1(started_cluster), ['--copy-fault-probability', str(COPYING_FAIL_PROBABILITY),
@ -329,7 +339,13 @@ def test_copy_with_recovering(started_cluster, use_sample_offset):
execute_task(Task1(started_cluster), ['--copy-fault-probability', str(COPYING_FAIL_PROBABILITY)])
@pytest.mark.parametrize(('use_sample_offset'),[False,True])
@pytest.mark.parametrize(
('use_sample_offset'),
[
False,
True
]
)
def test_copy_with_recovering_after_move_faults(started_cluster, use_sample_offset):
if use_sample_offset:
execute_task(Task1(started_cluster), ['--move-fault-probability', str(MOVING_FAIL_PROBABILITY),
@ -364,7 +380,11 @@ def test_no_index(started_cluster):
def test_no_arg(started_cluster):
execute_task(Task_no_arg(started_cluster), [])
def test_non_partitioned_table(started_cluster):
execute_task(Task_non_partitioned_table(started_cluster), [])
if __name__ == '__main__':
with contextmanager(started_cluster)() as cluster:
for name, instance in list(cluster.instances.items()):
print(name, instance.ip_address)
input("Cluster created, press any key to destroy...")

20
tests/integration/test_cluster_copier/test_big_cluster.py
View File

@ -52,8 +52,8 @@ def started_cluster():
}
}
for cluster_name, shards in clusters_schema.iteritems():
for shard_name, replicas in shards.iteritems():
for cluster_name, shards in clusters_schema.items():
for shard_name, replicas in shards.items():
for replica_name in replicas:
name = "s{}_{}_{}".format(cluster_name, shard_name, replica_name)
cluster.add_instance(name,
@ -100,16 +100,16 @@ def execute_task(task, cmd_options):
task.start()
zk = cluster.get_kazoo_client('zoo1')
print "Use ZooKeeper server: {}:{}".format(zk.hosts[0][0], zk.hosts[0][1])
print("Use ZooKeeper server: {}:{}".format(zk.hosts[0][0], zk.hosts[0][1]))
try:
zk.delete("/clickhouse-copier", recursive=True)
except kazoo.exceptions.NoNodeError:
print "No node /clickhouse-copier. It is Ok in first test."
print("No node /clickhouse-copier. It is Ok in first test.")
zk_task_path = task.zk_task_path
zk.ensure_path(zk_task_path)
zk.create(zk_task_path + "/description", task.copier_task_config)
zk.create(zk_task_path + "/description", task.copier_task_config.encode())
# Run cluster-copier processes on each node
docker_api = docker.from_env().api
@ -120,22 +120,20 @@ def execute_task(task, cmd_options):
'--task-path', zk_task_path,
'--base-dir', '/var/log/clickhouse-server/copier']
cmd += cmd_options
# copiers = []
copiers = random.sample(cluster.instances.keys(), 3)
copiers.append('s1_0_0')
copiers = random.sample(list(cluster.instances.keys()), 3)
for instance_name in copiers:
instance = cluster.instances[instance_name]
container = instance.get_docker_handle()
instance.copy_file_to_container(os.path.join(CURRENT_TEST_DIR, "configs/config-copier.xml"),
"/etc/clickhouse-server/config-copier.xml")
print "Copied copier config to {}".format(instance.name)
print("Copied copier config to {}".format(instance.name))
exec_id = docker_api.exec_create(container.id, cmd, stderr=True)
output = docker_api.exec_start(exec_id).decode('utf8')
print(output)
copiers_exec_ids.append(exec_id)
print "Copier for {} ({}) has started".format(instance.name, instance.ip_address)
print("Copier for {} ({}) has started".format(instance.name, instance.ip_address))
# Wait for copiers stopping and check their return codes
for exec_id, instance_name in zip(copiers_exec_ids, copiers):

66
tests/integration/test_cluster_copier/test_trivial.py
View File

@ -11,8 +11,8 @@ sys.path.insert(0, os.path.dirname(CURRENT_TEST_DIR))
from helpers.cluster import ClickHouseCluster
from helpers.test_tools import TSV
COPYING_FAIL_PROBABILITY = 0.2
MOVING_FAIL_PROBABILITY = 0.2
COPYING_FAIL_PROBABILITY = 0.33
MOVING_FAIL_PROBABILITY = 0.1
cluster = None
@ -85,11 +85,6 @@ def execute_task(task, cmd_options):
zk = cluster.get_kazoo_client('zoo1')
print("Use ZooKeeper server: {}:{}".format(zk.hosts[0][0], zk.hosts[0][1]))
try:
zk.delete("/clickhouse-copier", recursive=True)
except kazoo.exceptions.NoNodeError:
print "No node /clickhouse-copier. It is Ok in first test."
zk_task_path = task.zk_task_path
zk.ensure_path(zk_task_path)
zk.create(zk_task_path + "/description", task.copier_task_config)
@ -104,28 +99,23 @@ def execute_task(task, cmd_options):
'--base-dir', '/var/log/clickhouse-server/copier']
cmd += cmd_options
copiers = cluster.instances.keys()
print(cmd)
for instance_name in copiers:
instance = cluster.instances[instance_name]
for instance_name, instance in cluster.instances.items():
container = instance.get_docker_handle()
instance.copy_file_to_container(os.path.join(CURRENT_TEST_DIR, "configs/config-copier.xml"),
"/etc/clickhouse-server/config-copier.xml")
print "Copied copier config to {}".format(instance.name)
exec_id = docker_api.exec_create(container.id, cmd, stderr=True)
output = docker_api.exec_start(exec_id).decode('utf8')
print(output)
docker_api.exec_start(exec_id, detach=True)
copiers_exec_ids.append(exec_id)
print("Copier for {} ({}) has started".format(instance.name, instance.ip_address))
# Wait for copiers stopping and check their return codes
for exec_id, instance_name in zip(copiers_exec_ids, copiers):
instance = cluster.instances[instance_name]
for exec_id, instance in zip(copiers_exec_ids, iter(cluster.instances.values())):
while True:
res = docker_api.exec_inspect(exec_id)
if not res['Running']:
break
time.sleep(0.5)
time.sleep(1)
assert res['ExitCode'] == 0, "Instance: {} ({}). Info: {}".format(instance.name, instance.ip_address, repr(res))
@ -135,32 +125,56 @@ def execute_task(task, cmd_options):
zk.delete(zk_task_path, recursive=True)
# Tests
@pytest.mark.parametrize(('use_sample_offset'),[False,True])
@pytest.mark.parametrize(
('use_sample_offset'),
[
False,
True
]
)
def test_trivial_copy(started_cluster, use_sample_offset):
if use_sample_offset:
execute_task(TaskTrivial(started_cluster, use_sample_offset), ['--experimental-use-sample-offset', '1'])
else:
print("AAAAA")
execute_task(TaskTrivial(started_cluster, use_sample_offset), [])
@pytest.mark.parametrize(('use_sample_offset'),[False,True])
@pytest.mark.parametrize(
('use_sample_offset'),
[
False,
True
]
)
def test_trivial_copy_with_copy_fault(started_cluster, use_sample_offset):
if use_sample_offset:
execute_task(TaskTrivial(started_cluster, use_sample_offset), ['--copy-fault-probability', str(COPYING_FAIL_PROBABILITY),
execute_task(TaskTrivial(started_cluster), ['--copy-fault-probability', str(COPYING_FAIL_PROBABILITY),
'--experimental-use-sample-offset', '1'])
else:
execute_task(TaskTrivial(started_cluster, use_sample_offset), ['--copy-fault-probability', str(COPYING_FAIL_PROBABILITY)])
execute_task(TaskTrivial(started_cluster), ['--copy-fault-probability', str(COPYING_FAIL_PROBABILITY)])
@pytest.mark.parametrize(('use_sample_offset'),[False,True])
@pytest.mark.parametrize(
('use_sample_offset'),
[
False,
True
]
)
def test_trivial_copy_with_move_fault(started_cluster, use_sample_offset):
if use_sample_offset:
execute_task(TaskTrivial(started_cluster, use_sample_offset), ['--move-fault-probability', str(MOVING_FAIL_PROBABILITY),
execute_task(TaskTrivial(started_cluster), ['--move-fault-probability', str(MOVING_FAIL_PROBABILITY),
'--experimental-use-sample-offset', '1'])
else:
execute_task(TaskTrivial(started_cluster, use_sample_offset), ['--move-fault-probability', str(MOVING_FAIL_PROBABILITY)])
execute_task(TaskTrivial(started_cluster), ['--move-fault-probability', str(MOVING_FAIL_PROBABILITY)])
if __name__ == '__main__':
with contextmanager(started_cluster)() as cluster:
for name, instance in list(cluster.instances.items()):
print(name, instance.ip_address)
input("Cluster created, press any key to destroy...")