Add system nodes for keeper

src/Service/KeeperCommon.h

@@ -12,6 +12,23 @@ namespace RK
 using RunnerId = size_t;
 using ThreadPoolPtr = std::shared_ptr<ThreadPool>;
 
+#ifdef COMPATIBLE_MODE_ZOOKEEPER
+    const String ZOOKEEPER_SYSTEM_PATH = "/zookeeper";
+    const String ZOOKEEPER_CONFIG_NODE = ZOOKEEPER_SYSTEM_PATH + "/config";
+#else
+    const String CLICKHOUSE_KEEPER_SYSTEM_PATH = "/keeper";
+    const String CLICKHOUSE_KEEPER_API_VERSION_PATH = CLICKHOUSE_KEEPER_SYSTEM_PATH + "/api_version";
+
+    enum class KeeperApiVersion : uint8_t
+    {
+        ZOOKEEPER_COMPATIBLE = 0,
+        WITH_FILTERED_LIST,
+        WITH_MULTI_READ
+    };
+
+    inline constexpr auto CURRENT_KEEPER_API_VERSION = KeeperApiVersion::WITH_MULTI_READ;
+#endif
+
 struct RequestId;
 
 /// Attached session id to request

src/Service/KeeperServer.cpp

@@ -47,6 +47,19 @@ KeeperServer::KeeperServer(
         checkAndGetSuperdigest(settings->super_digest),
         MAX_OBJECT_NODE_SIZE,
         request_processor_);
+
+#ifdef COMPATIBLE_MODE_ZOOKEEPER
+    auto cluster_config = state_manager->getClusterConfig();
+
+    String data;
+    for (const auto & s : cluster_config->get_servers())
+    {
+        data += fmt::format("server.{}={}:participant\n", s->get_id(), s->get_endpoint());
+    }
+    data += "version=0";
+    state_machine->getStore().getNode(ZOOKEEPER_CONFIG_NODE)->data = data;
+#endif
+
 }
 namespace
 {

src/Service/KeeperStore.cpp

@@ -1596,4 +1596,27 @@ uint64_t KeeperStore::getApproximateDataSize() const
     return size_bytes;
 }
 
+void KeeperStore::initializeSystemNodes()
+{
+    auto add_node = [&](const String & path)
+    {
+        if (!data_tree.count(path))
+        {
+            data_tree.emplace(path, std::make_shared<KeeperNode>());
+            getNode(getParentPath(path))->children.insert(getBaseName(path));
+        }
+    };
+
+#ifdef COMPATIBLE_MODE_ZOOKEEPER
+    add_node(ZOOKEEPER_SYSTEM_PATH);
+    add_node(ZOOKEEPER_CONFIG_NODE);
+
+#else
+    add_node(CLICKHOUSE_KEEPER_SYSTEM_PATH);
+    add_node(CLICKHOUSE_KEEPER_API_VERSION_PATH);
+
+    data_tree.get(CLICKHOUSE_KEEPER_API_VERSION_PATH)->data = toString(static_cast<uint8_t>(CURRENT_KEEPER_API_VERSION));
+#endif
+}
+
 }

src/Service/KeeperStore.h

@@ -406,11 +406,14 @@ class KeeperStore
     {
         watch_manager.dumpWatches(buf);
     }
+
     void dumpWatchesByPath(WriteBufferFromOwnString & buf) const
     {
         watch_manager.dumpWatchesByPath(buf);
     }
 
+    void initializeSystemNodes();
+
     mutable std::shared_mutex auth_mutex;
     SessionAndAuth session_and_auth;
 

src/Service/NuRaftStateMachine.cpp

@@ -114,6 +114,8 @@ NuRaftStateMachine::NuRaftStateMachine(
         store.getSessionIDCounter(),
         store.getZxid());
 
+    store.initializeSystemNodes();
+
     LOG_INFO(log, "Starting background creating snapshot thread.");
     snap_thread = ThreadFromGlobalPool([this] { snapThread(); });
 }

src/Service/tests/gtest_raft_state_machine.cpp

