diff --git a/src/replica/CMakeLists.txt b/src/replica/CMakeLists.txt
index 5bcd3f57bc8..59ca1a90d49 100644
--- a/src/replica/CMakeLists.txt
+++ b/src/replica/CMakeLists.txt
@@ -58,6 +58,7 @@ set(MY_PROJ_SRC
 set(MY_SRC_SEARCH_MODE "GLOB")
 
 set(MY_PROJ_LIBS
+        absl::strings
         dsn_replication_common
         dsn.failure_detector
         dsn.block_service
@@ -66,9 +67,11 @@ set(MY_PROJ_LIBS
         dsn_nfs
         dsn_dist_cmd
         dsn_http
+        curl
         dsn_runtime
         dsn_aio
         dsn_meta_server
+        dsn.security
         rocksdb
         lz4
         zstd
diff --git a/src/replica/default_key_provider.h b/src/replica/default_key_provider.h
new file mode 100644
index 00000000000..4dabdfde699
--- /dev/null
+++ b/src/replica/default_key_provider.h
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <string>
+#include <openssl/rand.h>
+
+#include "absl/strings/escaping.h"
+#include "replica/key_provider.h"
+#include "utils/error_code.h"
+#include "utils/fmt_logging.h"
+
+namespace dsn {
+namespace security {
+
+class DefaultKeyProvider : public KeyProvider
+{
+public:
+    ~DefaultKeyProvider() override {}
+    dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
+                                      const std::string & /*iv*/,
+                                      const std::string & /*key_version*/,
+                                      std::string *decrypted_key) override
+    {
+        *decrypted_key = ::absl::HexStringToBytes(encryption_key);
+
+#ifdef __linux__
+        memfrob(decrypted_key->data(), decrypted_key->length());
+#else
+        // On Linux, memfrob() bitwise XORs the data with the magic number that is
+        // the answer to the ultimate question of life, the universe, and
+        // everything. On Mac, we do this manually.
+        const uint8_t kMagic = 42;
+        for (auto i = 0; i < decrypted_key->length(); ++i) {
+            decrypted_key->data()[i] ^= kMagic;
+        }
+#endif
+        *decrypted_key = ::absl::BytesToHexString(*decrypted_key);
+        return dsn::error_s::ok();
+    }
+
+    dsn::error_s GenerateEncryptionKey(std::string *encryption_key,
+                                       std::string *iv,
+                                       std::string *key_version) override
+    {
+        unsigned char key_bytes[32];
+        unsigned char iv_bytes[32];
+        int num_bytes = 16;
+        std::string dek;
+        RAND_bytes(key_bytes, num_bytes);
+        dek = ::absl::BytesToHexString(reinterpret_cast<const char *>(key_bytes));
+        RAND_bytes(iv_bytes, num_bytes);
+        *iv = ::absl::BytesToHexString(reinterpret_cast<const char *>(iv_bytes));
+        DecryptEncryptionKey(dek, *iv, *key_version, encryption_key);
+        *key_version = "encryptionkey@0";
+        return dsn::error_s::ok();
+    }
+};
+} // namespace security
+} // namespace dsn
diff --git a/src/replica/key_provider.h b/src/replica/key_provider.h
new file mode 100644
index 00000000000..c13ba6cb838
--- /dev/null
+++ b/src/replica/key_provider.h
@@ -0,0 +1,45 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <string>
+
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+
+// An interface for encrypting and decrypting Pegasus's encryption keys.
+class KeyProvider
+{
+public:
+    virtual ~KeyProvider() = default;
+
+    // Decrypts the encryption key.
+    virtual dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
+                                              const std::string &iv,
+                                              const std::string &key_version,
+                                              std::string *decrypted_key) = 0;
+
+    // Generates an encryption key (the generated key is encrypted).
+    virtual dsn::error_s GenerateEncryptionKey(std::string *encryption_key,
+                                               std::string *iv,
+                                               std::string *key_version) = 0;
+};
+} // namespace security
+} // namespace dsn
diff --git a/src/replica/pegasus_kms_key_provider.cpp b/src/replica/pegasus_kms_key_provider.cpp
new file mode 100644
index 00000000000..c6ce3791cef
--- /dev/null
+++ b/src/replica/pegasus_kms_key_provider.cpp
@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <string>
+
+#include "replica/pegasus_kms_key_provider.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+
+dsn::error_s PegasusKMSKeyProvider::DecryptEncryptionKey(const std::string &encryption_key,
+                                                         const std::string &iv,
+                                                         const std::string &key_version,
+                                                         std::string *decrypted_key)
+{
+    return client_.DecryptEncryptionKey(encryption_key, iv, key_version, decrypted_key);
+}
+
+dsn::error_s PegasusKMSKeyProvider::GenerateEncryptionKey(std::string *encryption_key,
+                                                          std::string *iv,
+                                                          std::string *key_version)
+{
+    return client_.GenerateEncryptionKey(encryption_key, iv, key_version);
+}
+
+} // namespace security
+} // namespace dsn
diff --git a/src/replica/pegasus_kms_key_provider.h b/src/replica/pegasus_kms_key_provider.h
new file mode 100644
index 00000000000..5aebe9523e7
--- /dev/null
+++ b/src/replica/pegasus_kms_key_provider.h
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <string>
+#include <utility>
+
+#include "replica/key_provider.h"
+#include "runtime/security/kms_client.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+class PegasusKMSKeyProvider : public KeyProvider
+{
+public:
+    ~PegasusKMSKeyProvider() override {}
+
+    PegasusKMSKeyProvider(const std::string &kms_url, std::string cluster_key_name)
+        : client_(kms_url, std::move(cluster_key_name))
+    {
+    }
+
+    // Decrypts the encryption key.
+    dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
+                                      const std::string &iv,
+                                      const std::string &key_version,
+                                      std::string *decrypted_key) override;
+
+    // Generates an encryption key (the generated key is encrypted).
+    dsn::error_s GenerateEncryptionKey(std::string *encryption_key,
+                                       std::string *iv,
+                                       std::string *key_version) override;
+
+private:
+    KMSClient client_;
+};
+} // namespace security
+} // namespace dsn
diff --git a/src/replica/replica_stub.cpp b/src/replica/replica_stub.cpp
index db029e8448a..6993c4f0d3e 100644
--- a/src/replica/replica_stub.cpp
+++ b/src/replica/replica_stub.cpp
@@ -57,12 +57,13 @@
 #include "nfs_types.h"
 #include "replica.h"
 #include "replica/duplication/replica_follower.h"
