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

#include <aws/s3/S3Client.h>

#include <aws/s3/model/ListObjectsV2Request.h>

#include <butil/endpoint.h>

#include <butil/strings/string_split.h>

#include <fmt/core.h>

#include <gen_cpp/cloud.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <algorithm>

#include <chrono>

#include <climits>

#include <cstdint>

#include <functional>

#include <memory>

#include <mutex>

#include <numeric>

#include <sstream>

#include <string_view>

#include <unordered_map>

#include <unordered_set>

#include <vector>

#include "common/bvars.h"

#include "common/config.h"

#include "common/defer.h"

#include "common/encryption_util.h"

#include "common/logging.h"

#include "common/util.h"

#include "cpp/sync_point.h"

#include "meta-service/meta_service.h"

#include "meta-service/meta_service_schema.h"

#include "meta-service/meta_service_tablet_stats.h"

#include "meta-store/blob_message.h"

#include "meta-store/keys.h"

#include "meta-store/txn_kv.h"

#include "snapshot/snapshot_manager_factory.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 "recycler/recycler.h"

#include "recycler/util.h"

namespace doris::cloud {

namespace config {

extern int32_t brpc_listen_port;

extern int32_t scan_instances_interval_seconds;

extern int32_t recycle_job_lease_expired_ms;

extern int32_t recycle_concurrency;

extern std::vector<std::string> recycle_whitelist;

extern std::vector<std::string> recycle_blacklist;

extern bool enable_inverted_check;

} // namespace config

using namespace std::chrono;

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

}

Checker::~Checker() {

if (!stopped()) {

stop();

}

}

int Checker::start() {

DCHECK(txn_kv_);

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

// launch instance scanner

auto scanner_func = [this]() {

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

LOG(INFO) << "Checker 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;

if (instance.status() == InstanceInfoPB::DELETED) continue;

using namespace std::chrono;

auto enqueue_time_s =

duration_cast<seconds>(system_clock::now().time_since_epoch()).count();

auto [_, success] =

pending_instance_map_.insert({instance.instance_id(), enqueue_time_s});

// 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::scan_instances_interval_seconds),

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

}

}

};

workers_.emplace_back(scanner_func);

// Launch lease thread

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

// Launch inspect thread

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

// launch check workers

