[feat][mdsv2] Adjust fsstat web page.

Author: rock-git

dev-scripts/test.py

@@ -0,0 +1,107 @@
+import os
+import shutil
+import stat
+import random
+import string
+import time
+import threading
+from concurrent.futures import ThreadPoolExecutor
+import queue
+import argparse
+
+def generate_random_string(length=10):
+    """生成随机字符串"""
+    return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
+
+
+def create_blank_file(file_path):
+    """创建一个空白文件"""
+    with open(file_path, 'w', encoding='utf-8') as f:
+        pass
+    return file_path
+
+def create_directory(dir_path):
+    """创建一个目录"""
+    os.makedirs(dir_path, exist_ok=True)
+    return dir_path
+
+
+
+# 递归创建目录和文件，每层创建10个子目录和1000个文件
+def create_nested_structure(base_path, depth, num_dirs=10, num_files=1000):
+    """递归创建目录和文件结构"""
+    if depth == 0:
+        return
+    
+    for i in range(num_dirs):
+        dir_name = f"dir_{i:05d}"
+        dir_path = os.path.join(base_path, dir_name)
+        create_directory(dir_path)
+        
+        # 在当前目录下创建文件
+        for j in range(num_files):
+
+            file_name = f"file_{j:010d}"
+            file_path = os.path.join(dir_path, file_name)
+            create_blank_file(file_path)
+        
+        # 递归创建更深层的目录结构
+        create_nested_structure(dir_path, depth - 1, num_dirs, num_files)
+    
+    return base_path
+
+# 递归删除目录和文件
+def delete_nested_structure(base_path):
+    """递归删除目录和文件结构"""
+    if not os.path.exists(base_path):
+        return
+    
+    for item in os.listdir(base_path):
+        item_path = os.path.join(base_path, item)
+        if os.path.isdir(item_path):
+            delete_nested_structure(item_path)
+        else:
+            os.remove(item_path)
+    
+    os.rmdir(base_path)
+
+# 多个线程同时调用create_nested_structure函数去创建目录和文件
+def threaded_create_structure(base_path, depth, num_threads=5, num_dirs=10, num_files=1000):
+    """使用多线程创建目录和文件结构"""
+    with ThreadPoolExecutor(max_workers=num_threads) as executor:
+        futures = []
+        for i in range(num_threads):
+            thread_base_path = f"{base_path}_thread_{i}"
+            futures.append(executor.submit(create_nested_structure, thread_base_path, depth, num_dirs, num_files))
+        
+        for future in futures:
+            future.result()  # 等待所有线程完成
+
+def main():
+    print("start test......")
+
+    # 解析命令行参数
+    parser = argparse.ArgumentParser(description="Test script for creating and deleting nested directory structures.")
+    parser.add_argument('--path', type=str, default='test_structure', help='Base path for the nested structure')
+    parser.add_argument('--depth', type=int, default=3, help='Depth of the nested structure')
+    parser.add_argument('--threads', type=int, default=5, help='Number of threads to use for creating structure')
+    args = parser.parse_args()
+
+    
+    try:
+        # 运行各种测试
+        base_path = args.path
+        depth = args.depth
+        print(f"Creating nested structure at {base_path} with depth {depth}...")
+
+        create_nested_structure(base_path, depth, num_dirs=5, num_files=1000)
+        
+        print("\ntest finish!")
+        
+    finally:
+        # delete_nested_structure(base_path)
+        # print("\ncleanup completed!")
+        pass
+
+if __name__ == "__main__":
+    main()

src/client/vfs/meta/v2/rpc.h

@@ -21,6 +21,7 @@
 #include <glog/logging.h>
 #include <json/config.h>
 
+#include <cstdint>
 #include <map>
 #include <memory>
 #include <string>
@@ -31,6 +32,7 @@
 #include "dingofs/error.pb.h"
 #include "dingofs/mdsv2.pb.h"
 #include "fmt/core.h"
