feat: add check for app envs

src/dist/replication/common/replication_common.cpp

@@ -600,6 +600,33 @@ const std::string backup_restore_constant::SKIP_BAD_PARTITION("restore.skip_bad_
 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");
+const uint64_t replica_envs::MIN_SLOW_QUERY_THRESHOLD_MS = 20;
+const std::string replica_envs::SLOW_QUERY_THRESHOLD("replica.slow_query_threshold");
+const std::string replica_envs::ROCKSDB_USAGE_SCENARIO("rocksdb.usage_scenario");
+const std::string replica_envs::TABLE_LEVEL_DEFAULT_TTL("default_ttl");
+const std::string MANUAL_COMPACT_PREFIX("manual_compact.");
+const std::string replica_envs::MANUAL_COMPACT_DISABLED(MANUAL_COMPACT_PREFIX + "disabled");
+const std::string replica_envs::MANUAL_COMPACT_MAX_CONCURRENT_RUNNING_COUNT(
+    MANUAL_COMPACT_PREFIX + "max_concurrent_running_count");
+const std::string MANUAL_COMPACT_ONCE_PREFIX(MANUAL_COMPACT_PREFIX + "once.");
+const std::string replica_envs::MANUAL_COMPACT_ONCE_TRIGGER_TIME(MANUAL_COMPACT_ONCE_PREFIX +
+                                                                 "trigger_time");
+const std::string replica_envs::MANUAL_COMPACT_ONCE_TARGET_LEVEL(MANUAL_COMPACT_ONCE_PREFIX +
+                                                                 "target_level");
+const std::string replica_envs::MANUAL_COMPACT_ONCE_BOTTOMMOST_LEVEL_COMPACTION(
+    MANUAL_COMPACT_ONCE_PREFIX + "bottommost_level_compaction");
+const std::string MANUAL_COMPACT_PERIODIC_PREFIX(MANUAL_COMPACT_PREFIX + "periodic.");
+const std::string replica_envs::MANUAL_COMPACT_PERIODIC_TRIGGER_TIME(
+    MANUAL_COMPACT_PERIODIC_PREFIX + "trigger_time");
+const std::string replica_envs::MANUAL_COMPACT_PERIODIC_TARGET_LEVEL(
+    MANUAL_COMPACT_PERIODIC_PREFIX + "target_level");
+const std::string replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION(
+    MANUAL_COMPACT_PERIODIC_PREFIX + "bottommost_level_compaction");
+const std::string
+    replica_envs::ROCKSDB_CHECKPOINT_RESERVE_MIN_COUNT("rocksdb.checkpoint.reserve_min_count");
+const std::string replica_envs::ROCKSDB_CHECKPOINT_RESERVE_TIME_SECONDS(
+    "rocksdb.checkpoint.reserve_time_seconds");
+const std::string replica_envs::BUSINESS_INFO("business.info");
 
 namespace cold_backup {
 std::string get_policy_path(const std::string &root, const std::string &policy_name)

src/dist/replication/common/replication_common.h

@@ -155,6 +155,21 @@ class replica_envs
     static const std::string DENY_CLIENT_WRITE;
     static const std::string WRITE_QPS_THROTTLING;
     static const std::string WRITE_SIZE_THROTTLING;
+    static const uint64_t MIN_SLOW_QUERY_THRESHOLD_MS;
+    static const std::string SLOW_QUERY_THRESHOLD;
+    static const std::string TABLE_LEVEL_DEFAULT_TTL;
+    static const std::string ROCKSDB_USAGE_SCENARIO;
+    static const std::string ROCKSDB_CHECKPOINT_RESERVE_MIN_COUNT;
+    static const std::string ROCKSDB_CHECKPOINT_RESERVE_TIME_SECONDS;
+    static const std::string MANUAL_COMPACT_DISABLED;
+    static const std::string MANUAL_COMPACT_MAX_CONCURRENT_RUNNING_COUNT;
+    static const std::string MANUAL_COMPACT_ONCE_TRIGGER_TIME;
+    static const std::string MANUAL_COMPACT_ONCE_TARGET_LEVEL;
+    static const std::string MANUAL_COMPACT_ONCE_BOTTOMMOST_LEVEL_COMPACTION;
+    static const std::string MANUAL_COMPACT_PERIODIC_TRIGGER_TIME;
+    static const std::string MANUAL_COMPACT_PERIODIC_TARGET_LEVEL;
+    static const std::string MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION;
+    static const std::string BUSINESS_INFO;
 };
 
 namespace cold_backup {

src/dist/replication/meta_server/server_state.cpp

@@ -56,10 +56,61 @@ 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;
+
+bool check_slow_query(const std::string &env_value, std::string &hint_message)
+{
+    uint64_t threshold = 0;
+    if (!dsn::buf2uint64(env_value, threshold) ||
+        threshold < replica_envs::MIN_SLOW_QUERY_THRESHOLD_MS) {
+        hint_message = fmt::format("Slow query threshold must be >= {}ms",
+                                   replica_envs::MIN_SLOW_QUERY_THRESHOLD_MS);
+        return false;
+    }
+    return true;
+}
+
+bool 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;
+}
 
 server_state::server_state()
     : _meta_svc(nullptr),
@@ -69,6 +120,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 +142,52 @@ server_state::~server_state()
     }
 }
 
