feat: add check for app envs

src/dist/replication/common/replication_common.cpp

@@ -598,8 +598,6 @@ const std::string backup_restore_constant::BACKUP_ID("restore.backup_id");
 const std::string backup_restore_constant::SKIP_BAD_PARTITION("restore.skip_bad_partition");
 
 const std::string replica_envs::DENY_CLIENT_WRITE("replica.deny_client_write");
-const std::string replica_envs::WRITE_QPS_THROTTLING("replica.write_throttling");
-const std::string replica_envs::WRITE_SIZE_THROTTLING("replica.write_throttling_by_size");
 
 namespace cold_backup {
 std::string get_policy_path(const std::string &root, const std::string &policy_name)

src/dist/replication/common/replication_common.h

@@ -153,8 +153,6 @@ class replica_envs
 {
 public:
     static const std::string DENY_CLIENT_WRITE;
-    static const std::string WRITE_QPS_THROTTLING;
-    static const std::string WRITE_SIZE_THROTTLING;
 };
 
 namespace cold_backup {

src/dist/replication/lib/replica_throttle.cpp

@@ -52,10 +52,9 @@ bool replica::throttle_request(throttling_controller &controller,
 
 void replica::update_throttle_envs(const std::map<std::string, std::string> &envs)
 {
+    update_throttle_env_internal(envs, ENV_WRITE_QPS_THROTTLING, _write_qps_throttling_controller);
     update_throttle_env_internal(
-        envs, replica_envs::WRITE_QPS_THROTTLING, _write_qps_throttling_controller);
-    update_throttle_env_internal(
-        envs, replica_envs::WRITE_SIZE_THROTTLING, _write_size_throttling_controller);
+        envs, ENV_WRITE_SIZE_THROTTLING, _write_size_throttling_controller);
 }
 
 void replica::update_throttle_env_internal(const std::map<std::string, std::string> &envs,

src/dist/replication/meta_server/server_state.cpp

@@ -56,10 +56,6 @@ 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),
@@ -69,6 +65,7 @@ server_state::server_state()
       _ctrl_add_secondary_enable_flow_control(nullptr),
       _ctrl_add_secondary_max_count_for_one_node(nullptr)
 {
+    init_env_check_functions();
 }
 
 server_state::~server_state()
@@ -90,6 +87,86 @@ server_state::~server_state()
     }
 }
 
+void server_state::init_env_check_functions()
+{
+    env_check_functions[ENV_SLOW_QUERY_THRESHOLD] = std::bind(
+        &server_state::check_slow_query, this, std::placeholders::_1, std::placeholders::_2);
+    env_check_functions[ENV_WRITE_QPS_THROTTLING] = std::bind(
+        &server_state::check_write_throttling, this, std::placeholders::_1, std::placeholders::_2);
+    env_check_functions[ENV_WRITE_SIZE_THROTTLING] = std::bind(
+        &server_state::check_write_throttling, this, std::placeholders::_1, std::placeholders::_2);
+}
+
+bool server_state::check_slow_query(const std::string &env_value, std::string &hint_message)
+{
+    uint64_t threshold = 0;
+    if (!dsn::buf2uint64(env_value, threshold) || threshold < MIN_SLOW_QUERY_THRESHOLD_MS) {
+        hint_message =
+            fmt::format("Slow query threshold must be >= {}ms", MIN_SLOW_QUERY_THRESHOLD_MS);
+        return false;
+    }
+    return true;
+}
+
+bool server_state::check_write_throttling(const std::string &env_value, std::string &hint_message)
+{
+    std::vector<std::string> sargs;
+    utils::split_args(env_value.c_str(), sargs, ',');
+    if (sargs.empty()) {
+        hint_message = "The value shouldn't be empty";
+        return false;
+    }
+
+    // example for arg: 100K*delay*100 / 100M*reject*100
+    for (std::string &sarg : sargs) {
+        std::vector<std::string> sub_sargs;
+        utils::split_args(sarg.c_str(), sub_sargs, '*');
+        if (sub_sargs.size() != 3) {
+            hint_message = fmt::format("The field count of {} should be 3", sarg);
+            return false;
+        }
+
+        int64_t units = 0;
+        if (!sub_sargs[0].empty() &&
+            ('M' == *sub_sargs[0].rbegin() || 'K' == *sub_sargs[0].rbegin())) {
+            sub_sargs[0].pop_back();
+        }
+        if (!buf2int64(sub_sargs[0], units) || units < 0) {
+            hint_message = fmt::format("{} should be non-negative int", sub_sargs[0]);
+            return false;
+        }
+
+        if (sub_sargs[1] != "delay" && sub_sargs[1] != "reject") {
+            hint_message = fmt::format("{} should be \"delay\" or \"reject\"", sub_sargs[1]);
+            return false;
+        }
+
+        int64_t ms = 0;
+        if (!buf2int64(sub_sargs[2], ms) || ms < 0) {
+            hint_message = fmt::format("{} should be non-negative int", sub_sargs[2]);
+            return false;
+        };
+    }
+
+    return true;
+}
+
+bool server_state::check_app_envs(const std::string &key,
+                                  const std::string &value,
+                                  std::string &hint_message)
+{
+
+    auto func_iter = env_check_functions.find(key);
+    if (func_iter != env_check_functions.end()) {
+        if (!func_iter->second(value, hint_message)) {
+            dwarn("{}={} is invalid.", key.c_str(), value.c_str());
+            return false;
+        }
+    }
+
+    return true;
+}
+
 void server_state::register_cli_commands()
 {
     _cli_dump_handle = dsn::command_manager::instance().register_app_command(
@@ -2617,20 +2694,14 @@ 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 << ", ";
+
+        if (!check_app_envs(keys[i], values[i], env_rpc.response().hint_message)) {
+            env_rpc.response().err = ERR_INVALID_PARAMETERS;
+            return;
+        }
+
         os << keys[i] << "=" << values[i];
     }
     ddebug("set app envs for app(%s) from remote(%s): kvs = {%s}",

src/dist/replication/meta_server/server_state.h

@@ -297,6 +297,12 @@ class server_state
     void process_one_partition(std::shared_ptr<app_state> &app);
     void transition_staging_state(std::shared_ptr<app_state> &app);
 
+    void init_env_check_functions();
+    bool check_slow_query(const std::string &env_value, std::string &hint_message);
+    bool check_write_throttling(const std::string &env_value, std::string &hint_message);
+    bool
+    check_app_envs(const std::string &key, const std::string &value, std::string &hint_message);
+
 private:
     friend class replication_checker;
     friend class test::test_checker;
@@ -340,6 +346,9 @@ class server_state
     perf_counter_wrapper _recent_update_config_count;
     perf_counter_wrapper _recent_partition_change_unwritable_count;
     perf_counter_wrapper _recent_partition_change_writable_count;
+
+    std::map<std::string, std::function<bool(const std::string &, std::string &)>>
+        env_check_functions;
 };
 
 } // namespace replication

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

