From 6b71cb1c350e23daacf3cc2c1131659e6c2f6003 Mon Sep 17 00:00:00 2001 From: Vitaly Baranov Date: Thu, 13 Apr 2023 12:11:58 +0200 Subject: [PATCH] Fix zero-copy-replication on encrypted disks. --- src/Disks/DiskEncrypted.cpp | 13 +++ src/Disks/DiskEncrypted.h | 7 ++ .../__init__.py | 0 .../configs/cluster.xml | 16 ++++ .../configs/disk_s3_encrypted.xml | 49 ++++++++++ .../configs/disk_s3_encrypted_node1.xml | 9 ++ .../configs/disk_s3_encrypted_node2.xml | 9 ++ .../test_encrypted_disk_replication/test.py | 92 +++++++++++++++++++ 8 files changed, 195 insertions(+) create mode 100644 tests/integration/test_encrypted_disk_replication/__init__.py create mode 100644 tests/integration/test_encrypted_disk_replication/configs/cluster.xml create mode 100644 tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted.xml create mode 100644 tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node1.xml create mode 100644 tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node2.xml create mode 100644 tests/integration/test_encrypted_disk_replication/test.py diff --git a/src/Disks/DiskEncrypted.cpp b/src/Disks/DiskEncrypted.cpp index 70624c1de84..db18e9652e7 100644 --- a/src/Disks/DiskEncrypted.cpp +++ b/src/Disks/DiskEncrypted.cpp @@ -360,6 +360,19 @@ SyncGuardPtr DiskEncrypted::getDirectorySyncGuard(const String & path) const return delegate->getDirectorySyncGuard(wrapped_path); } +std::unordered_map DiskEncrypted::getSerializedMetadata(const std::vector & paths) const +{ + std::vector wrapped_paths; + wrapped_paths.reserve(paths.size()); + for (const auto & path : paths) + wrapped_paths.emplace_back(wrappedPath(path)); + auto metadata = delegate->getSerializedMetadata(wrapped_paths); + std::unordered_map res; + for (size_t i = 0; i != paths.size(); ++i) + res.emplace(paths[i], metadata.at(wrapped_paths.at(i))); + return res; +} + void DiskEncrypted::applyNewSettings( const Poco::Util::AbstractConfiguration & config, ContextPtr /*context*/, diff --git a/src/Disks/DiskEncrypted.h b/src/Disks/DiskEncrypted.h index d37e3a560bc..8e824a1f7e5 100644 --- a/src/Disks/DiskEncrypted.h +++ b/src/Disks/DiskEncrypted.h @@ -225,6 +225,11 @@ public: return delegate->getUniqueId(wrapped_path); } + bool checkUniqueId(const String & id) const override + { + return delegate->checkUniqueId(id); + } + void onFreeze(const String & path) override { auto wrapped_path = wrappedPath(path); @@ -276,6 +281,8 @@ public: return delegate->getMetadataStorage(); } + std::unordered_map getSerializedMetadata(const std::vector & paths) const override; + DiskPtr getDelegateDiskIfExists() const override { return delegate; diff --git a/tests/integration/test_encrypted_disk_replication/__init__.py b/tests/integration/test_encrypted_disk_replication/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/integration/test_encrypted_disk_replication/configs/cluster.xml b/tests/integration/test_encrypted_disk_replication/configs/cluster.xml new file mode 100644 index 00000000000..84d16206080 --- /dev/null +++ b/tests/integration/test_encrypted_disk_replication/configs/cluster.xml @@ -0,0 +1,16 @@ + + + + + + node1 + 9000 + + + node2 + 9000 + + + + + diff --git a/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted.xml b/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted.xml new file mode 100644 index 00000000000..01720d6f9cb --- /dev/null +++ b/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted.xml @@ -0,0 +1,49 @@ + + + + + s3 + http://minio1:9001/root/data/ + minio + minio123 + + + encrypted + disk_s3 + 1234567812345678 + encrypted/ + + + encrypted + disk_s3 + encrypted_with_diff_keys/ + + + + + +
+ disk_s3 +
+ + disk_s3_encrypted + + + + + +
+ disk_s3_encrypted +
+ + + + +
+ disk_s3_encrypted_with_diff_keys +
+ + + + + diff --git a/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node1.xml b/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node1.xml new file mode 100644 index 00000000000..ee7ae5d719b --- /dev/null +++ b/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node1.xml @@ -0,0 +1,9 @@ + + + + + 1111111111111111 + + + + diff --git a/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node2.xml b/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node2.xml new file mode 100644 index 00000000000..c0b3ceebc25 --- /dev/null +++ b/tests/integration/test_encrypted_disk_replication/configs/disk_s3_encrypted_node2.xml @@ -0,0 +1,9 @@ + + + + + 2222222222222222 + + + + diff --git a/tests/integration/test_encrypted_disk_replication/test.py b/tests/integration/test_encrypted_disk_replication/test.py new file mode 100644 index 00000000000..f68c534ed43 --- /dev/null +++ b/tests/integration/test_encrypted_disk_replication/test.py @@ -0,0 +1,92 @@ +import pytest +from helpers.cluster import ClickHouseCluster +from helpers.test_tools import TSV + + +cluster = ClickHouseCluster(__file__) + +node1 = cluster.add_instance( + "node1", + main_configs=[ + "configs/cluster.xml", + "configs/disk_s3_encrypted.xml", + "configs/disk_s3_encrypted_node1.xml", + ], + macros={"replica": "node1"}, + with_zookeeper=True, + with_minio=True, +) + +node2 = cluster.add_instance( + "node2", + main_configs=[ + "configs/cluster.xml", + "configs/disk_s3_encrypted.xml", + "configs/disk_s3_encrypted_node2.xml", + ], + macros={"replica": "node2"}, + with_zookeeper=True, + with_minio=True, +) + + +@pytest.fixture(scope="module", autouse=True) +def start_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +@pytest.fixture(autouse=True) +def cleanup_after_test(): + try: + yield + finally: + node1.query("DROP TABLE IF EXISTS encrypted_test ON CLUSTER 'cluster' NO DELAY") + + +def create_table( + zero_copy_replication=False, storage_policy="s3_encrypted_policy_with_diff_keys" +): + engine = "ReplicatedMergeTree('/clickhouse/tables/encrypted_test/', '{replica}')" + + settings = f"storage_policy='{storage_policy}'" + if zero_copy_replication: + settings += ", allow_remote_fs_zero_copy_replication=true" + + node1.query( + f""" + CREATE TABLE encrypted_test ON CLUSTER 'cluster' ( + id Int64, + data String + ) ENGINE={engine} + ORDER BY id + SETTINGS {settings} + """ + ) + + +def check_replication(): + node1.query("INSERT INTO encrypted_test VALUES (0, 'a'), (1, 'b')") + node2.query("INSERT INTO encrypted_test VALUES (2, 'c'), (3, 'd')") + + node1.query("SYSTEM SYNC REPLICA ON CLUSTER 'cluster' encrypted_test") + + select_query = "SELECT * FROM encrypted_test ORDER BY id" + + assert node1.query(select_query) == TSV([[0, "a"], [1, "b"], [2, "c"], [3, "d"]]) + assert node2.query(select_query) == TSV([[0, "a"], [1, "b"], [2, "c"], [3, "d"]]) + + +def test_replication(): + create_table( + zero_copy_replication=False, storage_policy="s3_encrypted_policy_with_diff_keys" + ) + check_replication() + + +def test_zero_copy_replication(): + create_table(zero_copy_replication=True, storage_policy="s3_encrypted_policy") + check_replication()