+#include "replica/key_provider.h"
 #include "replica/log_file.h"
+#include "replica/pegasus_kms_key_provider.h"
 #include "replica/replica_context.h"
 #include "replica/replica_stub.h"
 #include "replica/replication_app_base.h"
 #include "replica_disk_migrator.h"
-#include "replica_stub.h"
 #include "runtime/api_layer1.h"
 #include "runtime/ranger/access_type.h"
 #include "runtime/rpc/rpc_message.h"
@@ -71,6 +72,7 @@
 #include "runtime/task/async_calls.h"
 #include "split/replica_split_manager.h"
 #include "utils/command_manager.h"
+#include "utils/errors.h"
 #include "utils/filesystem.h"
 #include "utils/fmt_logging.h"
 #include "utils/ports.h"
@@ -88,6 +90,7 @@
 #include "remote_cmd/remote_command.h"
 #include "utils/fail_point.h"
 
+<<<<<<< HEAD
 METRIC_DEFINE_gauge_int64(server,
                           total_replicas,
                           dsn::metric_unit::kReplicas,
@@ -216,6 +219,10 @@ METRIC_DEFINE_gauge_int64(server,
                           dsn::metric_unit::kBytes,
                           "The max size of copied files among all splitting replicas");
 
+=======
+DSN_DECLARE_bool(encrypt_data_at_rest);
+DSN_DECLARE_string(server_key);
+>>>>>>> 0055c71bb (feat(encryption): add kms key management)
 namespace dsn {
 namespace replication {
 DSN_DEFINE_bool(replication,
@@ -305,6 +312,14 @@ DSN_DEFINE_int32(
     10,
     "if tcmalloc reserved but not-used memory exceed this percentage of application allocated "
     "memory, replica server will release the exceeding memory back to operating system");
+DSN_DEFINE_string(pegasus.server,
+                  encryption_cluster_key_name,
+                  "pegasus",
+                  "The cluster name of encrypted server which use to get server key from kms.");
+DSN_DEFINE_string(pegasus.server,
+                  hadoop_kms_url,
+                  "",
+                  "Where the server encrypted key of file system can get from.");
 
 DSN_DECLARE_bool(duplication_enabled);
 DSN_DECLARE_int32(fd_beacon_interval_seconds);
@@ -368,6 +383,15 @@ replica_stub::replica_stub(replica_state_subscriber subscriber /*= nullptr*/,
     _state = NS_Disconnected;
     _log = nullptr;
     _primary_address_str[0] = '\0';
+<<<<<<< HEAD
+=======
+    install_perf_counters();
+    if (FLAGS_encrypt_data_at_rest) {
+        // TODO: check enable_acl whether be true
+        key_provider.reset(new dsn::security::PegasusKMSKeyProvider(
+            FLAGS_hadoop_kms_url, FLAGS_encryption_cluster_key_name));
+    }
+>>>>>>> 0055c71bb (feat(encryption): add kms key management)
 }
 
 replica_stub::~replica_stub(void) { close(); }
@@ -380,6 +404,32 @@ void replica_stub::initialize(bool clear /* = false*/)
     _access_controller = std::make_unique<dsn::security::access_controller>();
 }
 
+dsn::error_s store_kms_key(const std::string &data_dir,
+                           std::string encryption_key,
+                           std::string iv,
+                           std::string key_version)
+{
+    replica_kms_info kms_info(encryption_key, iv, key_version);
+    auto err = kms_info.store(data_dir);
+    if (!err.is_ok()) {
+        return dsn::error_s::make(err, "Can't open kms-info file to write");
+    }
+    return dsn::error_s::ok();
+}
+
+void get_kms_key(std::string data_dir,
+                 std::string *encryption_key,
+                 std::string *iv,
+                 std::string *key_version)
+{
+    replica_kms_info kms_info;
+    auto err = kms_info.load(data_dir);
+    *encryption_key = kms_info.encryption_key;
+    *iv = kms_info.iv;
+    *key_version = kms_info.key_version;
+    CHECK_OK(err, "Can't open replica_encrypted_key file to read");
+}
+
 void replica_stub::initialize(const replication_options &opts, bool clear /* = false*/)
 {
     _primary_address = dsn_primary_address();
@@ -407,9 +457,33 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
         }
     }
 
+    std::string encryption_key;
+    std::string iv;
+    std::string key_version;
+    std::string server_key;
+    // get and store Encrypted Encryption Key(eek),Initialization Vector(iv),Key Version from kms
+    if (key_provider && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        CHECK(_options.data_dirs[0], "data_dirs is empty");
+        get_kms_key(_options.data_dirs[0], &encryption_key, &iv, &key_version);
+        // The encryption key should empty when process upon the first launch. And the process will get eek,iv,kv from kms 
+        // After first launch, the encryption key should not empty. And the process get Decrypted Encryption Key(dek) from kms
+        if (encryption_key.empty()) {
+            CHECK(key_provider->GenerateEncryptionKey(&encryption_key, &iv, &key_version),
+                  "get encryption key failed");
+        }
+        CHECK(key_provider->DecryptEncryptionKey(encryption_key, iv, key_version, &server_key),
+              "get decryption key failed");
+        FLAGS_server_key = server_key.c_str();
+    }
+
     // Initialize the file system manager.
     _fs_manager.initialize(_options.data_dirs, _options.data_dir_tags);
 
+    if (key_provider && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        CHECK(store_kms_key(_options.data_dirs[0], encryption_key, iv, key_version),
+              "Cant store kms key");
+    }
+
     // TODO(yingchun): remove the slog related code.
     // Create slog directory if it does not exist.
     std::string cdir;
diff --git a/src/replica/replica_stub.h b/src/replica/replica_stub.h
index db88464a21e..afb670306aa 100644
--- a/src/replica/replica_stub.h
+++ b/src/replica/replica_stub.h
@@ -75,6 +75,10 @@ class command_deregister;
 class message_ex;
 class nfs_node;
 
+namespace security {
+class KeyProvider;
+} // namespace security
+
 namespace service {
 class copy_request;
 class copy_response;
@@ -488,6 +492,7 @@ class replica_stub : public serverlet<replica_stub>, public ref_counter
 
     std::unique_ptr<duplication_sync_timer> _duplication_sync_timer;
     std::unique_ptr<replica_backup_server> _backup_server;
+    std::unique_ptr<dsn::security::KeyProvider> key_provider;
 
     // command_handlers
     std::vector<std::unique_ptr<command_deregister>> _cmds;
diff --git a/src/replica/replication_app_base.cpp b/src/replica/replication_app_base.cpp
index 358222754e0..e9d5df20328 100644
--- a/src/replica/replication_app_base.cpp
+++ b/src/replica/replication_app_base.cpp
@@ -67,6 +67,7 @@ namespace dsn {
 namespace replication {
 
 const std::string replica_init_info::kInitInfo = ".init-info";
+const std::string replica_kms_info::kKmsInfo = ".kms-info";
 
 error_code replica_init_info::load(const std::string &dir)
 {
@@ -130,6 +131,56 @@ std::string replica_init_info::to_string()
     return oss.str();
 }
 
+error_code replica_kms_info::load(const std::string &dir)
+{
+    std::string info_path = utils::filesystem::path_combine(dir, kFileName);
+    LOG_AND_RETURN_NOT_TRUE(ERROR,
+                            utils::filesystem::path_exists(info_path),
+                            ERR_PATH_NOT_FOUND,
+                            "file({}) not exist",
+                            info_path);
+    LOG_AND_RETURN_NOT_OK(
+        ERROR, load_json(info_path), "load replica_kms_info from {} failed", info_path);
+    LOG_INFO("load replica_kms_info from {} ", info_path);
+    return ERR_OK;
+}
+
+error_code replica_kms_info::store(const std::string &dir)
+{
+    uint64_t start = dsn_now_ns();
+    std::string info_path = utils::filesystem::path_combine(dir, kFileName);
+    LOG_AND_RETURN_NOT_OK(ERROR,
+                          store_json(info_path),
+                          "store replica_kms_info to {} failed, time_used_ns = {}",
+                          info_path,
+                          dsn_now_ns() - start);
+    LOG_INFO(
+        "store replica_kms_info to {} succeed, time_used_ns = {}", info_path, dsn_now_ns() - start);
+    return ERR_OK;
+}
+
+error_code replica_kms_info::load_json(const std::string &fname)
+{
+    std::string data;
+    auto s = rocksdb::ReadFileToString(
+        dsn::utils::PegasusEnv(dsn::utils::FileDataType::kNonSensitive), fname, &data);
+    LOG_AND_RETURN_NOT_TRUE(ERROR, s.ok(), ERR_FILE_OPERATION_FAILED, "read file {} failed", fname);
+    LOG_AND_RETURN_NOT_TRUE(ERROR,
+                            json::json_forwarder<replica_kms_info>::decode(
+                                blob::create_from_bytes(std::move(data)), *this),
+                            ERR_FILE_OPERATION_FAILED,
+                            "decode json from file {} failed",
+                            fname);
+    return ERR_OK;
+}
+
+error_code replica_kms_info::store_json(const std::string &fname)
+{
+    return write_blob_to_file(fname,
+                              json::json_forwarder<replica_kms_info>::encode(*this),
+                              dsn::utils::FileDataType::kNonSensitive);
+}
+
 error_code replica_app_info::load(const std::string &fname)
 {
     std::string data;
diff --git a/src/replica/replication_app_base.h b/src/replica/replication_app_base.h
index d4736bc139a..e271fee778d 100644
--- a/src/replica/replication_app_base.h
+++ b/src/replica/replication_app_base.h
@@ -112,6 +112,30 @@ class replica_init_info
     error_code store_json(const std::string &fname);
 };
 
+class replica_kms_info
+{
+public:
+    std::string encryption_key;
+    std::string iv;
+    std::string key_version;
+    DEFINE_JSON_SERIALIZATION(encryption_key, iv, key_version)
+    static const std::string kFileName;
+
+public:
+    replica_kms_info(const std::string &e_key = "",
+                     const std::string &i = "",
+                     const std::string &k_version = "")
+        : encryption_key(e_key), iv(i), key_version(k_version)
+    {
+    }
+    error_code load(const std::string &dir) WARN_UNUSED_RESULT;
+    error_code store(const std::string &dir);
+
+private:
+    error_code load_json(const std::string &fname);
+    error_code store_json(const std::string &fname);
+};
+
 class replica_app_info
 {
 private:
diff --git a/src/replica/test/defaul_key_provider_test.cpp b/src/replica/test/defaul_key_provider_test.cpp
new file mode 100644
index 00000000000..16cc8da21f0
--- /dev/null
+++ b/src/replica/test/defaul_key_provider_test.cpp
@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <string>
+
+#include "gtest/gtest.h"
+#include "replica/default_key_provider.h"
+
+using std::string;
+
+namespace dsn {
+namespace security {
+
+class DefaultKeyProviderTest : public testing::Test
+{
+protected:
+    DefaultKeyProvider key_provider;
+};
+
+TEST_F(DefaultKeyProviderTest, TestEncryptAndDecrypt)
+{
+    string encryption_key;
+    string iv;
+    string version;
+    string decrypted_key;
+    ASSERT_TRUE(key_provider.GenerateEncryptionKey(&encryption_key, &iv, &version));
+    ASSERT_TRUE(key_provider.DecryptEncryptionKey(encryption_key, iv, version, &decrypted_key));
+    ASSERT_NE(encryption_key, decrypted_key);
+}
+
+} // namespace security
+} // namespace dsn
diff --git a/src/runtime/security/CMakeLists.txt b/src/runtime/security/CMakeLists.txt
index 02075bde912..a9781f59f61 100644
--- a/src/runtime/security/CMakeLists.txt
+++ b/src/runtime/security/CMakeLists.txt
@@ -32,7 +32,7 @@ set(MY_SRC_SEARCH_MODE "GLOB")
 
 set(MY_PROJ_INC_PATH "")
 
-set(MY_PROJ_LIBS "")
+set(MY_PROJ_LIBS absl::strings dsn_http curl gssapi_krb5)
 
 set(MY_PROJ_LIB_PATH "")
 
diff --git a/src/runtime/security/kms_client.cpp b/src/runtime/security/kms_client.cpp
new file mode 100644
index 00000000000..5aaaef758ff
--- /dev/null
+++ b/src/runtime/security/kms_client.cpp
@@ -0,0 +1,200 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <algorithm>
+#include <initializer_list>
+#include <map>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include "absl/strings/escaping.h"
+#include "absl/strings/substitute.h"
+#include "http/http_client.h"
+#include "http/http_method.h"
+#include "nlohmann/json.hpp"
+#include "nlohmann/json_fwd.hpp"
+#include "runtime/security/kms_client.h"
+#include "utils/error_code.h"
+#include "utils/fmt_logging.h"
+
+namespace dsn {
+namespace security {
+
+dsn::error_s KMSClient::DecryptEncryptionKey(const std::string &encryption_key,
+                                                    const std::string &iv,
+                                                    const std::string &key_version,
+                                                    std::string *decrypted_key)
+{
+    nlohmann::json post;
+    post["name"] = cluster_key_name_;
+    std::string iv_plain = ::absl::HexStringToBytes(iv);
+    std::string iv_b64;
+    ::absl::WebSafeBase64Escape(iv_plain, &iv_b64);
+    post["iv"] = iv_b64;
+    std::string eek_plain = ::absl::HexStringToBytes(encryption_key);
+    std::string eek_b64;
+    ::absl::WebSafeBase64Escape(eek_plain, &eek_b64);
+    post["material"] = eek_b64;
+
+    http_client client;
+    auto err = client.init();
+    if (!err.is_ok()) {
+        return dsn::error_s::make(ERR_CURL_FAILED, "Start http client failed");
+    }
+
+    err = client.set_auth(http_auth_type::SPNEGO);
+    if (!err.is_ok()) {
+        return dsn::error_s::make(ERR_CURL_FAILED, "http client set auth type failed");
+    }
+
+    std::vector<std::string> urls;
+    urls.reserve(kms_urls_.size());
+    for (const auto &url : kms_urls_) {
+        urls.emplace_back(
+            ::absl::Substitute("$0/v1/keyversion/$1/_eek?eek_op=decrypt", url, key_version));
+    }
+
+    client.clear_header_fields();
+    client.set_content_type("application/json");
+    client.set_accept("*/*");
+    err = client.with_post_method(post.dump());
+    if (!err.is_ok()) {
+        return dsn::error_s::make(ERR_CURL_FAILED, "http client set method failed");
+    }
+
+    nlohmann::json j;
+    for (const auto &url : urls) {
+        err = client.set_url(url);
+        if (!err.is_ok()) {
+            return dsn::error_s::make(ERR_CURL_FAILED, "http clientt set url failed");
+        }
+        std::string resp;
+        err = client.exec_method(&resp);
+        if (!err.is_ok()) {
+            return dsn::error_s::make(ERR_CURL_FAILED, "http client exec post method failed");
+        }
+        long http_status;
+        client.get_http_status(http_status);
+        if (http_status == 200) {
+            j = nlohmann::json::parse(resp);
+        }
+    }
+
+    std::string dek_b64;
+    if (j.contains("material")) {
+        dek_b64 = j.at("material");
+    } else {
+        return dsn::error_s::make(ERR_INVALID_DATA, "Null material received");
+    }
+    std::string dek_plain;
+    if (!::absl::WebSafeBase64Unescape(dek_b64, &dek_plain)) {
+        return dsn::error_s::make(ERR_INVALID_DATA, "Invalid IV received");
+    }
+    *decrypted_key = ::absl::BytesToHexString(dek_plain);
+    return dsn::error_s::ok();
+}
+
+dsn::error_s KMSClient::GenerateEncryptionKeyFromKMS(const std::string &key_name,
+                                                            std::string *encryption_key,
+                                                            std::string *iv,
+                                                            std::string *key_version)
+{
+    http_client client;
+    auto err = client.init();
+    if (!err.is_ok()) {
+        return dsn::error_s::make(ERR_CURL_FAILED, "Start http client failed");
+    }
+    err = client.set_auth(http_auth_type::SPNEGO);
+    if (!err.is_ok()) {
+        return dsn::error_s::make(ERR_CURL_FAILED, "http client set auth type failed");
+    }
+    std::vector<std::string> urls;
+    urls.reserve(kms_urls_.size());
+    for (const auto &url : kms_urls_) {
+        urls.emplace_back(
+            ::absl::Substitute("$0/v1/key/$1/_eek?eek_op=generate&num_keys=1", url, key_name));
+    }
+
+    nlohmann::json j = nlohmann::json::object();
+    for (const auto &url : urls) {
+        err = client.set_url(url);
+        if (!err.is_ok()) {
+            return dsn::error_s::make(ERR_CURL_FAILED, "http client set url failed");
+        }
+
+        err = client.with_get_method();
+        if (!err.is_ok()) {
+            return dsn::error_s::make(ERR_CURL_FAILED, "http client set get method failed");
+        }
+
+        std::string resp;
+        err = client.exec_method(&resp);
+        if (!err.is_ok()) {
+            return dsn::error_s::make(ERR_CURL_FAILED, "http client exec get method failed");
+        }
+
+        long http_status;
+        client.get_http_status(http_status);
+        if (http_status == 200) {
+            j = nlohmann::json::parse(resp);
+            nlohmann::json jsonObject = j.at(0);
+            std::string res = jsonObject.dump();
+            j = nlohmann::json::parse(res);
+        }
+    }
+
+    if (j.contains("versionName")) {
+        *key_version = j.at("versionName");
+    } else {
+        return dsn::error_s::make(ERR_INVALID_DATA, "Null versionName received");
+    }
+    std::string iv_b64;
+    if (j.contains("iv")) {
+        iv_b64 = j.at("iv");
+    } else {
+        return dsn::error_s::make(ERR_INVALID_DATA, "Null IV received");
+    }
+    std::string iv_plain;
+    if (!::absl::WebSafeBase64Unescape(iv_b64, &iv_plain)) {
+        return dsn::error_s::make(ERR_INVALID_DATA, "Invalid IV received");
+    }
+    *iv = ::absl::BytesToHexString(iv_plain);
+    std::string key_b64;
+    if (j.contains("encryptedKeyVersion") && j.at("encryptedKeyVersion").contains("material")) {
+        key_b64 = j.at("encryptedKeyVersion").at("material");
+    } else {
+        return dsn::error_s::make(ERR_INVALID_DATA,
+                                  "Null encryptedKeyVersion or material received");
+    }
+    std::string key_plain;
+    if (!::absl::WebSafeBase64Unescape(key_b64, &key_plain)) {
+        return dsn::error_s::make(ERR_INVALID_DATA, "Invalid encryption key received");
+    }
+    *encryption_key = ::absl::BytesToHexString(key_plain);
+    return dsn::error_s::ok();
+}
+
+dsn::error_s KMSClient::GenerateEncryptionKey(std::string *encryption_key,
+                                                     std::string *iv,
+                                                     std::string *key_version)
+{
+    return GenerateEncryptionKeyFromKMS(cluster_key_name_, encryption_key, iv, key_version);
+}
+
+} // namespace security
+} // namespace dsn
diff --git a/src/runtime/security/kms_client.h b/src/runtime/security/kms_client.h
new file mode 100644
index 00000000000..73ca13a305b
--- /dev/null
+++ b/src/runtime/security/kms_client.h
@@ -0,0 +1,96 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/strings/str_split.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+// A library for http client that provides convenient APIs to access http services, implemented
+// based on http client (https://curl.se/libcurl/c/).
+//
+// A class to implement 
+// This class is not thread-safe. Thus maintain one instance for each thread.
+//
+// Example of useing Kms client:
+// --------------------------------------------------------
+// Create an instance of http_client:
+// GenerateEncryptionKey
+//
+// It's necessary to initialize the new instance before coming into use:
+// DecryptEncryptionKey
+//
+// Specify the target url that you would request for:
+// err = client.set_url(method);
+//
+// If you would use GET method, call `with_get_method`:
+// err = client.with_get_method();
+//
+// If you would use POST method, call `with_post_method` with post data:
+// err = client.with_post_method(post_data);
+//
+// Submit the request to remote http service:
+// err = client.exec_method();
+//
+// If response data should be processed, use callback function:
+// auto callback = [...](const void *data, size_t length) {
+//     ......
+//     return true;
+// };
+// err = client.exec_method(callback);
+//
+// Or just provide a string pointer:
+// std::string response;
+// err = client.exec_method(&response);
+//
+// Get the http status code after requesting:
+// long http_status;
+// err = client.get_http_status(http_status);
+class KMSClient
+{
+public:
+    KMSClient(const std::string kms_url, std::string cluster_key_name)
+        : kms_urls_(::absl::StrSplit(kms_url, ",", ::absl::SkipEmpty())),
+          cluster_key_name_(std::move(cluster_key_name))
+    {
+    }
+
+    dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
+                                      const std::string &iv,
+                                      const std::string &key_version,
+                                      std::string *decrypted_key);
+
+    dsn::error_s
+    GenerateEncryptionKey(std::string *encryption_key, std::string *iv, std::string *key_version);
+
+private:
+    dsn::error_s GenerateEncryptionKeyFromKMS(const std::string &key_name,
+                                              std::string *encryption_key,
+                                              std::string *iv,
+                                              std::string *key_version);
+
+    std::vector<std::string> kms_urls_;
+    std::string cluster_key_name_;
+};
+} // namespace security
+} // namespace dsn
diff --git a/src/utils/env.cpp b/src/utils/env.cpp
index 444ac3784ad..9fca5271131 100644
--- a/src/utils/env.cpp
+++ b/src/utils/env.cpp
@@ -39,7 +39,7 @@ DSN_DEFINE_bool(pegasus.server,
                 "Whether the sensitive files should be encrypted on the file system.");
 
 DSN_DEFINE_string(pegasus.server,
-                  server_key_for_testing,
+                  server_key,
                   "0123456789ABCDEF0123456789ABCDEF",
                   "The encrypted server key to use in the filesystem. NOTE: only for testing.");
 
@@ -62,9 +62,8 @@ rocksdb::Env *NewEncryptedEnv()
 {
     // Create an encryption provider.
     std::shared_ptr<rocksdb::EncryptionProvider> provider;
-    auto provider_id = fmt::format("id=AES;hex_instance_key={};method={}",
-                                   FLAGS_server_key_for_testing,
-                                   FLAGS_encryption_method);
+    auto provider_id = fmt::format(
+        "id=AES;hex_instance_key={};method={}", FLAGS_server_key, FLAGS_encryption_method);
     auto s = rocksdb::EncryptionProvider::CreateFromString(
         rocksdb::ConfigOptions(), provider_id, &provider);
     CHECK(s.ok(), "Failed to create encryption provider: {}", s.ToString());