Skip to content

[feature](Vault) Support alter hdfs storage vault and alter vault's name #38685

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gavinchou merged 4 commits into from
Aug 13, 2024
Merged

[feature](Vault) Support alter hdfs storage vault and alter vault's name #38685

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b68cba9
alter s3 name
ByteYue Aug 1, 2024
dd90d5f
format
ByteYue Aug 9, 2024
7aa1f97
add ut
ByteYue Aug 9, 2024
9fa7542
resolve
ByteYue Aug 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 153 additions & 33 deletions cloud/src/meta-service/meta_service_resource.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <charconv>
#include <chrono>
#include <numeric>
#include <regex>
#include <string>
#include <tuple>

Expand All @@ -40,6 +41,14 @@

using namespace std::chrono;

namespace {
constexpr char pattern_str[] = "^[a-zA-Z][0-9a-zA-Z_]*$";
bool is_valid_storage_vault_name(const std::string& str) {
const std::regex pattern(pattern_str);
return std::regex_match(str, pattern);
}
} // namespace

namespace doris::cloud {

static void* run_bthread_work(void* arg) {
Expand Down Expand Up @@ -510,13 +519,114 @@ static void set_default_vault_log_helper(const InstanceInfoPB& instance,
LOG(INFO) << vault_msg;
}

static int alter_hdfs_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Transaction> txn,
Copy link
Contributor

@github-actions github-actions bot Aug 9, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

warning: function 'alter_hdfs_storage_vault' exceeds recommended size/complexity thresholds [readability-function-size]

static int alter_hdfs_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Transaction> txn,
           ^
Additional context

cloud/src/meta-service/meta_service_resource.cpp:521: 97 lines including whitespace and comments (threshold 80)

static int alter_hdfs_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Transaction> txn,
           ^

const StorageVaultPB& vault, MetaServiceCode& code,
std::string& msg) {
if (!vault.has_hdfs_info()) {
code = MetaServiceCode::INVALID_ARGUMENT;
std::stringstream ss;
ss << "There is no hdfs vault provided";
msg = ss.str();
return -1;
}
const auto& hdfs_info = vault.hdfs_info();
if (hdfs_info.has_prefix() || !hdfs_info.has_build_conf() ||
hdfs_info.build_conf().has_fs_name()) {
code = MetaServiceCode::INVALID_ARGUMENT;
std::stringstream ss;
ss << "You can not alter prefix or fs name because it might lose previoud written data";
msg = ss.str();
return -1;
}
const auto& name = vault.name();
// Here we try to get mutable iter since we might need to alter the vault name
auto name_itr = std::find_if(instance.mutable_storage_vault_names()->begin(),
instance.mutable_storage_vault_names()->end(),
[&](const auto& vault_name) { return name == vault_name; });
if (name_itr == instance.storage_vault_names().end()) {
code = MetaServiceCode::INVALID_ARGUMENT;
std::stringstream ss;
ss << "invalid storage vault name, not found, name =" << name;
msg = ss.str();
return -1;
}
auto pos = name_itr - instance.storage_vault_names().begin();
std::string vault_id = instance.resource_ids().begin()[pos];
auto vault_key = storage_vault_key({instance.instance_id(), vault_id});
std::string val;

auto err = txn->get(vault_key, &val);
LOG(INFO) << "get vault_key=" << hex(vault_key);

if (err != TxnErrorCode::TXN_OK) {
code = cast_as<ErrCategory::READ>(err);
std::stringstream ss;
ss << "failed to get storage vault, vault_id=" << vault_id << ", vault_name="
<< "" << name << " err=" << err;
msg = ss.str();
return -1;
}
StorageVaultPB new_vault;
new_vault.ParseFromString(val);
auto origin_vault_info = new_vault.DebugString();
if (vault.has_alter_name()) {
if (!is_valid_storage_vault_name(vault.alter_name())) {
code = MetaServiceCode::INVALID_ARGUMENT;
std::stringstream ss;
ss << "invalid storage vault name =" << vault.alter_name() << " the name must satisfy "
<< pattern_str;
msg = ss.str();
return -1;
}
new_vault.set_name(vault.alter_name());
*name_itr = vault.alter_name();
}
auto* alter_hdfs_info = new_vault.mutable_hdfs_info();
if (hdfs_info.build_conf().has_hdfs_kerberos_keytab()) {
alter_hdfs_info->mutable_build_conf()->set_hdfs_kerberos_keytab(
hdfs_info.build_conf().hdfs_kerberos_keytab());
}
if (hdfs_info.build_conf().has_hdfs_kerberos_principal()) {
alter_hdfs_info->mutable_build_conf()->set_hdfs_kerberos_principal(
hdfs_info.build_conf().hdfs_kerberos_principal());
}
if (hdfs_info.build_conf().has_user()) {
alter_hdfs_info->mutable_build_conf()->set_user(hdfs_info.build_conf().user());
}
if (0 != hdfs_info.build_conf().hdfs_confs_size()) {
alter_hdfs_info->mutable_build_conf()->mutable_hdfs_confs()->Add(
hdfs_info.build_conf().hdfs_confs().begin(),
hdfs_info.build_conf().hdfs_confs().end());
}
auto new_vault_info = new_vault.DebugString();

val = new_vault.SerializeAsString();
if (val.empty()) {
msg = "failed to serialize";
code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;
return -1;
}

txn->put(vault_key, val);
LOG(INFO) << "put vault_id=" << vault_id << ", vault_key=" << hex(vault_key)
<< ", origin vault=" << origin_vault_info << ", new_vault=" << new_vault_info;
err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
code = cast_as<ErrCategory::COMMIT>(err);
msg = fmt::format("failed to commit kv txn, err={}", err);
LOG(WARNING) << msg;
}

return 0;
}

