mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-27 01:51:59 +00:00
403 lines
14 KiB
Python
403 lines
14 KiB
Python
import pytest
|
|
from helpers.cluster import ClickHouseCluster
|
|
import random
|
|
import string
|
|
import os
|
|
import time
|
|
|
|
cluster = ClickHouseCluster(__file__)
|
|
node = cluster.add_instance('node', main_configs=['configs/enable_test_keeper.xml'], with_zookeeper=True)
|
|
from kazoo.client import KazooClient
|
|
|
|
_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")
|
|
_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
|