feat: support table-level slow query on meta server (#314)

include/dsn/dist/replication/duplication_common.h

@@ -38,6 +38,8 @@ typedef rpc_holder<duplication_status_change_request, duplication_status_change_
 typedef rpc_holder<duplication_add_request, duplication_add_response> duplication_add_rpc;
 typedef rpc_holder<duplication_query_request, duplication_query_response> duplication_query_rpc;
 typedef rpc_holder<duplication_sync_request, duplication_sync_response> duplication_sync_rpc;
+typedef rpc_holder<configuration_update_app_env_request, configuration_update_app_env_response>
+    update_app_env_rpc;
 
 typedef int32_t dupid_t;
 

include/dsn/dist/replication/replication_ddl_client.h

@@ -172,9 +172,10 @@ class replication_ddl_client
 
     dsn::error_code get_app_envs(const std::string &app_name,
                                  std::map<std::string, std::string> &envs);
-    dsn::error_code set_app_envs(const std::string &app_name,
-                                 const std::vector<std::string> &keys,
-                                 const std::vector<std::string> &values);
+    error_with<configuration_update_app_env_response>
+    set_app_envs(const std::string &app_name,
+                 const std::vector<std::string> &keys,
+                 const std::vector<std::string> &values);
     dsn::error_code del_app_envs(const std::string &app_name, const std::vector<std::string> &keys);
     // precondition:
     //  -- if clear_all = true, just ignore prefix

src/dist/replication/ddl_lib/replication_ddl_client.cpp

@@ -1496,36 +1496,17 @@ dsn::error_code replication_ddl_client::get_app_envs(const std::string &app_name
     return dsn::ERR_OBJECT_NOT_FOUND;
 }
 
-::dsn::error_code replication_ddl_client::set_app_envs(const std::string &app_name,
-                                                       const std::vector<std::string> &keys,
-                                                       const std::vector<std::string> &values)
+error_with<configuration_update_app_env_response>
+replication_ddl_client::set_app_envs(const std::string &app_name,
+                                     const std::vector<std::string> &keys,
+                                     const std::vector<std::string> &values)
 {
-    std::shared_ptr<configuration_update_app_env_request> req =
-        std::make_shared<configuration_update_app_env_request>();
+    auto req = make_unique<configuration_update_app_env_request>();
     req->__set_app_name(app_name);
-    req->__set_op(app_env_operation::type::APP_ENV_OP_SET);
     req->__set_keys(keys);
     req->__set_values(values);
-
-    auto resp_task = request_meta<configuration_update_app_env_request>(RPC_CM_UPDATE_APP_ENV, req);
-    resp_task->wait();
-
-    if (resp_task->error() != ERR_OK) {
-        return resp_task->error();
-    }
-    configuration_update_app_env_response response;
-    dsn::unmarshall(resp_task->get_response(), response);
-    if (response.err != ERR_OK) {
-        return response.err;
-    } else {
-        std::cout << "set app envs succeed" << std::endl;
-        if (!response.hint_message.empty()) {
-            std::cout << "=============================" << std::endl;
-            std::cout << response.hint_message << std::endl;
-            std::cout << "=============================" << std::endl;
-        }
-    }
-    return ERR_OK;
+    req->__set_op(app_env_operation::type::APP_ENV_OP_SET);
+    return call_rpc_sync(update_app_env_rpc(std::move(req), RPC_CM_UPDATE_APP_ENV));
 }
 
 ::dsn::error_code replication_ddl_client::del_app_envs(const std::string &app_name,

src/dist/replication/meta_server/server_state.cpp

@@ -56,6 +56,10 @@ namespace replication {
 
 static const char *lock_state = "lock";
 static const char *unlock_state = "unlock";
+// env name of slow query
+static const std::string ENV_SLOW_QUERY_THRESHOLD("replica.slow_query_threshold");
+// min value for slow query threshold, less than this value will be refused
+static const uint64_t MIN_SLOW_QUERY_THRESHOLD_MS = 20;
 
 server_state::server_state()
     : _meta_svc(nullptr),
@@ -2610,6 +2614,18 @@ void server_state::set_app_envs(const app_env_rpc &env_rpc)
 
     std::ostringstream os;
     for (int i = 0; i < keys.size(); i++) {
+        // check whether if slow query threshold is abnormal
+        if (0 == keys[i].compare(ENV_SLOW_QUERY_THRESHOLD)) {
+            uint64_t threshold = 0;
+            if (!dsn::buf2uint64(values[i], threshold) || threshold < MIN_SLOW_QUERY_THRESHOLD_MS) {
+                dwarn("{}={} is invalid.", keys[i].c_str(), threshold);
+                env_rpc.response().err = ERR_INVALID_PARAMETERS;
+                env_rpc.response().hint_message = fmt::format(
+                    "slow query threshold must be >= {}ms", MIN_SLOW_QUERY_THRESHOLD_MS);
+                return;
+            }
+        }
+
         if (i != 0)
             os << ", ";
         os << keys[i] << "=" << values[i];

src/dist/replication/test/meta_test/unit_test/meta_app_envs_test.cpp

@@ -0,0 +1,85 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2015 Microsoft Corporation
+ *
+ * -=- Robust Distributed System Nucleus (rDSN) -=-
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+#include "meta_test_base.h"
+
+namespace dsn {
+namespace replication {
+class meta_app_envs_test : public meta_test_base
+{
+public:
+    meta_app_envs_test() {}
+
+    void SetUp() override
+    {
+        meta_test_base::SetUp();
+        create_app(app_name);
+    }
+
+    error_code update_app_envs(const std::vector<std::string> &env_keys,
+                               const std::vector<std::string> &env_vals)
+    {
+        auto req = make_unique<configuration_update_app_env_request>();
+        req->__set_app_name(std::move(app_name));
+        req->__set_op(std::move(app_env_operation::type::APP_ENV_OP_SET));
+        req->__set_keys(env_keys);
+        req->__set_values(env_vals);
+
+        app_env_rpc rpc(std::move(req), RPC_CM_UPDATE_APP_ENV); // don't need reply
+        _ss->set_app_envs(rpc);
+        _ss->wait_all_task();
+        return rpc.response().err;
+    }
+
+    const std::string app_name = "test_app_env";
+    const std::string env_slow_query_threshold = "replica.slow_query_threshold";
+};
+
+TEST_F(meta_app_envs_test, set_slow_query_threshold)
+{
+    auto app = find_app(app_name);
+
+    struct test_case
+    {
+        error_code err;
+        std::string env_value;
+        std::string expect_env_value;
+    } tests[] = {{ERR_OK, "30", "30"},
+                 {ERR_OK, "21", "21"},
+                 {ERR_OK, "20", "20"},
+                 {ERR_INVALID_PARAMETERS, "19", "20"},
+                 {ERR_INVALID_PARAMETERS, "10", "20"},
+                 {ERR_INVALID_PARAMETERS, "0", "20"}};
+
+    for (auto test : tests) {
+        error_code err = update_app_envs({env_slow_query_threshold}, {test.env_value});
+
+        ASSERT_EQ(err, test.err);
+        ASSERT_EQ(app->envs.at(env_slow_query_threshold), test.expect_env_value);
+    }
+}
+} // namespace replication
+} // namespace dsn