mirror of
https://github.com/hanzoai/datastore.git
synced 2026-09-26 13:55:07 +00:00
zooqueen
0f1e1b5143
Merge upstream/changelog-26.6 (9593 commits) into our debranded fork. - 190 conflicts resolved: 24 delete-side kept-removed (grpc/arrowflight/gcloud), submodules+version→26.6, contrib/CMakeLists keeps zap + new libpng, drops capnproto (zap provides kj), Server.cpp/CMakeLists re-apply the ZAP wiring + gRPC/arrowflight/gcloud removal. - Full-tree debrand re-swept over the new 26.6 code (uniform Datastore naming; Azure fork Clickhouse* symbols preserved). 0 clickhouse in product src/programs/CMakeLists. - Version now 26.6.1.1. gRPC/google-cloud/chdig stay removed; ZAP remains the transport.
621 lines
18 KiB
Python
621 lines
18 KiB
Python
# pylint: disable=unused-argument
|
|
# pylint: disable=redefined-outer-name
|
|
# pylint: disable=line-too-long
|
|
|
|
# This test covers the following options:
|
|
# - max_backup_bandwidth
|
|
# - max_backup_bandwidth_for_server
|
|
# - max_local_read_bandwidth
|
|
# - max_local_read_bandwidth_for_server
|
|
# - max_local_write_bandwidth
|
|
# - max_local_write_bandwidth_for_server
|
|
# - max_remote_read_network_bandwidth
|
|
# - max_remote_read_network_bandwidth_for_server
|
|
# - max_remote_write_network_bandwidth
|
|
# - max_remote_write_network_bandwidth_for_server
|
|
# - and that max_backup_bandwidth from the query will override setting from the user profile
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from helpers.cluster import DatastoreCluster
|
|
|
|
cluster = DatastoreCluster(__file__)
|
|
|
|
|
|
def elapsed(node, query, **kwargs):
|
|
# Generate a unique query_id to reliably find this query in logs
|
|
query_id = f"elapsed_{uuid.uuid4().hex}"
|
|
|
|
# Execute the query with the unique query_id
|
|
ret = node.query(query, query_id=query_id, **kwargs)
|
|
|
|
# Flush logs to ensure the query appears in system.query_log
|
|
node.query("SYSTEM FLUSH LOGS query_log")
|
|
|
|
# Get the server-side query duration from system.query_log using the query_id
|
|
duration_result = node.query(
|
|
f"""
|
|
SELECT query_duration_ms / 1000.0 as duration
|
|
FROM system.query_log
|
|
WHERE type = 'QueryFinish'
|
|
AND query_id = '{query_id}'
|
|
LIMIT 1
|
|
"""
|
|
)
|
|
duration = float(duration_result.strip())
|
|
return ret, duration
|
|
|
|
|
|
node = cluster.add_instance(
|
|
"node",
|
|
stay_alive=True,
|
|
main_configs=[
|
|
"configs/static_overrides.xml",
|
|
"configs/dynamic_overrides.xml",
|
|
"configs/ssl.xml",
|
|
],
|
|
user_configs=[
|
|
"configs/users_overrides.xml",
|
|
"configs/users_overrides_persistent.xml",
|
|
],
|
|
with_minio=True,
|
|
minio_certs_dir="minio_certs",
|
|
# Disable `with_remote_database_disk` as the instances does not use the default minIO
|
|
with_remote_database_disk=False,
|
|
cpu_limit=6
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def start_cluster():
|
|
try:
|
|
cluster.start()
|
|
yield
|
|
finally:
|
|
cluster.shutdown()
|
|
|
|
|
|
@pytest.fixture(scope="function", autouse=True)
|
|
def revert_config():
|
|
# Revert configs after the test, not before
|
|
yield
|
|
node.exec_in_container(
|
|
[
|
|
"bash",
|
|
"-c",
|
|
"echo '<datastore></datastore>' > /etc/datastore-server/config.d/dynamic_overrides.xml",
|
|
]
|
|
)
|
|
node.exec_in_container(
|
|
[
|
|
"bash",
|
|
"-c",
|
|
"echo '<datastore></datastore>' > /etc/datastore-server/users.d/users_overrides.xml",
|
|
]
|
|
)
|
|
node.restart_datastore()
|
|
|
|
|
|
backup_id_counter = 0
|
|
|
|
|
|
def next_backup_name(storage):
|
|
global backup_id_counter
|
|
if storage == "local":
|
|
backup_id_counter += 1
|
|
return f"Disk('default', '{backup_id_counter}/')"
|
|
elif storage == "remote":
|
|
backup_id_counter += 1
|
|
return f"S3(s3, '{backup_id_counter}/')"
|
|
else:
|
|
raise Exception(storage)
|
|
|
|
|
|
def node_update_config(mode, setting, value=None, restart=True):
|
|
if mode is None:
|
|
return
|
|
if mode == "server":
|
|
config_path = "/etc/datastore-server/config.d/dynamic_overrides.xml"
|
|
config_content = f"""
|
|
<datastore><{setting}>{value}</{setting}></datastore>
|
|
"""
|
|
else:
|
|
config_path = "/etc/datastore-server/users.d/users_overrides.xml"
|
|
config_content = f"""
|
|
<datastore>
|
|
<profiles>
|
|
<default>
|
|
<{setting}>{value}</{setting}>
|
|
</default>
|
|
</profiles>
|
|
</datastore>
|
|
"""
|
|
node.exec_in_container(
|
|
[
|
|
"bash",
|
|
"-c",
|
|
f"echo '{config_content}' > {config_path}",
|
|
]
|
|
)
|
|
if restart:
|
|
node.restart_datastore()
|
|
|
|
|
|
def skip_if_sanitizer(instance):
|
|
if (
|
|
instance.is_built_with_thread_sanitizer()
|
|
or instance.is_built_with_address_sanitizer()
|
|
or instance.is_built_with_memory_sanitizer()
|
|
):
|
|
pytest.skip("throttling tests are unreliable under sanitizers")
|
|
|
|
|
|
def assert_took(took, should_take):
|
|
# we need to decrease the lower limit because the server limits could
|
|
# be enforced by throttling some server background IO instead of query IO
|
|
# and we have no control over it
|
|
assert took >= should_take * 0.80
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"policy,backup_storage,mode,setting,value,should_take",
|
|
[
|
|
#
|
|
# Local -> Local
|
|
#
|
|
pytest.param(
|
|
"default",
|
|
"local",
|
|
None,
|
|
None,
|
|
None,
|
|
0,
|
|
id="no_local_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"default",
|
|
"local",
|
|
"user",
|
|
"max_backup_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="user_local_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"default",
|
|
"local",
|
|
"server",
|
|
"max_backup_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="server_local_throttling",
|
|
),
|
|
#
|
|
# Remote -> Local
|
|
#
|
|
pytest.param(
|
|
"s3",
|
|
"local",
|
|
None,
|
|
None,
|
|
None,
|
|
0,
|
|
id="no_remote_to_local_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"s3",
|
|
"local",
|
|
"user",
|
|
"max_backup_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="user_remote_to_local_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"s3",
|
|
"local",
|
|
"server",
|
|
"max_backup_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="server_remote_to_local_throttling",
|
|
),
|
|
#
|
|
# Remote -> Remote
|
|
#
|
|
pytest.param(
|
|
"s3",
|
|
"remote",
|
|
None,
|
|
None,
|
|
None,
|
|
0,
|
|
id="no_remote_to_remote_throttling",
|
|
),
|
|
# No throttling for S3-to-S3, uses native copy
|
|
pytest.param(
|
|
"s3",
|
|
"remote",
|
|
"user",
|
|
"max_backup_bandwidth",
|
|
"1M",
|
|
0,
|
|
id="user_remote_to_remote_throttling",
|
|
),
|
|
# No throttling for S3-to-S3, uses native copy
|
|
pytest.param(
|
|
"s3",
|
|
"remote",
|
|
"server",
|
|
"max_backup_bandwidth_for_server",
|
|
"2M",
|
|
0,
|
|
id="server_remote_to_remote_throttling",
|
|
),
|
|
#
|
|
# Local -> Remote
|
|
#
|
|
# NOTE: S3 is complex, it will read file 3 times:
|
|
# - first for calculating the checksum
|
|
# - second for calculating the signature
|
|
# - and finally to write the payload to S3
|
|
# Hence the value should be multipled by 3.
|
|
#
|
|
# BUT: only in case of HTTP, HTTPS will not require this.
|
|
pytest.param(
|
|
"default",
|
|
"remote",
|
|
None,
|
|
None,
|
|
None,
|
|
0,
|
|
id="no_local_to_remote_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"default",
|
|
"remote",
|
|
"user",
|
|
"max_backup_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="user_local_to_remote_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"default",
|
|
"remote",
|
|
"server",
|
|
"max_backup_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="server_local_to_remote_throttling",
|
|
),
|
|
],
|
|
)
|
|
def test_backup_throttling(policy, backup_storage, mode, setting, value, should_take):
|
|
skip_if_sanitizer(node)
|
|
node_update_config(mode, setting, value)
|
|
node.query(
|
|
f"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='{policy}';
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
_, took = elapsed(node, f"backup table data to {next_backup_name(backup_storage)}")
|
|
assert_took(took, should_take)
|
|
|
|
|
|
def test_backup_throttling_override():
|
|
skip_if_sanitizer(node)
|
|
node_update_config("user", "max_backup_bandwidth", "1M")
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9;
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
|
|
backup_name = next_backup_name("local")
|
|
_, took = elapsed(
|
|
node,
|
|
f"backup table data to {backup_name}",
|
|
settings={
|
|
"max_backup_bandwidth": "500K",
|
|
},
|
|
)
|
|
# reading 1e6*8 bytes with 500Ki default bandwidth should take (8-0.5)/0.5=15 seconds
|
|
assert_took(took, 15)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"policy,mode,setting,value,should_take",
|
|
[
|
|
#
|
|
# Local
|
|
#
|
|
pytest.param("default", None, None, None, 0, id="no_local_throttling"),
|
|
# reading 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"default",
|
|
"user",
|
|
"max_local_read_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="user_local_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"default",
|
|
"server",
|
|
"max_local_read_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="server_local_throttling",
|
|
),
|
|
#
|
|
# Remote
|
|
#
|
|
pytest.param("s3", None, None, None, 0, id="no_remote_throttling"),
|
|
# reading 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"s3",
|
|
"user",
|
|
"max_remote_read_network_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="user_remote_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"s3",
|
|
"server",
|
|
"max_remote_read_network_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="server_remote_throttling",
|
|
),
|
|
],
|
|
)
|
|
def test_read_throttling(policy, mode, setting, value, should_take):
|
|
skip_if_sanitizer(node)
|
|
node_update_config(mode, setting, value)
|
|
node.query(
|
|
f"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='{policy}';
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
_, took = elapsed(node, "select * from data")
|
|
assert_took(took, should_take)
|
|
|
|
|
|
def test_remote_read_throttling_reload():
|
|
skip_if_sanitizer(node)
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='s3';
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
# without bandwidth limit
|
|
_, took = elapsed(node, "select * from data")
|
|
assert_took(took, 0)
|
|
|
|
# add bandwidth limit and reload config on fly
|
|
node_update_config(
|
|
"server", "max_remote_read_network_bandwidth_for_server", "2M", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
_, took = elapsed(node, "select * from data")
|
|
assert_took(took, 3)
|
|
|
|
# update bandwidth back to 0
|
|
node_update_config(
|
|
"server", "max_remote_read_network_bandwidth_for_server", "0", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
_, took = elapsed(node, "select * from data")
|
|
assert took < 3
|
|
|
|
def test_local_read_throttling_reload():
|
|
skip_if_sanitizer(node)
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='default';
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
# without bandwidth limit
|
|
_, took = elapsed(node, "select * from data")
|
|
assert_took(took, 0)
|
|
|
|
# add bandwidth limit and reload config on fly
|
|
node_update_config(
|
|
"server", "max_local_read_bandwidth_for_server", "2M", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
_, took = elapsed(node, "select * from data")
|
|
assert_took(took, 3)
|
|
|
|
# update bandwidth back to 0
|
|
node_update_config(
|
|
"server", "max_local_read_bandwidth_for_server", "0", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
_, took = elapsed(node, "select * from data")
|
|
assert took < 3
|
|
|
|
@pytest.mark.parametrize(
|
|
"policy,mode,setting,value,should_take",
|
|
[
|
|
#
|
|
# Local
|
|
#
|
|
pytest.param("default", None, None, None, 0, id="no_local_throttling"),
|
|
# reading 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"default",
|
|
"user",
|
|
"max_local_write_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="local_user_throttling",
|
|
),
|
|
# reading 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"default",
|
|
"server",
|
|
"max_local_write_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="local_server_throttling",
|
|
),
|
|
#
|
|
# Remote
|
|
#
|
|
pytest.param("s3", None, None, None, 0, id="no_remote_throttling"),
|
|
# writing 1e6*8 bytes with 1M default bandwidth should take (8-1)/1=7 seconds
|
|
pytest.param(
|
|
"s3",
|
|
"user",
|
|
"max_remote_write_network_bandwidth",
|
|
"1M",
|
|
7,
|
|
id="user_remote_throttling",
|
|
),
|
|
# writing 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
pytest.param(
|
|
"s3",
|
|
"server",
|
|
"max_remote_write_network_bandwidth_for_server",
|
|
"2M",
|
|
3,
|
|
id="server_remote_throttling",
|
|
),
|
|
],
|
|
)
|
|
def test_write_throttling(policy, mode, setting, value, should_take):
|
|
skip_if_sanitizer(node)
|
|
node_update_config(mode, setting, value)
|
|
node.query(
|
|
f"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='{policy}';
|
|
"""
|
|
)
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert_took(took, should_take)
|
|
|
|
|
|
def test_remote_write_throttling_reload():
|
|
skip_if_sanitizer(node)
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='s3';
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
# without bandwidth limit
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert_took(took, 0)
|
|
|
|
# add bandwidth limit and reload config on fly
|
|
node_update_config(
|
|
"server", "max_remote_write_network_bandwidth_for_server", "2M", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
# writing 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert_took(took, 3)
|
|
|
|
# update bandwidth back to 0
|
|
node_update_config(
|
|
"server", "max_remote_write_network_bandwidth_for_server", "0", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert took < 3
|
|
|
|
def test_local_write_throttling_reload():
|
|
skip_if_sanitizer(node)
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9, storage_policy='default';
|
|
insert into data select * from numbers(1e6);
|
|
"""
|
|
)
|
|
# without bandwidth limit
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert_took(took, 0)
|
|
|
|
# add bandwidth limit and reload config on fly
|
|
node_update_config(
|
|
"server", "max_local_write_bandwidth_for_server", "2M", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
# writing 1e6*8 bytes with 2M default bandwidth should take (8-2)/2=3 seconds
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert_took(took, 3)
|
|
|
|
# update bandwidth back to 0
|
|
node_update_config(
|
|
"server", "max_local_write_bandwidth_for_server", "0", False
|
|
)
|
|
node.query("SYSTEM RELOAD CONFIG")
|
|
|
|
_, took = elapsed(node, "insert into data select * from numbers(1e6)")
|
|
assert took < 3
|
|
|
|
def test_max_mutations_bandwidth_for_server():
|
|
skip_if_sanitizer(node)
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9;
|
|
"""
|
|
)
|
|
node.query("insert into data select * from numbers(1e6)")
|
|
_, took = elapsed(
|
|
node,
|
|
"alter table data update key = -key where 1 settings mutations_sync = 1",
|
|
)
|
|
# reading 1e6*8 bytes with 1M/s bandwidth should take (8-1)/1=7 seconds
|
|
assert_took(took, 7)
|
|
|
|
|
|
def test_max_merges_bandwidth_for_server():
|
|
skip_if_sanitizer(node)
|
|
node.query(
|
|
"""
|
|
drop table if exists data;
|
|
create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9;
|
|
"""
|
|
)
|
|
node.query("insert into data select * from numbers(1e6)")
|
|
_, took = elapsed(node, "optimize table data final")
|
|
# reading 1e6*8 bytes with 1M/s bandwidth should take (8-1)/1=7 seconds
|
|
assert_took(took, 7)
|