auto checker_func = [this]() {

while (!stopped()) {

// fetch instance to check

InstanceInfoPB instance;

long enqueue_time_s = 0;

{

std::unique_lock lock(mtx_);

pending_instance_cond_.wait(lock, [&]() -> bool {

return !pending_instance_queue_.empty() || stopped();

});

if (stopped()) {

return;

}

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

pending_instance_queue_.pop_front();

enqueue_time_s = pending_instance_map_[instance.instance_id()];

pending_instance_map_.erase(instance.instance_id());

}

const auto& instance_id = instance.instance_id();

{

std::lock_guard lock(mtx_);

// skip instance in recycling

if (working_instance_map_.count(instance_id)) {

LOG(INFO) << "checker skip instance already working, instance_id="

<< instance_id;

continue;

}

}

auto checker = std::make_shared<InstanceChecker>(txn_kv_, instance.instance_id());

if (checker->init(instance) != 0) {

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

<< instance.instance_id();

continue;

}

std::string check_job_key;

job_check_key({instance.instance_id()}, &check_job_key);

LOG(INFO) << "checker picked instance, instance_id=" << instance.instance_id()

<< " enqueue_time_s=" << enqueue_time_s;

int ret = prepare_instance_recycle_job(txn_kv_.get(), check_job_key,

instance.instance_id(), ip_port_,

config::check_object_interval_seconds * 1000);

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

LOG(WARNING) << "checker prepare job failed, instance_id=" << instance.instance_id()

<< " ret=" << ret;

continue;

} else {

std::lock_guard lock(mtx_);

working_instance_map_.emplace(instance_id, checker);

}

if (stopped()) return;

using namespace std::chrono;

auto ctime_ms =

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

g_bvar_checker_enqueue_cost_s.put(instance_id, ctime_ms / 1000 - enqueue_time_s);

bool success {true};

auto log_progress = [&](std::string_view stage) {

LOG(INFO) << "checker progress, instance_id=" << instance_id << " stage=" << stage;

};

log_progress("do_check");

if (int ret = checker->do_check(); ret != 0) {

success = false;

}

if (config::enable_inverted_check) {

log_progress("do_inverted_check");

if (int ret = checker->do_inverted_check(); ret != 0) {

success = false;

}

}

if (config::enable_delete_bitmap_inverted_check) {

log_progress("do_delete_bitmap_inverted_check");

if (int ret = checker->do_delete_bitmap_inverted_check(); ret != 0) {

success = false;

}

}

if (config::enable_mow_job_key_check) {

log_progress("do_mow_job_key_check");

if (int ret = checker->do_mow_job_key_check(); ret != 0) {

success = false;

}

}

if (config::enable_tablet_stats_key_check) {

log_progress("do_tablet_stats_key_check");

if (int ret = checker->do_tablet_stats_key_check(); ret != 0) {

success = false;

}

}

if (config::enable_restore_job_check) {

log_progress("do_restore_job_check");

if (int ret = checker->do_restore_job_check(); ret != 0) {

success = false;

}

}

if (config::enable_txn_key_check) {

log_progress("do_txn_key_check");

if (int ret = checker->do_txn_key_check(); ret != 0) {

success = false;

}

}

if (config::enable_meta_rowset_key_check) {

log_progress("do_meta_rowset_key_check");

if (int ret = checker->do_meta_rowset_key_check(); ret != 0) {

success = false;

}

}

if (config::enable_delete_bitmap_storage_optimize_v2_check) {

log_progress("do_delete_bitmap_storage_optimize_check_v2");

if (int ret = checker->do_delete_bitmap_storage_optimize_check(2 /*version*/);

ret != 0) {

success = false;

}

}

if (config::enable_version_key_check) {

log_progress("do_version_key_check");

if (int ret = checker->do_version_key_check(); ret != 0) {

success = false;

}

}

if (config::enable_snapshot_check) {

log_progress("do_snapshots_check");

if (int ret = checker->do_snapshots_check(); ret != 0) {

success = false;

}

}

if (config::enable_mvcc_meta_key_check) {

log_progress("do_mvcc_meta_key_check");

if (int ret = checker->do_mvcc_meta_key_check(); ret != 0) {

success = false;

}

}

if (config::enable_packed_file_check) {

log_progress("do_packed_file_check");

if (int ret = checker->do_packed_file_check(); ret != 0) {

success = false;

}

}

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

if (!checker->stopped()) {

finish_instance_recycle_job(txn_kv_.get(), check_job_key, instance.instance_id(),

ip_port_, success, ctime_ms);

}

LOG(INFO) << "checker finished instance, instance_id=" << instance.instance_id()

<< " success=" << success;

{

std::lock_guard lock(mtx_);

working_instance_map_.erase(instance.instance_id());

}

}

};

int num_threads = config::recycle_concurrency; // FIXME: use a new config entry?

for (int i = 0; i < num_threads; ++i) {

workers_.emplace_back(checker_func);

}

return 0;

}

void Checker::stop() {

stopped_ = true;

notifier_.notify_all();

pending_instance_cond_.notify_all();

{

std::lock_guard lock(mtx_);

for (auto& [_, checker] : working_instance_map_) {

checker->stop();

}

}

for (auto& w : workers_) {

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

}

}

void Checker::lease_check_jobs() {

while (!stopped()) {

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

instances.reserve(working_instance_map_.size());

{

std::lock_guard lock(mtx_);

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

instances.push_back(id);

}

}

for (auto& i : instances) {

std::string check_job_key;

job_check_key({i}, &check_job_key);

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

if (ret == 1) {

std::lock_guard lock(mtx_);

if (auto it = working_instance_map_.find(i); it != working_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(); });

}

}

}

#define LOG_CHECK_INTERVAL_ALARM LOG(WARNING) << "Err for check interval: "

