// 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 "recycler/recycler.h"

#include <brpc/server.h>

#include <butil/endpoint.h>

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <atomic>

#include <chrono>

#include <cstddef>

#include <cstdint>

#include <deque>

#include <numeric>

#include <string>

#include <string_view>

#include "common/stopwatch.h"

#include "meta-service/meta_service.h"

#include "meta-service/meta_service_helper.h"

#include "meta-service/meta_service_schema.h"

#include "meta-service/txn_kv.h"

#include "meta-service/txn_kv_error.h"

#include "recycler/checker.h"

#include "recycler/hdfs_accessor.h"

#include "recycler/s3_accessor.h"

#include "recycler/storage_vault_accessor.h"

#ifdef UNIT_TEST

#include "../test/mock_accessor.h"

#endif

#include "common/bvars.h"

#include "common/config.h"

#include "common/encryption_util.h"

#include "common/logging.h"

#include "common/simple_thread_pool.h"

#include "common/util.h"

#include "cpp/sync_point.h"

#include "meta-service/keys.h"

#include "recycler/recycler_service.h"

#include "recycler/sync_executor.h"

#include "recycler/util.h"

namespace doris::cloud {

using namespace std::chrono;

// return 0 for success get a key, 1 for key not found, negative for error

[[maybe_unused]] static int txn_get(TxnKv* txn_kv, std::string_view key, std::string& val) {

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

return -1;

}

switch (txn->get(key, &val, true)) {

case TxnErrorCode::TXN_OK:

return 0;

case TxnErrorCode::TXN_KEY_NOT_FOUND:

return 1;

default:

return -1;

};

}

// 0 for success, negative for error

static int txn_get(TxnKv* txn_kv, std::string_view begin, std::string_view end,

std::unique_ptr<RangeGetIterator>& it) {

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

return -1;

}

switch (txn->get(begin, end, &it, true)) {

case TxnErrorCode::TXN_OK:

return 0;

case TxnErrorCode::TXN_KEY_NOT_FOUND:

return 1;

default:

return -1;

};

}

// return 0 for success otherwise error

static int txn_remove(TxnKv* txn_kv, std::vector<std::string_view> keys) {

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

return -1;

}

for (auto k : keys) {

txn->remove(k);

}

switch (txn->commit()) {

case TxnErrorCode::TXN_OK:

return 0;

case TxnErrorCode::TXN_CONFLICT:

return -1;

default:

return -1;

}

}

// return 0 for success otherwise error

static int txn_remove(TxnKv* txn_kv, std::vector<std::string> keys) {

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

return -1;

}

for (auto& k : keys) {

txn->remove(k);

}

switch (txn->commit()) {

case TxnErrorCode::TXN_OK:

return 0;

case TxnErrorCode::TXN_CONFLICT:

return -1;

default:

return -1;

}

}

// return 0 for success otherwise error

[[maybe_unused]] static int txn_remove(TxnKv* txn_kv, std::string_view begin,

std::string_view end) {

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

return -1;

}

txn->remove(begin, end);

switch (txn->commit()) {

case TxnErrorCode::TXN_OK:

return 0;

case TxnErrorCode::TXN_CONFLICT:

return -1;

default:

return -1;

}

}

static inline void check_recycle_task(const std::string& instance_id, const std::string& task_name,

int64_t num_scanned, int64_t num_recycled,

int64_t start_time) {

if ((num_scanned % 10000) == 0 && (num_scanned > 0)) [[unlikely]] {

int64_t cost =

duration_cast<seconds>(steady_clock::now().time_since_epoch()).count() - start_time;

if (cost > config::recycle_task_threshold_seconds) {

LOG_INFO("recycle task cost too much time cost={}s", cost)

.tag("instance_id", instance_id)

.tag("task", task_name)

.tag("num_scanned", num_scanned)

.tag("num_recycled", num_recycled);

}

}

return;

}

Recycler::Recycler(std::shared_ptr<TxnKv> txn_kv) : txn_kv_(std::move(txn_kv)) {

ip_port_ = std::string(butil::my_ip_cstr()) + ":" + std::to_string(config::brpc_listen_port);

auto s3_producer_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism,

"s3_producer_pool");

s3_producer_pool->start();

auto recycle_tablet_pool = std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism,

"recycle_tablet_pool");

recycle_tablet_pool->start();

