Remove raftkeeper 1.0 only code (#352)

docs/how-to-monitor-and-manage.md

@@ -193,7 +193,6 @@ max_stored_snapshots=5
 shutdown_timeout=5000
 startup_timeout=6000000
 raft_logs_level=information
-rotate_log_storage_interval=100000
 log_fsync_mode=fsync_parallel
 log_fsync_interval=1000
 nuraft_thread_size=16

src/Service/ConnectionHandler.h

@@ -8,6 +8,7 @@
 #include <Poco/Thread.h>
 #include <Poco/Util/ServerApplication.h>
 
+#include <Common/IO/ReadBufferFromString.h>
 #include <Common/IO/WriteBufferFromString.h>
 #include <Network/SocketAcceptor.h>
 #include <Network/SocketNotification.h>
@@ -62,7 +63,6 @@ class ConnectionHandler
     void onSocketError(const Notification &);
 
     /// current connection statistics
-    ConnectionStats getConnectionStats() const;
     void dumpStats(WriteBufferFromOwnString & buf, bool brief);
 
     /// reset current connection statistics

src/Service/ForwardConnectionHandler.h

@@ -4,6 +4,7 @@
 #include <Poco/Net/StreamSocket.h>
 #include <Poco/Thread.h>
 
+#include <Common/IO/ReadBufferFromString.h>
 #include <Network/SocketAcceptor.h>
 #include <Network/SocketNotification.h>
 #include <Network/SocketReactor.h>

src/Service/FourLetterCommand.cpp

@@ -8,7 +8,6 @@
 #include <Poco/Environment.h>
 #include <Poco/Path.h>
 #include <Poco/String.h>
-#include "Common/StringUtils.h"
 #include <Common/config_version.h>
 #include <Common/getCurrentProcessFDCount.h>
 #include <Common/getMaxFileDescriptorCount.h>
@@ -258,7 +257,7 @@ String MonitorCommand::run()
     print(ret, "watch_count", state_machine.getTotalWatchesCount());
     print(ret, "ephemerals_count", state_machine.getTotalEphemeralNodesCount());
     print(ret, "approximate_data_size", state_machine.getApproximateDataSize());
-    print(ret, "in_snapshot", state_machine.getSnapshoting());
+    print(ret, "in_snapshot", state_machine.isCreatingSnapshot());
 
 #if defined(__linux__) || defined(__APPLE__)
     print(ret, "open_file_descriptor_count", getCurrentProcessFDCount());

src/Service/KeeperServer.cpp

@@ -135,10 +135,10 @@ ptr<nuraft::cmd_result<ptr<buffer>>> KeeperServer::pushRequestBatch(const std::v
 {
     LOG_DEBUG(log, "Push batch requests of size {}", request_batch.size());
     std::vector<ptr<buffer>> entries;
-    for (const auto & request_session : request_batch)
+    for (const auto & request : request_batch)
     {
-        LOG_TRACE(log, "Push request {}", request_session.toSimpleString());
-        entries.push_back(serializeKeeperRequest(request_session));
+        LOG_TRACE(log, "Push request {}", request.toSimpleString());
+        entries.push_back(serializeKeeperRequest(request));
     }
     /// append_entries write request
     ptr<nuraft::cmd_result<ptr<buffer>>> result = raft_instance->append_entries(entries);

src/Service/KeeperUtils.cpp

@@ -46,11 +46,11 @@ ptr<buffer> serializeKeeperRequest(const RequestForSession & request)
     return out.getBuffer();
 }
 
-RequestForSession deserializeKeeperRequest(nuraft::buffer & data)
+ptr<RequestForSession> deserializeKeeperRequest(nuraft::buffer & data)
 {
+    ptr<RequestForSession> request = cs_new<RequestForSession>();
     ReadBufferFromNuRaftBuffer buffer(data);
-    RequestForSession request_for_session;
-    readIntBinary(request_for_session.session_id, buffer);
+    readIntBinary(request->session_id, buffer);
 
     int32_t length;
     Coordination::read(length, buffer);
@@ -64,17 +64,13 @@ RequestForSession deserializeKeeperRequest(nuraft::buffer & data)
     //    bool is_internal;
     //    Coordination::read(is_internal, buffer);
 
-    request_for_session.request = Coordination::ZooKeeperRequestFactory::instance().get(opnum);
-    request_for_session.request->xid = xid;
-    request_for_session.request->readImpl(buffer);
+    request->request = Coordination::ZooKeeperRequestFactory::instance().get(opnum);
+    request->request->xid = xid;
+    request->request->readImpl(buffer);
 
-    if (!buffer.eof())
-        Coordination::read(request_for_session.create_time, buffer);
-    else /// backward compatibility
-        request_for_session.create_time
-            = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
+    Coordination::read(request->create_time, buffer);
 
-    return request_for_session;
+    return request;
 }
 
 ptr<log_entry> cloneLogEntry(const ptr<log_entry> & entry)

src/Service/KeeperUtils.h

@@ -1,22 +1,29 @@
 #pragma once
 
 #include <fstream>
-#include <time.h>
-#include <Service/Crc32.h>
 #include <ZooKeeper/IKeeper.h>
 #include <ZooKeeper/ZooKeeperCommon.h>
 #include <libnuraft/log_entry.hxx>
 #include <libnuraft/nuraft.hxx>
-#include <common/logger_useful.h>
 #include <Service/KeeperCommon.h>
 
 
 namespace RK
 {
 
+inline UInt64 getCurrentTimeMilliseconds()
+{
+    return duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+}
+
+inline UInt64 getCurrentTimeMicroseconds()
+{
+    return duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+}
+
 /// Serialize and deserialize ZooKeeper request to log
 nuraft::ptr<nuraft::buffer> serializeKeeperRequest(const RequestForSession & request);
-RequestForSession deserializeKeeperRequest(nuraft::buffer & data);
+nuraft::ptr<RequestForSession> deserializeKeeperRequest(nuraft::buffer & data);
 
 nuraft::ptr<nuraft::log_entry> cloneLogEntry(const nuraft::ptr<nuraft::log_entry> & entry);
 

src/Service/LastCommittedIndexManager.cpp

@@ -18,10 +18,6 @@ namespace ErrorCodes
     extern const int CANNOT_SEEK_THROUGH_FILE;
 }
 
-inline UInt64 getCurrentTimeMicroseconds()
-{
-    return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
-}
 
 LastCommittedIndexManager::LastCommittedIndexManager(const String & log_dir) : log(&Poco::Logger::get("LastCommittedIndexManager"))
 {
@@ -35,7 +31,7 @@ LastCommittedIndexManager::LastCommittedIndexManager(const String & log_dir) : l
         throwFromErrno("Failed to open committed log index file", ErrorCodes::CANNOT_OPEN_FILE);
 
     previous_persist_time = getCurrentTimeMicroseconds();
-    persist_thread = ThreadFromGlobalPool([this] { persistThread(); });
+    bg_persist_thread = ThreadFromGlobalPool([this] { persistThread(); });
 }
 
 LastCommittedIndexManager::~LastCommittedIndexManager()
@@ -117,7 +113,7 @@ void LastCommittedIndexManager::shutDown()
     if (!is_shut_down)
     {
         is_shut_down = true;
-        persist_thread.join();
+        bg_persist_thread.join();
 
         ::close(persist_file_fd);
     }

src/Service/LastCommittedIndexManager.h

@@ -38,7 +38,7 @@ class LastCommittedIndexManager
     UInt64 static constexpr PERSIST_INTERVAL_US = 100 * 1000;
     std::string_view static constexpr FILE_NAME = "last_committed_index.bin";
 
-    ThreadFromGlobalPool persist_thread;
+    ThreadFromGlobalPool bg_persist_thread;
     std::atomic<bool> is_shut_down{false};
 
     String persist_file_name;

src/Service/Metrics.h

@@ -13,11 +13,6 @@
 namespace RK
 {
 
-inline UInt64 getCurrentTimeMilliseconds()
-{
-    return duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
-};
-
 
 /**
  * Uses the reservoir sampling algorithm to sample statistical values

src/Service/NuRaftLogSnapshot.cpp

@@ -5,8 +5,6 @@
 #include <stdio.h>
 #include <unistd.h>
 
-#include <common/find_symbols.h>
-
 #include <Poco/DateTime.h>
 #include <Poco/DateTimeFormatter.h>
 #include <Poco/File.h>
@@ -15,6 +13,7 @@
 #include <Common/Exception.h>
 #include <Common/Stopwatch.h>
 
+#include <Service/Crc32.h>
 #include <Service/KeeperUtils.h>
 #include <Service/NuRaftLogSnapshot.h>
 #include <Service/ReadBufferFromNuRaftBuffer.h>
@@ -776,12 +775,12 @@ bool KeeperSnapshotManager::receiveSnapshotMeta(snapshot & meta)
     return true;
 }
 
-bool KeeperSnapshotManager::existSnapshot(const snapshot & meta)
+bool KeeperSnapshotManager::existSnapshot(const snapshot & meta) const
 {
     return snapshots.find(getSnapshotStoreMapKey(meta)) != snapshots.end();
 }
 
-bool KeeperSnapshotManager::existSnapshotObject(const snapshot & meta, ulong obj_id)
+bool KeeperSnapshotManager::existSnapshotObject(const snapshot & meta, ulong obj_id) const
 {
     auto it = snapshots.find(getSnapshotStoreMapKey(meta));
     if (it == snapshots.end())

src/Service/NuRaftLogSnapshot.h

@@ -327,10 +327,10 @@ class KeeperSnapshotManager
     bool saveSnapshotObject(snapshot & meta, ulong obj_id, buffer & buffer);
 
     /// whether snapshot exists
-    bool existSnapshot(const snapshot & meta);
+    bool existSnapshot(const snapshot & meta) const;
 
     /// whether snapshot object exists
-    bool existSnapshotObject(const snapshot & meta, ulong obj_id);
+    bool existSnapshotObject(const snapshot & meta, ulong obj_id) const;
 
     /// load snapshot object, invoked when leader should send snapshot to others.
     bool loadSnapshotObject(const snapshot & meta, ulong obj_id, ptr<buffer> & buffer);