void Checker::do_inspect(const InstanceInfoPB& instance) {

std::string check_job_key = job_check_key({instance.instance_id()});

std::unique_ptr<Transaction> txn;

std::string val;

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

if (err != TxnErrorCode::TXN_OK) {

LOG_CHECK_INTERVAL_ALARM << "failed to create txn";

return;

}

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

if (err != TxnErrorCode::TXN_OK && err != TxnErrorCode::TXN_KEY_NOT_FOUND) {

LOG_CHECK_INTERVAL_ALARM << "failed to get kv, err=" << err

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

return;

}

auto checker = InstanceChecker(txn_kv_, instance.instance_id());

if (checker.init(instance) != 0) {

LOG_CHECK_INTERVAL_ALARM << "failed to init instance checker, instance_id="

<< instance.instance_id();

return;

}

int64_t bucket_lifecycle_days = 0;

if (checker.get_bucket_lifecycle(&bucket_lifecycle_days) != 0) {

LOG_CHECK_INTERVAL_ALARM << "failed to get bucket lifecycle, instance_id="

<< instance.instance_id();

return;

}

DCHECK(bucket_lifecycle_days > 0);

if (bucket_lifecycle_days == INT64_MAX) {

// No s3 bucket (may all accessors are HdfsAccessor), skip inspect

return;

}

int64_t last_ctime_ms = -1;

auto job_status = JobRecyclePB::IDLE;

auto has_last_ctime = [&]() {

JobRecyclePB job_info;

if (!job_info.ParseFromString(val)) {

LOG_CHECK_INTERVAL_ALARM << "failed to parse JobRecyclePB, key=" << hex(check_job_key);

}

DCHECK(job_info.instance_id() == instance.instance_id());

if (!job_info.has_last_ctime_ms()) return false;

last_ctime_ms = job_info.last_ctime_ms();

job_status = job_info.status();

g_bvar_checker_last_success_time_ms.put(instance.instance_id(),

job_info.last_success_time_ms());

return true;

};

using namespace std::chrono;

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

if (err == TxnErrorCode::TXN_KEY_NOT_FOUND || !has_last_ctime()) {

// Use instance's ctime for instances that do not have job's last ctime

last_ctime_ms = instance.ctime();

}

DCHECK(now - last_ctime_ms >= 0);

int64_t expiration_ms =

bucket_lifecycle_days > config::reserved_buffer_days

? (bucket_lifecycle_days - config::reserved_buffer_days) * 86400000

: bucket_lifecycle_days * 86400000;

TEST_SYNC_POINT_CALLBACK("Checker:do_inspect", &last_ctime_ms);

if (now - last_ctime_ms >= expiration_ms) {

LOG_CHECK_INTERVAL_ALARM << "check risks, instance_id: " << instance.instance_id()

<< " last_ctime_ms: " << last_ctime_ms

<< " job_status: " << job_status

<< " bucket_lifecycle_days: " << bucket_lifecycle_days

<< " reserved_buffer_days: " << config::reserved_buffer_days

<< " expiration_ms: " << expiration_ms;

}

}

#undef LOG_CHECK_INTERVAL_ALARM

void Checker::inspect_instance_check_interval() {

while (!stopped()) {

LOG(INFO) << "start to inspect instance check interval";

std::vector<InstanceInfoPB> instances;

get_all_instances(txn_kv_.get(), instances);

for (const auto& instance : instances) {

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

if (stopped()) return;

if (instance.status() == InstanceInfoPB::DELETED) continue;

do_inspect(instance);

}

{

std::unique_lock lock(mtx_);

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

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

}

}

}

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

int key_exist(TxnKv* txn_kv, std::string_view key) {

std::unique_ptr<Transaction> txn;

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

if (err != TxnErrorCode::TXN_OK) {

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

return -1;

}

std::string val;

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

case TxnErrorCode::TXN_OK:

return 0;

case TxnErrorCode::TXN_KEY_NOT_FOUND:

return 1;

default:

return -1;

}

}

InstanceChecker::InstanceChecker(std::shared_ptr<TxnKv> txn_kv, const std::string& instance_id)

