// 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/builtin_service.pb.h>

#include <brpc/server.h>

#include <butil/endpoint.h>

#include <butil/strings/string_split.h>

#include <bvar/status.h>

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <algorithm>

#include <atomic>

#include <chrono>

#include <cstddef>

#include <cstdint>

#include <cstdlib>

#include <deque>

#include <functional>

#include <initializer_list>

#include <memory>

#include <numeric>

#include <optional>

#include <random>

#include <string>

#include <string_view>

#include <thread>

#include <unordered_map>

#include <utility>

#include <variant>

#include "common/defer.h"

#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-store/blob_message.h"

#include "meta-store/meta_reader.h"

#include "meta-store/txn_kv.h"

#include "meta-store/txn_kv_error.h"

#include "meta-store/versioned_value.h"

#include "recycler/checker.h"

#ifdef ENABLE_HDFS_STORAGE_VAULT

#include "recycler/hdfs_accessor.h"

#endif

#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-store/codec.h"

#include "meta-store/document_message.h"

#include "meta-store/keys.h"

#include "recycler/recycler_service.h"

#include "recycler/sync_executor.h"

#include "recycler/util.h"

#include "snapshot/snapshot_manager_factory.h"

namespace doris::cloud {

using namespace std::chrono;

namespace {

int64_t packed_file_retry_sleep_ms() {

const int64_t min_ms = std::max<int64_t>(0, config::packed_file_txn_retry_sleep_min_ms);

const int64_t max_ms = std::max<int64_t>(min_ms, config::packed_file_txn_retry_sleep_max_ms);

thread_local std::mt19937_64 gen(std::random_device {}());

std::uniform_int_distribution<int64_t> dist(min_ms, max_ms);

return dist(gen);

}

void sleep_for_packed_file_retry() {

std::this_thread::sleep_for(std::chrono::milliseconds(packed_file_retry_sleep_ms()));

}

bool is_packed_slice_path(const doris::RowsetMetaCloudPB& rowset, const std::string& path) {

const auto& locations = rowset.packed_slice_locations();

auto it = locations.find(path);

return it != locations.end() && it->second.has_packed_file_path() &&

!it->second.packed_file_path().empty();

}

void add_file_to_delete_if_not_packed(const doris::RowsetMetaCloudPB& rowset,

const std::string& path,

std::vector<std::string>* file_paths) {

if (!is_packed_slice_path(rowset, path)) {

file_paths->push_back(path);

}

}

bool filter_out_instance(const std::string& instance_id) {

if (config::recycle_whitelist.empty()) {

return std::ranges::find(config::recycle_blacklist, instance_id) !=

config::recycle_blacklist.end();

}

return std::ranges::find(config::recycle_whitelist, instance_id) ==

config::recycle_whitelist.end();

}

} // namespace

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

}

}