auto group_recycle_function_pool = std::make_shared<SimpleThreadPool>(

config::recycle_pool_parallelism, "group_recycle_function_pool");

group_recycle_function_pool->start();

_thread_pool_group =

RecyclerThreadPoolGroup(std::move(s3_producer_pool), std::move(recycle_tablet_pool),

std::move(group_recycle_function_pool));

txn_lazy_committer_ = std::make_shared<TxnLazyCommitter>(txn_kv_);

}

Recycler::~Recycler() {

if (!stopped()) {

stop();

}

}

void Recycler::instance_scanner_callback() {

// sleep 60 seconds before scheduling for the launch procedure to complete:

// some bad hdfs connection may cause some log to stdout stderr

// which may pollute .out file and affect the script to check success

std::this_thread::sleep_for(

std::chrono::seconds(config::recycler_sleep_before_scheduling_seconds));

while (!stopped()) {

std::vector<InstanceInfoPB> instances;

get_all_instances(txn_kv_.get(), instances);

// TODO(plat1ko): delete job recycle kv of non-existent instances

LOG(INFO) << "Recycler get instances: " << [&instances] {

std::stringstream ss;

for (auto& i : instances) ss << ' ' << i.instance_id();

return ss.str();

}();

if (!instances.empty()) {

// enqueue instances

std::lock_guard lock(mtx_);

for (auto& instance : instances) {

if (instance_filter_.filter_out(instance.instance_id())) continue;

auto [_, success] = pending_instance_set_.insert(instance.instance_id());

// skip instance already in pending queue

if (success) {

pending_instance_queue_.push_back(std::move(instance));

}

}

pending_instance_cond_.notify_all();

}

{

std::unique_lock lock(mtx_);

notifier_.wait_for(lock, std::chrono::seconds(config::recycle_interval_seconds),

[&]() { return stopped(); });

}

}

}

void Recycler::recycle_callback() {

while (!stopped()) {

InstanceInfoPB instance;

{

std::unique_lock lock(mtx_);

pending_instance_cond_.wait(

lock, [&]() { return !pending_instance_queue_.empty() || stopped(); });

if (stopped()) {

return;

}

instance = std::move(pending_instance_queue_.front());

pending_instance_queue_.pop_front();

pending_instance_set_.erase(instance.instance_id());

}

auto& instance_id = instance.instance_id();

{

std::lock_guard lock(mtx_);

// skip instance in recycling

if (recycling_instance_map_.count(instance_id)) continue;

}

auto instance_recycler = std::make_shared<InstanceRecycler>(

txn_kv_, instance, _thread_pool_group, txn_lazy_committer_);

if (int r = instance_recycler->init(); r != 0) {

LOG(WARNING) << "failed to init instance recycler, instance_id=" << instance_id

<< " ret=" << r;

continue;

}

std::string recycle_job_key;

job_recycle_key({instance_id}, &recycle_job_key);

int ret = prepare_instance_recycle_job(txn_kv_.get(), recycle_job_key, instance_id,

ip_port_, config::recycle_interval_seconds * 1000);

if (ret != 0) { // Prepare failed

LOG(WARNING) << "failed to prepare recycle_job, instance_id=" << instance_id

<< " ret=" << ret;

continue;

} else {

std::lock_guard lock(mtx_);

recycling_instance_map_.emplace(instance_id, instance_recycler);

}

if (stopped()) return;

LOG_INFO("begin to recycle instance").tag("instance_id", instance_id);

auto ctime_ms = duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();

ret = instance_recycler->do_recycle();

// If instance recycler has been aborted, don't finish this job

if (!instance_recycler->stopped()) {

finish_instance_recycle_job(txn_kv_.get(), recycle_job_key, instance_id, ip_port_,

ret == 0, ctime_ms);

}

{

std::lock_guard lock(mtx_);

recycling_instance_map_.erase(instance_id);

}

auto elpased_ms =

duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count() -

ctime_ms;

LOG_INFO("finish recycle instance")

.tag("instance_id", instance_id)

.tag("cost_ms", elpased_ms);

}

}