@@ -39,32 +39,92 @@ class meta_app_envs_test : public meta_test_base
         create_app(app_name);
     }
 
+    void TearDown() override { drop_app(app_name); }
+
     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)
 {
+    struct test_case
+    {
+        error_code err;
+        std::string env_value;
+        std::string hint;
+        std::string expect_value;
+    } tests[] = {{ERR_OK, "30", "", "30"},
+                 {ERR_OK, "20", "", "20"},
+                 {ERR_INVALID_PARAMETERS, "19", "Slow query threshold must be >= 20ms", "20"},
+                 {ERR_INVALID_PARAMETERS, "0", "Slow query threshold must be >= 20ms", "20"}};
+
+    const std::string env_slow_query_threshold = "replica.slow_query_threshold";
     auto app = find_app(app_name);
+    for (auto test : tests) {
+        configuration_update_app_env_response response =
+            update_app_envs(app_name, {env_slow_query_threshold}, {test.env_value});
 
+        ASSERT_EQ(response.err, test.err);
+        ASSERT_EQ(response.hint_message, test.hint);
+        ASSERT_EQ(app->envs.at(env_slow_query_threshold), test.expect_value);
+    }
+}
+
+TEST_F(meta_app_envs_test, set_write_throttling)
+{
     struct test_case
     {
-        error_code err;
+        std::string env_key;
         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"}};
+        error_code err;
+        std::string hint;
+        std::string expect_value;
+    } tests[] = {
+        {"replica.write_throttling", "100*delay*100", ERR_OK, "", "100*delay*100"},
+        {"replica.write_throttling", "20K*delay*100", ERR_OK, "", "20K*delay*100"},
+        {"replica.write_throttling", "20M*delay*100", ERR_OK, "", "20M*delay*100"},
+        {"replica.write_throttling",
+         "20A*delay*100",
+         ERR_INVALID_PARAMETERS,
+         "20A should be non-negative int",
+         "20M*delay*100"},
+        {"replica.write_throttling",
+         "-20*delay*100",
+         ERR_INVALID_PARAMETERS,
+         "-20 should be non-negative int",
+         "20M*delay*100"},
+        {"replica.write_throttling",
+         "",
+         ERR_INVALID_PARAMETERS,
+         "The value shouldn't be empty",
+         "20M*delay*100"},
+        {"replica.write_throttling",
+         "20A*delay",
+         ERR_INVALID_PARAMETERS,
+         "The field count of 20A*delay should be 3",
+         "20M*delay*100"},
+        {"replica.write_throttling",
+         "20K*pass*100",
+         ERR_INVALID_PARAMETERS,
+         "pass should be \"delay\" or \"reject\"",
+         "20M*delay*100"},
+        {"replica.write_throttling",
+         "20K*delay*-100",
+         ERR_INVALID_PARAMETERS,
+         "-100 should be non-negative int",
+         "20M*delay*100"},
+        {"replica.write_throttling", "20M*reject*100", ERR_OK, "", "20M*reject*100"},
+        {"replica.write_throttling_by_size", "300*delay*100", ERR_OK, "", "300*delay*100"}};
 
+    auto app = find_app(app_name);
     for (auto test : tests) {
-        error_code err = update_app_envs(app_name, {env_slow_query_threshold}, {test.env_value});
+        configuration_update_app_env_response response =
+            update_app_envs(app_name, {test.env_key}, {test.env_value});
 
-        ASSERT_EQ(err, test.err);
-        ASSERT_EQ(app->envs.at(env_slow_query_threshold), test.expect_env_value);
+        ASSERT_EQ(response.err, test.err);
+        ASSERT_EQ(response.hint_message, test.hint);
+        ASSERT_EQ(app->envs.at(test.env_key), test.expect_value);
     }
 }
+
 } // namespace replication
 } // namespace dsn

src/dist/replication/test/meta_test/unit_test/meta_test_base.h

@@ -101,9 +101,9 @@ class meta_test_base : public testing::Test
         ASSERT_TRUE(_ss->spin_wait_staging(30));
     }
 
-    error_code update_app_envs(const std::string &app_name,
-                               const std::vector<std::string> &env_keys,
-                               const std::vector<std::string> &env_vals)
+    configuration_update_app_env_response update_app_envs(const std::string &app_name,
+                                                          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));
@@ -114,7 +114,7 @@ class meta_test_base : public testing::Test
         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;
+        return rpc.response();
     }
 
     std::shared_ptr<app_state> find_app(const std::string &name) { return _ss->get_app(name); }