Add infinity admin API (#1997)

### What problem does this PR solve? _Briefly describe what this PR aims to solve. Include background context that will help reviewers understand the purpose of the PR._ ### Type of change - [x] New Feature (non-breaking change which adds functionality) Signed-off-by: Jin Hai <[email protected]>
infiniflow · Oct 9, 2024 · a1940fc · a1940fc
1 parent b604010
commit a1940fc
Show file tree

Hide file tree

Showing 4 changed files with 278 additions and 4 deletions.
diff --git a/example/http/cluster_admin.sh b/example/http/cluster_admin.sh
@@ -1,3 +1,3 @@
 curl --request GET \
-    --url http://localhost:23822/admin/node/current \
+    --url http://localhost:23820/admin/node/current \
     --header 'accept: application/json'
diff --git a/src/main/infinity.cpp b/src/main/infinity.cpp
@@ -48,6 +48,7 @@ import delete_statement;
 import optimize_statement;
 import alter_statement;
 import statement_common;
+import admin_statement;
 
 import create_schema_info;
 import drop_schema_info;
@@ -1127,4 +1128,248 @@ QueryResult Infinity::Cleanup() {
     return result;
 }
 
+QueryResult Infinity::AdminShowCatalogs() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kListCatalogs;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowCatalog(i64 index) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kShowCatalog;
+    admin_statement->catalog_file_index_ = index;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowLogs() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kListLogFiles;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowLog(i64 index) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kShowLogFile;
+    admin_statement->log_file_index_ = index;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowConfigs() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kListConfigs;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowVariables() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kListVariables;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowVariable(String var_name) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kShowVariable;
+    ToLower(var_name);
+    admin_statement->variable_name_ = var_name;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowNodes() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kListNodes;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowNode(String node_name) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kShowNode;
+    ToLower(node_name);
+    admin_statement->node_name_ = node_name;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminShowCurrentNode() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kShowCurrentNode;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminSetAdmin() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kSetRole;
+    admin_statement->admin_node_role_ = AdminNodeRole::kAdmin;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminSetStandalone() {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kSetRole;
+    admin_statement->admin_node_role_ = AdminNodeRole::kStandalone;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminSetLeader(String node_name) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kSetRole;
+    admin_statement->admin_node_role_ = AdminNodeRole::kLeader;
+    ToLower(node_name);
+    admin_statement->node_name_ = node_name;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminSetFollower(String node_name, const String& leader_address) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kSetRole;
+    admin_statement->admin_node_role_ = AdminNodeRole::kFollower;
+    admin_statement->leader_address_ = leader_address;
+    ToLower(node_name);
+    admin_statement->node_name_ = node_name;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
+QueryResult Infinity::AdminSetLearner(String node_name, const String& leader_address) {
+    auto query_context_ptr = MakeUnique<QueryContext>(session_.get());
+    query_context_ptr->Init(InfinityContext::instance().config(),
+                            InfinityContext::instance().task_scheduler(),
+                            InfinityContext::instance().storage(),
+                            InfinityContext::instance().resource_manager(),
+                            InfinityContext::instance().session_manager(),
+                            InfinityContext::instance().persistence_manager());
+
+    auto admin_statement = MakeUnique<AdminStatement>();
+    admin_statement->admin_type_ = AdminStmtType::kSetRole;
+    admin_statement->admin_node_role_ = AdminNodeRole::kLearner;
+    admin_statement->leader_address_ = leader_address;
+    ToLower(node_name);
+    admin_statement->node_name_ = node_name;
+    QueryResult result = query_context_ptr->QueryStatement(admin_statement.get());
+    return result;
+}
+
 } // namespace infinity
diff --git a/src/main/infinity.cppm b/src/main/infinity.cppm
@@ -175,6 +175,23 @@ public:
 
     QueryResult Cleanup();
 
+    // Admin interface
+    QueryResult AdminShowCatalogs();
+    QueryResult AdminShowCatalog(i64 index);
+    QueryResult AdminShowLogs();
+    QueryResult AdminShowLog(i64 index);
+    QueryResult AdminShowConfigs();
+    QueryResult AdminShowVariables();
+    QueryResult AdminShowVariable(String var_name);
+    QueryResult AdminShowNodes();
+    QueryResult AdminShowNode(String var_name);
+    QueryResult AdminShowCurrentNode();
+    QueryResult AdminSetAdmin();
+    QueryResult AdminSetStandalone();
+    QueryResult AdminSetLeader(String node_name);
+    QueryResult AdminSetFollower(String node_name, const String& leader_address);
+    QueryResult AdminSetLearner(String node_name, const String& leader_address);
+
 private:
     SharedPtr<BaseSession> session_{};
 };

diff --git a/src/network/http_server.cpp b/src/network/http_server.cpp
@@ -3472,16 +3472,28 @@ class ShowQueriesHandler final : public HttpRequestHandler {
 class ShowCurrentNodeHandler final : public HttpRequestHandler {
 public:
     SharedPtr<OutgoingResponse> handle(const SharedPtr<IncomingRequest> &request) final {
+
         auto infinity = Infinity::RemoteConnect();
         DeferFn defer_fn([&]() { infinity->RemoteDisconnect(); });
 
+        auto result = infinity->AdminShowCurrentNode();
+
         nlohmann::json json_response;
         HTTPStatus http_status;
 
-        http_status = HTTPStatus::CODE_200;
-        json_response["error_code"] = ErrorCode::kOk;
-        json_response["node_role"] = ToString(InfinityContext::instance().GetServerRole());
+        if (result.IsOk()) {
+            json_response["error_code"] = 0;
+            DataBlock *data_block = result.result_table_->GetDataBlockById(0).get();
+            Value value = data_block->GetValue(1, 0);
+            const String &variable_value = value.ToString();
+            json_response["role"] = variable_value;
 
+            http_status = HTTPStatus::CODE_200;
+        } else {
+            json_response["error_code"] = result.ErrorCode();
+            json_response["error_message"] = result.ErrorMsg();
+            http_status = HTTPStatus::CODE_500;
+        }
         return ResponseFactory::createResponse(http_status, json_response.dump());
     }
 };