Skip to content

[fix](cloud) Track deleted instance recycle status #65330

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

Open
wyxxxcat wants to merge 1 commit into
base: master
Choose a base branch
from
+251 −77
Open

[fix](cloud) Track deleted instance recycle status #65330

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
33 changes: 33 additions & 0 deletions cloud/src/common/http_helper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,12 @@ const std::unordered_map<std::string_view, HttpHandlerInfo>& get_http_handlers()
{.handler = [](void* s,
brpc::Controller* c) { return process_check_instance((RS*)s, c); },
.role = HttpRole::RECYCLER}},
{"set_instance_recycled_status",
{.handler =
[](void* s, brpc::Controller* c) {
return process_set_instance_recycled_status((RS*)s, c);
},
.role = HttpRole::RECYCLER}},
{"check_job_info",
{.handler = [](void* s,
brpc::Controller* c) { return process_check_job_info((RS*)s, c); },
Expand Down Expand Up @@ -561,6 +567,33 @@ HttpResponse process_alter_instance(MetaServiceImpl* service, brpc::Controller*
return http_json_reply(resp.status());
}

HttpResponse process_set_instance_recycled_status(RecyclerServiceImpl* service,
brpc::Controller* ctrl) {
auto& uri = ctrl->http_request().uri();
std::string instance_id(http_query(uri, "instance_id"));
std::string recycled_status_str(http_query(uri, "recycled_status"));
if (instance_id.empty()) {
return http_json_reply(MetaServiceCode::INVALID_ARGUMENT, "instance_id is empty");
}
if (recycled_status_str.empty()) {
return http_json_reply(MetaServiceCode::INVALID_ARGUMENT, "recycled_status is empty");
}

InstanceInfoPB::InstanceDeletedStatus recycled_status;
if (!InstanceInfoPB_InstanceDeletedStatus_Parse(recycled_status_str, &recycled_status) ||
(recycled_status != InstanceInfoPB::INSTANCE_RECYCLE_PENDING &&
recycled_status != InstanceInfoPB::INSTANCE_DATA_DELETED &&
recycled_status != InstanceInfoPB::INSTANCE_META_AND_DATA_DELETED)) {
return http_json_reply(MetaServiceCode::INVALID_ARGUMENT,
"invalid recycled_status, supported values: "
"INSTANCE_RECYCLE_PENDING, INSTANCE_DATA_DELETED, "
"INSTANCE_META_AND_DATA_DELETED");
}

auto [code, msg] = service->set_instance_recycled_status(instance_id, recycled_status);
return http_json_reply(code, msg);
}

HttpResponse process_abort_txn(MetaServiceImpl* service, brpc::Controller* ctrl) {
AbortTxnRequest req;
PARSE_MESSAGE_OR_RETURN(ctrl, req);
Expand Down
3 changes: 3 additions & 0 deletions cloud/src/common/http_helper.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,9 @@ const std::unordered_map<std::string_view, HttpHandlerInfo>& get_http_handlers()
[[maybe_unused]] HttpResponse process_check_instance(RecyclerServiceImpl* service,
brpc::Controller* cntl);

[[maybe_unused]] HttpResponse process_set_instance_recycled_status(RecyclerServiceImpl* service,
brpc::Controller* ctrl);

[[maybe_unused]] HttpResponse process_check_job_info(RecyclerServiceImpl* service,
brpc::Controller* cntl);

Expand Down
217 changes: 140 additions & 77 deletions cloud/src/recycler/recycler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -844,6 +844,10 @@ int InstanceRecycler::do_recycle() {
int InstanceRecycler::recycle_deleted_instance() {
LOG_WARNING("begin to recycle deleted instance").tag("instance_id", instance_id_);

if (instance_info_.recycled_status() == InstanceInfoPB::INSTANCE_META_AND_DATA_DELETED) {
return 0;
}

int ret = 0;
auto start_time = steady_clock::now();

Expand All @@ -854,39 +858,6 @@ int InstanceRecycler::recycle_deleted_instance() {
<< "s, instance_id=" << instance_id_;
};

// Step 1: Recycle tmp rowsets (contains ref count but txn is not committed)
auto recycle_tmp_rowsets_with_mark_delete_enabled = [&]() -> int {
int res = recycle_tmp_rowsets();
if (res == 0 && config::enable_mark_delete_rowset_before_recycle) {
// If mark_delete_rowset_before_recycle is enabled, we will mark delete rowsets before recycling them,
// so we need to recycle tmp rowsets again to make sure all rowsets in recycle space are marked for
// deletion, otherwise we may meet some corner cases that some rowsets are not marked for deletion
// and cannot be recycled.
res = recycle_tmp_rowsets();
}
return res;
};
if (recycle_tmp_rowsets_with_mark_delete_enabled() != 0) {
LOG_WARNING("failed to recycle tmp rowsets").tag("instance_id", instance_id_);
ret = -1;
return -1;
}

// Step 2: Recycle versioned rowsets in recycle space (already marked for deletion)
if (recycle_versioned_rowsets() != 0) {
LOG_WARNING("failed to recycle versioned rowsets").tag("instance_id", instance_id_);
ret = -1;
return -1;
}

// Step 3: Recycle operation logs (can recycle logs not referenced by snapshots)
if (recycle_operation_logs() != 0) {
LOG_WARNING("failed to recycle operation logs").tag("instance_id", instance_id_);
ret = -1;
return -1;
}

// Step 4: Check if there are still cluster snapshots
bool has_snapshots = false;
if (has_cluster_snapshots(&has_snapshots) != 0) {
LOG(WARNING) << "check instance cluster snapshots failed, instance_id=" << instance_id_;
Expand All @@ -897,41 +868,79 @@ int InstanceRecycler::recycle_deleted_instance() {
return 0;
}

bool snapshot_enabled = instance_info().has_snapshot_switch_status() &&
instance_info().snapshot_switch_status() !=
SnapshotSwitchStatus::SNAPSHOT_SWITCH_DISABLED;
if (snapshot_enabled) {
bool has_unrecycled_rowsets = false;
if (recycle_ref_rowsets(&has_unrecycled_rowsets) != 0) {
LOG_WARNING("failed to recycle ref rowsets").tag("instance_id", instance_id_);
if (instance_info_.recycled_status() < InstanceInfoPB::INSTANCE_DATA_DELETED) {
auto recycle_tmp_rowsets_with_mark_delete_enabled = [&]() -> int {
int res = recycle_tmp_rowsets();
if (res == 0 && config::enable_mark_delete_rowset_before_recycle) {
// If mark_delete_rowset_before_recycle is enabled, we will mark delete rowsets before recycling them,
// so we need to recycle tmp rowsets again to make sure all rowsets in recycle space are marked for
// deletion, otherwise we may meet some corner cases that some rowsets are not marked for deletion
// and cannot be recycled.
res = recycle_tmp_rowsets();
}
return res;
};
if (recycle_tmp_rowsets_with_mark_delete_enabled() != 0) {
LOG_WARNING("failed to recycle tmp rowsets").tag("instance_id", instance_id_);
ret = -1;
return -1;
} else if (has_unrecycled_rowsets) {
LOG_INFO("instance has referenced rowsets, skip recycling")
.tag("instance_id", instance_id_);
return ret;
}
} else { // delete all remote data if snapshot is disabled
for (auto& [_, accessor] : accessor_map_) {
if (stopped()) {

if (recycle_versioned_rowsets() != 0) {
LOG_WARNING("failed to recycle versioned rowsets").tag("instance_id", instance_id_);
ret = -1;
return -1;
}

if (recycle_operation_logs() != 0) {
LOG_WARNING("failed to recycle operation logs").tag("instance_id", instance_id_);
ret = -1;
return -1;
}

bool snapshot_enabled = instance_info().has_snapshot_switch_status() &&
instance_info().snapshot_switch_status() !=
SnapshotSwitchStatus::SNAPSHOT_SWITCH_DISABLED;
if (snapshot_enabled) {
bool has_unrecycled_rowsets = false;
if (recycle_ref_rowsets(&has_unrecycled_rowsets) != 0) {
LOG_WARNING("failed to recycle ref rowsets").tag("instance_id", instance_id_);
ret = -1;
return -1;
} else if (has_unrecycled_rowsets) {
LOG_INFO("instance has referenced rowsets, skip recycling")
.tag("instance_id", instance_id_);
return ret;
}
} else { // delete all remote data if snapshot is disabled
for (auto& [_, accessor] : accessor_map_) {
if (stopped()) {
return ret;
}

LOG(INFO) << "begin to delete all objects in " << accessor->uri();
int del_ret = accessor->delete_all();
if (del_ret == 0) {
LOG(INFO) << "successfully delete all objects in " << accessor->uri();
} else if (del_ret != 1) { // no need to log, because S3Accessor has logged this error
// If `del_ret == 1`, it can be considered that the object data has been recycled by cloud platform,
// so the recycling has been successful.
ret = -1;
LOG(INFO) << "begin to delete all objects in " << accessor->uri();
int del_ret = accessor->delete_all();
if (del_ret == 0) {
LOG(INFO) << "successfully delete all objects in " << accessor->uri();
} else if (del_ret !=
1) { // no need to log, because S3Accessor has logged this error
// If `del_ret == 1`, it can be considered that the object data has been recycled by cloud platform,
// so the recycling has been successful.
ret = -1;
}
}

if (ret != 0) {
LOG(WARNING) << "failed to delete all data of deleted instance=" << instance_id_;
return ret;
}
}

if (ret != 0) {
LOG(WARNING) << "failed to delete all data of deleted instance=" << instance_id_;
return ret;
if (update_instance_recycled_status(InstanceInfoPB::INSTANCE_DATA_DELETED) != 0) {
ret = -1;
return -1;
}
return 0;
}

// Check successor instance, if exists, skip deleting kv because successor instance may still need the data in kv
Expand Down Expand Up @@ -1026,29 +1035,83 @@ int InstanceRecycler::recycle_deleted_instance() {
ret = -1;
}

if (ret == 0) {
// remove instance kv
// ATTN: MUST ensure that cloud platform won't regenerate the same instance id
err = txn_kv_->create_txn(&txn);
if (err != TxnErrorCode::TXN_OK) {
LOG(WARNING) << "failed to create txn";
ret = -1;
return ret;
}
std::string key;
instance_key({instance_id_}, &key);
txn->atomic_add(system_meta_service_instance_update_key(), 1);
txn->remove(key);
err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
LOG(WARNING) << "failed to delete instance kv, instance_id=" << instance_id_
<< " err=" << err;
ret = -1;
}
if (ret == 0 &&
update_instance_recycled_status(InstanceInfoPB::INSTANCE_META_AND_DATA_DELETED) != 0) {
ret = -1;
}

return ret;
}

int InstanceRecycler::update_instance_recycled_status(
InstanceInfoPB::InstanceDeletedStatus recycled_status) {
std::unique_ptr<Transaction> txn;
TxnErrorCode err = txn_kv_->create_txn(&txn);
if (err != TxnErrorCode::TXN_OK) {
LOG_WARNING("failed to create txn when updating instance recycled status")
.tag("instance_id", instance_id_)
.tag("recycled_status", InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status))
.tag("err", err);
return -1;
}

std::string key = instance_key({instance_id_});
std::string value;
err = txn->get(key, &value);
if (err != TxnErrorCode::TXN_OK) {
LOG_WARNING("failed to get instance when updating instance recycled status")
.tag("instance_id", instance_id_)
.tag("recycled_status", InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status))
.tag("err", err);
return -1;
}

InstanceInfoPB instance;
if (!instance.ParseFromString(value)) {
LOG_WARNING("failed to parse InstanceInfoPB when updating instance recycled status")
.tag("instance_id", instance_id_)
.tag("recycled_status",
InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status));
return -1;
}
if (instance.status() != InstanceInfoPB::DELETED) {
LOG_WARNING("instance is not deleted when updating instance recycled status")
.tag("instance_id", instance_id_)
.tag("status", instance.status())
.tag("recycled_status",
InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status));
return -1;
}

instance.set_recycled_status(recycled_status);
instance.set_last_recycle_time_ms(
duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
if (!instance.SerializeToString(&value)) {
LOG_WARNING("failed to serialize InstanceInfoPB when updating instance recycled status")
.tag("instance_id", instance_id_)
.tag("recycled_status",
InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status));
return -1;
}

txn->atomic_add(system_meta_service_instance_update_key(), 1);
txn->put(key, value);
err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
LOG_WARNING("failed to commit instance recycled status")
.tag("instance_id", instance_id_)
.tag("recycled_status", InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status))
.tag("err", err);
return -1;
}

instance_info_.Swap(&instance);
LOG_INFO("updated instance recycled status")
.tag("instance_id", instance_id_)
.tag("recycled_status", InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status));
return 0;
}

int InstanceRecycler::check_rowset_exists(int64_t tablet_id, const std::string& rowset_id,
bool* exists, PackedFileRecycleStats* stats) {
if (exists == nullptr) {
Expand Down
2 changes: 2 additions & 0 deletions cloud/src/recycler/recycler.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -288,6 +288,8 @@ class InstanceRecycler {
// returns 0 for success otherwise error
int recycle_deleted_instance();

int update_instance_recycled_status(InstanceInfoPB::InstanceDeletedStatus recycled_status);

// scan and recycle expired indexes:
// 1. dropped table, dropped mv
// 2. half-successtable/index when create
Expand Down
59 changes: 59 additions & 0 deletions cloud/src/recycler/recycler_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
#include <rapidjson/stringbuffer.h>

#include <algorithm>
#include <chrono>
#include <functional>
#include <memory>
#include <numeric>
Expand Down Expand Up @@ -398,6 +399,64 @@ void RecyclerServiceImpl::check_instance(const std::string& instance_id, MetaSer
}
}

std::pair<MetaServiceCode, std::string> RecyclerServiceImpl::set_instance_recycled_status(
const std::string& instance_id, InstanceInfoPB::InstanceDeletedStatus recycled_status) {
std::unique_ptr<Transaction> txn;
TxnErrorCode err = txn_kv_->create_txn(&txn);
if (err != TxnErrorCode::TXN_OK) {
std::string msg = fmt::format("failed to create txn, err={}", err);
LOG(WARNING) << msg << " instance_id=" << instance_id;
return {MetaServiceCode::KV_TXN_CREATE_ERR, std::move(msg)};
}

std::string key = instance_key({instance_id});
std::string value;
err = txn->get(key, &value);
if (err != TxnErrorCode::TXN_OK) {
std::string msg =
fmt::format("failed to get instance, instance_id={}, err={}", instance_id, err);
LOG(WARNING) << msg;
return {MetaServiceCode::KV_TXN_GET_ERR, std::move(msg)};
}

InstanceInfoPB instance;
if (!instance.ParseFromString(value)) {
std::string msg = fmt::format("malformed instance info, key={}", hex(key));
LOG(WARNING) << msg;
return {MetaServiceCode::PROTOBUF_PARSE_ERR, std::move(msg)};
}
if (instance.status() != InstanceInfoPB::DELETED) {
std::string msg = fmt::format(
"failed to set instance recycle status, instance is not deleted, instance_id={}",
instance_id);
LOG(WARNING) << msg;
return {MetaServiceCode::INVALID_ARGUMENT, std::move(msg)};
}

instance.set_recycled_status(recycled_status);
instance.set_last_recycle_time_ms(std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count());
if (!instance.SerializeToString(&value)) {
std::string msg = "failed to serialize InstanceInfoPB";
LOG(WARNING) << msg << " instance_id=" << instance_id;
return {MetaServiceCode::PROTOBUF_SERIALIZE_ERR, std::move(msg)};
}

txn->atomic_add(system_meta_service_instance_update_key(), 1);
txn->put(key, value);
err = txn->commit();
if (err != TxnErrorCode::TXN_OK) {
std::string msg = fmt::format("failed to commit kv txn, err={}", err);
LOG(WARNING) << msg << " instance_id=" << instance_id;
return {MetaServiceCode::KV_TXN_COMMIT_ERR, std::move(msg)};
}

LOG(INFO) << "set instance recycle status, instance_id=" << instance_id
<< " recycled_status=" << InstanceInfoPB::InstanceDeletedStatus_Name(recycled_status);
return {MetaServiceCode::OK, "OK"};
}

void recycle_copy_jobs(const std::shared_ptr<TxnKv>& txn_kv, const std::string& instance_id,
MetaServiceCode& code, std::string& msg,
RecyclerThreadPoolGroup thread_pool_group,
Expand Down
Loading
Loading