void Recycler::lease_recycle_jobs() {

while (!stopped()) {

std::vector<std::string> instances;

instances.reserve(recycling_instance_map_.size());

{

std::lock_guard lock(mtx_);

for (auto& [id, _] : recycling_instance_map_) {

instances.push_back(id);

}

}

for (auto& i : instances) {

std::string recycle_job_key;

job_recycle_key({i}, &recycle_job_key);

int ret = lease_instance_recycle_job(txn_kv_.get(), recycle_job_key, i, ip_port_);

if (ret == 1) {

std::lock_guard lock(mtx_);

if (auto it = recycling_instance_map_.find(i);

it != recycling_instance_map_.end()) {

it->second->stop();

}

}

}

{

std::unique_lock lock(mtx_);

notifier_.wait_for(lock,

std::chrono::milliseconds(config::recycle_job_lease_expired_ms / 3),

[&]() { return stopped(); });

}

}

}

void Recycler::check_recycle_tasks() {

while (!stopped()) {

std::unordered_map<std::string, std::shared_ptr<InstanceRecycler>> recycling_instance_map;

{

std::lock_guard lock(mtx_);

recycling_instance_map = recycling_instance_map_;

}

for (auto& entry : recycling_instance_map) {

entry.second->check_recycle_tasks();

}

std::unique_lock lock(mtx_);

notifier_.wait_for(lock, std::chrono::seconds(config::check_recycle_task_interval_seconds),

[&]() { return stopped(); });

}

}

int Recycler::start(brpc::Server* server) {

instance_filter_.reset(config::recycle_whitelist, config::recycle_blacklist);

if (config::enable_checker) {

checker_ = std::make_unique<Checker>(txn_kv_);

int ret = checker_->start();

std::string msg;

if (ret != 0) {

msg = "failed to start checker";

LOG(ERROR) << msg;

std::cerr << msg << std::endl;

return ret;

}

msg = "checker started";

LOG(INFO) << msg;

std::cout << msg << std::endl;

}

if (server) {

// Add service

auto recycler_service =

new RecyclerServiceImpl(txn_kv_, this, checker_.get(), txn_lazy_committer_);

server->AddService(recycler_service, brpc::SERVER_OWNS_SERVICE);

}

workers_.emplace_back([this] { instance_scanner_callback(); });

for (int i = 0; i < config::recycle_concurrency; ++i) {

workers_.emplace_back([this] { recycle_callback(); });

}

workers_.emplace_back(std::mem_fn(&Recycler::lease_recycle_jobs), this);

workers_.emplace_back(std::mem_fn(&Recycler::check_recycle_tasks), this);

return 0;

}

void Recycler::stop() {

stopped_ = true;

notifier_.notify_all();

pending_instance_cond_.notify_all();

{

std::lock_guard lock(mtx_);

for (auto& [_, recycler] : recycling_instance_map_) {

recycler->stop();

}

}

for (auto& w : workers_) {

if (w.joinable()) w.join();

}

if (checker_) {

checker_->stop();

}

}

class InstanceRecycler::InvertedIndexIdCache {

public:

InvertedIndexIdCache(std::string instance_id, std::shared_ptr<TxnKv> txn_kv)

: instance_id_(std::move(instance_id)), txn_kv_(std::move(txn_kv)) {}

// Return 0 if success, 1 if schema kv not found, negative for error

int get(int64_t index_id, int32_t schema_version, InvertedIndexInfo& res) {

{

std::lock_guard lock(mtx_);

if (schemas_without_inverted_index_.count({index_id, schema_version})) {

return 0;

}

if (auto it = inverted_index_id_map_.find({index_id, schema_version});

it != inverted_index_id_map_.end()) {

res = it->second;

return 0;

}

}

// Get schema from kv

// TODO(plat1ko): Single flight

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv_->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

LOG(WARNING) << "failed to create txn, err=" << err;

return -1;

}

auto schema_key = meta_schema_key({instance_id_, index_id, schema_version});

ValueBuf val_buf;

err = cloud::get(txn.get(), schema_key, &val_buf);

if (err != TxnErrorCode::TXN_OK) {

LOG(WARNING) << "failed to get schema, err=" << err;

return static_cast<int>(err);

}

doris::TabletSchemaCloudPB schema;

if (!parse_schema_value(val_buf, &schema)) {

LOG(WARNING) << "malformed schema value, key=" << hex(schema_key);

return -1;

}

if (schema.index_size() > 0) {

InvertedIndexStorageFormatPB index_format = InvertedIndexStorageFormatPB::V1;

if (schema.has_inverted_index_storage_format()) {

index_format = schema.inverted_index_storage_format();

}

res.first = index_format;

res.second.reserve(schema.index_size());

for (auto& i : schema.index()) {

if (i.has_index_type() && i.index_type() == IndexType::INVERTED) {

res.second.push_back(std::make_pair(i.index_id(), i.index_suffix_name()));

}

}

}

insert(index_id, schema_version, res);

return 0;

}