+#include "mdsv2/common/helper.h"
 #include "utils/concurrent/concurrent.h"
 
 namespace dingofs {
@@ -152,30 +154,33 @@ Status RPC::SendRequest(const EndPoint& endpoint,
     cntl.set_timeout_ms(FLAGS_rpc_timeout_ms);
     cntl.set_log_id(butil::fast_rand());
 
+    uint64_t start_us = mdsv2::Helper::TimestampUs();
     channel->CallMethod(method, &cntl, &request, &response, nullptr);
+    uint64_t elapsed_us = mdsv2::Helper::TimestampUs() - start_us;
     if (cntl.Failed()) {
-      LOG(ERROR) << fmt::format("[rpc][{}][{}] fail, {} {} {} request({}).",
-                                EndPointToStr(endpoint), api_name,
-                                cntl.log_id(), cntl.ErrorCode(),
-                                cntl.ErrorText(), request.ShortDebugString());
+      LOG(ERROR) << fmt::format(
+          "[rpc][{}][{}][{}us] fail, {} {} {} request({}).",
+          EndPointToStr(endpoint), api_name, elapsed_us, cntl.log_id(),
+          cntl.ErrorCode(), cntl.ErrorText(), request.ShortDebugString());
       DeleteChannel(endpoint);
       return Status::NetError(cntl.ErrorCode(), cntl.ErrorText());
     }
 
     if (response.error().errcode() == pb::error::OK) {
       LOG(INFO) << fmt::format(
-          "[rpc][{}][{}] success, request({}) response({}).",
-          EndPointToStr(endpoint), api_name, request.ShortDebugString(),
-          response.ShortDebugString());
+          "[rpc][{}][{}][{}us] success, request({}) response({}).",
+          EndPointToStr(endpoint), api_name, elapsed_us,
+          request.ShortDebugString(), response.ShortDebugString());
       return Status();
     }
 
     ++retry_count;
 
     LOG(ERROR) << fmt::format(
-        "[rpc][{}][{}] fail, request({}) retry_count({}) error({} {}).",
-        EndPointToStr(endpoint), api_name, request.ShortDebugString(),
-        retry_count, pb::error::Errno_Name(response.error().errcode()),
+        "[rpc][{}][{}][{}us] fail, request({}) retry_count({}) error({} {}).",
+        EndPointToStr(endpoint), api_name, elapsed_us,
+        request.ShortDebugString(), retry_count,
+        pb::error::Errno_Name(response.error().errcode()),
         response.error().errmsg());
 
     // the errno of need retry

src/mdsv2/common/constant.h

@@ -31,6 +31,9 @@ const uint32_t kSetAttrNlink = 1 << 7;
 
 const int kEmptyDirMinLinkNum = 2;
 
+const uint64_t kRootIno = 1;
+const uint64_t kRootParentIno = 0;
+
 }  // namespace mdsv2
 }  // namespace dingofs
 

src/mdsv2/filesystem/filesystem.cc

@@ -55,9 +55,6 @@ static const int64_t kInoTableId = 1001;
 static const int64_t kInoBatchSize = 32;
 static const int64_t kInoStartId = 100000;
 
-static const uint64_t kRootIno = 1;
-static const uint64_t kRootParentIno = 0;
-
 static const std::string kFsTableName = "dingofs";
 
 static const std::string kStatsName = ".stats";

src/mdsv2/filesystem/fs_utils.cc

@@ -14,25 +14,29 @@
 
 #include "mdsv2/filesystem/fs_utils.h"
 
-#include <fmt/format.h>
-#include <glog/logging.h>
-
 #include <cstdint>
 #include <map>
+#include <string>
+#include <vector>
 
 #include "dingofs/mdsv2.pb.h"
-#include "fmt/core.h"
+#include "fmt/format.h"
+#include "glog/logging.h"
 #include "mdsv2/common/codec.h"
+#include "mdsv2/common/constant.h"
 #include "mdsv2/common/helper.h"
 #include "mdsv2/common/logging.h"
 #include "mdsv2/common/status.h"
