feat(bulk_load): add query compaction status http interface

src/replica/replica.cpp

@@ -443,6 +443,33 @@ std::string replica::query_compact_state() const
     return _app->query_compact_state();
 }
 
+const char *manual_compaction_status_to_string(manual_compaction_status status)
+{
+    switch (status) {
+    case CompactionFinish:
+        return "CompactionFinish";
+    case CompactionRunning:
+        return "CompactionRunning";
+    case CompactionQueue:
+        return "CompactionQueue";
+    default:
+        dassert(false, "");
+    }
+    return "wrongCompactionStatus";
+}
+
+manual_compaction_status replica::get_compact_status() const
+{
+    std::string compact_state = query_compact_state();
+    if (compact_state.find("recent start at") != std::string::npos) {
+        return CompactionRunning;
+    } else if (compact_state.find("recent enqueue at") != std::string::npos) {
+        return CompactionQueue;
+    } else {
+        return CompactionFinish;
+    }
+}
+
 // Replicas on the server which serves for the same table will share the same perf-counter.
 // For example counter `table.level.RPC_RRDB_RRDB_MULTI_PUT.latency(ns)@test_table` is shared by
 // all the replicas for `test_table`.

src/replica/replica.h

@@ -78,6 +78,14 @@ namespace test {
 class test_checker;
 }
 
+enum manual_compaction_status
+{
+    CompactionFinish = 0,
+    CompactionRunning,
+    CompactionQueue
+};
+const char *manual_compaction_status_to_string(manual_compaction_status status);
+
 class replica : public serverlet<replica>, public ref_counter, public replica_base
 {
 public:
@@ -392,8 +400,12 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
 
     void update_restore_progress(uint64_t f_size);
 
+    // Used for remote command
     std::string query_compact_state() const;
 
+    // Used for http interface
+    manual_compaction_status get_compact_status() const;
+
     void init_table_level_latency_counters();
 
     void on_detect_hotkey(const detect_hotkey_request &req, /*out*/ detect_hotkey_response &resp);

src/replica/replica_http_service.cpp

@@ -4,6 +4,7 @@
 
 #include <nlohmann/json.hpp>
 #include <fmt/format.h>
+#include <dsn/utility/output_utils.h>
 #include "replica_http_service.h"
 #include "duplication/duplication_sync_timer.h"
 
@@ -55,5 +56,46 @@ void replica_http_service::query_duplication_handler(const http_request &req, ht
     resp.body = json.dump();
 }
 
+void replica_http_service::query_compaction_handler(const http_request &req, http_response &resp)
+{
+    auto it = req.query_args.find("app_id");
+    if (it == req.query_args.end()) {
+        resp.body = "app_id should not be empty";
+        resp.status_code = http_status_code::bad_request;
+        return;
+    }
+
+    int32_t app_id = -1;
+    if (!buf2int32(it->second, app_id) || app_id < 0) {
+        resp.body = fmt::format("invalid app_id={}", it->second);
+        resp.status_code = http_status_code::bad_request;
+        return;
+    }
+
+    std::unordered_map<gpid, manual_compaction_status> partition_compaction_status;
+    _stub->query_app_compact_status(app_id, partition_compaction_status);
+
+    int32_t running_count = 0;
+    int32_t queue_count = 0;
+    int32_t finish_count = 0;
+    for (const auto &kv : partition_compaction_status) {
+        if (kv.second == CompactionRunning) {
+            running_count++;
+        } else if (kv.second == CompactionQueue) {
+            queue_count++;
+        } else if (kv.second == CompactionFinish) {
+            finish_count++;
+        }
+    }
+    dsn::utils::table_printer tp("status");
+    tp.add_row_name_and_data(manual_compaction_status_to_string(CompactionRunning), running_count);
+    tp.add_row_name_and_data(manual_compaction_status_to_string(CompactionQueue), queue_count);
+    tp.add_row_name_and_data(manual_compaction_status_to_string(CompactionFinish), finish_count);
+    std::ostringstream out;
+    tp.output(out, dsn::utils::table_printer::output_format::kJsonCompact);
+    resp.body = out.str();
+    resp.status_code = http_status_code::ok;
+}
+
 } // namespace replication
 } // namespace dsn