// Empty `ids` means this schema has no inverted index

void insert(int64_t index_id, int32_t schema_version, const InvertedIndexInfo& index_info) {

if (index_info.second.empty()) {

TEST_SYNC_POINT("InvertedIndexIdCache::insert1");

std::lock_guard lock(mtx_);

schemas_without_inverted_index_.emplace(index_id, schema_version);

} else {

TEST_SYNC_POINT("InvertedIndexIdCache::insert2");

std::lock_guard lock(mtx_);

inverted_index_id_map_.try_emplace({index_id, schema_version}, index_info);

}

}

private:

std::string instance_id_;

std::shared_ptr<TxnKv> txn_kv_;

std::mutex mtx_;

using Key = std::pair<int64_t, int32_t>; // <index_id, schema_version>

struct HashOfKey {

size_t operator()(const Key& key) const {

size_t seed = 0;

seed = std::hash<int64_t> {}(key.first);

seed = std::hash<int32_t> {}(key.second);

return seed;

}

};

// <index_id, schema_version> -> inverted_index_ids

std::unordered_map<Key, InvertedIndexInfo, HashOfKey> inverted_index_id_map_;

// Store <index_id, schema_version> of schema which doesn't have inverted index

std::unordered_set<Key, HashOfKey> schemas_without_inverted_index_;

};

InstanceRecycler::InstanceRecycler(std::shared_ptr<TxnKv> txn_kv, const InstanceInfoPB& instance,

RecyclerThreadPoolGroup thread_pool_group,

std::shared_ptr<TxnLazyCommitter> txn_lazy_committer)

: txn_kv_(std::move(txn_kv)),

instance_id_(instance.instance_id()),

instance_info_(instance),

inverted_index_id_cache_(std::make_unique<InvertedIndexIdCache>(instance_id_, txn_kv_)),

_thread_pool_group(std::move(thread_pool_group)),

txn_lazy_committer_(std::move(txn_lazy_committer)) {};

InstanceRecycler::~InstanceRecycler() = default;

int InstanceRecycler::init_obj_store_accessors() {

for (const auto& obj_info : instance_info_.obj_info()) {

#ifdef UNIT_TEST

auto accessor = std::make_shared<MockAccessor>();

#else

auto s3_conf = S3Conf::from_obj_store_info(obj_info);

if (!s3_conf) {

LOG(WARNING) << "failed to init object accessor, instance_id=" << instance_id_;

return -1;

}

std::shared_ptr<S3Accessor> accessor;

int ret = S3Accessor::create(std::move(*s3_conf), &accessor);

if (ret != 0) {

LOG(WARNING) << "failed to init s3 accessor. instance_id=" << instance_id_

<< " resource_id=" << obj_info.id();

return ret;

}

#endif

accessor_map_.emplace(obj_info.id(), std::move(accessor));

}

return 0;

}