static int alter_s3_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Transaction> txn,
const StorageVaultPB& vault, MetaServiceCode& code,
std::string& msg) {
if (!vault.has_obj_info()) {
code = MetaServiceCode::INVALID_ARGUMENT;
std::stringstream ss;
ss << "Only s3 vault can be altered";
ss << "There is no s3 vault provided";
msg = ss.str();
return -1;
}
Expand All @@ -530,8 +640,9 @@ static int alter_s3_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Tran
return -1;
}
const auto& name = vault.name();
auto name_itr = std::find_if(instance.storage_vault_names().begin(),
instance.storage_vault_names().end(),
// Here we try to get mutable iter since we might need to alter the vault name
auto name_itr = std::find_if(instance.mutable_storage_vault_names()->begin(),
instance.mutable_storage_vault_names()->end(),
[&](const auto& vault_name) { return name == vault_name; });
if (name_itr == instance.storage_vault_names().end()) {
code = MetaServiceCode::INVALID_ARGUMENT;
Expand All @@ -541,35 +652,39 @@ static int alter_s3_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Tran
return -1;
}
auto pos = name_itr - instance.storage_vault_names().begin();
auto id_itr = instance.resource_ids().begin() + pos;
auto vault_key = storage_vault_key({instance.instance_id(), *id_itr});
std::string vault_id = instance.resource_ids().begin()[pos];
auto vault_key = storage_vault_key({instance.instance_id(), vault_id});
std::string val;

auto err = txn->get(vault_key, &val);
LOG(INFO) << "get instance_key=" << hex(vault_key);
LOG(INFO) << "get vault_key=" << hex(vault_key);

if (err != TxnErrorCode::TXN_OK) {
code = cast_as<ErrCategory::READ>(err);
std::stringstream ss;
ss << "failed to get storage vault, vault_id=" << *name_itr << ", vault_name="
ss << "failed to get storage vault, vault_id=" << vault_id << ", vault_name="
<< "" << name << " err=" << err;
msg = ss.str();
return -1;
}
StorageVaultPB alter;
alter.ParseFromString(val);
AkSkPair pre {alter.obj_info().ak(), alter.obj_info().sk()};
const auto& plain_ak = obj_info.has_ak() ? obj_info.ak() : alter.obj_info().ak();
const auto& plain_sk = obj_info.has_ak() ? obj_info.sk() : alter.obj_info().sk();
auto obfuscating_sk = [](const auto& sk) -> std::string {
if (sk.empty()) {
return "";
}
std::string result(sk.length(), '*');
result.replace(0, 2, sk, 0, 2);
result.replace(result.length() - 2, 2, sk, sk.length() - 2, 2);
return result;
};
StorageVaultPB new_vault;
new_vault.ParseFromString(val);
if (vault.has_alter_name()) {
if (!is_valid_storage_vault_name(vault.alter_name())) {
code = MetaServiceCode::INVALID_ARGUMENT;
std::stringstream ss;
ss << "invalid storage vault name =" << vault.alter_name() << " the name must satisfy "
<< pattern_str;
msg = ss.str();
return -1;
}
new_vault.set_name(vault.alter_name());
*name_itr = vault.alter_name();
}
auto origin_vault_info = new_vault.DebugString();
AkSkPair pre {new_vault.obj_info().ak(), new_vault.obj_info().sk()};
const auto& plain_ak = obj_info.has_ak() ? obj_info.ak() : new_vault.obj_info().ak();
const auto& plain_sk = obj_info.has_ak() ? obj_info.sk() : new_vault.obj_info().sk();
AkSkPair plain_ak_sk_pair {plain_ak, plain_sk};
AkSkPair cipher_ak_sk_pair;
EncryptionInfoPB encryption_info;
Expand All @@ -581,22 +696,21 @@ static int alter_s3_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Tran
LOG(WARNING) << msg;
return -1;
}
alter.mutable_obj_info()->set_ak(cipher_ak_sk_pair.first);
alter.mutable_obj_info()->set_sk(cipher_ak_sk_pair.second);
alter.mutable_obj_info()->mutable_encryption_info()->CopyFrom(encryption_info);
new_vault.mutable_obj_info()->set_ak(cipher_ak_sk_pair.first);
new_vault.mutable_obj_info()->set_sk(cipher_ak_sk_pair.second);
new_vault.mutable_obj_info()->mutable_encryption_info()->CopyFrom(encryption_info);

