[feat][mds] Support tikv as storage.

Author: rock-git

.github/workflows/ci-build.yml

@@ -42,7 +42,7 @@ jobs:
 
     - name: Pull dingodatabase/dingo-eureka:rocky9-fs
       run: |
-        docker pull dingodatabase/dingo-eureka:rocky9-fs-78d3a4f
+        docker pull dingodatabase/dingo-eureka:rocky9-fs-ac7e2b9
 
     - name: Init build script
       run: |
@@ -141,7 +141,7 @@ jobs:
     - name: Build DingoFS
       run: |
         echo "Build DingoFS"
-        docker run --name release-dingofs --rm -v /mnt/dingofs:/opt/dingofs/ dingodatabase/dingo-eureka:rocky9-fs-78d3a4f /opt/dingofs/build.sh
+        docker run --name release-dingofs --rm -v /mnt/dingofs:/opt/dingofs/ dingodatabase/dingo-eureka:rocky9-fs-ac7e2b9 /opt/dingofs/build.sh
 
     - name: Copy artifactory to another workflow
       uses: actions/upload-artifact@v4

CMakeLists.txt

@@ -237,6 +237,9 @@ message("Using rados: ${RADOS_LIBRARIES}")
 find_package(dingosdk REQUIRED)
 message(STATUS "Found dingosdk: ${dingosdk_VERSION}")
 
+find_package(tikvcpp REQUIRED)
+message(STATUS "Found tikvcpp: ${tikvcpp_VERSION}")
+
 # ------------------------ related to source code ------------------------
 
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/src)

conf/mds.template.conf

@@ -1,8 +1,12 @@
 
-# storage [dingo-store|dummy]
+# storage [dingo-store|tikv|dummy]
+# storage_url depends on storage engine
+# dingo-store: coordinator address 
+# tikv: pd address
+# dummy: no need address
+# format: file://./conf/coor_list | list://ip1:port1,ip2:port2,... | ip1:port1,ip2:port2,...
 --mds_storage_engine=dingo-store
-# dingo-store coordinator address
---coor_url=file://./conf/coor_list
+--storage_url=file://./conf/coor_list
 
 # server
 --mds_server_id=$INSTANCE_ID$

src/mds/client/br.cc