: txn_kv_(txn_kv), instance_id_(instance_id) {

snapshot_manager_ = create_snapshot_manager(txn_kv);

resource_mgr_ = std::make_shared<ResourceManager>(std::move(txn_kv));

resource_mgr_->init();

}

int InstanceChecker::init(const InstanceInfoPB& instance) {

int ret = init_obj_store_accessors(instance);

if (ret != 0) {

return ret;

}

return init_storage_vault_accessors(instance);

}

int InstanceChecker::init_obj_store_accessors(const InstanceInfoPB& instance) {

for (const auto& obj_info : instance.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 object 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 InstanceChecker::init_storage_vault_accessors(const InstanceInfoPB& instance) {

if (instance.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;

}

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

return ret;

}

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

#ifdef UNIT_TEST

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

#else

auto s3_conf = S3Conf::from_obj_store_info(vault.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=" << vault.id() << " name=" << vault.name();

return ret;

}

#endif

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

}

}

if (!it->is_valid()) {

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

return -1;

}

return 0;

}

int InstanceChecker::do_check() {

TEST_SYNC_POINT("InstanceChecker.do_check");

LOG(INFO) << "begin to check instance objects instance_id=" << instance_id_;

int check_ret = 0;

long num_scanned = 0;

long num_scanned_with_segment = 0;

long num_rowset_loss = 0;

long instance_volume = 0;

using namespace std::chrono;

auto start_time = steady_clock::now();

DORIS_CLOUD_DEFER {

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

LOG(INFO) << "check instance objects finished, cost=" << cost

<< "s. instance_id=" << instance_id_ << " num_scanned=" << num_scanned

<< " num_scanned_with_segment=" << num_scanned_with_segment

<< " num_rowset_loss=" << num_rowset_loss

<< " instance_volume=" << instance_volume;

g_bvar_checker_num_scanned.put(instance_id_, num_scanned);

g_bvar_checker_num_scanned_with_segment.put(instance_id_, num_scanned_with_segment);

g_bvar_checker_num_check_failed.put(instance_id_, num_rowset_loss);

g_bvar_checker_check_cost_s.put(instance_id_, static_cast<long>(cost));

// FIXME(plat1ko): What if some list operation failed?

g_bvar_checker_instance_volume.put(instance_id_, instance_volume);

};

struct TabletFiles {

int64_t tablet_id {0};

std::unordered_set<std::string> files;

};

TabletFiles tablet_files_cache;

auto check_rowset_objects = [&, this](doris::RowsetMetaCloudPB& rs_meta, std::string_view key) {

if (rs_meta.num_segments() == 0) {

return;

}

bool data_loss = false;

bool segment_file_loss = false;

bool index_file_loss = false;

DORIS_CLOUD_DEFER {

if (data_loss) {

LOG(INFO) << "segment file is" << (segment_file_loss ? "" : " not") << " loss, "

<< "index file is" << (index_file_loss ? "" : " not") << " loss, "

<< "rowset.tablet_id = " << rs_meta.tablet_id();

num_rowset_loss++;

}

};

++num_scanned_with_segment;

if (tablet_files_cache.tablet_id != rs_meta.tablet_id()) {

long tablet_volume = 0;

// Clear cache

tablet_files_cache.tablet_id = 0;

tablet_files_cache.files.clear();

// Get all file paths under this tablet directory

auto find_it = accessor_map_.find(rs_meta.resource_id());

if (find_it == accessor_map_.end()) {

LOG_WARNING("resource id not found in accessor map")

.tag("resource_id", rs_meta.resource_id())

.tag("tablet_id", rs_meta.tablet_id())

.tag("rowset_id", rs_meta.rowset_id_v2());

check_ret = -1;

return;

}

std::unique_ptr<ListIterator> list_iter;

int ret = find_it->second->list_directory(tablet_path_prefix(rs_meta.tablet_id()),

&list_iter);

if (ret != 0) { // No need to log, because S3Accessor has logged this error

check_ret = -1;

return;

}

for (auto file = list_iter->next(); file.has_value(); file = list_iter->next()) {

tablet_files_cache.files.insert(std::move(file->path));

tablet_volume += file->size;

}

tablet_files_cache.tablet_id = rs_meta.tablet_id();

instance_volume += tablet_volume;

}

for (int i = 0; i < rs_meta.num_segments(); ++i) {

auto path = segment_path(rs_meta.tablet_id(), rs_meta.rowset_id_v2(), i);

// Skip check if segment is already packed into a larger file

const auto& index_map = rs_meta.packed_slice_locations();

if (index_map.find(path) != index_map.end()) {

continue;

}

if (tablet_files_cache.files.contains(path)) {

continue;

}

if (1 == key_exist(txn_kv_.get(), key)) {

// Rowset has been deleted instead of data loss

break;

}

data_loss = true;

segment_file_loss = true;

TEST_SYNC_POINT_CALLBACK("InstanceChecker.do_check1", &path);

LOG(WARNING) << "object not exist, path=" << path

<< ", rs_meta=" << rs_meta.ShortDebugString() << " key=" << hex(key);

}

std::unique_ptr<Transaction> txn;

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

if (err != TxnErrorCode::TXN_OK) {

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

check_ret = -1;

return;

}

TabletIndexPB tablet_index;

if (get_tablet_idx(txn_kv_.get(), instance_id_, rs_meta.tablet_id(), tablet_index) == -1) {

LOG(WARNING) << "failed to get tablet index, tablet_id= " << rs_meta.tablet_id();

check_ret = -1;

return;

}

auto tablet_schema_key =

meta_schema_key({instance_id_, tablet_index.index_id(), rs_meta.schema_version()});

ValueBuf tablet_schema_val;

err = cloud::blob_get(txn.get(), tablet_schema_key, &tablet_schema_val);

if (err != TxnErrorCode::TXN_OK) {

check_ret = -1;

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

return;

}

auto* schema = rs_meta.mutable_tablet_schema();

if (!parse_schema_value(tablet_schema_val, schema)) {

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

return;

}

std::vector<std::pair<int64_t, std::string>> index_ids;

for (const auto& i : rs_meta.tablet_schema().index()) {

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

index_ids.emplace_back(i.index_id(), i.index_suffix_name());

}

}

if (!index_ids.empty()) {

const auto& index_map = rs_meta.packed_slice_locations();

for (int i = 0; i < rs_meta.num_segments(); ++i) {

std::vector<std::string> index_path_v;

if (rs_meta.tablet_schema().inverted_index_storage_format() ==

InvertedIndexStorageFormatPB::V1) {

for (const auto& index_id : index_ids) {

LOG(INFO) << "check inverted index, tablet_id=" << rs_meta.tablet_id()

<< " rowset_id=" << rs_meta.rowset_id_v2() << " segment_id=" << i

<< " index_id=" << index_id.first

<< " index_suffix_name=" << index_id.second;

index_path_v.emplace_back(

inverted_index_path_v1(rs_meta.tablet_id(), rs_meta.rowset_id_v2(),

i, index_id.first, index_id.second));

}

} else {

index_path_v.emplace_back(

inverted_index_path_v2(rs_meta.tablet_id(), rs_meta.rowset_id_v2(), i));

}

if (std::ranges::all_of(index_path_v, [&](const auto& idx_file_path) {

// Skip check if inverted index file is already packed into a larger file

if (index_map.find(idx_file_path) != index_map.end()) {

return true;

}

if (!tablet_files_cache.files.contains(idx_file_path)) {

LOG(INFO) << "loss index file: " << idx_file_path;

return false;

}

return true;

})) {

continue;

}

index_file_loss = true;

data_loss = true;

}

}

};