@@ -156,7 +156,7 @@ TEST(RaftStateMachine, createSnapshot)
     UInt64 term = 1;
     snapshot meta(last_index, term, config);
     machine.create_snapshot(meta);
-    ASSERT_EQ(machine.getStore().getNodesCount(), 36);
+    ASSERT_EQ(machine.getStore().getNodesCount(), 38);
     machine.shutdown();
 
     cleanDirectory(snap_dir);
@@ -209,7 +209,7 @@ TEST(RaftStateMachine, syncSnapshot)
         machine_target.save_logical_snp_obj(meta, obj_id, *(data_out.get()), is_first, is_last_obj);
     }
     machine_target.apply_snapshot(meta);
-    ASSERT_EQ(machine_target.getStore().getNodesCount(), last_index + 1);
+    ASSERT_EQ(machine_target.getStore().getNodesCount(), last_index + 3);
 
     for (auto i = 1; i < obj_id; i++)
     {
@@ -276,7 +276,7 @@ TEST(RaftStateMachine, initStateMachine)
         LOG_INFO(log, "get sm/tm last commit index {},{}", machine.last_commit_index(), machine.getLastCommittedIndex());
 
 
-        ASSERT_EQ(machine.getStore().getNodesCount(), 257);
+        ASSERT_EQ(machine.getStore().getNodesCount(), 259);
         machine.shutdown();
     }
 

tests/integration/test_back_to_back/test.py

@@ -718,3 +718,15 @@ def call(total):
         assert dumb_watch_triggered_counter == watches_must_be_triggered
     finally:
         close_zk_clients([fake_zk])
+
+
+def test_system_nodes(started_cluster):
+    genuine_zk = fake_zk = None
+    try:
+        genuine_zk = get_genuine_zk()
+        fake_zk = get_fake_zk()
+        cluster_config = (b'server.1=node1:8103:participant\nserver.2=node2:8103:participant\nserver.3=node3:8103'
+                          b':participant\nversion=0')
+        assert fake_zk.get('/zookeeper/config')[0] == cluster_config
+    finally:
+        close_zk_clients([genuine_zk, fake_zk])

tests/integration/test_converter/test.py

@@ -105,9 +105,10 @@ def compare_states(zk1, zk2, path="/"):
     data2, stat2 = zk2.get(path)
     print("Left Stat", stat1)
     print("Right Stat", stat2)
-    assert data1 == data2, "Data not equal on path " + str(path)
+    if path not in ("/zookeeper/config",):
+        assert data1 == data2, "Data not equal on path " + str(path)
     # both paths have strange stats
-    if path not in ("/", "/zookeeper"):
+    if path not in ("/", "/zookeeper", "/zookeeper/config"):
         compare_stats(stat1, stat2, path)
 
     first_children = list(sorted(zk1.get_children(path)))

tests/integration/test_four_word_command/test.py

@@ -180,7 +180,7 @@ def test_cmd_mntr(started_cluster):
         # contains:
         #   10 nodes created by test
         #   1 root node
-        assert int(result["zk_znode_count"]) == 11
+        assert int(result["zk_znode_count"]) == 13
         assert int(result["zk_watch_count"]) == 2
         assert int(result["zk_ephemerals_count"]) == 2
         assert int(result["zk_approximate_data_size"]) > 0
@@ -354,7 +354,7 @@ def test_cmd_srvr(started_cluster):
         assert int(result['Connections']) == 1
         assert int(result['Zxid']) > 14
         assert result['Mode'] == 'leader'
-        assert result['Node count'] == '11'
+        assert result['Node count'] == '13'
 
     finally:
         close_zk_client(zk)
@@ -392,7 +392,7 @@ def test_cmd_stat(started_cluster):
         assert int(result['Connections']) == 1
         assert int(result['Zxid']) > 14
         assert result['Mode'] == 'leader'
-        assert result['Node count'] == '11'
+        assert result['Node count'] == '13'
 
         # filter connection statistics
         cons = [n for n in data.split('\n') if '=' in n]