int InstanceRecycler::init_storage_vault_accessors() {

if (instance_info_.resource_ids().empty()) {

return 0;

}

FullRangeGetIteratorOptions opts(txn_kv_);

opts.prefetch = true;

auto it = txn_kv_->full_range_get(storage_vault_key({instance_id_, ""}),

storage_vault_key({instance_id_, "\xff"}), std::move(opts));

for (auto kv = it->next(); kv.has_value(); kv = it->next()) {

auto [k, v] = *kv;

StorageVaultPB vault;

if (!vault.ParseFromArray(v.data(), v.size())) {

LOG(WARNING) << "malformed storage vault, unable to deserialize key=" << hex(k);

return -1;

}

std::string recycler_storage_vault_white_list = accumulate(

config::recycler_storage_vault_white_list.begin(),

config::recycler_storage_vault_white_list.end(), std::string(),

[](std::string a, std::string b) { return a + (a.empty() ? "" : ",") + b; });

LOG_INFO("config::recycler_storage_vault_white_list")

.tag("", recycler_storage_vault_white_list);

if (!config::recycler_storage_vault_white_list.empty()) {

if (auto it = std::find(config::recycler_storage_vault_white_list.begin(),

config::recycler_storage_vault_white_list.end(), vault.name());

it == config::recycler_storage_vault_white_list.end()) {

LOG_WARNING(

"failed to init accessor for vault because this vault is not in "

"config::recycler_storage_vault_white_list. ")

.tag(" vault name:", vault.name())

.tag(" config::recycler_storage_vault_white_list:",

recycler_storage_vault_white_list);

continue;

}

}

TEST_SYNC_POINT_CALLBACK("InstanceRecycler::init_storage_vault_accessors.mock_vault",

&accessor_map_, &vault);

if (vault.has_hdfs_info()) {

auto accessor = std::make_shared<HdfsAccessor>(vault.hdfs_info());

int ret = accessor->init();

if (ret != 0) {

LOG(WARNING) << "failed to init hdfs accessor. instance_id=" << instance_id_

<< " resource_id=" << vault.id() << " name=" << vault.name()

<< " hdfs_vault=" << vault.hdfs_info().ShortDebugString();

continue;

}

LOG(INFO) << "succeed to init hdfs accessor. instance_id=" << instance_id_

<< " resource_id=" << vault.id() << " name=" << vault.name()

<< " hdfs_vault=" << vault.hdfs_info().ShortDebugString();

accessor_map_.emplace(vault.id(), std::move(accessor));

} else if (vault.has_obj_info()) {

auto s3_conf = S3Conf::from_obj_store_info(vault.obj_info());

if (!s3_conf) {

LOG(WARNING) << "failed to init object accessor, invalid conf, instance_id="

<< instance_id_ << " s3_vault=" << vault.obj_info().ShortDebugString();

continue;

}

std::shared_ptr<S3Accessor> accessor;

int ret = S3Accessor::create(*s3_conf, &accessor);

if (ret != 0) {

LOG(WARNING) << "failed to init s3 accessor. instance_id=" << instance_id_

<< " resource_id=" << vault.id() << " name=" << vault.name()

<< " ret=" << ret

<< " s3_vault=" << encryt_sk(vault.obj_info().ShortDebugString());

continue;

}

LOG(INFO) << "succeed to init s3 accessor. instance_id=" << instance_id_

<< " resource_id=" << vault.id() << " name=" << vault.name() << " ret=" << ret

<< " s3_vault=" << encryt_sk(vault.obj_info().ShortDebugString());

accessor_map_.emplace(vault.id(), std::move(accessor));

}

}

if (!it->is_valid()) {

LOG_WARNING("failed to get storage vault kv");

return -1;

}

if (accessor_map_.empty()) {

LOG(WARNING) << "no accessors for instance=" << instance_id_;

return -2;

}

LOG_INFO("finish init instance recycler number_accessors={} instance=", accessor_map_.size(),

instance_id_);

return 0;

}

int InstanceRecycler::init() {

int ret = init_obj_store_accessors();

if (ret != 0) {

return ret;

}

return init_storage_vault_accessors();

}

template <typename... Func>

auto task_wrapper(Func... funcs) -> std::function<int()> {

return [funcs...]() {

return [](std::initializer_list<int> ret_vals) {

int i = 0;

for (int ret : ret_vals) {

if (ret != 0) {

i = ret;

}

}

return i;

}({funcs()...});

};

}

int InstanceRecycler::do_recycle() {

TEST_SYNC_POINT("InstanceRecycler.do_recycle");

if (instance_info_.status() == InstanceInfoPB::DELETED) {

return recycle_deleted_instance();

} else if (instance_info_.status() == InstanceInfoPB::NORMAL) {

SyncExecutor<int> sync_executor(_thread_pool_group.group_recycle_function_pool,

fmt::format("instance id {}", instance_id_),

[](int r) { return r != 0; });

sync_executor

.add(task_wrapper( // dropped table and dropped partition need to be recycled in series

// becase they may both recycle the same set of tablets

// recycle dropped table or idexes(mv, rollup)

[this]() -> int { return InstanceRecycler::recycle_indexes(); },

// recycle dropped partitions

[this]() -> int { return InstanceRecycler::recycle_partitions(); }))

.add(task_wrapper(

[this]() -> int { return InstanceRecycler::recycle_tmp_rowsets(); }))

.add(task_wrapper([this]() -> int { return InstanceRecycler::recycle_rowsets(); }))

.add(task_wrapper(

[this]() { return InstanceRecycler::abort_timeout_txn(); },

[this]() { return InstanceRecycler::recycle_expired_txn_label(); }))

.add(task_wrapper([this]() { return InstanceRecycler::recycle_copy_jobs(); }))

.add(task_wrapper([this]() { return InstanceRecycler::recycle_stage(); }))

.add(task_wrapper(

[this]() { return InstanceRecycler::recycle_expired_stage_objects(); }))

.add(task_wrapper([this]() { return InstanceRecycler::recycle_versions(); }));

bool finished = true;

std::vector<int> rets = sync_executor.when_all(&finished);

for (int ret : rets) {

if (ret != 0) {

return ret;

}

}

return finished ? 0 : -1;

} else {

LOG(WARNING) << "invalid instance status: " << instance_info_.status()

<< " instance_id=" << instance_id_;

return -1;

}

}