// scan visible rowsets

auto start_key = meta_rowset_key({instance_id_, 0, 0});

auto end_key = meta_rowset_key({instance_id_, INT64_MAX, 0});

std::unique_ptr<RangeGetIterator> it;

while (it == nullptr /* may be not init */ || (it->more() && !stopped())) {

std::unique_ptr<Transaction> txn;

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

if (err != TxnErrorCode::TXN_OK) {

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

return -1;

}

err = txn->get(start_key, end_key, &it);

if (err != TxnErrorCode::TXN_OK) {

LOG(WARNING) << "internal error, failed to get rowset meta, err=" << err;

return -1;

}

num_scanned += it->size();

while (it->has_next() && !stopped()) {

auto [k, v] = it->next();

if (!it->has_next()) {

start_key = k;

}

doris::RowsetMetaCloudPB rs_meta;

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

++num_rowset_loss;

LOG(WARNING) << "malformed rowset meta. key=" << hex(k) << " val=" << hex(v);

continue;

}

check_rowset_objects(rs_meta, k);

}

start_key.push_back('\x00'); // Update to next smallest key for iteration

}

return num_rowset_loss > 0 ? 1 : check_ret;

}

int InstanceChecker::get_bucket_lifecycle(int64_t* lifecycle_days) {

// If there are multiple buckets, return the minimum lifecycle.

int64_t min_lifecycle_days = INT64_MAX;

int64_t tmp_liefcycle_days = 0;

for (const auto& [id, accessor] : accessor_map_) {

if (accessor->type() != AccessorType::S3) {

continue;

}

auto* s3_accessor = static_cast<S3Accessor*>(accessor.get());

if (s3_accessor->check_versioning() != 0) {

return -1;

}

if (s3_accessor->get_life_cycle(&tmp_liefcycle_days) != 0) {

return -1;

}

if (tmp_liefcycle_days < min_lifecycle_days) {

min_lifecycle_days = tmp_liefcycle_days;

}

}

*lifecycle_days = min_lifecycle_days;

return 0;

}