+void server_state::init_env_check_functions()
+{
+    env_check_functions[replica_envs::SLOW_QUERY_THRESHOLD] =
+        std::bind(&check_slow_query, std::placeholders::_1, std::placeholders::_2);
+    env_check_functions[replica_envs::WRITE_QPS_THROTTLING] =
+        std::bind(&check_write_throttling, std::placeholders::_1, std::placeholders::_2);
+    env_check_functions[replica_envs::WRITE_SIZE_THROTTLING] =
+        std::bind(&check_write_throttling, std::placeholders::_1, std::placeholders::_2);
+    // TODO: not implemented
+    env_check_functions[replica_envs::BUSINESS_INFO] = nullptr;
+    env_check_functions[replica_envs::DENY_CLIENT_WRITE] = nullptr;
+    env_check_functions[replica_envs::TABLE_LEVEL_DEFAULT_TTL] = nullptr;
+    env_check_functions[replica_envs::ROCKSDB_USAGE_SCENARIO] = nullptr;
+    env_check_functions[replica_envs::ROCKSDB_CHECKPOINT_RESERVE_MIN_COUNT] = nullptr;
+    env_check_functions[replica_envs::ROCKSDB_CHECKPOINT_RESERVE_TIME_SECONDS] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_DISABLED] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_MAX_CONCURRENT_RUNNING_COUNT] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_ONCE_TRIGGER_TIME] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_ONCE_TARGET_LEVEL] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_ONCE_BOTTOMMOST_LEVEL_COMPACTION] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_PERIODIC_TRIGGER_TIME] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_PERIODIC_TARGET_LEVEL] = nullptr;
+    env_check_functions[replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION] =
+        nullptr;
+}
+
+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()) {
+        // check function == nullptr means no check
+        if (nullptr != func_iter->second && !func_iter->second(value, hint_message)) {
+            dwarn("{}={} is invalid.", key.c_str(), value.c_str());
+            return false;
+        }
+
+        return true;
+    }
+
+    hint_message = fmt::format("The env of {} is not supported", key);
+    return false;
+}
+
 void server_state::register_cli_commands()
 {
     _cli_dump_handle = dsn::command_manager::instance().register_app_command(
@@ -2627,20 +2725,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,10 @@ 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_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 +344,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,103 @@ 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)
+TEST_F(meta_app_envs_test, update_app_envs_test)
 {
-    auto app = find_app(app_name);
-
     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_envs::WRITE_QPS_THROTTLING, "100*delay*100", ERR_OK, "", "100*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING, "20K*delay*100", ERR_OK, "", "20K*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING, "20M*delay*100", ERR_OK, "", "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING,
+         "20A*delay*100",
+         ERR_INVALID_PARAMETERS,
+         "20A should be non-negative int",
+         "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING,
+         "-20*delay*100",
+         ERR_INVALID_PARAMETERS,
+         "-20 should be non-negative int",
+         "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING,
+         "",
+         ERR_INVALID_PARAMETERS,
+         "The value shouldn't be empty",
+         "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING,
+         "20A*delay",
+         ERR_INVALID_PARAMETERS,
+         "The field count of 20A*delay should be 3",
+         "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING,
+         "20K*pass*100",
+         ERR_INVALID_PARAMETERS,
+         "pass should be \"delay\" or \"reject\"",
+         "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING,
+         "20K*delay*-100",
+         ERR_INVALID_PARAMETERS,
+         "-100 should be non-negative int",
+         "20M*delay*100"},
+        {replica_envs::WRITE_QPS_THROTTLING, "20M*reject*100", ERR_OK, "", "20M*reject*100"},
+        {replica_envs::WRITE_SIZE_THROTTLING, "300*delay*100", ERR_OK, "", "300*delay*100"},
+        {replica_envs::SLOW_QUERY_THRESHOLD, "30", ERR_OK, "", "30"},
+        {replica_envs::SLOW_QUERY_THRESHOLD, "20", ERR_OK, "", "20"},
+        {replica_envs::SLOW_QUERY_THRESHOLD,
+         "19",
+         ERR_INVALID_PARAMETERS,
+         "Slow query threshold must be >= 20ms",
+         "20"},
+        {replica_envs::SLOW_QUERY_THRESHOLD,
+         "0",
+         ERR_INVALID_PARAMETERS,
+         "Slow query threshold must be >= 20ms",
+         "20"},
+        {replica_envs::TABLE_LEVEL_DEFAULT_TTL, "10", ERR_OK, "", "10"},
+        {replica_envs::ROCKSDB_USAGE_SCENARIO, "20", ERR_OK, "", "20"},
+        {replica_envs::ROCKSDB_CHECKPOINT_RESERVE_MIN_COUNT, "30", ERR_OK, "", "30"},
+        {replica_envs::ROCKSDB_CHECKPOINT_RESERVE_TIME_SECONDS, "40", ERR_OK, "", "40"},
+        {replica_envs::MANUAL_COMPACT_DISABLED, "50", ERR_OK, "", "50"},
+        {replica_envs::MANUAL_COMPACT_MAX_CONCURRENT_RUNNING_COUNT, "60", ERR_OK, "", "60"},
+        {replica_envs::MANUAL_COMPACT_ONCE_TRIGGER_TIME, "70", ERR_OK, "", "70"},
+        {replica_envs::MANUAL_COMPACT_ONCE_TARGET_LEVEL, "80", ERR_OK, "", "80"},
+        {replica_envs::MANUAL_COMPACT_PERIODIC_TRIGGER_TIME, "90", ERR_OK, "", "90"},
+        {replica_envs::MANUAL_COMPACT_PERIODIC_TARGET_LEVEL, "100", ERR_OK, "", "100"},
+        {replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION,
+         "200",
+         ERR_OK,
+         "",
+         "200"},
+        {replica_envs::BUSINESS_INFO, "300", ERR_OK, "", "300"},
+        {replica_envs::DENY_CLIENT_WRITE, "400", ERR_OK, "", "400"},
+        {"not_exist_env",
+         "500",
+         ERR_INVALID_PARAMETERS,
+         "The env of not_exist_env is not supported",
+         ""}};
 
+    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);
+        if (app->envs.find(test.env_key) != app->envs.end()) {
+            ASSERT_EQ(app->envs.at(test.env_key), test.expect_value);
+        }
     }
 }
+
 } // namespace replication
 } // namespace dsn

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

@@ -41,8 +41,8 @@ class meta_http_service_test : public meta_test_base
     void test_get_app_envs()
     {
         // set app env
-        std::string env_key = "test_env";
-        std::string env_value = "test_value";
+        std::string env_key = "replica.slow_query_threshold";
+        std::string env_value = "100";
         update_app_envs(test_app, {env_key}, {env_value});
 
         // http get app envs

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); }