/**

* 1. delete all remote data

* 2. delete all kv

* 3. remove instance kv

*/

int InstanceRecycler::recycle_deleted_instance() {

LOG_INFO("begin to recycle deleted instance").tag("instance_id", instance_id_);

int ret = 0;

auto start_time = steady_clock::now();

std::unique_ptr<int, std::function<void(int*)>> defer_log_statistics((int*)0x01, [&](int*) {

auto cost = duration<float>(steady_clock::now() - start_time).count();

LOG(INFO) << (ret == 0 ? "successfully" : "failed to")

<< " recycle deleted instance, cost=" << cost

<< "s, instance_id=" << instance_id_;

});

// delete all remote data

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;

}

}

if (ret != 0) {

LOG(WARNING) << "failed to delete all data of deleted instance=" << instance_id_;

return ret;

}

// delete all kv

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv_->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

LOG(WARNING) << "failed to create txn";

ret = -1;

return -1;

}

LOG(INFO) << "begin to delete all kv, instance_id=" << instance_id_;

// delete kv before deleting objects to prevent the checker from misjudging data loss

std::string start_txn_key = txn_key_prefix(instance_id_);

std::string end_txn_key = txn_key_prefix(instance_id_ + '\x00');

txn->remove(start_txn_key, end_txn_key);

std::string start_version_key = version_key_prefix(instance_id_);

std::string end_version_key = version_key_prefix(instance_id_ + '\x00');

txn->remove(start_version_key, end_version_key);

std::string start_meta_key = meta_key_prefix(instance_id_);

std::string end_meta_key = meta_key_prefix(instance_id_ + '\x00');

txn->remove(start_meta_key, end_meta_key);

std::string start_recycle_key = recycle_key_prefix(instance_id_);

std::string end_recycle_key = recycle_key_prefix(instance_id_ + '\x00');

txn->remove(start_recycle_key, end_recycle_key);

std::string start_stats_tablet_key = stats_tablet_key({instance_id_, 0, 0, 0, 0});

std::string end_stats_tablet_key = stats_tablet_key({instance_id_, INT64_MAX, 0, 0, 0});

txn->remove(start_stats_tablet_key, end_stats_tablet_key);

std::string start_copy_key = copy_key_prefix(instance_id_);

std::string end_copy_key = copy_key_prefix(instance_id_ + '\x00');

txn->remove(start_copy_key, end_copy_key);

// should not remove job key range, because we need to reserve job recycle kv

// 0:instance_id 1:table_id 2:index_id 3:part_id 4:tablet_id

std::string start_job_tablet_key = job_tablet_key({instance_id_, 0, 0, 0, 0});

std::string end_job_tablet_key = job_tablet_key({instance_id_, INT64_MAX, 0, 0, 0});

txn->remove(start_job_tablet_key, end_job_tablet_key);

StorageVaultKeyInfo key_info0 {instance_id_, ""};

StorageVaultKeyInfo key_info1 {instance_id_, "\xff"};

std::string start_vault_key = storage_vault_key(key_info0);

std::string end_vault_key = storage_vault_key(key_info1);

txn->remove(start_vault_key, end_vault_key);

err = txn->commit();

if (err != TxnErrorCode::TXN_OK) {

LOG(WARNING) << "failed to delete all kv, instance_id=" << instance_id_ << ", err=" << err;

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

}

}

return ret;

}