@@ -284,7 +284,7 @@ bool Backup::Init(const std::string& coor_addr) {
   auto kv_storage = DingodbStorage::New();
   CHECK(kv_storage != nullptr) << "new DingodbStorage fail.";
 
-  std::string store_addrs = Helper::ParseCoorAddr(coor_addr);
+  std::string store_addrs = Helper::ParseStorageAddr(coor_addr);
   if (store_addrs.empty()) {
     return false;
   }
@@ -458,7 +458,7 @@ bool Restore::Init(const std::string& coor_addr) {
   auto kv_storage = DingodbStorage::New();
   CHECK(kv_storage != nullptr) << "new DingodbStorage fail.";
 
-  std::string store_addrs = Helper::ParseCoorAddr(coor_addr);
+  std::string store_addrs = Helper::ParseStorageAddr(coor_addr);
   if (store_addrs.empty()) {
     return false;
   }

src/mds/client/store.cc

@@ -40,7 +40,7 @@ bool StoreClient::Init(const std::string& coor_addr) {
   kv_storage_ = DingodbStorage::New();
   CHECK(kv_storage_ != nullptr) << "new DingodbStorage fail.";
 
-  std::string store_addrs = Helper::ParseCoorAddr(coor_addr);
+  std::string store_addrs = Helper::ParseStorageAddr(coor_addr);
   if (store_addrs.empty()) {
     return false;
   }

src/mds/common/helper.cc

@@ -23,6 +23,7 @@
 #include <filesystem>
 #include <fstream>
 #include <random>
+#include <regex>
 #include <string>
 #include <thread>
 
@@ -431,8 +432,13 @@ std::string Helper::EndPointToString(const butil::EndPoint& endpoint) {
   return std::string(butil::endpoint2str(endpoint).c_str());
 }
 
-static bool IsValidFileAddr(const std::string& coor_url) { return coor_url.substr(0, 7) == "file://"; }
-static bool IsValidListAddr(const std::string& coor_url) { return coor_url.substr(0, 7) == "list://"; }
+static bool IsValidFileAddr(const std::string& url) { return url.substr(0, 7) == "file://"; }
+static bool IsValidListAddr(const std::string& url) { return url.substr(0, 7) == "list://"; }
+static bool IsValidBareAddr(const std::string& url) {
+  // check 127.0.0.1:80 or 127.0.0.1:80,127.0.0.1:81 use regex
+  std::regex ip_port_list_regex(R"(^((\d{1,3}\.){3}\d{1,3}:\d{1,5})(,(\d{1,3}\.){3}\d{1,3}:\d{1,5})*$)");
+  return std::regex_match(url, ip_port_list_regex);
+}
 
 static std::string ParseFileUrl(const std::string& coor_url) {
   CHECK(coor_url.substr(0, 7) == "file://") << "Invalid coor_url: " << coor_url;
@@ -461,25 +467,25 @@ static std::string ParseFileUrl(const std::string& coor_url) {
   return addrs.empty() ? "" : addrs.substr(0, addrs.size() - 1);
 }
 
-static std::string ParseListUrl(const std::string& coor_url) {
-  CHECK(coor_url.substr(0, 7) == "list://") << "Invalid coor_url: " << coor_url;
+static std::string ParseListUrl(const std::string& url) {
+  CHECK(url.substr(0, 7) == "list://") << "invalid url: " << url;
 
-  return coor_url.substr(7);
+  return url.substr(7);
 }
 
-std::string Helper::ParseCoorAddr(const std::string& coor_url) {
-  std::string coor_addrs;
-  if (IsValidFileAddr(coor_url)) {
-    coor_addrs = ParseFileUrl(coor_url);
+std::string Helper::ParseStorageAddr(const std::string& url) {
+  std::string storage_addrs;
+  if (IsValidFileAddr(url)) {
+    storage_addrs = ParseFileUrl(url);
 
-  } else if (IsValidListAddr(coor_url)) {
-    coor_addrs = ParseListUrl(coor_url);
+  } else if (IsValidListAddr(url)) {
+    storage_addrs = ParseListUrl(url);
 
-  } else {
-    DINGO_LOG(ERROR) << "Invalid coor_url: " << coor_url;
+  } else if (IsValidBareAddr(url)) {
+    storage_addrs = url;
   }
 
-  return coor_addrs;
+  return storage_addrs;
 }
 
 bool Helper::SaveFile(const std::string& filepath, const std::string& data) {

src/mds/common/helper.h

@@ -100,7 +100,7 @@ class Helper {
 
   static std::string EndPointToString(const butil::EndPoint& endpoint);
 
-  static std::string ParseCoorAddr(const std::string& coor_url);
+  static std::string ParseStorageAddr(const std::string& url);
 
   static bool SaveFile(const std::string& filepath, const std::string& data);
 

src/mds/filesystem/filesystem.cc

@@ -87,8 +87,14 @@ DEFINE_validator(mds_transfer_max_slice_num, brpc::PassValidate);
 DEFINE_uint32(mds_cache_expire_interval_s, 7200, "Cache expire interval in seconds.");
 DEFINE_validator(mds_cache_expire_interval_s, brpc::PassValidate);
 
-DEFINE_string(mds_storage_engine, "dummy", "mds storage engine, e.g dingo-store|dummy");
+DEFINE_string(mds_storage_engine, "dummy", "mds storage engine, e.g dingo-store|tikv|dummy");
+DEFINE_validator(mds_storage_engine, [](const char*, const std::string& value) -> bool {
+  return value == "dingo-store" || value == "tikv" || value == "dummy";
+});
+
 DEFINE_string(mds_id_generator_type, "coor", "id generator type, e.g coor|store");
+DEFINE_validator(mds_id_generator_type,
+                 [](const char*, const std::string& value) -> bool { return value == "coor" || value == "store"; });
 
 static bool IsInvalidName(const std::string& name) {
   return name.empty() || name.size() > FLAGS_mds_filesystem_name_max_size;
@@ -2835,7 +2841,7 @@ IdGeneratorUPtr FileSystemSet::NewInoGenerator(uint32_t fs_id) {
     ino_id_generator = (FLAGS_mds_id_generator_type == "coor") ? NewInodeIdGenerator(fs_id, coordinator_client_)
                                                                : NewInodeIdGenerator(fs_id, kv_storage_);
 
-  } else if (FLAGS_mds_storage_engine == "dummy") {
+  } else {
     ino_id_generator = NewInodeIdGenerator(fs_id, kv_storage_);
   }
 
@@ -2850,7 +2856,7 @@ void FileSystemSet::DestroyInoGenerator(uint32_t fs_id) {
       DestroyInodeIdGenerator(fs_id, kv_storage_);
     }
 
-  } else if (FLAGS_mds_storage_engine == "dummy") {
+  } else {
     DestroyInodeIdGenerator(fs_id, kv_storage_);
   }
 }

src/mds/filesystem/store_operation.cc

@@ -13,6 +13,7 @@
 
 #include "mds/filesystem/store_operation.h"
 
+#include <bthread/types.h>
 #include <fcntl.h>
 
 #include <algorithm>
@@ -2690,7 +2691,7 @@ void OperationProcessor::LaunchExecuteBatchOperation(const BatchOperation& batch
   Params* params = new Params({.self = this, .batch_operation = batch_operation});
 
   bthread_t tid;
-  bthread_attr_t attr = BTHREAD_ATTR_SMALL;
+  bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
   if (bthread_start_background(
           &tid, &attr,
           [](void* arg) -> void* {

src/mds/main.cc

@@ -30,7 +30,7 @@
 #include "mds/server.h"
 
 DEFINE_string(conf, "./conf/mds.conf", "mds config path");
-DEFINE_string(coor_url, "file://./conf/coor_list", "coor service url, e.g. file://<path> or list://<addr1>");
+DEFINE_string(storage_url, "file://./conf/coor_list", "storage url, e.g. file://<path> or list://<addr1>");
 
 const int kMaxStacktraceSize = 128;
 
@@ -218,7 +218,7 @@ static std::vector<gflags::CommandLineFlagInfo> GetFlags(const std::string& pref
   gflags::CommandLineFlagInfo conf_flag;
   if (gflags::GetCommandLineFlagInfo("conf", &conf_flag)) dist_flags.push_back(conf_flag);
   gflags::CommandLineFlagInfo coor_url_flag;
-  if (gflags::GetCommandLineFlagInfo("coor_url", &coor_url_flag)) dist_flags.push_back(coor_url_flag);
+  if (gflags::GetCommandLineFlagInfo("storage_url", &coor_url_flag)) dist_flags.push_back(coor_url_flag);
 
   std::vector<gflags::CommandLineFlagInfo> flags;
   gflags::GetAllFlags(&flags);
@@ -238,8 +238,8 @@ static std::string GetUsage(char* program_name) {
   oss << fmt::format("\t{} --help\n", program_name);
   oss << fmt::format("\t{} --mds_server_port=7801", program_name);
   oss << fmt::format("\t{} --conf=./conf/mds.conf", program_name);
-  oss << fmt::format("\t{} --conf=./conf/mds.conf --coor_url=file://./conf/coor_list\n", program_name);
-  oss << fmt::format("\t{} --conf=./conf/mds.conf --coor_url=list://127.0.0.1:22001\n", program_name);
+  oss << fmt::format("\t{} --conf=./conf/mds.conf --storage_url=file://./conf/coor_list\n", program_name);
+  oss << fmt::format("\t{} --conf=./conf/mds.conf --storage_url=list://127.0.0.1:22001\n", program_name);
   oss << fmt::format("\t{} [OPTIONS]\n", program_name);
 
   auto flags = GetFlags("mds_");
@@ -287,20 +287,15 @@ static bool ParseOption(int argc, char** argv) {
   return false;
 }
 
-static bool CheckCoorUrl(const std::string& coor_url) {
-  if (coor_url.empty()) {
-    std::cerr << "coor url is empty.\n";
+static bool CheckStorageUrl(const std::string& storage_url) {
+  if (storage_url.empty()) {
+    std::cerr << "storage url is empty.\n";
     return false;
   }
 
-  if (coor_url.substr(0, 7) != "file://" && coor_url.substr(0, 7) != "list://") {
-    std::cerr << "coor url must start with file:// or list://\n";
-    return false;
-  }
-
-  auto coor_addr = dingofs::mds::Helper::ParseCoorAddr(coor_url);
-  if (coor_addr.empty()) {
-    std::cerr << "coor addr is invalid, please check your coor url: " << coor_url << '\n';
+  auto storage_addr = dingofs::mds::Helper::ParseStorageAddr(storage_url);
+  if (storage_addr.empty()) {
+    std::cerr << "storage addr is invalid, please check your storage url: " << storage_url << '\n';
     return false;
   }
 
@@ -324,7 +319,7 @@ int main(int argc, char* argv[]) {
 
   gflags::ParseCommandLineNonHelpFlags(&argc, &argv, false);
 
-  if (!CheckCoorUrl(FLAGS_coor_url)) return -1;
+  if (dingofs::mds::FLAGS_mds_storage_engine != "dummy" && !CheckStorageUrl(FLAGS_storage_url)) return -1;
 
   SetupSignalHandler();
 
@@ -334,8 +329,7 @@ int main(int argc, char* argv[]) {
   CHECK(server.InitConfig(FLAGS_conf)) << fmt::format("init config({}) error.", FLAGS_conf);
   CHECK(GeneratePidFile(server.GetPidFilePath())) << "generate pid file error.";
   CHECK(server.InitMDSMeta()) << "init mds meta error.";
-  CHECK(server.InitCoordinatorClient(FLAGS_coor_url)) << "init coordinator client error.";
-  CHECK(server.InitStorage(FLAGS_coor_url)) << "init storage error.";
+  CHECK(server.InitStorage(FLAGS_storage_url)) << "init storage error.";
   CHECK(server.InitOperationProcessor()) << "init operation processor error.";
   CHECK(server.InitCacheGroupMemberManager()) << "init cache group member manager error.";
   CHECK(server.InitNotifyBuddy()) << "init notify buddy error.";