+#include "mdsv2/common/type.h"
 #include "nlohmann/json.hpp"
 
 namespace dingofs {
 namespace mdsv2 {
 
 DEFINE_int32(fs_scan_batch_size, 10000, "fs scan batch size");
 
+static const uint32_t kBatchGetSize = 1000;
+
 void FreeFsTree(FsTreeNode* root) {
   if (root == nullptr) {
     return;
@@ -237,5 +241,139 @@ std::string FsUtils::GenFsTreeJsonString() {
   return doc.dump();
 }
 
+Status FsUtils::GenRootDirJsonString(std::string& result) {
+  const uint32_t fs_id = fs_info_.fs_id();
+
+  auto txn = kv_storage_->NewTxn();
+
+  std::string value;
+  auto status = txn->Get(MetaCodec::EncodeInodeKey(fs_id, kRootIno), value);
+  if (!status.ok()) {
+    DINGO_LOG(ERROR) << fmt::format("[fsutils] get root inode fail, {}.", status.error_str());
+    return status;
+  }
+
+  auto attr = MetaCodec::DecodeInodeValue(value);
+
+  nlohmann::json doc = nlohmann::json::array();
+
+  nlohmann::json item;
+  item["ino"] = attr.ino();
+  item["name"] = "/";
+  item["type"] = "directory";
+  if (fs_info_.partition_policy().type() == pb::mdsv2::PartitionType::MONOLITHIC_PARTITION) {
+    item["node"] = fs_info_.partition_policy().mono().mds_id();
+  } else {
+    item["node"] = hash_router_->GetMDS(attr.ino());
+  }
+
+  // mode,nlink,uid,gid,size,ctime,mtime,atime
+  item["description"] =
+      fmt::format("{},{}/{},{},{},{},{},{},{},{}", attr.version(), attr.mode(), Helper::FsModeToString(attr.mode()),
+                  attr.nlink(), attr.uid(), attr.gid(), attr.length(), FormatTime(attr.ctime()),
+                  FormatTime(attr.mtime()), FormatTime(attr.atime()));
+
+  doc.push_back(item);
+
+  result = doc.dump();
+  return Status::OK();
+}
+
+Status FsUtils::GenDirJsonString(Ino parent, std::string& result) {
+  if (parent == kRootParentIno) {
+    return GenRootDirJsonString(result);
+  }
+
+  const uint32_t fs_id = fs_info_.fs_id();
+  Range range;
+  MetaCodec::EncodeDentryRange(fs_id, parent, range.start_key, range.end_key);
+
+  auto txn = kv_storage_->NewTxn();
+
+  std::vector<KeyValue> kvs;
+  auto status = txn->Scan(range, 100000, kvs);
+  if (!status.ok()) {
+    DINGO_LOG(ERROR) << fmt::format("[fsutils] scan dentry table fail, {}.", status.error_str());
+    return status;
+  }
+
+  if (kvs.empty()) {
+    return Status(pb::error::ENOT_FOUND, "not found kv");
+  }
+
+  // add child dentry
+  std::map<Ino, DentryType> dentries;
+  for (size_t i = 1; i < kvs.size(); ++i) {
+    const auto& kv = kvs.at(i);
+    auto dentry = MetaCodec::DecodeDentryValue(kv.value);
+    dentries.insert(std::make_pair(dentry.ino(), dentry));
+  }
+
+  // batch get inode attrs
+  std::map<Ino, AttrType> attrs;
+  uint32_t count = 0;
+  std::vector<std::string> keys;
+  keys.reserve(kBatchGetSize);
+  for (auto& [ino, dentry] : dentries) {
+    keys.push_back(MetaCodec::EncodeInodeKey(fs_id, ino));
+
+    if (++count == dentries.size() || keys.size() == kBatchGetSize) {
+      std::vector<KeyValue> inode_kvs;
+      status = txn->BatchGet(keys, inode_kvs);
+      if (!status.ok()) {
+        DINGO_LOG(ERROR) << fmt::format("[fsutils] batch get inode attrs fail, {}.", status.error_str());
+        return status;
+      }
+
+      if (inode_kvs.size() != keys.size()) {
+        DINGO_LOG(WARNING) << fmt::format("[fsutils] batch get inode attrs size({}) not match keys size({}).",
+                                          inode_kvs.size(), keys.size());
+      }
+
+      for (const auto& kv : inode_kvs) {
+        auto attr = MetaCodec::DecodeInodeValue(kv.value);
+        attrs.insert(std::make_pair(attr.ino(), attr));
+      }
+
+      keys.clear();
+    }
+  }
+
+  // gen json
+  nlohmann::json doc = nlohmann::json::array();
+  for (auto& [ino, dentry] : dentries) {
+    auto it = attrs.find(ino);
+    if (it == attrs.end()) {
+      DINGO_LOG(ERROR) << fmt::format("[fsutils] not found attr for dentry({}/{})", dentry.ino(), dentry.name());
+      continue;
+    }
+
+    const auto& attr = it->second;
+
+    nlohmann::json item;
+    item["ino"] = dentry.ino();
+    item["name"] = dentry.name();
+    item["type"] = dentry.type() == pb::mdsv2::FileType::DIRECTORY ? "directory" : "file";
+    if (fs_info_.partition_policy().type() == pb::mdsv2::PartitionType::MONOLITHIC_PARTITION) {
+      item["node"] = fs_info_.partition_policy().mono().mds_id();
+    } else {
+      item["node"] = (dentry.type() == pb::mdsv2::FileType::DIRECTORY) ? hash_router_->GetMDS(dentry.ino())
+                                                                       : hash_router_->GetMDS(dentry.parent());
+    }
+
+    // mode,nlink,uid,gid,size,ctime,mtime,atime
+    item["description"] =
+        fmt::format("{},{}/{},{},{},{},{},{},{},{}", attr.version(), attr.mode(), Helper::FsModeToString(attr.mode()),
+                    attr.nlink(), attr.uid(), attr.gid(), attr.length(), FormatTime(attr.ctime()),
+                    FormatTime(attr.mtime()), FormatTime(attr.atime()));
+
+    doc.push_back(item);
+  }
+
+  result = doc.dump();
+
+  return Status::OK();
+}
+
 }  // namespace mdsv2
 }  // namespace dingofs

src/mdsv2/filesystem/fs_utils.h

@@ -74,9 +74,11 @@ class FsUtils {
 
   FsTreeNode* GenFsTree(uint32_t fs_id);
   std::string GenFsTreeJsonString();
+  Status GenDirJsonString(Ino parent, std::string& result);
 
  private:
   void GenFsTreeJson(FsTreeNode* node, nlohmann::json& doc);
+  Status GenRootDirJsonString(std::string& result);
 
   FsInfoType fs_info_;
   KVStorageSPtr kv_storage_;