val = alter.SerializeAsString();
auto new_vault_info = new_vault.DebugString();
val = new_vault.SerializeAsString();
if (val.empty()) {
msg = "failed to serialize";
code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;
return -1;
}

txn->put(vault_key, val);
LOG(INFO) << "put vault_id=" << *id_itr << ", instance_key=" << hex(vault_key)
<< ", previous ak=" << pre.first << ", previous sk=" << obfuscating_sk(pre.second)
<< ", new ak=" << cipher_ak_sk_pair.first
<< ", new sk=" << obfuscating_sk(cipher_ak_sk_pair.second);
LOG(INFO) << "put vault_id=" << vault_id << ", vault_key=" << hex(vault_key)
<< ", origin vault=" << origin_vault_info << ", new vault=" << new_vault_info;
err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
code = cast_as<ErrCategory::COMMIT>(err);
Expand Down Expand Up @@ -738,6 +852,8 @@ void MetaServiceImpl::alter_storage_vault(google::protobuf::RpcController* contr
}
case AlterObjStoreInfoRequest::ALTER_S3_VAULT:
break;
case AlterObjStoreInfoRequest::ALTER_HDFS_VAULT:
break;
case AlterObjStoreInfoRequest::UNSET_DEFAULT_VAULT:
break;
case AlterObjStoreInfoRequest::UNKNOWN: {
Expand Down Expand Up @@ -925,12 +1041,12 @@ void MetaServiceImpl::alter_storage_vault(google::protobuf::RpcController* contr
return;
}
auto pos = name_itr - instance.storage_vault_names().begin();
auto id_itr = instance.resource_ids().begin() + pos;
std::string vault_id = instance.resource_ids().begin()[pos];
response->set_default_storage_vault_replaced(instance.has_default_storage_vault_id());
set_default_vault_log_helper(instance, name, *id_itr);
instance.set_default_storage_vault_id(*id_itr);
set_default_vault_log_helper(instance, name, vault_id);
instance.set_default_storage_vault_id(vault_id);
instance.set_default_storage_vault_name(name);
response->set_storage_vault_id(*id_itr);
response->set_storage_vault_id(vault_id);
break;
}
case AlterObjStoreInfoRequest::UNSET_DEFAULT_VAULT: {
Expand All @@ -945,6 +1061,10 @@ void MetaServiceImpl::alter_storage_vault(google::protobuf::RpcController* contr
alter_s3_storage_vault(instance, std::move(txn), request->vault(), code, msg);
return;
}
case AlterObjStoreInfoRequest::ALTER_HDFS_VAULT: {
alter_hdfs_storage_vault(instance, std::move(txn), request->vault(), code, msg);
return;
}
case AlterObjStoreInfoRequest::DROP_S3_VAULT:
[[fallthrough]];
default: {
Expand Down
Loading