int InstanceChecker::do_inverted_check() {

if (accessor_map_.size() > 1) {

LOG(INFO) << "currently not support inverted check for multi accessor. instance_id="

<< instance_id_;

return 0;

}

LOG(INFO) << "begin to inverted check objects instance_id=" << instance_id_;

int check_ret = 0;

long num_scanned = 0;

long num_file_leak = 0;

using namespace std::chrono;

auto start_time = steady_clock::now();

DORIS_CLOUD_DEFER {

g_bvar_inverted_checker_num_scanned.put(instance_id_, num_scanned);

g_bvar_inverted_checker_num_check_failed.put(instance_id_, num_file_leak);

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

LOG(INFO) << "inverted check instance objects finished, cost=" << cost

<< "s. instance_id=" << instance_id_ << " num_scanned=" << num_scanned

<< " num_file_leak=" << num_file_leak;

};

struct TabletRowsets {

int64_t tablet_id {0};

std::unordered_set<std::string> rowset_ids;

};

TabletRowsets tablet_rowsets_cache;

RowsetIndexesFormatV1 rowset_index_cache_v1;

RowsetIndexesFormatV2 rowset_index_cache_v2;

// Return 0 if check success, return 1 if file is garbage data, negative if error occurred

auto check_segment_file = [&](const std::string& obj_key) {

std::vector<std::string> str;

butil::SplitString(obj_key, '/', &str);

// data/{tablet_id}/{rowset_id}_{seg_num}.dat

if (str.size() < 3) {

// clang-format off

LOG(WARNING) << "split obj_key error, str.size() should be less than 3,"

<< " value = " << str.size();

// clang-format on

return -1;

}

int64_t tablet_id = atol(str[1].c_str());

if (tablet_id <= 0) {

LOG(WARNING) << "failed to parse tablet_id, key=" << obj_key;

return -1;

}

if (!str[2].ends_with(".dat")) {

// skip check not segment file

return 0;

}

std::string rowset_id;

if (auto pos = str.back().find('_'); pos != std::string::npos) {

rowset_id = str.back().substr(0, pos);

} else {

LOG(WARNING) << "failed to parse rowset_id, key=" << obj_key;

return -1;

}

if (tablet_rowsets_cache.tablet_id == tablet_id) {

if (tablet_rowsets_cache.rowset_ids.contains(rowset_id)) {

return 0;

} else {

LOG(WARNING) << "rowset not exists, key=" << obj_key;

return -1;

}

}

// Get all rowset id of this tablet

tablet_rowsets_cache.tablet_id = tablet_id;

tablet_rowsets_cache.rowset_ids.clear();

std::unique_ptr<Transaction> txn;

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

if (err != TxnErrorCode::TXN_OK) {

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

return -1;

}

std::unique_ptr<RangeGetIterator> it;

auto begin = meta_rowset_key({instance_id_, tablet_id, 0});

auto end = meta_rowset_key({instance_id_, tablet_id, INT64_MAX});

while (it == nullptr /* may be not init */ || (it->more() && !stopped())) {

TxnErrorCode err = txn->get(begin, end, &it);

if (err != TxnErrorCode::TXN_OK) {

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

return -1;

}

if (!it->has_next()) {

break;

}

while (it->has_next()) {

// recycle corresponding resources

auto [k, v] = it->next();

doris::RowsetMetaCloudPB rowset;

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

LOG(WARNING) << "malformed rowset meta value, key=" << hex(k);

return -1;

}

tablet_rowsets_cache.rowset_ids.insert(rowset.rowset_id_v2());

if (!it->has_next()) {

begin = k;

begin.push_back('\x00'); // Update to next smallest key for iteration

break;

}

}

}

if (!tablet_rowsets_cache.rowset_ids.contains(rowset_id)) {

// Garbage data leak

LOG(WARNING) << "rowset should be recycled, key=" << obj_key;

return 1;

}

return 0;

};