int InstanceRecycler::recycle_indexes() {

const std::string task_name = "recycle_indexes";

int64_t num_scanned = 0;

int64_t num_expired = 0;

int64_t num_recycled = 0;

RecycleIndexKeyInfo index_key_info0 {instance_id_, 0};

RecycleIndexKeyInfo index_key_info1 {instance_id_, INT64_MAX};

std::string index_key0;

std::string index_key1;

recycle_index_key(index_key_info0, &index_key0);

recycle_index_key(index_key_info1, &index_key1);

LOG_INFO("begin to recycle indexes").tag("instance_id", instance_id_);

int64_t start_time = duration_cast<seconds>(steady_clock::now().time_since_epoch()).count();

register_recycle_task(task_name, start_time);

std::unique_ptr<int, std::function<void(int*)>> defer_log_statistics((int*)0x01, [&](int*) {

unregister_recycle_task(task_name);

int64_t cost =

duration_cast<seconds>(steady_clock::now().time_since_epoch()).count() - start_time;

LOG_INFO("recycle indexes finished, cost={}s", cost)

.tag("instance_id", instance_id_)

.tag("num_scanned", num_scanned)

.tag("num_expired", num_expired)

.tag("num_recycled", num_recycled);

});

int64_t earlest_ts = std::numeric_limits<int64_t>::max();

auto calc_expiration = [&earlest_ts, this](const RecycleIndexPB& index) {

if (config::force_immediate_recycle) {

return 0L;

}

int64_t expiration = index.expiration() > 0 ? index.expiration() : index.creation_time();

int64_t retention_seconds = config::retention_seconds;

if (index.state() == RecycleIndexPB::DROPPED) {

retention_seconds =

std::min(config::dropped_index_retention_seconds, retention_seconds);

}

int64_t final_expiration = expiration + retention_seconds;

if (earlest_ts > final_expiration) {

earlest_ts = final_expiration;

g_bvar_recycler_recycle_index_earlest_ts.put(instance_id_, earlest_ts);

}

return final_expiration;

};

// Elements in `index_keys` has the same lifetime as `it` in `scan_and_recycle`

std::vector<std::string_view> index_keys;

auto recycle_func = [&, this](std::string_view k, std::string_view v) -> int {

++num_scanned;

RecycleIndexPB index_pb;

if (!index_pb.ParseFromArray(v.data(), v.size())) {

LOG_WARNING("malformed recycle index value").tag("key", hex(k));

return -1;

}

int64_t current_time = ::time(nullptr);

if (current_time < calc_expiration(index_pb)) { // not expired

return 0;

}

++num_expired;

// decode index_id

auto k1 = k;

k1.remove_prefix(1);

std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> out;

decode_key(&k1, &out);

// 0x01 "recycle" ${instance_id} "index" ${index_id} -> RecycleIndexPB

auto index_id = std::get<int64_t>(std::get<0>(out[3]));

LOG(INFO) << "begin to recycle index, instance_id=" << instance_id_

<< " table_id=" << index_pb.table_id() << " index_id=" << index_id

<< " state=" << RecycleIndexPB::State_Name(index_pb.state());

// Change state to RECYCLING

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv_->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

LOG_WARNING("failed to create txn").tag("err", err);

return -1;

}

std::string val;

err = txn->get(k, &val);

if (err ==

TxnErrorCode::TXN_KEY_NOT_FOUND) { // UNKNOWN, maybe recycled or committed, skip it

LOG_INFO("index {} has been recycled or committed", index_id);

return 0;

}

if (err != TxnErrorCode::TXN_OK) {

LOG_WARNING("failed to get kv").tag("key", hex(k)).tag("err", err);

return -1;

}

index_pb.Clear();

if (!index_pb.ParseFromString(val)) {

LOG_WARNING("malformed recycle index value").tag("key", hex(k));

return -1;

}

if (index_pb.state() != RecycleIndexPB::RECYCLING) {

index_pb.set_state(RecycleIndexPB::RECYCLING);

txn->put(k, index_pb.SerializeAsString());

err = txn->commit();

if (err != TxnErrorCode::TXN_OK) {

LOG_WARNING("failed to commit txn").tag("err", err);

return -1;

}

}

if (recycle_tablets(index_pb.table_id(), index_id) != 0) {

LOG_WARNING("failed to recycle tablets under index")

.tag("table_id", index_pb.table_id())

.tag("instance_id", instance_id_)

.tag("index_id", index_id);

return -1;

}

++num_recycled;

check_recycle_task(instance_id_, task_name, num_scanned, num_recycled, start_time);

index_keys.push_back(k);

return 0;

};