void scan_restore_job_rowset(

Transaction* txn, const std::string& instance_id, int64_t tablet_id, MetaServiceCode& code,

std::string& msg,

std::vector<std::pair<std::string, doris::RowsetMetaCloudPB>>* restore_job_rs_metas);

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_WARNING("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));

auto resource_mgr = std::make_shared<ResourceManager>(txn_kv_);

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

snapshot_manager_ = create_snapshot_manager(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()) {

if (config::enable_recycler) {

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 (filter_out_instance(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();

}

} else {

LOG(WARNING) << "Skip recycler since enable_recycler is false";

}

{

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;

}

if (!config::enable_recycler) {

LOG(WARNING) << "Skip recycle instance_id=" << instance_id

<< " since enable_recycler is false";

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_WARNING("begin to recycle instance").tag("instance_id", instance_id);

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

g_bvar_recycler_instance_recycle_start_ts.put({instance_id}, ctime_ms);

g_bvar_recycler_instance_recycle_task_status.put({"submitted"}, 1);

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

}

if (instance_recycler->stopped() || ret != 0) {

g_bvar_recycler_instance_recycle_task_status.put({"error"}, 1);

}

{

std::lock_guard lock(mtx_);

recycling_instance_map_.erase(instance_id);

}

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

auto elpased_ms = now - ctime_ms;

g_bvar_recycler_instance_recycle_end_ts.put({instance_id}, now);

g_bvar_recycler_instance_last_round_recycle_duration.put({instance_id}, elpased_ms);

g_bvar_recycler_instance_next_ts.put({instance_id},

now + config::recycle_interval_seconds * 1000);

g_bvar_recycler_instance_recycle_task_status.put({"completed"}, 1);

LOG(INFO) << "recycle instance done, "

<< "instance_id=" << instance_id << " ret=" << ret << " ctime_ms: " << ctime_ms

<< " now: " << now;

g_bvar_recycler_instance_recycle_last_success_ts.put({instance_id}, now);

LOG_WARNING("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) {

g_bvar_recycler_task_max_concurrency.set_value(config::recycle_concurrency);

S3Environment::getInstance();

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

if (config::enable_snapshot_data_migrator) {

snapshot_data_migrator_ = std::make_shared<SnapshotDataMigrator>(txn_kv_);

int ret = snapshot_data_migrator_->start();

if (ret != 0) {

LOG(ERROR) << "failed to start snapshot data migrator";

return ret;

}

LOG(INFO) << "snapshot data migrator started";

}

if (config::enable_snapshot_chain_compactor) {

snapshot_chain_compactor_ = std::make_shared<SnapshotChainCompactor>(txn_kv_);

int ret = snapshot_chain_compactor_->start();

if (ret != 0) {

LOG(ERROR) << "failed to start snapshot chain compactor";

return ret;

}

LOG(INFO) << "snapshot chain compactor started";

}

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

}

if (snapshot_data_migrator_) {

snapshot_data_migrator_->stop();

}

if (snapshot_chain_compactor_) {

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

// For the same index_id, schema_version, res, since `get` is not completely atomic

// one thread has not finished inserting, and another thread has not get the index_id and schema_version,

// resulting in repeated addition and inaccuracy.

// however, this approach can reduce the lock range and sacrifice a bit of meta repeated get to improve concurrency performance.

// repeated addition does not affect correctness.

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::blob_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)),

delete_bitmap_lock_white_list_(std::make_shared<DeleteBitmapLockWhiteList>()),

resource_mgr_(std::make_shared<ResourceManager>(txn_kv_)) {

delete_bitmap_lock_white_list_->init();

resource_mgr_->init();

snapshot_manager_ = create_snapshot_manager(txn_kv_);

// Since the recycler's resource manager could not be notified when instance info changes,

// we need to refresh the instance info here to ensure the resource manager has the latest info.

txn_lazy_committer_->resource_manager()->refresh_instance(instance_id_, instance);

};

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;

}

FullRangeGetOptions 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()) {

#ifdef ENABLE_HDFS_STORAGE_VAULT

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

LOG(ERROR) << "HDFS is disabled (via the ENABLE_HDFS_STORAGE_VAULT build option), "

<< "but HDFS storage vaults were detected";

#endif

} 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");

tablet_metrics_context_.reset();

segment_metrics_context_.reset();

DORIS_CLOUD_DEFER {

tablet_metrics_context_.finish_report();

segment_metrics_context_.finish_report();

};

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

int res = recycle_cluster_snapshots();

if (res != 0) {

return -1;

}

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(

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

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

.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]() -> int { return InstanceRecycler::recycle_packed_files(); }))

.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(); }))

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

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_WARNING("begin to recycle deleted instance").tag("instance_id", instance_id_);

int ret = 0;

auto start_time = steady_clock::now();

DORIS_CLOUD_DEFER {

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

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

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

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

ret = -1;

return -1;

} else if (has_snapshots) {

LOG(INFO) << "instance has cluster snapshots, skip recycling, instance_id=" << instance_id_;

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

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;

}

}

if (ret != 0) {

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

return ret;

}

}

// Check successor instance, if exists, skip deleting kv because successor instance may still need the data in kv

if (instance_info_.has_successor_instance_id() &&

!instance_info_.successor_instance_id().empty()) {

std::string key = instance_key(instance_info_.successor_instance_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_

<< " successor_instance_id=" << instance_info_.successor_instance_id()

<< " err=" << err;

ret = -1;

return -1;

}

std::string value;

err = txn->get(key, &value);

if (err == TxnErrorCode::TXN_OK) {

LOG(INFO) << "instance successor instance is still exist, skip deleting kv,"

<< " instance_id=" << instance_id_

<< " successor_instance_id=" << instance_info_.successor_instance_id();

return 0;

} else if (err != TxnErrorCode::TXN_KEY_NOT_FOUND) {

LOG(WARNING) << "failed to get successor instance, instance_id=" << instance_id_

<< " successor_instance_id=" << instance_info_.successor_instance_id()

<< " err=" << err;

ret = -1;

return -1;

}

}

// 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_, ""};