auto check_inverted_index_file = [&](const std::string& obj_key) {

std::vector<std::string> str;

butil::SplitString(obj_key, '/', &str);

// format v1: data/{tablet_id}/{rowset_id}_{seg_num}_{idx_id}{idx_suffix}.idx

// format v2: data/{tablet_id}/{rowset_id}_{seg_num}.idx

if (str.size() < 3) {

// clang-format off

LOG(WARNING) << "split obj_key error, str.size() should be less than 3,"

<< " value = " << str.size();

// clang-format on

return -1;

}

int64_t tablet_id = atol(str[1].c_str());

if (tablet_id <= 0) {

LOG(WARNING) << "failed to parse tablet_id, key=" << obj_key;

return -1;

}

// v1: {rowset_id}_{seg_num}_{idx_id}{idx_suffix}.idx

// v2: {rowset_id}_{seg_num}.idx

std::string rowset_info = str.back();

if (!rowset_info.ends_with(".idx")) {

return 0; // Not an index file

}

InvertedIndexStorageFormatPB inverted_index_storage_format =

std::count(rowset_info.begin(), rowset_info.end(), '_') > 1

? InvertedIndexStorageFormatPB::V1

: InvertedIndexStorageFormatPB::V2;

size_t pos = rowset_info.find_last_of('_');

if (pos == std::string::npos || pos + 1 >= str.back().size() - 4) {

LOG(WARNING) << "Invalid index_id format, key=" << obj_key;

return -1;

}

if (inverted_index_storage_format == InvertedIndexStorageFormatPB::V1) {

return check_inverted_index_file_storage_format_v1(tablet_id, obj_key, rowset_info,

rowset_index_cache_v1);

} else {

return check_inverted_index_file_storage_format_v2(tablet_id, obj_key, rowset_info,

rowset_index_cache_v2);

}

};

// so we choose to skip here.

TEST_SYNC_POINT_RETURN_WITH_VALUE("InstanceChecker::do_inverted_check", (int)0);

for (auto& [_, accessor] : accessor_map_) {

std::unique_ptr<ListIterator> list_iter;

int ret = accessor->list_directory("data", &list_iter);

if (ret != 0) {

return -1;

}

for (auto file = list_iter->next(); file.has_value(); file = list_iter->next()) {

const auto& path = file->path;

if (path == "data/packed_file" || path.starts_with("data/packed_file/")) {

continue; // packed_file has dedicated check logic

}

++num_scanned;

int ret = check_segment_file(path);