auto loop_done = [&index_keys, this]() -> int {

if (index_keys.empty()) return 0;

std::unique_ptr<int, std::function<void(int*)>> defer((int*)0x01,

[&](int*) { index_keys.clear(); });

if (0 != txn_remove(txn_kv_.get(), index_keys)) {

LOG(WARNING) << "failed to delete recycle index kv, instance_id=" << instance_id_;

return -1;

}

return 0;

};

return scan_and_recycle(index_key0, index_key1, std::move(recycle_func), std::move(loop_done));

}

bool check_lazy_txn_finished(std::shared_ptr<TxnKv> txn_kv, const std::string instance_id,

int64_t tablet_id) {

std::unique_ptr<Transaction> txn;

TxnErrorCode err = txn_kv->create_txn(&txn);

if (err != TxnErrorCode::TXN_OK) {

LOG(WARNING) << "failed to create txn, instance_id=" << instance_id

<< " tablet_id=" << tablet_id << " err=" << err;

return false;

}

std::string tablet_idx_key = meta_tablet_idx_key({instance_id, tablet_id});

std::string tablet_idx_val;

err = txn->get(tablet_idx_key, &tablet_idx_val);

if (TxnErrorCode::TXN_OK != err) {

LOG(WARNING) << "failed to get tablet index, instance_id=" << instance_id

<< " tablet_id=" << tablet_id << " err=" << err

<< " key=" << hex(tablet_idx_key);

return false;

}

TabletIndexPB tablet_idx_pb;

if (!tablet_idx_pb.ParseFromString(tablet_idx_val)) {

LOG(WARNING) << "failed to parse tablet_idx_pb, instance_id=" << instance_id

<< " tablet_id=" << tablet_id;

return false;

}

if (!tablet_idx_pb.has_db_id()) {

// In the previous version, the db_id was not set in the index_pb.

// If updating to the version which enable txn lazy commit, the db_id will be set.

LOG(INFO) << "txn index has no db_id, tablet_id=" << tablet_id

<< " instance_id=" << instance_id

<< " tablet_idx_pb=" << tablet_idx_pb.ShortDebugString();

return true;

}

std::string ver_val;

std::string ver_key =

partition_version_key({instance_id, tablet_idx_pb.db_id(), tablet_idx_pb.table_id(),

tablet_idx_pb.partition_id()});

err = txn->get(ver_key, &ver_val);

if (TxnErrorCode::TXN_KEY_NOT_FOUND == err) {

LOG(INFO) << ""

"partition version not found, instance_id="

<< instance_id << " db_id=" << tablet_idx_pb.db_id()

<< " table_id=" << tablet_idx_pb.table_id()

<< " partition_id=" << tablet_idx_pb.partition_id() << " tablet_id=" << tablet_id

<< " key=" << hex(ver_key);

return true;

}

if (TxnErrorCode::TXN_OK != err) {

LOG(WARNING) << "failed to get partition version, instance_id=" << instance_id

<< " db_id=" << tablet_idx_pb.db_id()

<< " table_id=" << tablet_idx_pb.table_id()

<< " partition_id=" << tablet_idx_pb.partition_id()

<< " tablet_id=" << tablet_id << " key=" << hex(ver_key) << " err=" << err;

return false;

}

VersionPB version_pb;

if (!version_pb.ParseFromString(ver_val)) {

LOG(WARNING) << "failed to parse version_pb, instance_id=" << instance_id

<< " db_id=" << tablet_idx_pb.db_id()

<< " table_id=" << tablet_idx_pb.table_id()

<< " partition_id=" << tablet_idx_pb.partition_id()

<< " tablet_id=" << tablet_id << " key=" << hex(ver_key);

return false;

}

if (version_pb.pending_txn_ids_size() > 0) {

TEST_SYNC_POINT_CALLBACK("check_lazy_txn_finished::txn_not_finished");

DCHECK(version_pb.pending_txn_ids_size() == 1);

LOG(WARNING) << "lazy txn not finished, instance_id=" << instance_id

<< " db_id=" << tablet_idx_pb.db_id()

<< " table_id=" << tablet_idx_pb.table_id()

<< " partition_id=" << tablet_idx_pb.partition_id()

<< " tablet_id=" << tablet_id << " txn_id=" << version_pb.pending_txn_ids(0)

<< " key=" << hex(ver_key);

return false;

}

return true;

}