src/replica/replica_http_service.h

@@ -20,11 +20,18 @@ class replica_http_service : public http_service
                                    std::placeholders::_1,
                                    std::placeholders::_2),
                          "ip:port/replica/duplication?appid=<appid>");
+        register_handler("compaction",
+                         std::bind(&replica_http_service::query_compaction_handler,
+                                   this,
+                                   std::placeholders::_1,
+                                   std::placeholders::_2),
+                         "ip:port/replica/compaction?app_id=<app_id>");
     }
 
     std::string path() const override { return "replica"; }
 
     void query_duplication_handler(const http_request &req, http_response &resp);
+    void query_compaction_handler(const http_request &req, http_response &resp);
 
 private:
     replica_stub *_stub;

src/replica/replica_stub.cpp

@@ -2808,5 +2808,17 @@ void replica_stub::on_detect_hotkey(detect_hotkey_rpc rpc)
         response.err_hint = fmt::format("not find the replica {} \n", request.pid);
     }
 }
+
+void replica_stub::query_app_compact_status(
+    int32_t app_id, std::unordered_map<gpid, manual_compaction_status> &status)
+{
+    zauto_read_lock l(_replicas_lock);
+    for (auto it = _replicas.begin(); it != _replicas.end(); ++it) {
+        if (it->first.get_app_id() == app_id) {
+            status[it->first] = it->second->get_compact_status();
+        }
+    }
+}
+
 } // namespace replication
 } // namespace dsn

src/replica/replica_stub.h

@@ -217,6 +217,11 @@ class replica_stub : public serverlet<replica_stub>, public ref_counter
     void on_query_disk_info(query_disk_info_rpc rpc);
     void on_disk_migrate(replica_disk_migrate_rpc rpc);
 
+    // query partitions compact status by app_id
+    void
+    query_app_compact_status(int32_t app_id,
+                             /*out*/ std::unordered_map<gpid, manual_compaction_status> &status);
+
 private:
     enum replica_node_state
     {

src/replica/test/replica_test.cpp

@@ -7,6 +7,7 @@
 #include <dsn/utility/fail_point.h>
 #include "replica_test_base.h"
 #include <dsn/utility/defer.h>
+#include "replica/replica_http_service.h"
 
 namespace dsn {
 namespace replication {
@@ -21,6 +22,7 @@ class replica_test : public replica_test_base
 public:
     void SetUp() override
     {
+        FLAGS_enable_http_server = false;
         stub->install_perf_counters();
         mock_app_info();
         _mock_replica = stub->generate_replica(_app_info, pid, partition_status::PS_PRIMARY, 1);
@@ -81,5 +83,35 @@ TEST_F(replica_test, backup_request_qps)
     ASSERT_GT(get_table_level_backup_request_qps(), 0);
 }
 
+TEST_F(replica_test, query_compaction_test)
+{
+    replica_http_service http_svc(stub.get());
+    struct query_compaction_test
+    {
+        std::string app_id;
+        http_status_code expected_code;
+        std::string expected_response_json;
+    } tests[] = {{"", http_status_code::bad_request, "app_id should not be empty"},
+                 {"xxx", http_status_code::bad_request, "invalid app_id=xxx"},
+                 {"2",
+                  http_status_code::ok,
+                  "{\"status\":{\"CompactionRunning\":\"0\",\"CompactionQueue\":"
+                  "\"0\",\"CompactionFinish\":\"1\"}}\n"},
+                 {"4",
+                  http_status_code::ok,
+                  "{\"status\":{\"CompactionRunning\":\"0\",\"CompactionQueue\":"
+                  "\"0\",\"CompactionFinish\":\"0\"}}\n"}};
+    for (const auto &test : tests) {
+        http_request req;
+        http_response resp;
+        if (!test.app_id.empty()) {
+            req.query_args["app_id"] = test.app_id;
+        }
+        http_svc.query_compaction_handler(req, resp);
+        ASSERT_EQ(resp.status_code, test.expected_code);
+        ASSERT_EQ(resp.body, test.expected_response_json);
+    }
+}
+
 } // namespace replication
 } // namespace dsn