diff --git a/src/replica/CMakeLists.txt b/src/replica/CMakeLists.txt
index fcf7891017..aab1a4ee24 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)
 
 set(MY_BOOST_LIBS Boost::filesystem)
diff --git a/src/replica/key_provider.h b/src/replica/key_provider.h
new file mode 100644
index 0000000000..c13ba6cb83
--- /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 0000000000..c6ce3791ce
--- /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 0000000000..bd9f90fe77
--- /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:
+    PegasusKMSClient client_;
+};
+} // namespace security
+} // namespace dsn
diff --git a/src/replica/replica_stub.cpp b/src/replica/replica_stub.cpp
index e48a2e15ea..74d6a041cb 100644
--- a/src/replica/replica_stub.cpp
+++ b/src/replica/replica_stub.cpp
@@ -41,6 +41,7 @@
 #include <set>
 #include <vector>
 
+#include "absl/strings/string_view.h"
 #include "backup/replica_backup_server.h"
 #include "bulk_load/replica_bulk_loader.h"
 #include "common/backup_common.h"
@@ -57,12 +58,13 @@
 #include "perf_counter/perf_counter.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,13 +73,13 @@
 #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"
 #include "utils/process_utils.h"
 #include "utils/rand.h"
 #include "utils/string_conv.h"
-#include "absl/strings/string_view.h"
 #include "utils/strings.h"
 #include "utils/synchronize.h"
 #ifdef DSN_ENABLE_GPERF
@@ -89,6 +91,8 @@
 #include "remote_cmd/remote_command.h"
 #include "utils/fail_point.h"
 
+DSN_DECLARE_bool(encrypt_data_at_rest);
+DSN_DECLARE_string(server_key);
 namespace dsn {
 namespace replication {
 DSN_DEFINE_bool(replication,
@@ -178,6 +182,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);
@@ -215,6 +227,11 @@ replica_stub::replica_stub(replica_state_subscriber subscriber /*= nullptr*/,
     _log = nullptr;
     _primary_address_str[0] = '\0';
     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));
+    }
 }
 
 replica_stub::~replica_stub(void) { close(); }
@@ -573,6 +590,35 @@ void replica_stub::initialize(bool clear /* = false*/)
     _access_controller = std::make_unique<dsn::security::access_controller>();
 }
 
+dsn::error_s store_kms_key(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 (dsn::ERR_OK == err) {
+        return dsn::error_s::ok();
+    } else {
+        return dsn::error_s::make(err, "Can't open replica_encrypted_key file to write");
+    }
+}
+
+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;
+    if (dsn::ERR_OK != err) {
+        CHECK(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();
@@ -600,9 +646,30 @@ 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 eek from kms
+    if (key_provider && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        get_kms_key(_options.data_dirs[0], &encryption_key, &iv, &key_version);
+        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 03a6333aa2..bd3a09c8df 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;
@@ -489,6 +493,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 6085bffc18..af180e5ac4 100644
--- a/src/replica/replication_app_base.cpp
+++ b/src/replica/replication_app_base.cpp
@@ -65,6 +65,7 @@ namespace dsn {
 namespace replication {
 
 const std::string replica_init_info::kInitInfo = ".init-info";
+const std::string replica_kms_info::kFileName = "/replica_encrypted_key";
 
 namespace {
 error_code write_blob_to_file(const std::string &fname, const blob &data)
@@ -149,6 +150,70 @@ 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)
+{
+    const blob &data = json::json_forwarder<replica_kms_info>::encode(*this);
+    std::string tmp_fname = fname + ".tmp";
+    auto cleanup = defer([tmp_fname]() { utils::filesystem::remove_path(tmp_fname); });
+    auto s =
+        rocksdb::WriteStringToFile(dsn::utils::PegasusEnv(dsn::utils::FileDataType::kNonSensitive),
+                                   rocksdb::Slice(data.data(), data.length()),
+                                   tmp_fname,
+                                   /* should_sync */ true);
+    LOG_AND_RETURN_NOT_TRUE(
+        ERROR, s.ok(), ERR_FILE_OPERATION_FAILED, "write file {} failed", tmp_fname);
+    LOG_AND_RETURN_NOT_TRUE(ERROR,
+                            utils::filesystem::rename_path(tmp_fname, fname),
+                            ERR_FILE_OPERATION_FAILED,
+                            "move file from {} to {} failed",
+                            tmp_fname,
+                            fname);
+    return ERR_OK;
+}
+
 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 5ae162a1bb..5cd2bc8abe 100644
--- a/src/replica/replication_app_base.h
+++ b/src/replica/replication_app_base.h
@@ -80,6 +80,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/runtime/security/CMakeLists.txt b/src/runtime/security/CMakeLists.txt
index 02075bde91..a9781f59f6 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 0000000000..80319d1798
--- /dev/null
+++ b/src/runtime/security/kms_client.cpp
@@ -0,0 +1,202 @@
+// 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 PegasusKMSClient::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);
+        LOG_INFO("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 PegasusKMSClient::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);
+        LOG_INFO("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 PegasusKMSClient::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 0000000000..3d51fd08b6
--- /dev/null
+++ b/src/runtime/security/kms_client.h
@@ -0,0 +1,56 @@
+// 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 {
+class PegasusKMSClient
+{
+public:
+    PegasusKMSClient(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 444ac3784a..531c715a79 100644
--- a/src/utils/env.cpp
+++ b/src/utils/env.cpp
@@ -49,6 +49,11 @@ DSN_DEFINE_string(pegasus.server,
                   "The encryption method to use in the filesystem. Now "
                   "supports AES128CTR, AES192CTR, AES256CTR and SM4CTR.");
 
+DSN_DEFINE_string(pegasus.server,
+                  server_key,
+                  "0123456789ABCDEF0123456789ABCDEF",
+                  "The encrypted server key to use in the filesystem.");
+
 DSN_DEFINE_bool(replication,
                 enable_direct_io,
                 false,
@@ -62,9 +67,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());