mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-09-20 16:50:48 +00:00
533 lines
18 KiB
Python
533 lines
18 KiB
Python
import pytest
|
|
from helpers.cluster import ClickHouseCluster
|
|
import random
|
|
import string
|
|
import os
|
|
import time
|
|
from multiprocessing.dummy import Pool
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
node = cluster.add_instance('node', main_configs=['configs/enable_test_keeper.xml', 'configs/logs_conf.xml'], with_zookeeper=True)
|
|
from kazoo.client import KazooClient, KazooState
|
|
|
|
_genuine_zk_instance = None
|
|
_fake_zk_instance = None
|
|
|
|
def get_genuine_zk():
|
|
global _genuine_zk_instance
|
|
if not _genuine_zk_instance:
|
|
print("Zoo1", cluster.get_instance_ip("zoo1"))
|
|
_genuine_zk_instance = cluster.get_kazoo_client('zoo1')
|
|
return _genuine_zk_instance
|
|
|
|
|
|
def get_fake_zk():
|
|
global _fake_zk_instance
|
|
if not _fake_zk_instance:
|
|
print("node", cluster.get_instance_ip("node"))
|
|
_fake_zk_instance = KazooClient(hosts=cluster.get_instance_ip("node") + ":9181", timeout=30.0)
|
|
def reset_last_zxid_listener(state):
|
|
print("Fake zk callback called for state", state)
|
|
global _fake_zk_instance
|
|
if state != KazooState.CONNECTED:
|
|
_fake_zk_instance._reset()
|
|
|
|
_fake_zk_instance.add_listener(reset_last_zxid_listener)
|
|
_fake_zk_instance.start()
|
|
return _fake_zk_instance
|
|
|
|
def random_string(length):
|
|
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
|
|
|
|
def create_random_path(prefix="", depth=1):
|
|
if depth == 0:
|
|
return prefix
|
|
return create_random_path(os.path.join(prefix, random_string(3)), depth - 1)
|
|
|
|
@pytest.fixture(scope="module")
|
|
def started_cluster():
|
|
try:
|
|
cluster.start()
|
|
|
|
yield cluster
|
|
|
|
finally:
|
|
cluster.shutdown()
|
|
if _genuine_zk_instance:
|
|
_genuine_zk_instance.stop()
|
|
_genuine_zk_instance.close()
|
|
if _fake_zk_instance:
|
|
_fake_zk_instance.stop()
|
|
_fake_zk_instance.close()
|
|
|
|
|
|
def test_simple_commands(started_cluster):
|
|
genuine_zk = get_genuine_zk()
|
|
fake_zk = get_fake_zk()
|
|
|
|
for zk in [genuine_zk, fake_zk]:
|
|
zk.create("/test_simple_commands", b"")
|
|
zk.create("/test_simple_commands/somenode1", b"hello")
|
|
zk.set("/test_simple_commands/somenode1", b"world")
|
|
|
|
for zk in [genuine_zk, fake_zk]:
|
|
assert zk.exists("/test_simple_commands")
|
|
assert zk.exists("/test_simple_commands/somenode1")
|
|
print(zk.get("/test_simple_commands/somenode1"))
|
|
assert zk.get("/test_simple_commands/somenode1")[0] == b"world"
|
|
|
|
|
|
def test_sequential_nodes(started_cluster):
|
|
genuine_zk = get_genuine_zk()
|
|
fake_zk = get_fake_zk()
|
|
genuine_zk.create("/test_sequential_nodes")
|
|
fake_zk.create("/test_sequential_nodes")
|
|
for i in range(1, 11):
|
|
genuine_zk.create("/test_sequential_nodes/" + ("a" * i) + "-", sequence=True)
|
|
genuine_zk.create("/test_sequential_nodes/" + ("b" * i))
|
|
fake_zk.create("/test_sequential_nodes/" + ("a" * i) + "-", sequence=True)
|
|
fake_zk.create("/test_sequential_nodes/" + ("b" * i))
|
|
|
|
genuine_childs = list(sorted(genuine_zk.get_children("/test_sequential_nodes")))
|
|
fake_childs = list(sorted(fake_zk.get_children("/test_sequential_nodes")))
|
|
assert genuine_childs == fake_childs
|
|
|
|
|
|
def assert_eq_stats(stat1, stat2):
|
|
assert stat1.version == stat2.version
|
|
assert stat1.cversion == stat2.cversion
|
|
assert stat1.aversion == stat2.aversion
|
|
assert stat1.aversion == stat2.aversion
|
|
assert stat1.dataLength == stat2.dataLength
|
|
assert stat1.numChildren == stat2.numChildren
|
|
|
|
def test_stats(started_cluster):
|
|
genuine_zk = get_genuine_zk()
|
|
fake_zk = get_fake_zk()
|
|
genuine_zk.create("/test_stats_nodes")
|
|
fake_zk.create("/test_stats_nodes")
|
|
genuine_stats = genuine_zk.exists("/test_stats_nodes")
|
|
fake_stats = fake_zk.exists("/test_stats_nodes")
|
|
assert_eq_stats(genuine_stats, fake_stats)
|
|
for i in range(1, 11):
|
|
genuine_zk.create("/test_stats_nodes/" + ("a" * i) + "-", sequence=True)
|
|
genuine_zk.create("/test_stats_nodes/" + ("b" * i))
|
|
fake_zk.create("/test_stats_nodes/" + ("a" * i) + "-", sequence=True)
|
|
fake_zk.create("/test_stats_nodes/" + ("b" * i))
|
|
|
|
genuine_stats = genuine_zk.exists("/test_stats_nodes")
|
|
fake_stats = fake_zk.exists("/test_stats_nodes")
|
|
assert_eq_stats(genuine_stats, fake_stats)
|
|
for i in range(1, 11):
|
|
print("/test_stats_nodes/" + ("a" * i) + "-" + "{:010d}".format((i - 1) * 2))
|
|
genuine_zk.delete("/test_stats_nodes/" + ("a" * i) + "-" + "{:010d}".format((i - 1) * 2))
|
|
genuine_zk.delete("/test_stats_nodes/" + ("b" * i))
|
|
fake_zk.delete("/test_stats_nodes/" + ("a" * i) + "-" + "{:010d}".format((i - 1) * 2))
|
|
fake_zk.delete("/test_stats_nodes/" + ("b" * i))
|
|
|
|
genuine_stats = genuine_zk.exists("/test_stats_nodes")
|
|
fake_stats = fake_zk.exists("/test_stats_nodes")
|
|
print(genuine_stats)
|
|
print(fake_stats)
|
|
assert_eq_stats(genuine_stats, fake_stats)
|
|
for i in range(100):
|
|
genuine_zk.set("/test_stats_nodes", ("q" * i).encode())
|
|
fake_zk.set("/test_stats_nodes", ("q" * i).encode())
|
|
|
|
genuine_stats = genuine_zk.exists("/test_stats_nodes")
|
|
fake_stats = fake_zk.exists("/test_stats_nodes")
|
|
print(genuine_stats)
|
|
print(fake_stats)
|
|
assert_eq_stats(genuine_stats, fake_stats)
|
|
|
|
def test_watchers(started_cluster):
|
|
genuine_zk = get_genuine_zk()
|
|
fake_zk = get_fake_zk()
|
|
genuine_zk.create("/test_data_watches")
|
|
fake_zk.create("/test_data_watches")
|
|
genuine_data_watch_data = None
|
|
|
|
def genuine_callback(event):
|
|
print("Genuine data watch called")
|
|
nonlocal genuine_data_watch_data
|
|
genuine_data_watch_data = event
|
|
|
|
fake_data_watch_data = None
|
|
def fake_callback(event):
|
|
print("Fake data watch called")
|
|
nonlocal fake_data_watch_data
|
|
fake_data_watch_data = event
|
|
|
|
genuine_zk.get("/test_data_watches", watch=genuine_callback)
|
|
fake_zk.get("/test_data_watches", watch=fake_callback)
|
|
|
|
print("Calling set genuine")
|
|
genuine_zk.set("/test_data_watches", b"a")
|
|
print("Calling set fake")
|
|
fake_zk.set("/test_data_watches", b"a")
|
|
time.sleep(3)
|
|
|
|
print("Genuine data", genuine_data_watch_data)
|
|
print("Fake data", fake_data_watch_data)
|
|
assert genuine_data_watch_data == fake_data_watch_data
|
|
|
|
genuine_children = None
|
|
def genuine_child_callback(event):
|
|
print("Genuine child watch called")
|
|
nonlocal genuine_children
|
|
genuine_children = event
|
|
|
|
fake_children = None
|
|
def fake_child_callback(event):
|
|
print("Fake child watch called")
|
|
nonlocal fake_children
|
|
fake_children = event
|
|
|
|
genuine_zk.get_children("/test_data_watches", watch=genuine_child_callback)
|
|
fake_zk.get_children("/test_data_watches", watch=fake_child_callback)
|
|
|
|
print("Calling genuine child")
|
|
genuine_zk.create("/test_data_watches/child", b"b")
|
|
print("Calling fake child")
|
|
fake_zk.create("/test_data_watches/child", b"b")
|
|
|
|
time.sleep(3)
|
|
|
|
print("Genuine children", genuine_children)
|
|
print("Fake children", fake_children)
|
|
assert genuine_children == fake_children
|
|
|
|
def test_multitransactions(started_cluster):
|
|
genuine_zk = get_genuine_zk()
|
|
fake_zk = get_fake_zk()
|
|
for zk in [genuine_zk, fake_zk]:
|
|
zk.create('/test_multitransactions')
|
|
t = zk.transaction()
|
|
t.create('/test_multitransactions/freddy')
|
|
t.create('/test_multitransactions/fred', ephemeral=True)
|
|
t.create('/test_multitransactions/smith', sequence=True)
|
|
results = t.commit()
|
|
assert len(results) == 3
|
|
assert results[0] == '/test_multitransactions/freddy'
|
|
assert results[2].startswith('/test_multitransactions/smith0') is True
|
|
|
|
from kazoo.exceptions import RolledBackError, NoNodeError
|
|
for i, zk in enumerate([genuine_zk, fake_zk]):
|
|
print("Processing ZK", i)
|
|
t = zk.transaction()
|
|
t.create('/test_multitransactions/q')
|
|
t.delete('/test_multitransactions/a')
|
|
t.create('/test_multitransactions/x')
|
|
results = t.commit()
|
|
print("Results", results)
|
|
assert results[0].__class__ == RolledBackError
|
|
assert results[1].__class__ == NoNodeError
|
|
assert zk.exists('/test_multitransactions/q') is None
|
|
assert zk.exists('/test_multitransactions/a') is None
|
|
assert zk.exists('/test_multitransactions/x') is None
|
|
|
|
|
|
def exists(zk, path):
|
|
result = zk.exists(path)
|
|
return result is not None
|
|
|
|
def get(zk, path):
|
|
result = zk.get(path)
|
|
return result[0]
|
|
|
|
def get_children(zk, path):
|
|
return [elem for elem in list(sorted(zk.get_children(path))) if elem not in ('clickhouse', 'zookeeper')]
|
|
|
|
READ_REQUESTS = [
|
|
("exists", exists),
|
|
("get", get),
|
|
("get_children", get_children),
|
|
]
|
|
|
|
|
|
def create(zk, path, data):
|
|
zk.create(path, data.encode())
|
|
|
|
|
|
def set_data(zk, path, data):
|
|
zk.set(path, data.encode())
|
|
|
|
|
|
WRITE_REQUESTS = [
|
|
("create", create),
|
|
("set_data", set_data),
|
|
]
|
|
|
|
|
|
def delete(zk, path):
|
|
zk.delete(path)
|
|
|
|
DELETE_REQUESTS = [
|
|
("delete", delete)
|
|
]
|
|
|
|
|
|
class Request(object):
|
|
def __init__(self, name, arguments, callback, is_return):
|
|
self.name = name
|
|
self.arguments = arguments
|
|
self.callback = callback
|
|
self.is_return = is_return
|
|
|
|
def __str__(self):
|
|
arg_str = ', '.join([str(k) + "=" + str(v) for k, v in self.arguments.items()])
|
|
return "ZKRequest name {} with arguments {}".format(self.name, arg_str)
|
|
|
|
def generate_requests(iters=1):
|
|
requests = []
|
|
existing_paths = []
|
|
for i in range(iters):
|
|
for _ in range(100):
|
|
rand_length = random.randint(0, 10)
|
|
path = "/"
|
|
for j in range(1, rand_length):
|
|
path = create_random_path(path, 1)
|
|
existing_paths.append(path)
|
|
value = random_string(1000)
|
|
request = Request("create", {"path" : path, "value": value[0:10]}, lambda zk, path=path, value=value: create(zk, path, value), False)
|
|
requests.append(request)
|
|
|
|
for _ in range(100):
|
|
path = random.choice(existing_paths)
|
|
value = random_string(100)
|
|
request = Request("set", {"path": path, "value": value[0:10]}, lambda zk, path=path, value=value: set_data(zk, path, value), False)
|
|
requests.append(request)
|
|
|
|
for _ in range(100):
|
|
path = random.choice(existing_paths)
|
|
callback = random.choice(READ_REQUESTS)
|
|
def read_func1(zk, path=path, callback=callback):
|
|
return callback[1](zk, path)
|
|
|
|
request = Request(callback[0], {"path": path}, read_func1, True)
|
|
requests.append(request)
|
|
|
|
for _ in range(30):
|
|
path = random.choice(existing_paths)
|
|
request = Request("delete", {"path": path}, lambda zk, path=path: delete(zk, path), False)
|
|
|
|
for _ in range(100):
|
|
path = random.choice(existing_paths)
|
|
callback = random.choice(READ_REQUESTS)
|
|
def read_func2(zk, path=path, callback=callback):
|
|
return callback[1](zk, path)
|
|
request = Request(callback[0], {"path": path}, read_func2, True)
|
|
requests.append(request)
|
|
return requests
|
|
|
|
|
|
def test_random_requests(started_cluster):
|
|
requests = generate_requests(10)
|
|
genuine_zk = get_genuine_zk()
|
|
fake_zk = get_fake_zk()
|
|
for i, request in enumerate(requests):
|
|
genuine_throw = False
|
|
fake_throw = False
|
|
fake_result = None
|
|
genuine_result = None
|
|
try:
|
|
genuine_result = request.callback(genuine_zk)
|
|
except Exception as ex:
|
|
genuine_throw = True
|
|
|
|
try:
|
|
fake_result = request.callback(fake_zk)
|
|
except Exception as ex:
|
|
fake_throw = True
|
|
|
|
assert fake_throw == genuine_throw, "Fake throw genuine not or vise versa"
|
|
assert fake_result == genuine_result, "Zookeeper results differ"
|
|
root_children_genuine = [elem for elem in list(sorted(genuine_zk.get_children("/"))) if elem not in ('clickhouse', 'zookeeper')]
|
|
root_children_fake = [elem for elem in list(sorted(fake_zk.get_children("/"))) if elem not in ('clickhouse', 'zookeeper')]
|
|
assert root_children_fake == root_children_genuine
|
|
|
|
def test_end_of_session(started_cluster):
|
|
fake_zk1 = None
|
|
fake_zk2 = None
|
|
genuine_zk1 = None
|
|
genuine_zk2 = None
|
|
|
|
try:
|
|
fake_zk1 = KazooClient(hosts=cluster.get_instance_ip("node") + ":9181")
|
|
fake_zk1.start()
|
|
fake_zk2 = KazooClient(hosts=cluster.get_instance_ip("node") + ":9181")
|
|
fake_zk2.start()
|
|
genuine_zk1 = cluster.get_kazoo_client('zoo1')
|
|
genuine_zk1.start()
|
|
genuine_zk2 = cluster.get_kazoo_client('zoo1')
|
|
genuine_zk2.start()
|
|
|
|
fake_zk1.create("/test_end_of_session")
|
|
genuine_zk1.create("/test_end_of_session")
|
|
fake_ephemeral_event = None
|
|
def fake_ephemeral_callback(event):
|
|
print("Fake watch triggered")
|
|
nonlocal fake_ephemeral_event
|
|
fake_ephemeral_event = event
|
|
|
|
genuine_ephemeral_event = None
|
|
def genuine_ephemeral_callback(event):
|
|
print("Genuine watch triggered")
|
|
nonlocal genuine_ephemeral_event
|
|
genuine_ephemeral_event = event
|
|
|
|
assert fake_zk2.exists("/test_end_of_session") is not None
|
|
assert genuine_zk2.exists("/test_end_of_session") is not None
|
|
|
|
fake_zk1.create("/test_end_of_session/ephemeral_node", ephemeral=True)
|
|
genuine_zk1.create("/test_end_of_session/ephemeral_node", ephemeral=True)
|
|
|
|
assert fake_zk2.exists("/test_end_of_session/ephemeral_node", watch=fake_ephemeral_callback) is not None
|
|
assert genuine_zk2.exists("/test_end_of_session/ephemeral_node", watch=genuine_ephemeral_callback) is not None
|
|
|
|
print("Stopping genuine zk")
|
|
genuine_zk1.stop()
|
|
print("Closing genuine zk")
|
|
genuine_zk1.close()
|
|
|
|
print("Stopping fake zk")
|
|
fake_zk1.stop()
|
|
print("Closing fake zk")
|
|
fake_zk1.close()
|
|
|
|
assert fake_zk2.exists("/test_end_of_session/ephemeral_node") is None
|
|
assert genuine_zk2.exists("/test_end_of_session/ephemeral_node") is None
|
|
|
|
assert fake_ephemeral_event == genuine_ephemeral_event
|
|
|
|
finally:
|
|
try:
|
|
for zk in [fake_zk1, fake_zk2, genuine_zk1, genuine_zk2]:
|
|
if zk:
|
|
zk.stop()
|
|
zk.close()
|
|
except:
|
|
pass
|
|
|
|
def test_end_of_watches_session(started_cluster):
|
|
fake_zk1 = None
|
|
fake_zk2 = None
|
|
try:
|
|
fake_zk1 = KazooClient(hosts=cluster.get_instance_ip("node") + ":9181")
|
|
fake_zk1.start()
|
|
|
|
fake_zk2 = KazooClient(hosts=cluster.get_instance_ip("node") + ":9181")
|
|
fake_zk2.start()
|
|
|
|
fake_zk1.create("/test_end_of_watches_session")
|
|
|
|
dummy_set = 0
|
|
def dummy_callback(event):
|
|
nonlocal dummy_set
|
|
dummy_set += 1
|
|
print(event)
|
|
|
|
for child_node in range(100):
|
|
fake_zk1.create("/test_end_of_watches_session/" + str(child_node))
|
|
fake_zk1.get_children("/test_end_of_watches_session/" + str(child_node), watch=dummy_callback)
|
|
|
|
fake_zk2.get_children("/test_end_of_watches_session/" + str(0), watch=dummy_callback)
|
|
fake_zk2.get_children("/test_end_of_watches_session/" + str(1), watch=dummy_callback)
|
|
|
|
fake_zk1.stop()
|
|
fake_zk1.close()
|
|
|
|
for child_node in range(100):
|
|
fake_zk2.create("/test_end_of_watches_session/" + str(child_node) + "/" + str(child_node), b"somebytes")
|
|
|
|
assert dummy_set == 2
|
|
finally:
|
|
try:
|
|
for zk in [fake_zk1, fake_zk2]:
|
|
if zk:
|
|
zk.stop()
|
|
zk.close()
|
|
except:
|
|
pass
|
|
|
|
def test_concurrent_watches(started_cluster):
|
|
fake_zk = get_fake_zk()
|
|
fake_zk.restart()
|
|
global_path = "/test_concurrent_watches_0"
|
|
fake_zk.create(global_path)
|
|
|
|
dumb_watch_triggered_counter = 0
|
|
all_paths_triggered = []
|
|
|
|
existing_path = []
|
|
all_paths_created = []
|
|
watches_created = 0
|
|
def create_path_and_watch(i):
|
|
nonlocal watches_created
|
|
nonlocal all_paths_created
|
|
fake_zk.ensure_path(global_path + "/" + str(i))
|
|
# new function each time
|
|
def dumb_watch(event):
|
|
nonlocal dumb_watch_triggered_counter
|
|
dumb_watch_triggered_counter += 1
|
|
nonlocal all_paths_triggered
|
|
all_paths_triggered.append(event.path)
|
|
|
|
fake_zk.get(global_path + "/" + str(i), watch=dumb_watch)
|
|
all_paths_created.append(global_path + "/" + str(i))
|
|
watches_created += 1
|
|
existing_path.append(i)
|
|
|
|
trigger_called = 0
|
|
def trigger_watch(i):
|
|
nonlocal trigger_called
|
|
trigger_called += 1
|
|
fake_zk.set(global_path + "/" + str(i), b"somevalue")
|
|
try:
|
|
existing_path.remove(i)
|
|
except:
|
|
pass
|
|
|
|
def call(total):
|
|
for i in range(total):
|
|
create_path_and_watch(random.randint(0, 1000))
|
|
time.sleep(random.random() % 0.5)
|
|
try:
|
|
rand_num = random.choice(existing_path)
|
|
trigger_watch(rand_num)
|
|
except:
|
|
pass
|
|
while existing_path:
|
|
try:
|
|
rand_num = random.choice(existing_path)
|
|
trigger_watch(rand_num)
|
|
except:
|
|
pass
|
|
|
|
p = Pool(10)
|
|
arguments = [100] * 10
|
|
watches_must_be_created = sum(arguments)
|
|
watches_trigger_must_be_called = sum(arguments)
|
|
watches_must_be_triggered = sum(arguments)
|
|
p.map(call, arguments)
|
|
p.close()
|
|
|
|
# waiting for late watches
|
|
for i in range(50):
|
|
if dumb_watch_triggered_counter == watches_must_be_triggered:
|
|
break
|
|
|
|
time.sleep(0.1)
|
|
|
|
assert watches_created == watches_must_be_created
|
|
assert trigger_called >= watches_trigger_must_be_called
|
|
assert len(existing_path) == 0
|
|
if dumb_watch_triggered_counter != watches_must_be_triggered:
|
|
print("All created paths", all_paths_created)
|
|
print("All triggerred paths", all_paths_triggered)
|
|
print("All paths len", len(all_paths_created))
|
|
print("All triggered len", len(all_paths_triggered))
|
|
print("Diff", list(set(all_paths_created) - set(all_paths_triggered)))
|
|
|
|
assert dumb_watch_triggered_counter == watches_must_be_triggered
|