feat: add table level latency perf counters

include/dsn/tool-api/task_spec.h

@@ -139,6 +139,8 @@ class message_ex;
 class admission_controller;
 typedef void (*task_rejection_handler)(task *, admission_controller *);
 
+extern std::vector<dsn::task_code> storage_rpc_req_codes;
+
 class task_spec : public extensible_object<task_spec, 4>
 {
 public:

src/core/core/task_spec.cpp

@@ -42,6 +42,8 @@ namespace dsn {
 
 constexpr int TASK_SPEC_STORE_CAPACITY = 512;
 
+std::vector<dsn::task_code> storage_rpc_req_codes;
+
 // A sequential storage maps task_code to task_spec.
 static std::array<std::unique_ptr<task_spec>, TASK_SPEC_STORE_CAPACITY> s_task_spec_store;
 
@@ -112,6 +114,9 @@ void task_spec::register_storage_task_code(task_code code,
     spec->rpc_request_is_write_operation = is_write_operation;
     spec->rpc_request_is_write_allow_batch = allow_batch;
     spec->rpc_request_is_write_idempotent = is_idempotent;
+    if (TASK_TYPE_RPC_REQUEST == type) {
+        storage_rpc_req_codes.push_back(code);
+    }
 }
 
 task_spec *task_spec::get(int code)

src/dist/replication/lib/replica.cpp

@@ -77,6 +77,15 @@ replica::replica(
     _counter_recent_write_throttling_reject_count.init_app_counter(
         "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
 
+    // init table level latency perf counters
+    for (auto code : storage_rpc_req_codes) {
+        counter_str = fmt::format("table.level.{}.latency(us)@{}", code, app.app_name);
+        _counters_table_level_latency[code].init_app_counter("eon.replica",
+                                                             counter_str.c_str(),
+                                                             COUNTER_TYPE_NUMBER_PERCENTILES,
+                                                             counter_str.c_str());
+    }
+
     if (need_restore) {
         // add an extra env for restore
         _extra_envs.insert(
@@ -163,8 +172,14 @@ void replica::on_client_read(dsn::message_ex *request)
         }
     }
 
+    uint64_t start_time_ns = dsn_now_ns();
     dassert(_app != nullptr, "");
     _app->on_request(request);
+
+    auto iter = _counters_table_level_latency.find(request->rpc_code());
+    if (iter != _counters_table_level_latency.end()) {
+        iter->second->set((dsn_now_ns() - start_time_ns) * 1e-3);
+    }
 }
 
 void replica::response_client_read(dsn::message_ex *request, error_code error)
@@ -284,6 +299,17 @@ void replica::execute_mutation(mutation_ptr &mu)
             init_prepare(next, false);
         }
     }
+
+    // update table level latency perf-counters for primary partition
+    if (partition_status::PS_PRIMARY == status()) {
+        uint64_t now_ns = dsn_now_ns();
+        for (auto update : mu->data.updates) {
+            auto iter = _counters_table_level_latency.find(update.code);
+            if (iter != _counters_table_level_latency.end()) {
+                iter->second->set((now_ns - update.start_time_ns) * 1e-3);
+            }
+        }
+    }
 }
 
 mutation_ptr replica::new_mutation(decree decree)

src/dist/replication/lib/replica.h

@@ -470,6 +470,7 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     perf_counter_wrapper _counter_private_log_size;
     perf_counter_wrapper _counter_recent_write_throttling_delay_count;
     perf_counter_wrapper _counter_recent_write_throttling_reject_count;
+    std::map<dsn::task_code, perf_counter_wrapper> _counters_table_level_latency;
 
     dsn::task_tracker _tracker;
     // the thread access checker