diff --git a/cloud/src/common/bvars.cpp b/cloud/src/common/bvars.cpp index 4d39a60892b932..1a5e367d36c87e 100644 --- a/cloud/src/common/bvars.cpp +++ b/cloud/src/common/bvars.cpp @@ -166,21 +166,33 @@ mBvarStatus g_bvar_recycler_instance_recycle_last_success_ts("recycler_ // resource_id: unique identifier for the repository // status: status of the recycle task (submitted, completed, error) mBvarIntAdder g_bvar_recycler_vault_recycle_task_status("recycler_vault_recycle_task_status", {"instance_id", "resource_id", "status"}); -// current concurrency of vault delete task -mBvarStatus g_bvar_recycler_instance_last_round_recycled_num("recycler_instance_last_round_recycled_num", {"instance_id", "resource_type"}); -mBvarStatus g_bvar_recycler_instance_last_round_to_recycle_num("recycler_instance_last_round_to_recycle_num", {"instance_id", "resource_type"}); -mBvarStatus g_bvar_recycler_instance_last_round_recycled_bytes("recycler_instance_last_round_recycled_bytes", {"instance_id", "resource_type"}); -mBvarStatus g_bvar_recycler_instance_last_round_to_recycle_bytes("recycler_instance_last_round_to_recycle_bytes", {"instance_id", "resource_type"}); -mBvarStatus g_bvar_recycler_instance_last_round_recycle_elpased_ts("recycler_instance_last_round_recycle_elpased_ts", {"instance_id", "resource_type"}); +mBvarStatus g_bvar_recycler_instance_last_round_recycled_num( + "recycler_instance_last_round_recycled_num", {"instance_id", "resource_type"}); +mBvarStatus g_bvar_recycler_instance_last_round_recycled_bytes( + "recycler_instance_last_round_recycled_bytes", {"instance_id", "resource_type"}); +// Elapsed time in milliseconds for the instance's current recycle round. +mBvarStatus g_bvar_recycler_instance_last_round_elapsed_ms( + "recycler_instance_last_round_elapsed_ms", {"instance_id"}); + +// Current round TxnKV statistics. +mBvarStatus g_bvar_recycler_instance_recycle_current_round_kv_scanned( + "instance_recycle_current_round_kv_scanned", + {"instance_id", "task_type"}); +mBvarStatus g_bvar_recycler_instance_recycle_current_round_kv_expired( + "instance_recycle_current_round_kv_expired", + {"instance_id", "task_type"}); +mBvarStatus g_bvar_recycler_instance_recycle_current_round_kv_recycled( + "instance_recycle_current_round_kv_recycled", + {"instance_id", "task_type"}); +mBvarStatus g_bvar_recycler_instance_current_round_recycled_object_bytes( + "instance_current_round_recycled_object_bytes", + {"instance_id", "task_type"}); +mBvarStatus g_bvar_recycler_instance_current_round_task_elapsed_ms( + "recycler_instance_current_round_task_elapsed_ms", {"instance_id", "task_type"}); // total recycled num and bytes of resources since recycler started -mBvarInt64Adder g_bvar_recycler_instance_recycle_total_num_since_started("recycler_instance_recycle_total_num_since_started", {"instance_id", "resource_type"}); + + mBvarInt64Adder g_bvar_recycler_instance_recycle_total_num_since_started("recycler_instance_recycle_total_num_since_started", {"instance_id", "resource_type"}); mBvarInt64Adder g_bvar_recycler_instance_recycle_total_bytes_since_started("recycler_instance_recycle_total_bytes_since_started", {"instance_id", "resource_type"}); -mBvarIntAdder g_bvar_recycler_instance_recycle_round("recycler_instance_recycle_round", {"instance_id", "resource_type"}); -// represents the ms required per resource to be recycled -// value of -1 means no resource recycled -mBvarStatus g_bvar_recycler_instance_recycle_time_per_resource("recycler_instance_recycle_time_per_resource", {"instance_id", "resource_type"}); -// represents the bytes of resources that can be recycled per ms -mBvarStatus g_bvar_recycler_instance_recycle_bytes_per_ms("recycler_instance_recycle_bytes_per_ms", {"instance_id", "resource_type"}); BvarStatusWithTag g_bvar_recycler_packed_file_recycled_kv_num("recycler", "packed_file_recycled_kv_num"); BvarStatusWithTag g_bvar_recycler_packed_file_recycled_kv_bytes( diff --git a/cloud/src/common/bvars.h b/cloud/src/common/bvars.h index 44e2c66d1ee912..8b168e60aba208 100644 --- a/cloud/src/common/bvars.h +++ b/cloud/src/common/bvars.h @@ -714,6 +714,7 @@ extern BvarStatusWithTag g_bvar_recycler_recycle_expired_txn_label_earl extern BvarStatusWithTag g_bvar_recycler_recycle_restore_job_earlest_ts; // recycler's mbvars +// instance extern bvar::Status g_bvar_recycler_task_max_concurrency; extern mBvarIntAdder g_bvar_recycler_instance_recycle_task_status; extern mBvarStatus g_bvar_recycler_instance_last_round_recycle_duration; @@ -721,18 +722,19 @@ extern mBvarStatus g_bvar_recycler_instance_next_ts; extern mBvarStatus g_bvar_recycler_instance_recycle_start_ts; extern mBvarStatus g_bvar_recycler_instance_recycle_end_ts; extern mBvarStatus g_bvar_recycler_instance_recycle_last_success_ts; +extern mBvarStatus g_bvar_recycler_instance_last_round_elapsed_ms; +// instance, task_type extern mBvarIntAdder g_bvar_recycler_vault_recycle_task_status; extern mBvarStatus g_bvar_recycler_instance_last_round_recycled_num; -extern mBvarStatus g_bvar_recycler_instance_last_round_to_recycle_num; extern mBvarStatus g_bvar_recycler_instance_last_round_recycled_bytes; -extern mBvarStatus g_bvar_recycler_instance_last_round_to_recycle_bytes; -extern mBvarStatus g_bvar_recycler_instance_last_round_recycle_elpased_ts; +extern mBvarStatus g_bvar_recycler_instance_recycle_current_round_kv_scanned; +extern mBvarStatus g_bvar_recycler_instance_recycle_current_round_kv_expired; +extern mBvarStatus g_bvar_recycler_instance_recycle_current_round_kv_recycled; +extern mBvarStatus g_bvar_recycler_instance_current_round_recycled_object_bytes; +extern mBvarStatus g_bvar_recycler_instance_current_round_task_elapsed_ms; extern mBvarInt64Adder g_bvar_recycler_instance_recycle_total_num_since_started; extern mBvarInt64Adder g_bvar_recycler_instance_recycle_total_bytes_since_started; -extern mBvarIntAdder g_bvar_recycler_instance_recycle_round; -extern mBvarStatus g_bvar_recycler_instance_recycle_time_per_resource; -extern mBvarStatus g_bvar_recycler_instance_recycle_bytes_per_ms; extern BvarStatusWithTag g_bvar_recycler_packed_file_recycled_kv_num; extern BvarStatusWithTag g_bvar_recycler_packed_file_recycled_kv_bytes; extern BvarStatusWithTag g_bvar_recycler_packed_file_recycle_cost_ms; @@ -746,7 +748,7 @@ extern BvarStatusWithTag g_bvar_recycler_packed_file_bytes_object_delet extern BvarStatusWithTag g_bvar_recycler_packed_file_rowset_scanned_num; // Operation Log Recycler BVars -// Note: generic metrics (last_round_to_recycle_num/bytes, last_round_recycled_num/bytes, etc.) +// Note: generic metrics (last_round_recycled_num/bytes, etc.) // are reported by RecyclerMetricsContext with operation_type = "recycle_operation_logs". extern mBvarStatus g_bvar_recycler_oplog_last_round_total_num; extern mBvarStatus g_bvar_recycler_oplog_last_round_not_recycled_num; diff --git a/cloud/src/common/config.h b/cloud/src/common/config.h index 949f46339604be..9f0da4f41b7f81 100644 --- a/cloud/src/common/config.h +++ b/cloud/src/common/config.h @@ -114,8 +114,6 @@ CONF_mInt32(instance_recycler_worker_pool_size, "32"); // Max number of delete tasks per batch when recycling objects. // Each task deletes up to 1000 files. Controls memory usage during large-scale deletion. CONF_Int32(recycler_max_tasks_per_batch, "1000"); -// The worker pool size for http api `statistics_recycle` worker pool -CONF_mInt32(instance_recycler_statistics_recycle_worker_pool_size, "5"); CONF_Bool(enable_checker, "false"); // The parallelism for parallel recycle operation // s3_producer_pool recycle_tablet_pool, delete single object in this pool @@ -130,8 +128,6 @@ CONF_mInt64(delete_bitmap_storage_optimize_v2_check_skip_seconds, "300"); // 5mi CONF_mInt32(scan_instances_interval_seconds, "60"); // 1min // interval for check object CONF_mInt32(check_object_interval_seconds, "43200"); // 12hours -// enable recycler metrics statistics -CONF_Bool(enable_recycler_stats_metrics, "false"); CONF_mBool(retain_deleted_instance_tombstone, "true"); CONF_mInt64(check_recycle_task_interval_seconds, "600"); // 10min diff --git a/cloud/src/common/http_helper.cpp b/cloud/src/common/http_helper.cpp index fea8c7aa371c7c..7010c5397671ae 100644 --- a/cloud/src/common/http_helper.cpp +++ b/cloud/src/common/http_helper.cpp @@ -383,12 +383,6 @@ const std::unordered_map& get_http_handlers() return process_recycle_instance((RS*)s, c); }, .role = HttpRole::RECYCLER}}, - {"statistics_recycle", - {.handler = - [](void* s, brpc::Controller* c) { - return process_statistics_recycle((RS*)s, c); - }, - .role = HttpRole::RECYCLER}}, {"recycle_copy_jobs", {.handler = [](void* s, brpc::Controller* c) { @@ -775,20 +769,6 @@ HttpResponse process_recycle_instance(RecyclerServiceImpl* service, brpc::Contro return http_text_reply(res.status(), res.status().msg()); } -HttpResponse process_statistics_recycle(RecyclerServiceImpl* service, brpc::Controller* cntl) { - std::string request_body = cntl->request_attachment().to_string(); - StatisticsRecycleRequest req; - auto st = google::protobuf::util::JsonStringToMessage(request_body, &req); - if (!st.ok()) { - return http_json_reply(MetaServiceCode::INVALID_ARGUMENT, - "failed to parse StatisticsRecycleRequest"); - } - MetaServiceCode code = MetaServiceCode::OK; - std::string msg = "OK"; - service->statistics_recycle(req, code, msg); - return http_text_reply(code, msg, msg); -} - HttpResponse process_recycle_copy_jobs(RecyclerServiceImpl* service, brpc::Controller* cntl) { const auto* instance_id = cntl->http_request().uri().GetQuery("instance_id"); if (!instance_id || instance_id->empty()) { diff --git a/cloud/src/common/http_helper.h b/cloud/src/common/http_helper.h index f7cd58084d8667..fd634a70d45037 100644 --- a/cloud/src/common/http_helper.h +++ b/cloud/src/common/http_helper.h @@ -192,9 +192,6 @@ const std::unordered_map& get_http_handlers() [[maybe_unused]] HttpResponse process_recycle_instance(RecyclerServiceImpl* service, brpc::Controller* cntl); -[[maybe_unused]] HttpResponse process_statistics_recycle(RecyclerServiceImpl* service, - brpc::Controller* cntl); - [[maybe_unused]] HttpResponse process_recycle_copy_jobs(RecyclerServiceImpl* service, brpc::Controller* cntl); diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index a27fea2e212cd8..a7b1a8f4505cdf 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -125,8 +125,99 @@ bool filter_out_instance(const std::string& instance_id) { config::recycle_whitelist.end(); } +void publish_recycled_current(const std::string& instance_id, const std::string& type, uint64_t num, + uint64_t bytes, int64_t elapsed_ms) { + g_bvar_recycler_instance_current_round_task_elapsed_ms.put({instance_id, type}, elapsed_ms); + g_bvar_recycler_instance_recycle_current_round_kv_recycled.put({instance_id, type}, num); + g_bvar_recycler_instance_current_round_recycled_object_bytes.put({instance_id, type}, bytes); +} + +void publish_recycled_finished(const std::string& instance_id, const std::string& type, + uint64_t num, uint64_t bytes) { + if (num > 0) { + g_bvar_recycler_instance_recycle_total_num_since_started.put({instance_id, type}, + static_cast(num)); + } + if (bytes > 0) { + g_bvar_recycler_instance_recycle_total_bytes_since_started.put({instance_id, type}, + static_cast(bytes)); + } + g_bvar_recycler_instance_last_round_recycled_num.put({instance_id, type}, + static_cast(num)); + g_bvar_recycler_instance_last_round_recycled_bytes.put({instance_id, type}, + static_cast(bytes)); +} + } // namespace +RecyclerMetricsContext::RecyclerMetricsContext(std::string instance_id, std::string operation_type) + : operation_type(std::move(operation_type)), instance_id(std::move(instance_id)) { + reset(); +} + +RecyclerMetricsContext::~RecyclerMetricsContext() { + finish(); +} + +void RecyclerMetricsContext::update_metrics() { + const auto elapsed_ms = duration_cast(steady_clock::now() - start_time_).count(); + publish_current_round_metrics(elapsed_ms); +} + +void RecyclerMetricsContext::reset() { + start_time_ = steady_clock::now(); + kv_scanned_num.reset(); + kv_expired_num.reset(); + kv_recycled_num.reset(); + kv_recycled_bytes.reset(); + publish_current_round_metrics(0); +} + +void RecyclerMetricsContext::finish() { + publish_recycled_finished(instance_id, operation_type, kv_recycled_num.value(), + kv_recycled_bytes.value()); + reset(); +} + +void RecyclerMetricsContext::publish_current_round_metrics(int64_t elapsed_ms) { + publish_recycled_current(instance_id, operation_type, kv_recycled_num.value(), + kv_recycled_bytes.value(), elapsed_ms); + g_bvar_recycler_instance_recycle_current_round_kv_scanned.put({instance_id, operation_type}, + kv_scanned_num.value()); + g_bvar_recycler_instance_recycle_current_round_kv_expired.put({instance_id, operation_type}, + kv_expired_num.value()); +} + +RecyclerResourceMetricsContext::RecyclerResourceMetricsContext(std::string instance_id, + std::string resource_type) + : instance_id_(std::move(instance_id)), resource_type_(std::move(resource_type)) {} + +void RecyclerResourceMetricsContext::start_round() { + start_time_ = steady_clock::now(); + recycled_num.reset(); + recycled_bytes.reset(); + update_metrics(); +} + +void RecyclerResourceMetricsContext::update_metrics() { + const auto elapsed_ms = + start_time_.has_value() + ? duration_cast(steady_clock::now() - *start_time_).count() + : 0; + publish_recycled_current(instance_id_, resource_type_, recycled_num.value(), + recycled_bytes.value(), elapsed_ms); +} + +void RecyclerResourceMetricsContext::finish_round() { + update_metrics(); + publish_recycled_finished(instance_id_, resource_type_, recycled_num.value(), + recycled_bytes.value()); + recycled_num.reset(); + recycled_bytes.reset(); + start_time_.reset(); + update_metrics(); +} + // 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 txn; @@ -677,7 +768,9 @@ InstanceRecycler::InstanceRecycler(std::shared_ptr txn_kv, const Instance _thread_pool_group(std::move(thread_pool_group)), txn_lazy_committer_(std::move(txn_lazy_committer)), delete_bitmap_lock_white_list_(std::make_shared()), - resource_mgr_(std::make_shared(txn_kv_)) { + resource_mgr_(std::make_shared(txn_kv_)), + tablet_metrics_context_(instance_id_, "recycle_tablet"), + segment_metrics_context_(instance_id_, "recycle_segment") { delete_bitmap_lock_white_list_->init(); resource_mgr_->init(); @@ -842,13 +935,25 @@ auto task_wrapper(Func... funcs) -> std::function { } int InstanceRecycler::do_recycle() { - TEST_SYNC_POINT("InstanceRecycler.do_recycle"); - tablet_metrics_context_.reset(); - segment_metrics_context_.reset(); + // The monitor publishes these metrics under the same lock. + { + std::lock_guard lock(recycle_tasks_mutex); + tablet_metrics_context_.start_round(); + segment_metrics_context_.start_round(); + } + DORIS_CLOUD_DEFER { + std::lock_guard lock(recycle_tasks_mutex); + tablet_metrics_context_.finish_round(); + segment_metrics_context_.finish_round(); + }; + const auto start_time = steady_clock::now(); DORIS_CLOUD_DEFER { - tablet_metrics_context_.finish_report(); - segment_metrics_context_.finish_report(); + g_bvar_recycler_instance_last_round_elapsed_ms.put( + {instance_id_}, + duration_cast(steady_clock::now() - start_time).count()); }; + + TEST_SYNC_POINT("InstanceRecycler.do_recycle"); if (instance_info_.status() == InstanceInfoPB::DELETED) { int res = recycle_cluster_snapshots(); if (res != 0) { @@ -2553,9 +2658,9 @@ int InstanceRecycler::batch_abort_txn_or_job_for_recycle( void InstanceRecycler::submit_recycle_prepare_rowsets_job( SimpleThreadPool& worker_pool, std::vector rowset_keys_to_abort, - std::atomic_long* num_recycled) { + std::atomic_long* num_recycled, RecyclerMetricsContext* metrics_context) { int ret = worker_pool.submit([this, rowset_keys_to_abort = std::move(rowset_keys_to_abort), - num_recycled]() mutable { + num_recycled, metrics_context]() mutable { std::vector> keys_to_recheck; int abort_ret = batch_abort_txn_or_job_for_recycle(rowset_keys_to_abort, keys_to_recheck); @@ -2592,6 +2697,9 @@ void InstanceRecycler::submit_recycle_prepare_rowsets_job( LOG(WARNING) << "failed to delete recycle rowset kv, instance_id=" << instance_id_; continue; } + metrics_context->kv_recycled_bytes += current_meta.total_disk_size(); + segment_metrics_context_.recycled_num += current_meta.num_segments(); + segment_metrics_context_.recycled_bytes += current_meta.total_disk_size(); num_recycled->fetch_add(1, std::memory_order_relaxed); } }); @@ -2690,6 +2798,10 @@ int InstanceRecycler::recycle_ref_rowsets(bool* has_unrecycled_rowsets) { // Phase 1: Scan to collect all tablet_ids that have rowset ref counts std::set tablets_with_refs; int64_t num_scanned = 0; + RecyclerMetricsContext metrics_context(instance_id_, task_name); + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + }; auto scan_func = [&](std::string_view k, std::string_view v) -> int { ++num_scanned; @@ -2726,13 +2838,14 @@ int InstanceRecycler::recycle_ref_rowsets(bool* has_unrecycled_rowsets) { break; } - RecyclerMetricsContext metrics_context(instance_id_, task_name); if (recycle_versioned_tablet(tablet_id, metrics_context) != 0) { LOG_WARNING("failed to recycle tablet") .tag("instance_id", instance_id_) .tag("tablet_id", tablet_id); return -1; } + tablet_metrics_context_.update_metrics(); + segment_metrics_context_.update_metrics(); ++num_recycled_tablets; } @@ -2773,11 +2886,12 @@ int InstanceRecycler::recycle_table_stream_offset_prefix(std::string prefix, std::string end = prefix; end.push_back('\xff'); std::vector keys; - auto collect_key = [&keys](std::string_view key, std::string_view) -> int { + auto collect_key = [&keys, &metrics_context](std::string_view key, std::string_view) -> int { + metrics_context->kv_scanned_num++; keys.push_back(key); return 0; }; - auto remove_keys = [this, &keys, metrics_context]() -> int { + auto remove_keys = [this, &keys, &metrics_context]() -> int { if (keys.empty()) { return 0; } @@ -2791,8 +2905,8 @@ int InstanceRecycler::recycle_table_stream_offset_prefix(std::string prefix, .tag("num_keys", num_keys); return -1; } - metrics_context->total_recycled_num += num_keys; - metrics_context->report(); + metrics_context->kv_expired_num += keys.size(); + metrics_context->kv_recycled_num += keys.size(); return 0; }; return scan_and_recycle(std::move(prefix), end, std::move(collect_key), std::move(remove_keys)); @@ -2849,7 +2963,8 @@ int InstanceRecycler::finalize_recycle_stream(int64_t stream_id, } int InstanceRecycler::recycle_stream(int64_t stream_id, const RecycleIndexPB& recycle_index, - std::string_view recycle_key) { + std::string_view recycle_key, + RecyclerMetricsContext& metrics_context) { if (!recycle_index.has_db_id() || !recycle_index.has_stream_db_id()) { LOG_WARNING("table stream recycle index is missing binding") .tag("instance_id", instance_id_) @@ -2857,10 +2972,6 @@ int InstanceRecycler::recycle_stream(int64_t stream_id, const RecycleIndexPB& re return -1; } - RecyclerMetricsContext metrics_context(instance_id_, "recycle_stream"); - DORIS_CLOUD_DEFER { - metrics_context.finish_report(); - }; const std::string latest_offset_prefix = table_stream_offset_key_prefix( instance_id_, recycle_index.db_id(), recycle_index.table_id(), recycle_index.stream_db_id(), stream_id); @@ -2879,11 +2990,10 @@ int InstanceRecycler::recycle_stream(int64_t stream_id, const RecycleIndexPB& re int InstanceRecycler::recycle_partition_table_stream_offsets( int64_t db_id, int64_t table_id, int64_t partition_id, - const google::protobuf::RepeatedPtrField& table_streams) { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_stream_partition_offsets"); - DORIS_CLOUD_DEFER { - metrics_context.finish_report(); - }; + const google::protobuf::RepeatedPtrField& table_streams, + RecyclerMetricsContext& metrics_context) { + metrics_context.kv_scanned_num += table_streams.size(); + metrics_context.kv_expired_num += table_streams.size(); const size_t batch_size = std::max(1, config::recycler_max_tasks_per_batch); for (size_t begin_index = 0; begin_index < table_streams.size(); begin_index += batch_size) { const size_t end_index = @@ -2923,8 +3033,7 @@ int InstanceRecycler::recycle_partition_table_stream_offsets( .tag("err", err); return -1; } - metrics_context.total_recycled_num += end_index - begin_index; - metrics_context.report(); + metrics_context.kv_recycled_num += end_index - begin_index; } return 0; } @@ -2935,6 +3044,7 @@ int InstanceRecycler::recycle_indexes() { int64_t num_expired = 0; int64_t num_recycled = 0; RecyclerMetricsContext metrics_context(instance_id_, task_name); + RecyclerMetricsContext stream_metrics_context(instance_id_, "recycle_stream"); RecycleIndexKeyInfo index_key_info0 {instance_id_, 0}; RecycleIndexKeyInfo index_key_info1 {instance_id_, INT64_MAX}; @@ -2949,10 +3059,12 @@ int InstanceRecycler::recycle_indexes() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle indexes finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -3020,14 +3132,13 @@ int InstanceRecycler::recycle_indexes() { } } if (index_pb.object_type() == IndexObjectTypePB::TABLE_STREAM) { - if (recycle_stream(index_id, index_pb, k) != 0) { + if (recycle_stream(index_id, index_pb, k, stream_metrics_context) != 0) { LOG_WARNING("failed to recycle table stream") .tag("instance_id", instance_id_) .tag("stream_id", index_id); return -1; } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); + ++num_recycled; check_recycle_task(instance_id_, task_name, num_scanned, num_recycled, start_time); return 0; } @@ -3061,14 +3172,19 @@ int InstanceRecycler::recycle_indexes() { } } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); 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 { + auto loop_done = [&index_keys, this, &metrics_context, &num_scanned, &num_expired, + &num_recycled]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; if (index_keys.empty()) return 0; DORIS_CLOUD_DEFER { index_keys.clear(); @@ -3077,12 +3193,10 @@ int InstanceRecycler::recycle_indexes() { LOG(WARNING) << "failed to delete recycle index kv, instance_id=" << instance_id_; return -1; } + num_recycled += index_keys.size(); return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_indexes(); - } // recycle_func and loop_done for scan and recycle return scan_and_recycle(index_key0, index_key1, std::move(recycle_func), std::move(loop_done)); } @@ -3180,6 +3294,7 @@ int InstanceRecycler::recycle_partitions() { int64_t num_expired = 0; int64_t num_recycled = 0; RecyclerMetricsContext metrics_context(instance_id_, task_name); + RecyclerMetricsContext stream_metrics_context(instance_id_, "recycle_stream_partition_offsets"); RecyclePartKeyInfo part_key_info0 {instance_id_, 0}; RecyclePartKeyInfo part_key_info1 {instance_id_, INT64_MAX}; @@ -3194,10 +3309,12 @@ int InstanceRecycler::recycle_partitions() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle partitions finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -3280,7 +3397,8 @@ int InstanceRecycler::recycle_partitions() { } if (ret == 0 && part_pb.has_db_id() && recycle_partition_table_stream_offsets(part_pb.db_id(), part_pb.table_id(), - partition_id, part_pb.table_streams()) != 0) { + partition_id, part_pb.table_streams(), + stream_metrics_context) != 0) { LOG_WARNING("failed to recycle table stream offsets under partition") .tag("instance_id", instance_id_) .tag("table_id", part_pb.table_id()) @@ -3313,20 +3431,24 @@ int InstanceRecycler::recycle_partitions() { } if (ret == 0) { - ++num_recycled; check_recycle_task(instance_id_, task_name, num_scanned, num_recycled, start_time); partition_keys.push_back(k); if (part_pb.db_id() > 0) { partition_version_keys.push_back(partition_version_key( {instance_id_, part_pb.db_id(), part_pb.table_id(), partition_id})); } - metrics_context.total_recycled_num = num_recycled; - metrics_context.report(); } return ret; }; - auto loop_done = [&partition_keys, &partition_version_keys, this]() -> int { + auto loop_done = [&partition_keys, &partition_version_keys, this, &metrics_context, + &num_scanned, &num_expired, &num_recycled]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; if (partition_keys.empty()) return 0; DORIS_CLOUD_DEFER { partition_keys.clear(); @@ -3350,12 +3472,10 @@ int InstanceRecycler::recycle_partitions() { << " err=" << err; return -1; } + num_recycled += partition_keys.size(); return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_partitions(); - } // recycle_func and loop_done for scan and recycle return scan_and_recycle(part_key0, part_key1, std::move(recycle_func), std::move(loop_done)); } @@ -3366,6 +3486,7 @@ int InstanceRecycler::recycle_versions() { } int64_t num_scanned = 0; + int64_t num_expired = 0; int64_t num_recycled = 0; RecyclerMetricsContext metrics_context(instance_id_, "recycle_versions"); @@ -3374,8 +3495,10 @@ int InstanceRecycler::recycle_versions() { auto start_time = steady_clock::now(); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); auto cost = duration(steady_clock::now() - start_time).count(); - metrics_context.finish_report(); LOG_WARNING("recycle table and partition versions finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -3386,8 +3509,8 @@ int InstanceRecycler::recycle_versions() { auto version_key_end = partition_version_key({instance_id_, INT64_MAX, 0, 0}); int64_t last_scanned_table_id = 0; bool is_recycled = false; // Is last scanned kv recycled - auto recycle_func = [&num_scanned, &num_recycled, &last_scanned_table_id, &is_recycled, - &metrics_context, this](std::string_view k, std::string_view) { + auto recycle_func = [&num_scanned, &num_expired, &num_recycled, &last_scanned_table_id, + &is_recycled, this](std::string_view k, std::string_view) { ++num_scanned; auto k1 = k; k1.remove_prefix(1); @@ -3417,6 +3540,7 @@ int InstanceRecycler::recycle_versions() { if (iter->has_next()) { // Table is useful, should not recycle table and partition versions return 0; } + ++num_expired; auto db_id = std::get(std::get<0>(out[3])); // 1. Remove all partition version kvs of this table auto partition_version_key_begin = @@ -3445,21 +3569,29 @@ int InstanceRecycler::recycle_versions() { if (err != TxnErrorCode::TXN_OK) { return -1; } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); + ++num_recycled; is_recycled = true; return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_versions(); - } + auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; + return 0; + }; + // recycle_func and loop_done for scan and recycle - return scan_and_recycle(version_key_begin, version_key_end, std::move(recycle_func)); + return scan_and_recycle(version_key_begin, version_key_end, std::move(recycle_func), + std::move(loop_done)); } int InstanceRecycler::recycle_orphan_partitions() { int64_t num_scanned = 0; + int64_t num_expired = 0; int64_t num_recycled = 0; RecyclerMetricsContext metrics_context(instance_id_, "recycle_orphan_partitions"); @@ -3469,8 +3601,10 @@ int InstanceRecycler::recycle_orphan_partitions() { auto start_time = steady_clock::now(); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); auto cost = duration(steady_clock::now() - start_time).count(); - metrics_context.finish_report(); LOG_WARNING("recycle orphan table and partition versions finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -3480,7 +3614,7 @@ int InstanceRecycler::recycle_orphan_partitions() { bool is_empty_table = false; // whether the table has no indexes bool is_table_kvs_recycled = false; // whether the table related kvs have been recycled int64_t current_table_id = 0; // current scanning table id - auto recycle_func = [&num_scanned, &num_recycled, &metrics_context, &is_empty_table, + auto recycle_func = [&num_scanned, &num_expired, &num_recycled, &is_empty_table, ¤t_table_id, &is_table_kvs_recycled, this](std::string_view k, std::string_view) { ++num_scanned; @@ -3507,6 +3641,7 @@ int InstanceRecycler::recycle_orphan_partitions() { // table is not empty, skip recycle return 0; } + num_expired++; std::unique_ptr txn; TxnErrorCode err = txn_kv_->create_txn(&txn); @@ -3556,8 +3691,17 @@ int InstanceRecycler::recycle_orphan_partitions() { if (err != TxnErrorCode::TXN_OK) { return -1; } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); + ++num_recycled; + return 0; + }; + + auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; return 0; }; @@ -3565,7 +3709,7 @@ int InstanceRecycler::recycle_orphan_partitions() { return scan_and_recycle( versioned::partition_inverted_index_key({instance_id_, 0, 0, 0}), versioned::partition_inverted_index_key({instance_id_, INT64_MAX, 0, 0}), - std::move(recycle_func)); + std::move(recycle_func), std::move(loop_done)); } int InstanceRecycler::recycle_tablets(int64_t table_id, int64_t index_id, @@ -3671,6 +3815,8 @@ int InstanceRecycler::recycle_tablets(int64_t table_id, int64_t index_id, DORIS_CLOUD_DEFER { init_rs_keys.clear(); has_failure = false; + tablet_metrics_context_.update_metrics(); + segment_metrics_context_.update_metrics(); }; auto tablets_info = sync_executor.when_all(&finished); if (!finished) { @@ -4577,8 +4723,7 @@ int InstanceRecycler::delete_rowset_data( // that marks the entire batch as failed and prevents txn_remove from // cleaning up recycle KV keys. if (num_segments <= 0) { - metrics_context.total_recycled_num++; - metrics_context.total_recycled_data_size += rs.total_disk_size(); + metrics_context.kv_recycled_bytes += rs.total_disk_size(); continue; } @@ -4764,13 +4909,12 @@ int InstanceRecycler::delete_rowset_data( if (rs_meta != rowsets.end() && !deleted_rowset_id.contains(rowset_id)) { deleted_rowset_id.emplace(rowset_id); - metrics_context.total_recycled_data_size += + metrics_context.kv_recycled_bytes += rs_meta->second.total_disk_size(); - segment_metrics_context_.total_recycled_num += + segment_metrics_context_.recycled_num += rs_meta->second.num_segments(); - segment_metrics_context_.total_recycled_data_size += + segment_metrics_context_.recycled_bytes += rs_meta->second.total_disk_size(); - metrics_context.total_recycled_num++; } }); } @@ -4786,10 +4930,9 @@ int InstanceRecycler::delete_rowset_data( int ret = delete_rowset_data(resource_id, tablet_id, rowset_id); if (!ret) { auto rs = rowsets.at(rowset_id); - metrics_context.total_recycled_data_size += rs.total_disk_size(); - metrics_context.total_recycled_num++; - segment_metrics_context_.total_recycled_data_size += rs.total_disk_size(); - segment_metrics_context_.total_recycled_num += rs.num_segments(); + metrics_context.kv_recycled_bytes += rs.total_disk_size(); + segment_metrics_context_.recycled_bytes += rs.total_disk_size(); + segment_metrics_context_.recycled_num += rs.num_segments(); } return ret; }); @@ -5001,126 +5144,6 @@ int InstanceRecycler::recycle_packed_files() { return ret; } -int InstanceRecycler::scan_tablets_and_statistics(int64_t table_id, int64_t index_id, - RecyclerMetricsContext& metrics_context, - int64_t partition_id, bool is_empty_tablet) { - std::string tablet_key_begin, tablet_key_end; - - if (partition_id > 0) { - meta_tablet_key({instance_id_, table_id, index_id, partition_id, 0}, &tablet_key_begin); - meta_tablet_key({instance_id_, table_id, index_id, partition_id + 1, 0}, &tablet_key_end); - } else { - meta_tablet_key({instance_id_, table_id, index_id, 0, 0}, &tablet_key_begin); - meta_tablet_key({instance_id_, table_id, index_id + 1, 0, 0}, &tablet_key_end); - } - // for calculate the total num or bytes of recyled objects - auto scan_and_statistics = [&, is_empty_tablet, this](std::string_view k, - std::string_view v) -> int { - doris::TabletMetaCloudPB tablet_meta_pb; - if (!tablet_meta_pb.ParseFromArray(v.data(), v.size())) { - return 0; - } - int64_t tablet_id = tablet_meta_pb.tablet_id(); - - if (config::enable_recycler_check_lazy_txn_finished && - !check_lazy_txn_finished(txn_kv_, instance_id_, tablet_meta_pb.tablet_id())) { - return 0; - } - - if (!is_empty_tablet) { - if (scan_tablet_and_statistics(tablet_id, metrics_context) != 0) { - return 0; - } - tablet_metrics_context_.total_need_recycle_num++; - } - return 0; - }; - int ret = scan_and_recycle(tablet_key_begin, tablet_key_end, std::move(scan_and_statistics)); - metrics_context.report(true); - tablet_metrics_context_.report(true); - segment_metrics_context_.report(true); - return ret; -} - -int InstanceRecycler::scan_tablet_and_statistics(int64_t tablet_id, - RecyclerMetricsContext& metrics_context) { - int ret = 0; - std::map rowset_meta_map; - std::unique_ptr txn; - if (txn_kv_->create_txn(&txn) != TxnErrorCode::TXN_OK) { - LOG_WARNING("failed to recycle tablet ") - .tag("tablet id", tablet_id) - .tag("instance_id", instance_id_) - .tag("reason", "failed to create txn"); - ret = -1; - } - GetRowsetResponse resp; - std::string msg; - MetaServiceCode code = MetaServiceCode::OK; - // get rowsets in tablet - internal_get_rowset(txn.get(), 0, std::numeric_limits::max() - 1, instance_id_, - tablet_id, code, msg, &resp); - if (code != MetaServiceCode::OK) { - LOG_WARNING("failed to get rowsets of tablet when recycle tablet") - .tag("tablet id", tablet_id) - .tag("msg", msg) - .tag("code", code) - .tag("instance id", instance_id_); - ret = -1; - } - for (const auto& rs_meta : resp.rowset_meta()) { - /* - * For compatibility, we skip the loop for [0-1] here. - * The purpose of this loop is to delete object files, - * and since [0-1] only has meta and doesn't have object files, - * skipping it doesn't affect system correctness. - * - * If not skipped, the check "if (!rs_meta.has_resource_id())" below - * would return error -1 directly, causing the recycle operation to fail. - * - * [0-1] doesn't have resource id is a bug. - * In the future, we will fix this problem, after that, - * we can remove this if statement. - * - * TODO(Yukang-Lian): remove this if statement when [0-1] has resource id in the future. - */ - - if (rs_meta.end_version() == 1) { - // Assert that [0-1] has no resource_id to make sure - // this if statement will not be forgetted to remove - // when the resource id bug is fixed - DCHECK(!rs_meta.has_resource_id()) << "rs_meta" << rs_meta.ShortDebugString(); - continue; - } - if (!rs_meta.has_resource_id()) { - LOG_WARNING("rowset meta does not have a resource id, impossible!") - .tag("rs_meta", rs_meta.ShortDebugString()) - .tag("instance_id", instance_id_) - .tag("tablet_id", tablet_id); - continue; - } - DCHECK(rs_meta.has_resource_id()) << "rs_meta" << rs_meta.ShortDebugString(); - auto it = accessor_map_.find(rs_meta.resource_id()); - // possible if the accessor is not initilized correctly - if (it == accessor_map_.end()) [[unlikely]] { - LOG_WARNING( - "failed to find resource id when recycle tablet, skip this vault accessor " - "recycle process") - .tag("tablet id", tablet_id) - .tag("instance_id", instance_id_) - .tag("resource_id", rs_meta.resource_id()) - .tag("rowset meta pb", rs_meta.ShortDebugString()); - continue; - } - - metrics_context.total_need_recycle_data_size += rs_meta.total_disk_size(); - tablet_metrics_context_.total_need_recycle_data_size += rs_meta.total_disk_size(); - segment_metrics_context_.total_need_recycle_data_size += rs_meta.total_disk_size(); - segment_metrics_context_.total_need_recycle_num += rs_meta.num_segments(); - } - return ret; -} - int InstanceRecycler::recycle_tablet(int64_t tablet_id, RecyclerMetricsContext& metrics_context) { LOG_INFO("begin to recycle rowsets in a dropped tablet") .tag("instance_id", instance_id_) @@ -5379,18 +5402,6 @@ int InstanceRecycler::recycle_tablet(int64_t tablet_id, RecyclerMetricsContext& return ret; } - tablet_metrics_context_.total_recycled_data_size += - recycle_rowsets_data_size + recycle_rowsets_index_size; - tablet_metrics_context_.total_recycled_num += 1; - segment_metrics_context_.total_recycled_num += recycle_segments_number; - segment_metrics_context_.total_recycled_data_size += - recycle_rowsets_data_size + recycle_rowsets_index_size; - metrics_context.total_recycled_data_size += - recycle_rowsets_data_size + recycle_rowsets_index_size; - tablet_metrics_context_.report(); - segment_metrics_context_.report(); - metrics_context.report(); - txn.reset(); if (txn_kv_->create_txn(&txn) != TxnErrorCode::TXN_OK) { LOG_WARNING("failed to recycle tablet ") @@ -5425,6 +5436,12 @@ int InstanceRecycler::recycle_tablet(int64_t tablet_id, RecyclerMetricsContext& if (ret == 0) { // All object files under tablet have been deleted + const auto recycled_bytes = recycle_rowsets_data_size + recycle_rowsets_index_size; + tablet_metrics_context_.recycled_num += 1; + tablet_metrics_context_.recycled_bytes += recycled_bytes; + segment_metrics_context_.recycled_num += recycle_segments_number; + segment_metrics_context_.recycled_bytes += recycled_bytes; + metrics_context.kv_recycled_bytes += recycled_bytes; std::lock_guard lock(recycled_tablets_mtx_); recycled_tablets_.insert(tablet_id); } @@ -5611,6 +5628,8 @@ int InstanceRecycler::recycle_versioned_tablet(int64_t tablet_id, // Phase 1: Classify tasks by ref_count std::vector batch_delete_tasks; + // Fallback path counts bytes via a separate context; the recycled bytes of all + // load/compact rowsets are published once below from the pre-aggregated totals. for (auto& task : all_tasks) { int classify_ret = classify_rowset_task_by_ref_count(task, batch_delete_tasks); if (classify_ret < 0) { @@ -5618,8 +5637,8 @@ int InstanceRecycler::recycle_versioned_tablet(int64_t tablet_id, .tag("instance_id", instance_id_) .tag("tablet_id", tablet_id) .tag("rowset_id", task.rowset_meta.rowset_id_v2()); - concurrent_delete_executor.add([this, t = std::move(task)]() mutable { - return recycle_rowset_meta_and_data(t); + concurrent_delete_executor.add([this, t = std::move(task), &metrics_context]() mutable { + return recycle_rowset_meta_and_data(t, metrics_context); }); } } @@ -5649,10 +5668,8 @@ int InstanceRecycler::recycle_versioned_tablet(int64_t tablet_id, // Only call delete_rowset_data if there are rowsets with actual data to delete bool delete_success = true; if (!rowsets_to_delete.empty()) { - RecyclerMetricsContext batch_metrics_context(instance_id_, - "batch_delete_versioned_tablet"); int delete_ret = delete_rowset_data( - rowsets_to_delete, RowsetRecyclingState::FORMAL_ROWSET, batch_metrics_context); + rowsets_to_delete, RowsetRecyclingState::FORMAL_ROWSET, metrics_context); if (delete_ret != 0) { LOG_WARNING("batch delete execution failed") .tag("instance_id", instance_id_) @@ -5696,18 +5713,6 @@ int InstanceRecycler::recycle_versioned_tablet(int64_t tablet_id, return ret; } - tablet_metrics_context_.total_recycled_data_size += - recycle_rowsets_data_size + recycle_rowsets_index_size; - tablet_metrics_context_.total_recycled_num += 1; - segment_metrics_context_.total_recycled_num += recycle_segments_number; - segment_metrics_context_.total_recycled_data_size += - recycle_rowsets_data_size + recycle_rowsets_index_size; - metrics_context.total_recycled_data_size += - recycle_rowsets_data_size + recycle_rowsets_index_size; - tablet_metrics_context_.report(); - segment_metrics_context_.report(); - metrics_context.report(); - txn.reset(); if (txn_kv_->create_txn(&txn) != TxnErrorCode::TXN_OK) { LOG_WARNING("failed to recycle tablet ") @@ -5740,7 +5745,10 @@ int InstanceRecycler::recycle_versioned_tablet(int64_t tablet_id, } if (ret == 0) { - // All object files under tablet have been deleted + // All object files under tablet have been deleted, and all KV keys have been deleted + tablet_metrics_context_.recycled_num += 1; + tablet_metrics_context_.recycled_bytes += + recycle_rowsets_data_size + recycle_rowsets_index_size; std::lock_guard lock(recycled_tablets_mtx_); recycled_tablets_.insert(tablet_id); } @@ -5778,10 +5786,12 @@ int InstanceRecycler::recycle_rowsets() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle rowsets finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -5809,14 +5819,19 @@ int InstanceRecycler::recycle_rowsets() { worker_pool->start(); auto delete_rowset_data_by_prefix = [&](std::string key, const std::string& resource_id, int64_t partition_id, int64_t tablet_id, - const std::string& rowset_id) { + const std::string& rowset_id, int64_t num_segments, + int64_t recycled_bytes) { // Try to delete rowset data in background thread int ret = worker_pool->submit_with_timeout( - [&, resource_id, partition_id, tablet_id, rowset_id, key]() mutable { + [&, resource_id, partition_id, tablet_id, rowset_id, num_segments, recycled_bytes, + key]() mutable { if (delete_rowset_data(resource_id, tablet_id, rowset_id) != 0) { LOG(WARNING) << "failed to delete rowset data, key=" << hex(key); return; } + metrics_context.kv_recycled_bytes += recycled_bytes; + segment_metrics_context_.recycled_num += num_segments; + segment_metrics_context_.recycled_bytes += recycled_bytes; if (delete_versioned_delete_bitmap_kvs(partition_id, tablet_id, rowset_id) != 0) { return; @@ -5846,6 +5861,9 @@ int InstanceRecycler::recycle_rowsets() { LOG(WARNING) << "failed to delete rowset data, key=" << hex(key); return -1; } + metrics_context.kv_recycled_bytes += recycled_bytes; + segment_metrics_context_.recycled_num += num_segments; + segment_metrics_context_.recycled_bytes += recycled_bytes; if (delete_versioned_delete_bitmap_kvs(partition_id, tablet_id, rowset_id) != 0) { return -1; } @@ -5901,14 +5919,11 @@ int InstanceRecycler::recycle_rowsets() { << " tablet_id=" << rowset.tablet_id() << " rowset_id=" << rowset_id << " task_type=" << metrics_context.operation_type; if (delete_rowset_data_by_prefix(std::string(k), rowset.resource_id(), -1, - rowset.tablet_id(), rowset_id) != 0) { + rowset.tablet_id(), rowset_id, + rowset.rowset_meta().num_segments(), + rowset.rowset_meta().total_disk_size()) != 0) { return -1; } - metrics_context.total_recycled_data_size += rowset.rowset_meta().total_disk_size(); - metrics_context.total_recycled_num++; - segment_metrics_context_.total_recycled_data_size += - rowset.rowset_meta().total_disk_size(); - segment_metrics_context_.total_recycled_num += rowset.rowset_meta().num_segments(); return 0; } auto* rowset_meta = rowset.mutable_rowset_meta(); @@ -5959,9 +5974,10 @@ int InstanceRecycler::recycle_rowsets() { return 0; } } - if (delete_rowset_data_by_prefix(std::string(k), rowset_meta->resource_id(), - rowset_meta->partition_id(), rowset_meta->tablet_id(), - rowset_meta->rowset_id_v2()) != 0) { + if (delete_rowset_data_by_prefix( + std::string(k), rowset_meta->resource_id(), rowset_meta->partition_id(), + rowset_meta->tablet_id(), rowset_meta->rowset_id_v2(), + rowset_meta->num_segments(), rowset_meta->total_disk_size()) != 0) { return -1; } } else { @@ -5976,6 +5992,13 @@ int InstanceRecycler::recycle_rowsets() { }; auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + segment_metrics_context_.update_metrics(); + }; std::vector rowset_keys_to_delete; // rowset_id -> rowset_meta // store rowset id and meta for statistics rs size when delete @@ -5992,7 +6015,7 @@ int InstanceRecycler::recycle_rowsets() { } if (!abort_job_keys_to_process.empty()) { submit_recycle_prepare_rowsets_job(*worker_pool, std::move(abort_job_keys_to_process), - &num_recycled); + &num_recycled, &metrics_context); } worker_pool->submit([&, rowset_keys_to_delete = std::move(rowset_keys_to_delete), rowsets_to_delete = std::move(rowsets_to_delete)]() mutable { @@ -6018,14 +6041,12 @@ int InstanceRecycler::recycle_rowsets() { return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_rowsets(); - } // recycle_func and loop_done for scan and recycle int ret = scan_and_recycle(recyc_rs_key0, recyc_rs_key1, std::move(handle_rowset_kv), std::move(loop_done)); worker_pool->stop(); + segment_metrics_context_.update_metrics(); if (!async_recycled_rowset_keys.empty()) { if (txn_remove(txn_kv_.get(), async_recycled_rowset_keys) != 0) { @@ -6035,11 +6056,6 @@ int InstanceRecycler::recycle_rowsets() { num_recycled.fetch_add(async_recycled_rowset_keys.size(), std::memory_order_relaxed); } } - - // Report final metrics after all concurrent tasks completed - segment_metrics_context_.report(); - metrics_context.report(); - return ret; } @@ -6075,11 +6091,12 @@ int InstanceRecycler::recycle_restore_jobs() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); - LOG_INFO("recycle restore jobs finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -6195,8 +6212,6 @@ int InstanceRecycler::recycle_restore_jobs() { return -1; } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); check_recycle_task(instance_id_, task_name, num_scanned, num_recycled, start_time); restore_job_keys.push_back(k); @@ -6205,7 +6220,16 @@ int InstanceRecycler::recycle_restore_jobs() { return 0; }; - auto loop_done = [&restore_job_keys, this]() -> int { + auto loop_done = [&restore_job_keys, &metrics_context, &num_scanned, &num_expired, + &num_recycled, this]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + tablet_metrics_context_.update_metrics(); + segment_metrics_context_.update_metrics(); + }; if (restore_job_keys.empty()) return 0; DORIS_CLOUD_DEFER { restore_job_keys.clear(); @@ -6231,13 +6255,11 @@ int InstanceRecycler::recycle_restore_jobs() { .tag("reason", "failed to commit txn"); return -1; } + // Count after the restore job KV keys are successfully deleted. + num_recycled += restore_job_keys.size(); return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_restore_jobs(); - } - return scan_and_recycle(restore_job_key0, restore_job_key1, std::move(recycle_func), std::move(loop_done)); } @@ -6268,10 +6290,12 @@ int InstanceRecycler::recycle_versioned_rowsets() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle rowsets finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -6294,14 +6318,19 @@ int InstanceRecycler::recycle_versioned_rowsets() { config::instance_recycler_worker_pool_size, "recycle_rowsets"); worker_pool->start(); auto delete_rowset_data_by_prefix = [&](std::string key, const std::string& resource_id, - int64_t tablet_id, const std::string& rowset_id) { + int64_t tablet_id, const std::string& rowset_id, + int64_t num_segments, int64_t recycled_bytes) { // Try to delete rowset data in background thread int ret = worker_pool->submit_with_timeout( - [&, resource_id, tablet_id, rowset_id, key]() mutable { + [&, resource_id, tablet_id, rowset_id, num_segments, recycled_bytes, + key]() mutable { if (delete_rowset_data(resource_id, tablet_id, rowset_id) != 0) { LOG(WARNING) << "failed to delete rowset data, key=" << hex(key); return; } + metrics_context.kv_recycled_bytes += recycled_bytes; + segment_metrics_context_.recycled_num += num_segments; + segment_metrics_context_.recycled_bytes += recycled_bytes; // The async recycled rowsets are staled format or has not been used, // so we don't need to check the rowset ref count key. std::vector keys; @@ -6329,6 +6358,9 @@ int InstanceRecycler::recycle_versioned_rowsets() { LOG(WARNING) << "failed to delete rowset data, key=" << hex(key); return -1; } + metrics_context.kv_recycled_bytes += recycled_bytes; + segment_metrics_context_.recycled_num += num_segments; + segment_metrics_context_.recycled_bytes += recycled_bytes; orphan_rowset_keys.push_back(std::move(key)); return 0; }; @@ -6379,7 +6411,9 @@ int InstanceRecycler::recycle_versioned_rowsets() { LOG(INFO) << "delete rowset data, instance_id=" << instance_id_ << " tablet_id=" << rowset.tablet_id() << " rowset_id=" << rowset_id; if (delete_rowset_data_by_prefix(std::string(k), rowset.resource_id(), - rowset.tablet_id(), rowset_id) != 0) { + rowset.tablet_id(), rowset_id, + rowset.rowset_meta().num_segments(), + rowset.rowset_meta().total_disk_size()) != 0) { return -1; } return 0; @@ -6407,8 +6441,9 @@ int InstanceRecycler::recycle_versioned_rowsets() { // unable to calculate file path, can only be deleted by rowset id prefix num_prepare += 1; if (delete_rowset_data_by_prefix(std::string(k), rowset_meta->resource_id(), - rowset_meta->tablet_id(), - rowset_meta->rowset_id_v2()) != 0) { + rowset_meta->tablet_id(), rowset_meta->rowset_id_v2(), + rowset_meta->num_segments(), + rowset_meta->total_disk_size()) != 0) { return -1; } } else { @@ -6419,7 +6454,7 @@ int InstanceRecycler::recycle_versioned_rowsets() { RowsetDeleteTask task; task.rowset_meta = rowset_meta; task.recycle_rowset_key = k; - if (recycle_rowset_meta_and_data(task) != 0) { + if (recycle_rowset_meta_and_data(task, metrics_context) != 0) { return; } num_compacted += is_compacted; @@ -6432,13 +6467,18 @@ int InstanceRecycler::recycle_versioned_rowsets() { return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_rowsets(); - } - auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + segment_metrics_context_.update_metrics(); + }; if (txn_remove(txn_kv_.get(), orphan_rowset_keys)) { LOG(WARNING) << "failed to delete recycle rowset kv, instance_id=" << instance_id_; + } else { + num_recycled.fetch_add(orphan_rowset_keys.size(), std::memory_order_relaxed); } orphan_rowset_keys.clear(); return 0; @@ -6449,6 +6489,7 @@ int InstanceRecycler::recycle_versioned_rowsets() { std::move(loop_done)); worker_pool->stop(); + segment_metrics_context_.update_metrics(); if (!async_recycled_rowset_keys.empty()) { if (txn_remove(txn_kv_.get(), async_recycled_rowset_keys) != 0) { @@ -6459,14 +6500,11 @@ int InstanceRecycler::recycle_versioned_rowsets() { } } - // Report final metrics after all concurrent tasks completed - segment_metrics_context_.report(); - metrics_context.report(); - return ret; } -int InstanceRecycler::recycle_rowset_meta_and_data(const RowsetDeleteTask& task) { +int InstanceRecycler::recycle_rowset_meta_and_data(const RowsetDeleteTask& task, + RecyclerMetricsContext& metrics_context) { constexpr int MAX_RETRY = 10; const RowsetMetaCloudPB& rowset_meta = task.rowset_meta; int64_t tablet_id = rowset_meta.tablet_id(); @@ -6507,6 +6545,7 @@ int InstanceRecycler::recycle_rowset_meta_and_data(const RowsetDeleteTask& task) } } + int64_t recycled_bytes = 0; if (ref_count == 1) { // It would not be added since it is recycling. if (delete_rowset_data(rowset_meta) != 0) { @@ -6514,6 +6553,8 @@ int InstanceRecycler::recycle_rowset_meta_and_data(const RowsetDeleteTask& task) return -1; } + recycled_bytes = rowset_meta.total_disk_size(); + // Reset the transaction to avoid timeout. err = txn_kv_->create_txn(&txn); if (err != TxnErrorCode::TXN_OK) { @@ -6585,6 +6626,11 @@ int InstanceRecycler::recycle_rowset_meta_and_data(const RowsetDeleteTask& task) LOG_WARNING("failed to recycle rowset meta and data").tag("err", err); return -1; } + metrics_context.kv_recycled_bytes += recycled_bytes; + if (ref_count == 1) { + segment_metrics_context_.recycled_num += rowset_meta.num_segments(); + segment_metrics_context_.recycled_bytes += recycled_bytes; + } LOG_INFO("recycle rowset meta and data success"); return 0; } @@ -6618,10 +6664,12 @@ int InstanceRecycler::recycle_tmp_rowsets() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle tmp rowsets finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -6725,6 +6773,13 @@ int InstanceRecycler::recycle_tmp_rowsets() { }; auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + segment_metrics_context_.update_metrics(); + }; std::vector tmp_rowset_keys_to_delete; std::vector tmp_rowset_ref_count_keys_to_delete; std::vector mark_keys_to_process; @@ -6781,18 +6836,12 @@ int InstanceRecycler::recycle_tmp_rowsets() { return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_tmp_rowsets(); - } // recycle_func and loop_done for scan and recycle int ret = scan_and_recycle(tmp_rs_key0, tmp_rs_key1, std::move(handle_rowset_kv), std::move(loop_done)); worker_pool->stop(); - - // Report final metrics after all concurrent tasks completed - segment_metrics_context_.report(); - metrics_context.report(); + segment_metrics_context_.update_metrics(); return ret; } @@ -6902,9 +6951,11 @@ int InstanceRecycler::abort_timeout_txn() { DORIS_CLOUD_DEFER { unregister_recycle_task(task_name); + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_timeout); + metrics_context.kv_recycled_num.set(num_abort); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("end to abort timeout txn, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -6917,10 +6968,9 @@ int InstanceRecycler::abort_timeout_txn() { duration_cast(system_clock::now().time_since_epoch()).count(); auto handle_txn_running_kv = [&num_scanned, &num_timeout, &num_abort, &num_advance, - ¤t_time, &metrics_context, + ¤t_time, this](std::string_view k, std::string_view v) -> int { ++num_scanned; - std::unique_ptr txn; TxnErrorCode err = txn_kv_->create_txn(&txn); if (err != TxnErrorCode::TXN_OK) { @@ -7012,19 +7062,25 @@ int InstanceRecycler::abort_timeout_txn() { .tag("txn_id", txn_id); return -1; } - metrics_context.total_recycled_num = ++num_abort; - metrics_context.report(); + ++num_abort; } return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_abort_timeout_txn(); - } + auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_timeout); + metrics_context.kv_recycled_num.set(num_abort); + metrics_context.update_metrics(); + }; + return 0; + }; + // recycle_func and loop_done for scan and recycle return scan_and_recycle(begin_txn_running_key, end_txn_running_key, - std::move(handle_txn_running_kv)); + std::move(handle_txn_running_kv), std::move(loop_done)); } int InstanceRecycler::recycle_expired_txn_label() { @@ -7048,10 +7104,12 @@ int InstanceRecycler::recycle_expired_txn_label() { int64_t start_time = duration_cast(steady_clock::now().time_since_epoch()).count(); register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("end to recycle expired txn, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -7201,6 +7259,10 @@ int InstanceRecycler::recycle_expired_txn_label() { auto loop_done = [&]() -> int { DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); recycle_txn_keys_by_label.clear(); }; TEST_SYNC_POINT_CALLBACK( @@ -7256,11 +7318,6 @@ int InstanceRecycler::recycle_expired_txn_label() { } ret = finished ? ret : -1; - - // Update metrics after all concurrent tasks completed - metrics_context.total_recycled_num = num_recycled.load(); - metrics_context.report(); - TEST_SYNC_POINT_CALLBACK("InstanceRecycler::recycle_expired_txn_label.failure", &ret); if (ret != 0) { @@ -7273,9 +7330,6 @@ int InstanceRecycler::recycle_expired_txn_label() { return ret; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_expired_txn_label(); - } // recycle_func and loop_done for scan and recycle return scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, std::move(handle_recycle_txn_kv), std::move(loop_done)); @@ -7411,9 +7465,11 @@ int InstanceRecycler::recycle_copy_jobs() { DORIS_CLOUD_DEFER { unregister_recycle_task(task_name); + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle copy jobs finished, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -7430,7 +7486,7 @@ int InstanceRecycler::recycle_copy_jobs() { copy_job_key(key_info1, &key1); std::unordered_map> stage_accessor_map; auto recycle_func = [&start_time, &num_scanned, &num_finished, &num_expired, &num_recycled, - &batch_count, &stage_accessor_map, &task_name, &metrics_context, + &batch_count, &stage_accessor_map, &task_name, this](std::string_view k, std::string_view v) -> int { ++num_scanned; CopyJobPB copy_job; @@ -7507,8 +7563,8 @@ int InstanceRecycler::recycle_copy_jobs() { if (!config::force_immediate_recycle && current_time <= copy_job.timeout_time_ms()) { return 0; } - ++num_expired; } + ++num_expired; // delete all copy files std::vector copy_file_keys; @@ -7537,17 +7593,23 @@ int InstanceRecycler::recycle_copy_jobs() { return -1; } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); + ++num_recycled; check_recycle_task(instance_id_, task_name, num_scanned, num_recycled, start_time); return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_copy_jobs(); - } + auto loop_done = [&]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; + return 0; + }; + // recycle_func and loop_done for scan and recycle - return scan_and_recycle(key0, key1, std::move(recycle_func)); + return scan_and_recycle(key0, key1, std::move(recycle_func), std::move(loop_done)); } int InstanceRecycler::init_copy_job_accessor(const std::string& stage_id, @@ -7648,6 +7710,7 @@ int InstanceRecycler::init_copy_job_accessor(const std::string& stage_id, int InstanceRecycler::recycle_stage() { int64_t num_scanned = 0; + int64_t num_expired = 0; int64_t num_recycled = 0; const std::string task_name = "recycle_stage"; RecyclerMetricsContext metrics_context(instance_id_, task_name); @@ -7658,10 +7721,12 @@ int InstanceRecycler::recycle_stage() { register_recycle_task(task_name, start_time); DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); unregister_recycle_task(task_name); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle stage, cost={}s", cost) .tag("instance_id", instance_id_) .tag("num_scanned", num_scanned) @@ -7674,8 +7739,8 @@ int InstanceRecycler::recycle_stage() { std::string key1 = recycle_stage_key(key_info1); std::vector stage_keys; - auto recycle_func = [&start_time, &num_scanned, &num_recycled, &stage_keys, &metrics_context, - this](std::string_view k, std::string_view v) -> int { + auto recycle_func = [&start_time, &num_scanned, &num_expired, &num_recycled, &stage_keys, this]( + std::string_view k, std::string_view v) -> int { ++num_scanned; RecycleStagePB recycle_stage; if (!recycle_stage.ParseFromArray(v.data(), v.size())) { @@ -7689,6 +7754,8 @@ int InstanceRecycler::recycle_stage() { return -1; } + ++num_expired; + std::shared_ptr accessor; int ret = SYNC_POINT_HOOK_RETURN_VALUE( [&] { @@ -7730,14 +7797,19 @@ int InstanceRecycler::recycle_stage() { << ", ret=" << ret; return -1; } - metrics_context.total_recycled_num = ++num_recycled; - metrics_context.report(); check_recycle_task(instance_id_, "recycle_stage", num_scanned, num_recycled, start_time); stage_keys.push_back(k); return 0; }; - auto loop_done = [&stage_keys, this]() -> int { + auto loop_done = [&stage_keys, &metrics_context, &num_scanned, &num_expired, &num_recycled, + this]() -> int { + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; if (stage_keys.empty()) return 0; DORIS_CLOUD_DEFER { stage_keys.clear(); @@ -7746,11 +7818,10 @@ int InstanceRecycler::recycle_stage() { LOG(WARNING) << "failed to delete recycle partition kv, instance_id=" << instance_id_; return -1; } + // Count only after the recycle KV keys are successfully deleted. + num_recycled += stage_keys.size(); return 0; }; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_stage(); - } // recycle_func and loop_done for scan and recycle return scan_and_recycle(key0, key1, std::move(recycle_func), std::move(loop_done)); } @@ -7760,21 +7831,22 @@ int InstanceRecycler::recycle_expired_stage_objects() { int64_t start_time = duration_cast(steady_clock::now().time_since_epoch()).count(); RecyclerMetricsContext metrics_context(instance_id_, "recycle_expired_stage_objects"); + int64_t num_scanned = 0; + int64_t num_expired = 0; + int64_t num_recycled = 0; DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); int64_t cost = duration_cast(steady_clock::now().time_since_epoch()).count() - start_time; - metrics_context.finish_report(); LOG_WARNING("recycle expired stage objects, cost={}s", cost) .tag("instance_id", instance_id_); }; int ret = 0; - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_expired_stage_objects(); - } - for (const auto& stage : instance_info_.stages()) { std::stringstream ss; ss << "instance_id=" << instance_id_ << ", stage_id=" << stage.stage_id() << ", user_name=" @@ -7782,12 +7854,20 @@ int InstanceRecycler::recycle_expired_stage_objects() { << ", user_id=" << (stage.mysql_user_id().empty() ? "null" : stage.mysql_user_id().at(0)) << ", prefix=" << stage.obj_info().prefix(); + DORIS_CLOUD_DEFER { + metrics_context.kv_scanned_num.set(num_scanned); + metrics_context.kv_expired_num.set(num_expired); + metrics_context.kv_recycled_num.set(num_recycled); + metrics_context.update_metrics(); + }; + if (stopped()) { break; } if (stage.type() == StagePB::EXTERNAL) { continue; } + int idx = stoi(stage.obj_info().id()); if (idx > instance_info_.obj_info().size() || idx < 1) { LOG(WARNING) << "invalid idx: " << idx << ", id: " << stage.obj_info().id(); @@ -7815,6 +7895,7 @@ int InstanceRecycler::recycle_expired_stage_objects() { ret = -1; continue; } + ++num_scanned; LOG(INFO) << "recycle expired stage objects, " << ss.str(); int64_t expiration_time = @@ -7830,8 +7911,9 @@ int InstanceRecycler::recycle_expired_stage_objects() { ret = -1; continue; } - metrics_context.total_recycled_num++; - metrics_context.report(); + ++num_expired; + ++num_recycled; + // Note: this function only deletes S3 objects, not KV keys; no kv_recycled_num update. } return ret; } @@ -7851,6 +7933,8 @@ bool InstanceRecycler::check_recycle_tasks() { std::map tmp_running_recycle_tasks; { std::lock_guard lock(recycle_tasks_mutex); + tablet_metrics_context_.update_metrics(); + segment_metrics_context_.update_metrics(); tmp_running_recycle_tasks = running_recycle_tasks; } @@ -7869,650 +7953,6 @@ bool InstanceRecycler::check_recycle_tasks() { return found; } -// Scan and statistics indexes that need to be recycled -int InstanceRecycler::scan_and_statistics_indexes() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_indexes"); - RecyclerMetricsContext stream_metrics_context(instance_id_, "recycle_stream"); - - 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); - int64_t earlest_ts = std::numeric_limits::max(); - - auto handle_index_kv = [&, this](std::string_view k, std::string_view v) -> int { - RecycleIndexPB index_pb; - if (!index_pb.ParseFromArray(v.data(), v.size())) { - return 0; - } - int64_t current_time = ::time(nullptr); - if (current_time < - calculate_index_expired_time(instance_id_, index_pb, &earlest_ts)) { // not expired - return 0; - } - // decode index_id - auto k1 = k; - k1.remove_prefix(1); - std::vector, int, int>> out; - decode_key(&k1, &out); - // 0x01 "recycle" ${instance_id} "index" ${index_id} -> RecycleIndexPB - auto index_id = std::get(std::get<0>(out[3])); - std::unique_ptr txn; - TxnErrorCode err = txn_kv_->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - std::string val; - err = txn->get(k, &val); - if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { - return 0; - } - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - index_pb.Clear(); - if (!index_pb.ParseFromString(val)) { - return 0; - } - if (index_pb.object_type() == IndexObjectTypePB::TABLE_STREAM) { - if (!index_pb.has_db_id() || !index_pb.has_stream_db_id()) { - LOG_WARNING("table stream recycle index is missing binding") - .tag("instance_id", instance_id_) - .tag("stream_id", index_id); - return 0; - } - auto scan_offset_prefix = [this, &stream_metrics_context](std::string prefix) { - std::string end = prefix; - end.push_back('\xff'); - auto count_offset = [&stream_metrics_context](std::string_view key, - std::string_view value) { - stream_metrics_context.total_need_recycle_num++; - stream_metrics_context.total_need_recycle_data_size += - key.size() + value.size(); - return 0; - }; - return scan_and_recycle(std::move(prefix), end, std::move(count_offset)); - }; - const std::string latest_prefix = table_stream_offset_key_prefix( - instance_id_, index_pb.db_id(), index_pb.table_id(), index_pb.stream_db_id(), - index_id); - const std::string versioned_prefix = versioned::table_stream_offset_key_prefix( - instance_id_, index_pb.db_id(), index_pb.table_id(), index_pb.stream_db_id(), - index_id); - if (scan_offset_prefix(latest_prefix) != 0 || - scan_offset_prefix(versioned_prefix) != 0) { - LOG_WARNING("failed to scan table stream offsets for recycle statistics") - .tag("instance_id", instance_id_) - .tag("stream_id", index_id); - } - return 0; - } - if (scan_tablets_and_statistics(index_pb.table_id(), index_id, metrics_context) != 0) { - return 0; - } - metrics_context.total_need_recycle_num++; - return 0; - }; - - int ret = scan_and_recycle(index_key0, index_key1, std::move(handle_index_kv)); - metrics_context.report(true); - stream_metrics_context.report(true); - segment_metrics_context_.report(true); - tablet_metrics_context_.report(true); - return ret; -} - -// Scan and statistics partitions that need to be recycled -int InstanceRecycler::scan_and_statistics_partitions() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_partitions"); - - RecyclePartKeyInfo part_key_info0 {instance_id_, 0}; - RecyclePartKeyInfo part_key_info1 {instance_id_, INT64_MAX}; - std::string part_key0; - std::string part_key1; - int64_t earlest_ts = std::numeric_limits::max(); - - recycle_partition_key(part_key_info0, &part_key0); - recycle_partition_key(part_key_info1, &part_key1); - auto handle_partition_kv = [&, this](std::string_view k, std::string_view v) -> int { - RecyclePartitionPB part_pb; - if (!part_pb.ParseFromArray(v.data(), v.size())) { - return 0; - } - int64_t current_time = ::time(nullptr); - if (current_time < - calculate_partition_expired_time(instance_id_, part_pb, &earlest_ts)) { // not expired - return 0; - } - // decode partition_id - auto k1 = k; - k1.remove_prefix(1); - std::vector, int, int>> out; - decode_key(&k1, &out); - // 0x01 "recycle" ${instance_id} "partition" ${partition_id} -> RecyclePartitionPB - auto partition_id = std::get(std::get<0>(out[3])); - // Change state to RECYCLING - std::unique_ptr txn; - TxnErrorCode err = txn_kv_->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - std::string val; - err = txn->get(k, &val); - if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) { - return 0; - } - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - part_pb.Clear(); - if (!part_pb.ParseFromString(val)) { - return 0; - } - // Partitions with PREPARED state MUST have no data - bool is_empty_tablet = part_pb.state() == RecyclePartitionPB::PREPARED; - int ret = 0; - for (int64_t index_id : part_pb.index_id()) { - if (scan_tablets_and_statistics(part_pb.table_id(), index_id, metrics_context, - partition_id, is_empty_tablet) != 0) { - ret = 0; - } - } - metrics_context.total_need_recycle_num++; - return ret; - }; - - int ret = scan_and_recycle(part_key0, part_key1, std::move(handle_partition_kv)); - metrics_context.report(true); - segment_metrics_context_.report(true); - tablet_metrics_context_.report(true); - return ret; -} - -// Scan and statistics rowsets that need to be recycled -int InstanceRecycler::scan_and_statistics_rowsets() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_rowsets"); - RecycleRowsetKeyInfo recyc_rs_key_info0 {instance_id_, 0, ""}; - RecycleRowsetKeyInfo recyc_rs_key_info1 {instance_id_, INT64_MAX, ""}; - std::string recyc_rs_key0; - std::string recyc_rs_key1; - recycle_rowset_key(recyc_rs_key_info0, &recyc_rs_key0); - recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1); - int64_t earlest_ts = std::numeric_limits::max(); - - auto handle_rowset_kv = [&, this](std::string_view k, std::string_view v) -> int { - RecycleRowsetPB rowset; - if (!rowset.ParseFromArray(v.data(), v.size())) { - return 0; - } - auto* rowset_meta = rowset.mutable_rowset_meta(); - int64_t current_time = ::time(nullptr); - if (current_time < - calculate_rowset_expired_time(instance_id_, rowset, &earlest_ts)) { // not expired - return 0; - } - - if (!rowset.has_type()) { - if (!rowset.has_resource_id()) [[unlikely]] { - return 0; - } - if (rowset.resource_id().empty()) [[unlikely]] { - return 0; - } - metrics_context.total_need_recycle_num++; - metrics_context.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size(); - segment_metrics_context_.total_need_recycle_num += rowset.rowset_meta().num_segments(); - segment_metrics_context_.total_need_recycle_data_size += - rowset.rowset_meta().total_disk_size(); - return 0; - } - - if (config::enable_mark_delete_rowset_before_recycle && - rowset.type() == RecycleRowsetPB::PREPARE && - (!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled())) { - return 0; - } - - if (!rowset_meta->has_resource_id()) [[unlikely]] { - if (rowset.type() == RecycleRowsetPB::PREPARE || rowset_meta->num_segments() != 0) { - return 0; - } - } - metrics_context.total_need_recycle_num++; - metrics_context.total_need_recycle_data_size += rowset_meta->total_disk_size(); - segment_metrics_context_.total_need_recycle_num += rowset_meta->num_segments(); - segment_metrics_context_.total_need_recycle_data_size += rowset_meta->total_disk_size(); - return 0; - }; - int ret = scan_and_recycle(recyc_rs_key0, recyc_rs_key1, std::move(handle_rowset_kv)); - metrics_context.report(true); - segment_metrics_context_.report(true); - return ret; -} - -// Scan and statistics tmp_rowsets that need to be recycled -int InstanceRecycler::scan_and_statistics_tmp_rowsets() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_tmp_rowsets"); - MetaRowsetTmpKeyInfo tmp_rs_key_info0 {instance_id_, 0, 0}; - MetaRowsetTmpKeyInfo tmp_rs_key_info1 {instance_id_, INT64_MAX, 0}; - std::string tmp_rs_key0; - std::string tmp_rs_key1; - meta_rowset_tmp_key(tmp_rs_key_info0, &tmp_rs_key0); - meta_rowset_tmp_key(tmp_rs_key_info1, &tmp_rs_key1); - - int64_t earlest_ts = std::numeric_limits::max(); - - auto handle_tmp_rowsets_kv = [&, this](std::string_view k, std::string_view v) -> int { - doris::RowsetMetaCloudPB rowset; - if (!rowset.ParseFromArray(v.data(), v.size())) { - return 0; - } - int64_t expiration = calculate_tmp_rowset_expired_time(instance_id_, rowset, &earlest_ts); - int64_t current_time = ::time(nullptr); - if (current_time < expiration) { - return 0; - } - - DCHECK_GT(rowset.txn_id(), 0) - << "txn_id=" << rowset.txn_id() << " rowset=" << rowset.ShortDebugString(); - - if (!rowset.has_is_recycled() || !rowset.is_recycled()) { - return 0; - } - - if (!rowset.has_resource_id()) { - if (rowset.num_segments() > 0) [[unlikely]] { // impossible - return 0; - } - return 0; - } - - metrics_context.total_need_recycle_num++; - metrics_context.total_need_recycle_data_size += rowset.total_disk_size(); - segment_metrics_context_.total_need_recycle_data_size += rowset.total_disk_size(); - segment_metrics_context_.total_need_recycle_num += rowset.num_segments(); - return 0; - }; - int ret = scan_and_recycle(tmp_rs_key0, tmp_rs_key1, std::move(handle_tmp_rowsets_kv)); - metrics_context.report(true); - segment_metrics_context_.report(true); - return ret; -} - -// Scan and statistics abort_timeout_txn that need to be recycled -int InstanceRecycler::scan_and_statistics_abort_timeout_txn() { - RecyclerMetricsContext metrics_context(instance_id_, "abort_timeout_txn"); - - TxnRunningKeyInfo txn_running_key_info0 {instance_id_, 0, 0}; - TxnRunningKeyInfo txn_running_key_info1 {instance_id_, INT64_MAX, INT64_MAX}; - std::string begin_txn_running_key; - std::string end_txn_running_key; - txn_running_key(txn_running_key_info0, &begin_txn_running_key); - txn_running_key(txn_running_key_info1, &end_txn_running_key); - - int64_t current_time = - duration_cast(system_clock::now().time_since_epoch()).count(); - - auto handle_abort_timeout_txn_kv = [&metrics_context, ¤t_time, this]( - std::string_view k, std::string_view v) -> int { - std::unique_ptr txn; - TxnErrorCode err = txn_kv_->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - std::string_view k1 = k; - k1.remove_prefix(1); - std::vector, int, int>> out; - if (decode_key(&k1, &out) != 0) { - return 0; - } - int64_t db_id = std::get(std::get<0>(out[3])); - int64_t txn_id = std::get(std::get<0>(out[4])); - // Update txn_info - std::string txn_inf_key, txn_inf_val; - txn_info_key({instance_id_, db_id, txn_id}, &txn_inf_key); - err = txn->get(txn_inf_key, &txn_inf_val); - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - TxnInfoPB txn_info; - if (!txn_info.ParseFromString(txn_inf_val)) { - return 0; - } - - if (TxnStatusPB::TXN_STATUS_COMMITTED != txn_info.status()) { - TxnRunningPB txn_running_pb; - if (!txn_running_pb.ParseFromArray(v.data(), v.size())) { - return 0; - } - if (!config::force_immediate_recycle && txn_running_pb.timeout_time() > current_time) { - return 0; - } - metrics_context.total_need_recycle_num++; - } - return 0; - }; - - int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, - std::move(handle_abort_timeout_txn_kv)); - metrics_context.report(true); - return ret; -} - -// Scan and statistics expired_txn_label that need to be recycled -int InstanceRecycler::scan_and_statistics_expired_txn_label() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_expired_txn_label"); - - RecycleTxnKeyInfo recycle_txn_key_info0 {instance_id_, 0, 0}; - RecycleTxnKeyInfo recycle_txn_key_info1 {instance_id_, INT64_MAX, INT64_MAX}; - std::string begin_recycle_txn_key; - std::string end_recycle_txn_key; - recycle_txn_key(recycle_txn_key_info0, &begin_recycle_txn_key); - recycle_txn_key(recycle_txn_key_info1, &end_recycle_txn_key); - int64_t earlest_ts = std::numeric_limits::max(); - int64_t current_time_ms = - duration_cast(system_clock::now().time_since_epoch()).count(); - - // for calculate the total num or bytes of recyled objects - auto handle_expired_txn_label_kv = [&, this](std::string_view k, std::string_view v) -> int { - RecycleTxnPB recycle_txn_pb; - if (!recycle_txn_pb.ParseFromArray(v.data(), v.size())) { - return 0; - } - if ((config::force_immediate_recycle) || - (recycle_txn_pb.has_immediate() && recycle_txn_pb.immediate()) || - (calculate_txn_expired_time(instance_id_, recycle_txn_pb, &earlest_ts) <= - current_time_ms)) { - metrics_context.total_need_recycle_num++; - } - return 0; - }; - - int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, - std::move(handle_expired_txn_label_kv)); - metrics_context.report(true); - return ret; -} - -// Scan and statistics copy_jobs that need to be recycled -int InstanceRecycler::scan_and_statistics_copy_jobs() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_copy_jobs"); - CopyJobKeyInfo key_info0 {instance_id_, "", 0, "", 0}; - CopyJobKeyInfo key_info1 {instance_id_, "\xff", 0, "", 0}; - std::string key0; - std::string key1; - copy_job_key(key_info0, &key0); - copy_job_key(key_info1, &key1); - - // for calculate the total num or bytes of recyled objects - auto scan_and_statistics = [&metrics_context](std::string_view k, std::string_view v) -> int { - CopyJobPB copy_job; - if (!copy_job.ParseFromArray(v.data(), v.size())) { - LOG_WARNING("malformed copy job").tag("key", hex(k)); - return 0; - } - - if (copy_job.job_status() == CopyJobPB::FINISH) { - if (copy_job.stage_type() == StagePB::EXTERNAL) { - int64_t current_time = - duration_cast(system_clock::now().time_since_epoch()).count(); - if (copy_job.finish_time_ms() > 0) { - if (!config::force_immediate_recycle && - current_time < copy_job.finish_time_ms() + - config::copy_job_max_retention_second * 1000) { - return 0; - } - } else { - if (!config::force_immediate_recycle && - current_time < copy_job.start_time_ms() + - config::copy_job_max_retention_second * 1000) { - return 0; - } - } - } - } else if (copy_job.job_status() == CopyJobPB::LOADING) { - int64_t current_time = - duration_cast(system_clock::now().time_since_epoch()).count(); - if (!config::force_immediate_recycle && current_time <= copy_job.timeout_time_ms()) { - return 0; - } - } - metrics_context.total_need_recycle_num++; - return 0; - }; - - int ret = scan_and_recycle(key0, key1, std::move(scan_and_statistics)); - metrics_context.report(true); - return ret; -} - -// Scan and statistics stage that need to be recycled -int InstanceRecycler::scan_and_statistics_stage() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_stage"); - RecycleStageKeyInfo key_info0 {instance_id_, ""}; - RecycleStageKeyInfo key_info1 {instance_id_, "\xff"}; - std::string key0 = recycle_stage_key(key_info0); - std::string key1 = recycle_stage_key(key_info1); - - // for calculate the total num or bytes of recyled objects - auto scan_and_statistics = [&metrics_context, this](std::string_view k, - std::string_view v) -> int { - RecycleStagePB recycle_stage; - if (!recycle_stage.ParseFromArray(v.data(), v.size())) { - LOG_WARNING("malformed recycle stage").tag("key", hex(k)); - return 0; - } - - int idx = stoi(recycle_stage.stage().obj_info().id()); - if (idx > instance_info_.obj_info().size() || idx < 1) { - LOG(WARNING) << "invalid idx: " << idx; - return 0; - } - - std::shared_ptr accessor; - int ret = SYNC_POINT_HOOK_RETURN_VALUE( - [&] { - auto& old_obj = instance_info_.obj_info()[idx - 1]; - auto s3_conf = S3Conf::from_obj_store_info(old_obj); - if (!s3_conf) { - return 0; - } - - s3_conf->prefix = recycle_stage.stage().obj_info().prefix(); - std::shared_ptr s3_accessor; - int ret = S3Accessor::create(std::move(s3_conf.value()), &s3_accessor); - if (ret != 0) { - return 0; - } - - accessor = std::move(s3_accessor); - return 0; - }(), - "recycle_stage:get_accessor", &accessor); - - if (ret != 0) { - LOG(WARNING) << "failed to init accessor ret=" << ret; - return 0; - } - - metrics_context.total_need_recycle_num++; - return 0; - }; - - int ret = scan_and_recycle(key0, key1, std::move(scan_and_statistics)); - metrics_context.report(true); - return ret; -} - -// Scan and statistics expired_stage_objects that need to be recycled -int InstanceRecycler::scan_and_statistics_expired_stage_objects() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_expired_stage_objects"); - - // for calculate the total num or bytes of recyled objects - auto scan_and_statistics = [&metrics_context, this]() { - for (const auto& stage : instance_info_.stages()) { - if (stopped()) { - break; - } - if (stage.type() == StagePB::EXTERNAL) { - continue; - } - int idx = stoi(stage.obj_info().id()); - if (idx > instance_info_.obj_info().size() || idx < 1) { - continue; - } - const auto& old_obj = instance_info_.obj_info()[idx - 1]; - auto s3_conf = S3Conf::from_obj_store_info(old_obj); - if (!s3_conf) { - continue; - } - s3_conf->prefix = stage.obj_info().prefix(); - std::shared_ptr accessor; - int ret1 = S3Accessor::create(*s3_conf, &accessor); - if (ret1 != 0) { - continue; - } - if (s3_conf->prefix.find("/stage/") == std::string::npos) { - continue; - } - metrics_context.total_need_recycle_num++; - } - }; - - scan_and_statistics(); - metrics_context.report(true); - return 0; -} - -// Scan and statistics versions that need to be recycled -int InstanceRecycler::scan_and_statistics_versions() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_versions"); - auto version_key_begin = partition_version_key({instance_id_, 0, 0, 0}); - auto version_key_end = partition_version_key({instance_id_, INT64_MAX, 0, 0}); - - int64_t last_scanned_table_id = 0; - bool is_recycled = false; // Is last scanned kv recycled - // for calculate the total num or bytes of recyled objects - auto scan_and_statistics = [&metrics_context, &last_scanned_table_id, &is_recycled, this]( - std::string_view k, std::string_view) { - auto k1 = k; - k1.remove_prefix(1); - // 0x01 "version" ${instance_id} "partition" ${db_id} ${tbl_id} ${partition_id} - std::vector, int, int>> out; - decode_key(&k1, &out); - DCHECK_EQ(out.size(), 6) << k; - auto table_id = std::get(std::get<0>(out[4])); - if (table_id == last_scanned_table_id) { // Already handle kvs of this table - metrics_context.total_need_recycle_num += - is_recycled; // Version kv of this table has been recycled - return 0; - } - last_scanned_table_id = table_id; - is_recycled = false; - auto tablet_key_begin = stats_tablet_key({instance_id_, table_id, 0, 0, 0}); - auto tablet_key_end = stats_tablet_key({instance_id_, table_id, INT64_MAX, 0, 0}); - std::unique_ptr txn; - TxnErrorCode err = txn_kv_->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - std::unique_ptr iter; - err = txn->get(tablet_key_begin, tablet_key_end, &iter, false, 1); - if (err != TxnErrorCode::TXN_OK) { - return 0; - } - if (iter->has_next()) { // Table is useful, should not recycle table and partition versions - return 0; - } - metrics_context.total_need_recycle_num++; - is_recycled = true; - return 0; - }; - - int ret = scan_and_recycle(version_key_begin, version_key_end, std::move(scan_and_statistics)); - metrics_context.report(true); - return ret; -} - -// Scan and statistics restore jobs that need to be recycled -int InstanceRecycler::scan_and_statistics_restore_jobs() { - RecyclerMetricsContext metrics_context(instance_id_, "recycle_restore_jobs"); - JobRestoreTabletKeyInfo restore_job_key_info0 {instance_id_, 0}; - JobRestoreTabletKeyInfo restore_job_key_info1 {instance_id_, INT64_MAX}; - std::string restore_job_key0; - std::string restore_job_key1; - job_restore_tablet_key(restore_job_key_info0, &restore_job_key0); - job_restore_tablet_key(restore_job_key_info1, &restore_job_key1); - - int64_t earlest_ts = std::numeric_limits::max(); - - // for calculate the total num or bytes of recyled objects - auto scan_and_statistics = [&](std::string_view k, std::string_view v) -> int { - RestoreJobCloudPB restore_job_pb; - if (!restore_job_pb.ParseFromArray(v.data(), v.size())) { - LOG_WARNING("malformed recycle partition value").tag("key", hex(k)); - return 0; - } - int64_t expiration = - calculate_restore_job_expired_time(instance_id_, restore_job_pb, &earlest_ts); - int64_t current_time = ::time(nullptr); - if (current_time < expiration) { // not expired - return 0; - } - metrics_context.total_need_recycle_num++; - if (restore_job_pb.need_recycle_data()) { - scan_tablet_and_statistics(restore_job_pb.tablet_id(), metrics_context); - } - return 0; - }; - - int ret = scan_and_recycle(restore_job_key0, restore_job_key1, std::move(scan_and_statistics)); - metrics_context.report(true); - return ret; -} - -void InstanceRecycler::scan_and_statistics_operation_logs() { - if (!should_recycle_versioned_keys()) { - return; - } - - RecyclerMetricsContext metrics_context(instance_id_, "recycle_operation_logs"); - - OperationLogRecycleChecker recycle_checker(instance_id_, txn_kv_.get(), instance_info_); - if (recycle_checker.init() != 0) { - return; - } - - std::string log_key_prefix = versioned::log_key(instance_id_); - std::string begin_key = encode_versioned_key(log_key_prefix, Versionstamp::min()); - std::string end_key = encode_versioned_key(log_key_prefix, Versionstamp::max()); - - std::unique_ptr iter = blob_get_range(txn_kv_, begin_key, end_key); - for (; iter->valid(); iter->next()) { - OperationLogPB operation_log; - if (!iter->parse_value(&operation_log)) { - continue; - } - - std::string_view key = iter->key(); - Versionstamp log_versionstamp; - if (!decode_versioned_key(&key, &log_versionstamp)) { - continue; - } - - OperationLogReferenceInfo ref_info; - if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(), - &ref_info)) { - metrics_context.total_need_recycle_num++; - metrics_context.total_need_recycle_data_size += operation_log.ByteSizeLong(); - } - } - - metrics_context.report(true); -} - int InstanceRecycler::classify_rowset_task_by_ref_count( RowsetDeleteTask& task, std::vector& batch_delete_tasks) { constexpr int MAX_RETRY = 10; diff --git a/cloud/src/recycler/recycler.h b/cloud/src/recycler/recycler.h index 77b017d3e404ec..fd20d2e4ef870d 100644 --- a/cloud/src/recycler/recycler.h +++ b/cloud/src/recycler/recycler.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -28,6 +29,8 @@ #include #include #include +#include +#include #include #include #include @@ -55,9 +58,6 @@ class InstanceRecycler; class StorageVaultAccessor; class Checker; class SimpleThreadPool; -class RecyclerMetricsContext; -class TabletRecyclerMetricsContext; -class SegmentRecyclerMetricsContext; int64_t calculate_tmp_rowset_expired_time( const std::string& instance_id_, const doris::RowsetMetaCloudPB& tmp_rowset_meta_pb, @@ -151,107 +151,69 @@ struct RowsetDeleteTask { std::string rowset_ref_count_key; }; -class RecyclerMetricsContext { +class RecyclerMetricValue { public: - RecyclerMetricsContext() = default; + // Workers update counters concurrently; publishers read relaxed snapshots. + RecyclerMetricValue& operator+=(uint64_t delta) { + value_.fetch_add(delta, std::memory_order_relaxed); + return *this; + } - RecyclerMetricsContext(std::string instance_id, std::string operation_type) - : operation_type(std::move(operation_type)), instance_id(std::move(instance_id)) { - start(); + RecyclerMetricValue& operator++() { + *this += 1; + return *this; } - ~RecyclerMetricsContext() = default; + uint64_t operator++(int) { return value_.fetch_add(1, std::memory_order_relaxed); } - std::atomic_ullong total_need_recycle_data_size = 0; - std::atomic_ullong total_need_recycle_num = 0; + void reset() { value_.store(0, std::memory_order_relaxed); } + void set(uint64_t v) { value_.store(v, std::memory_order_relaxed); } + uint64_t value() const { return value_.load(std::memory_order_relaxed); } - std::atomic_ullong total_recycled_data_size = 0; - std::atomic_ullong total_recycled_num = 0; +private: + std::atomic_ullong value_ = 0; +}; - std::string operation_type; - std::string instance_id; +class RecyclerMetricsContext { +public: + explicit RecyclerMetricsContext(std::string instance_id, std::string operation_type); + ~RecyclerMetricsContext(); - double start_time = 0; + void update_metrics(); - void start() { - start_time = duration_cast( - std::chrono::system_clock::now().time_since_epoch()) - .count(); - } + RecyclerMetricValue kv_scanned_num; + RecyclerMetricValue kv_expired_num; + RecyclerMetricValue kv_recycled_num; + RecyclerMetricValue kv_recycled_bytes; - double duration() const { - return duration_cast( - std::chrono::system_clock::now().time_since_epoch()) - .count() - - start_time; - } + const std::string operation_type; + const std::string instance_id; - void reset() { - total_need_recycle_data_size = 0; - total_need_recycle_num = 0; - total_recycled_data_size = 0; - total_recycled_num = 0; - start_time = duration_cast( - std::chrono::system_clock::now().time_since_epoch()) - .count(); - } - - void finish_report() { - if (!operation_type.empty()) { - double cost = duration(); - g_bvar_recycler_instance_last_round_recycle_elpased_ts.put( - {instance_id, operation_type}, cost); - g_bvar_recycler_instance_recycle_round.put({instance_id, operation_type}, 1); - g_bvar_recycler_instance_recycle_total_bytes_since_started.put( - {instance_id, operation_type}, total_recycled_data_size.load()); - g_bvar_recycler_instance_recycle_total_num_since_started.put( - {instance_id, operation_type}, total_recycled_num.load()); - LOG(INFO) << "recycle instance: " << instance_id - << ", operation type: " << operation_type << ", cost: " << cost - << " ms, total recycled num: " << total_recycled_num.load() - << ", total recycled data size: " << total_recycled_data_size.load() - << " bytes"; - if (cost != 0) { - if (total_recycled_num.load() != 0) { - g_bvar_recycler_instance_recycle_time_per_resource.put( - {instance_id, operation_type}, cost / total_recycled_num.load()); - } - g_bvar_recycler_instance_recycle_bytes_per_ms.put( - {instance_id, operation_type}, total_recycled_data_size.load() / cost); - } - } - } +private: + std::chrono::steady_clock::time_point start_time_; - // `is_begin` is used to initialize total num of items need to be recycled - void report(bool is_begin = false) { - if (!operation_type.empty()) { - // is init - if (is_begin) { - auto value = total_need_recycle_num.load(); - - g_bvar_recycler_instance_last_round_to_recycle_bytes.put( - {instance_id, operation_type}, total_need_recycle_data_size.load()); - g_bvar_recycler_instance_last_round_to_recycle_num.put( - {instance_id, operation_type}, value); - } else { - g_bvar_recycler_instance_last_round_recycled_bytes.put( - {instance_id, operation_type}, total_recycled_data_size.load()); - g_bvar_recycler_instance_last_round_recycled_num.put({instance_id, operation_type}, - total_recycled_num.load()); - } - } - } + void reset(); + void finish(); + void publish_current_round_metrics(int64_t elapsed_ms); }; -class TabletRecyclerMetricsContext : public RecyclerMetricsContext { +// Per-instance tablet or segment totals for one do_recycle() round. +class RecyclerResourceMetricsContext { public: - TabletRecyclerMetricsContext() : RecyclerMetricsContext("global_recycler", "recycle_tablet") {} -}; + RecyclerResourceMetricsContext(std::string instance_id, std::string resource_type); -class SegmentRecyclerMetricsContext : public RecyclerMetricsContext { -public: - SegmentRecyclerMetricsContext() - : RecyclerMetricsContext("global_recycler", "recycle_segment") {} + void start_round(); + void update_metrics(); + void finish_round(); + + RecyclerMetricValue recycled_num; + RecyclerMetricValue recycled_bytes; + +private: + const std::string instance_id_; + const std::string resource_type_; + // Empty between recycle rounds, including monitor updates after finish_round(). + std::optional start_time_; }; struct OplogRecycleStats; @@ -413,30 +375,6 @@ class InstanceRecycler { bool check_recycle_tasks(); - int scan_and_statistics_indexes(); - - int scan_and_statistics_partitions(); - - int scan_and_statistics_rowsets(); - - int scan_and_statistics_tmp_rowsets(); - - int scan_and_statistics_abort_timeout_txn(); - - int scan_and_statistics_expired_txn_label(); - - int scan_and_statistics_copy_jobs(); - - int scan_and_statistics_stage(); - - int scan_and_statistics_expired_stage_objects(); - - int scan_and_statistics_versions(); - - int scan_and_statistics_restore_jobs(); - - void scan_and_statistics_operation_logs(); - /** * Decode the key of a packed-file metadata record into the persisted object path. * @@ -467,7 +405,7 @@ class InstanceRecycler { int init_storage_vault_accessors(); int recycle_stream(int64_t stream_id, const RecycleIndexPB& recycle_index, - std::string_view recycle_key); + std::string_view recycle_key, RecyclerMetricsContext& metrics_context); int recycle_table_stream_offset_prefix(std::string prefix, RecyclerMetricsContext* metrics_context); @@ -477,7 +415,8 @@ class InstanceRecycler { int recycle_partition_table_stream_offsets( int64_t db_id, int64_t table_id, int64_t partition_id, - const google::protobuf::RepeatedPtrField& table_streams); + const google::protobuf::RepeatedPtrField& table_streams, + RecyclerMetricsContext& metrics_context); /** * Scan key-value pairs between [`begin`, `end`) with multiple rounds of range get(`RangeGetIterator`), @@ -583,7 +522,8 @@ class InstanceRecycler { // Recycle rowset meta and data, return 0 for success otherwise error // // This function will decrease the rowset ref count and remove the rowset meta and data if the ref count is 1. - int recycle_rowset_meta_and_data(const RowsetDeleteTask& task); + int recycle_rowset_meta_and_data(const RowsetDeleteTask& task, + RecyclerMetricsContext& metrics_context); // Classify rowset task by ref_count, return 0 to add to batch delete, 1 if handled (ref>1), -1 on error int classify_rowset_task_by_ref_count(RowsetDeleteTask& task, @@ -673,7 +613,8 @@ class InstanceRecycler { void submit_recycle_prepare_rowsets_job(SimpleThreadPool& worker_pool, std::vector rowset_keys_to_abort, - std::atomic_long* num_recycled); + std::atomic_long* num_recycled, + RecyclerMetricsContext* metrics_context); void submit_recycle_tmp_rowsets_job(SimpleThreadPool& worker_pool, std::vector rowset_keys_to_abort, @@ -709,8 +650,8 @@ class InstanceRecycler { std::shared_ptr delete_bitmap_lock_white_list_; std::shared_ptr resource_mgr_; - TabletRecyclerMetricsContext tablet_metrics_context_; - SegmentRecyclerMetricsContext segment_metrics_context_; + RecyclerResourceMetricsContext tablet_metrics_context_; + RecyclerResourceMetricsContext segment_metrics_context_; // Data tablets in the same partition have the same MoW setting. Cache both true and false so // subsequent rowsets can avoid reading the tablet index and tablet meta. Row-binlog tablets diff --git a/cloud/src/recycler/recycler_operation_log.cpp b/cloud/src/recycler/recycler_operation_log.cpp index 0e4d94bf5d0641..6ecf4b7be45d20 100644 --- a/cloud/src/recycler/recycler_operation_log.cpp +++ b/cloud/src/recycler/recycler_operation_log.cpp @@ -767,14 +767,6 @@ int InstanceRecycler::recycle_operation_logs() { RecyclerMetricsContext metrics_context(instance_id_, task_name); OplogRecycleStats oplog_stats; - // scan_and_statistics_operation_logs() is expensive (scans lots of KVs), - // so it's controlled by enable_recycler_stats_metrics. - // The other stats (counting what was actually recycled) are lightweight - // and always collected. - if (config::enable_recycler_stats_metrics) { - scan_and_statistics_operation_logs(); - } - StopWatch stop_watch; size_t total_operation_logs = 0; size_t recycled_operation_logs = 0; @@ -783,7 +775,6 @@ int InstanceRecycler::recycle_operation_logs() { size_t recycled_operation_log_data_size = 0; DORIS_CLOUD_DEFER { - metrics_context.finish_report(); report_oplog_recycle_stats(instance_id_, oplog_stats); int64_t cost = stop_watch.elapsed_us() / 1000'000; @@ -817,10 +808,11 @@ int InstanceRecycler::recycle_operation_logs() { size_t value_size = operation_log.ByteSizeLong(); OperationLogReferenceInfo reference_info; + metrics_context.kv_scanned_num++; + if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(), &reference_info)) { - metrics_context.total_need_recycle_num++; - metrics_context.total_need_recycle_data_size += value_size; + metrics_context.kv_expired_num++; AnnotateTag tag("log_key", hex(key)); int res = recycle_operation_log(log_versionstamp, raw_keys, std::move(operation_log), @@ -833,8 +825,7 @@ int InstanceRecycler::recycle_operation_logs() { recycled_operation_logs++; recycled_operation_log_data_size += value_size; - metrics_context.total_recycled_num++; - metrics_context.total_recycled_data_size += value_size; + metrics_context.kv_recycled_num++; } else { oplog_stats.not_recycled_num.fetch_add(1, std::memory_order_relaxed); int res = calculator.calculate_operation_log_data_size(key, operation_log, @@ -849,7 +840,6 @@ int InstanceRecycler::recycle_operation_logs() { operation_log_data_size += value_size; max_operation_log_data_size = std::max(max_operation_log_data_size, value_size); oplog_stats.total_num.fetch_add(1, std::memory_order_relaxed); - metrics_context.report(); return 0; }; @@ -905,8 +895,11 @@ int InstanceRecycler::recycle_operation_logs() { return res; } - if (i % 1000 == 0 && is_multi_version_status_changed() != 0) { - return -1; + if (i % 1000 == 0) { + metrics_context.update_metrics(); + if (is_multi_version_status_changed() != 0) { + return -1; + } } } if (iter->error_code() != TxnErrorCode::TXN_OK) { diff --git a/cloud/src/recycler/recycler_service.cpp b/cloud/src/recycler/recycler_service.cpp index c9c68004f8aef5..656a18f8eea974 100644 --- a/cloud/src/recycler/recycler_service.cpp +++ b/cloud/src/recycler/recycler_service.cpp @@ -64,240 +64,6 @@ RecyclerServiceImpl::RecyclerServiceImpl(std::shared_ptr txn_kv, Recycler RecyclerServiceImpl::~RecyclerServiceImpl() = default; -void RecyclerServiceImpl::statistics_recycle(StatisticsRecycleRequest& req, MetaServiceCode& code, - std::string& msg) { - std::unique_ptr txn; - TxnErrorCode err = txn_kv_->create_txn(&txn); - if (err != TxnErrorCode::TXN_OK) { - code = MetaServiceCode::KV_TXN_CREATE_ERR; - msg = "failed to create txn"; - return; - } - - static std::map> resource_handlers = { - {"recycle_indexes", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_indexes(); - }}, - {"recycle_partitions", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_partitions(); - }}, - {"recycle_tmp_rowsets", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_tmp_rowsets(); - }}, - {"recycle_rowsets", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_rowsets(); - }}, - {"abort_timeout_txn", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_abort_timeout_txn(); - }}, - {"recycle_expired_txn_label", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_expired_txn_label(); - }}, - {"recycle_versions", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_versions(); - }}, - {"recycle_copy_jobs", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_copy_jobs(); - }}, - {"recycle_stage", - [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_stage(); - }}, - {"recycle_expired_stage_objects", [](InstanceRecycler& instance_recycler) { - instance_recycler.scan_and_statistics_expired_stage_objects(); - }}}; - - std::set resource_types; - for (const auto& resource_type : req.resource_type()) { - if (resource_type == "*") { - std::ranges::for_each(resource_handlers, - [&](const auto& it) { resource_types.emplace(it.first); }); - break; - } else { - if (!resource_handlers.contains(resource_type)) { - code = MetaServiceCode::INVALID_ARGUMENT; - msg = fmt::format( - "invalid resource type: {}, valid resource_type have [{}]", resource_type, - std::accumulate(resource_handlers.begin(), resource_handlers.end(), - std::string(), [](const std::string& acc, const auto& it) { - return acc.empty() ? it.first : acc + ", " + it.first; - })); - LOG_WARNING(msg); - return; - } else { - resource_types.emplace(resource_type); - } - } - } - - std::set instance_ids; - std::vector instances; - get_all_instances(txn_kv_.get(), instances); - - for (const auto& instance_id : req.instance_ids()) { - if (instance_id == "*") { - std::ranges::for_each(instances, [&](const InstanceInfoPB& instance) { - instance_ids.emplace(instance.instance_id()); - }); - break; - } else { - if (std::ranges::find_if(instances, [&](const InstanceInfoPB& instance) { - return instance.instance_id() == instance_id; - }) == instances.end()) { - code = MetaServiceCode::INVALID_ARGUMENT; - msg = fmt::format("invalid instance id: {}", instance_id); - LOG_WARNING(msg); - return; - } else { - instance_ids.emplace(instance_id); - } - } - } - - LOG(INFO) << "begin to statistics recycle for " - << std::accumulate(instance_ids.begin(), instance_ids.end(), std::string(), - [](const std::string& acc, const std::string& id) { - return acc.empty() ? id : acc + ", " + id; - }); - - auto worker_pool = std::make_unique( - config::instance_recycler_statistics_recycle_worker_pool_size, "statistics_recycle"); - worker_pool->start(); - - for (const auto& id : instance_ids) { - InstanceKeyInfo key_info {id}; - std::string key; - instance_key(key_info, &key); - std::string val; - err = txn->get(key, &val); - if (err != TxnErrorCode::TXN_OK) { - code = MetaServiceCode::KV_TXN_GET_ERR; - msg = fmt::format("failed to get instance, instance_id={}, err={}", id, err); - LOG_WARNING(msg); - continue; - } - InstanceInfoPB instance; - if (!instance.ParseFromString(val)) { - code = MetaServiceCode::PROTOBUF_PARSE_ERR; - msg = fmt::format("malformed instance info, key={}, val={}", hex(key), hex(val)); - LOG_WARNING(msg); - continue; - } - auto instance_recycler = std::make_shared( - txn_kv_, instance, recycler_->_thread_pool_group, txn_lazy_committer_); - - if (int r = instance_recycler->init(); r != 0) { - LOG(WARNING) << "failed to init instance recycler, instance_id=" << id << " ret=" << r; - continue; - } - // if empty, statistics all resources - if (resource_types.empty()) { - for (const auto& [_, func] : resource_handlers) { - worker_pool->submit([&instance_recycler, &func]() { func(*instance_recycler); }); - } - } else { - for (const auto& resource_type : resource_types) { - if (auto it = resource_handlers.find(resource_type); - it != resource_handlers.end()) { - worker_pool->submit( - [&it, &instance_recycler]() { it->second(*instance_recycler); }); - } - } - } - } - - worker_pool->stop(); - std::stringstream ss; - std::ranges::for_each(instance_ids, [&](const std::string& id) { - ss << "Instance ID: " << id << "\n"; - ss << "----------------------------------------\n"; - - // tablet and segment statistics - int64_t tablet_num = g_bvar_recycler_instance_last_round_to_recycle_num.get( - {"global_recycler", "recycle_tablet"}); - int64_t tablet_bytes = g_bvar_recycler_instance_last_round_to_recycle_num.get( - {"global_recycler", "recycle_tablet"}); - int64_t segment_num = g_bvar_recycler_instance_last_round_to_recycle_num.get( - {"global_recycler", "recycle_segment"}); - int64_t segment_bytes = g_bvar_recycler_instance_last_round_to_recycle_num.get( - {"global_recycler", "recycle_segment"}); - // clang-format off - ss << "Global recycler: " << "tablet and segment" << "\n"; - ss << " • Need to recycle tablet count: " << tablet_num << " items\n"; - ss << " • Need to recycle tablet size: " << tablet_bytes << " bytes\n"; - ss << " • Need to recycle segment count: " << segment_num << " items\n"; - ss << " • Need to recycle segment size: " << segment_bytes << " bytes\n"; - // clang-format on - - std::ranges::for_each(resource_types, [&](const auto& resource_type) { - int64_t to_recycle_num = - g_bvar_recycler_instance_last_round_to_recycle_num.get({id, resource_type}); - int64_t to_recycle_bytes = - g_bvar_recycler_instance_last_round_to_recycle_bytes.get({id, resource_type}); - - ss << "Task Type: " << resource_type << "\n"; - - // Add specific counts for different resource types - if (resource_type == "recycle_partitions") { - ss << " • Need to recycle partition count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle partition size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_rowsets") { - ss << " • Need to recycle rowset count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle rowset size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_tmp_rowsets") { - ss << " • Need to recycle tmp rowset count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle tmp rowset size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_indexes") { - ss << " • Need to recycle index count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle index size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_segment") { - ss << " • Need to recycle segment count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle segment size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_tablet") { - ss << " • Need to recycle tablet count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle tablet size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_versions") { - ss << " • Need to recycle version count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle version size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "abort_timeout_txn") { - ss << " • Need to abort timeout txn count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle timeout txn size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_expired_txn_label") { - ss << " • Need to recycle expired txn label count: " << to_recycle_num - << " items\n"; - ss << " • Need to recycle expired txn label size: " << to_recycle_bytes - << " bytes\n"; - } else if (resource_type == "recycle_copy_jobs") { - ss << " • Need to recycle copy job count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle copy job size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_stage") { - ss << " • Need to recycle stage count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle stage size: " << to_recycle_bytes << " bytes\n"; - } else if (resource_type == "recycle_expired_stage_objects") { - ss << " • Need to recycle expired stage object count: " << to_recycle_num - << " items\n"; - ss << " • Need to recycle expired stage object size: " << to_recycle_bytes - << " bytes\n"; - } else { - ss << " • Need to recycle count: " << to_recycle_num << " items\n"; - ss << " • Need to recycle size: " << to_recycle_bytes << " bytes\n"; - } - - ss << "----------------------------------------\n"; - }); - ss << "\n"; - }); - msg = ss.str(); -} - void RecyclerServiceImpl::recycle_instance(::google::protobuf::RpcController* controller, const ::doris::cloud::RecycleInstanceRequest* request, ::doris::cloud::RecycleInstanceResponse* response, diff --git a/cloud/src/recycler/recycler_service.h b/cloud/src/recycler/recycler_service.h index 01849dd8b08f51..43894c53906b68 100644 --- a/cloud/src/recycler/recycler_service.h +++ b/cloud/src/recycler/recycler_service.h @@ -46,8 +46,6 @@ class RecyclerServiceImpl : public cloud::RecyclerService { ::doris::cloud::MetaServiceHttpResponse* response, ::google::protobuf::Closure* done) override; - void statistics_recycle(StatisticsRecycleRequest& req, MetaServiceCode& code, std::string& msg); - void check_instance(const std::string& instance_id, MetaServiceCode& code, std::string& msg); std::pair skip_instance_data_cleanup( diff --git a/cloud/test/recycle_versioned_keys_test.cpp b/cloud/test/recycle_versioned_keys_test.cpp index 763749dafb4ff8..546618e9dffb9f 100644 --- a/cloud/test/recycle_versioned_keys_test.cpp +++ b/cloud/test/recycle_versioned_keys_test.cpp @@ -1236,7 +1236,7 @@ TEST(RecycleVersionedKeysTest, RecycleTabletWithRowsetRefCount) { InstanceInfoPB instance_info; ASSERT_NO_FATAL_FAILURE(get_instance(meta_service.get(), cloud_unique_id, instance_info)); auto recycler = get_instance_recycler(meta_service.get(), instance_info, accessor); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); ASSERT_EQ(0, recycler->recycle_tablet(tablet_id, ctx)); { @@ -1476,7 +1476,7 @@ void recycle_tablet_with_rowset_ref_count_concurrent() { recycler_threads.emplace_back([&, i]() { InstanceInfoPB instance_info = instance_infos[i]; auto recycler = get_instance_recycler(meta_service.get(), instance_info, accessor); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); recycler_results[i] = recycler->recycle_tablet(tablet_id, ctx); completed_recyclers.fetch_add(1); }); @@ -1600,7 +1600,7 @@ TEST(RecycleVersionedKeysTest, RecycleTabletMetadataAndIndexesAtomically) { }); sp->enable_processing(); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "recycle_tablets"); ASSERT_EQ(recycler->recycle_tablets(table_id, index_id, ctx), -1); ASSERT_TRUE(commit_attempted); ASSERT_NO_FATAL_FAILURE(check_tablet_keys(true)); @@ -2037,7 +2037,7 @@ TEST(RecycleVersionedKeysTest, BatchDeleteRefCountOne) { InstanceInfoPB instance_info; ASSERT_NO_FATAL_FAILURE(get_instance(meta_service.get(), cloud_unique_id, instance_info)); auto recycler = get_instance_recycler(meta_service.get(), instance_info, accessor); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); ASSERT_EQ(0, recycler->recycle_tablet(tablet_id, ctx)); { @@ -2134,7 +2134,7 @@ TEST(RecycleVersionedKeysTest, BatchDeleteRefCountGreaterThanOne) { InstanceInfoPB instance_info; ASSERT_NO_FATAL_FAILURE(get_instance(meta_service.get(), cloud_unique_id, instance_info)); auto recycler = get_instance_recycler(meta_service.get(), instance_info, accessor); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); ASSERT_EQ(0, recycler->recycle_tablet(tablet_id, ctx)); { @@ -2238,7 +2238,7 @@ TEST(RecycleVersionedKeysTest, BatchDeleteMixedRefCount) { InstanceInfoPB instance_info; ASSERT_NO_FATAL_FAILURE(get_instance(meta_service.get(), cloud_unique_id, instance_info)); auto recycler = get_instance_recycler(meta_service.get(), instance_info, accessor); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); ASSERT_EQ(0, recycler->recycle_tablet(tablet_id, ctx)); { @@ -2340,7 +2340,7 @@ TEST(RecycleVersionedKeysTest, BatchDeleteMultipleVaults) { recycler->TEST_add_accessor(resource_id_1, accessor_1); recycler->TEST_add_accessor(resource_id_2, accessor_2); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); ASSERT_EQ(0, recycler->recycle_tablet(tablet_id, ctx)); { @@ -2410,7 +2410,7 @@ TEST(RecycleVersionedKeysTest, BatchDeleteExceedsMaxBatchSize) { InstanceInfoPB instance_info; ASSERT_NO_FATAL_FAILURE(get_instance(meta_service.get(), cloud_unique_id, instance_info)); auto recycler = get_instance_recycler(meta_service.get(), instance_info, accessor); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx(instance_id, "test"); ASSERT_EQ(0, recycler->recycle_tablet(tablet_id, ctx)); { diff --git a/cloud/test/recycler_operation_log_test.cpp b/cloud/test/recycler_operation_log_test.cpp index fcacb503882322..687adb937b9da2 100644 --- a/cloud/test/recycler_operation_log_test.cpp +++ b/cloud/test/recycler_operation_log_test.cpp @@ -3309,12 +3309,6 @@ TEST(RecycleOperationLogTest, OplogRecycleStatsPerTypeCounters) { // Test recycle_operation_logs with stats enabled: drop partition + drop index logs // Verifies per-type recycled counts and bvar reporting TEST(RecycleOperationLogTest, RecycleOperationLogsWithStatsEnabled) { - auto old_flag = config::enable_recycler_stats_metrics; - config::enable_recycler_stats_metrics = true; - DORIS_CLOUD_DEFER { - config::enable_recycler_stats_metrics = old_flag; - }; - auto txn_kv = std::make_shared(); txn_kv->update_commit_version(1000); ASSERT_EQ(txn_kv->init(), 0); @@ -3455,12 +3449,6 @@ TEST(RecycleOperationLogTest, RecycleOperationLogsWithStatsEnabled) { // Test recycle_operation_logs with snapshot protection: some logs skipped TEST(RecycleOperationLogTest, RecycleOperationLogsSkippedBySnapshot) { - auto old_flag = config::enable_recycler_stats_metrics; - config::enable_recycler_stats_metrics = true; - DORIS_CLOUD_DEFER { - config::enable_recycler_stats_metrics = old_flag; - }; - auto txn_kv = std::make_shared(); txn_kv->update_commit_version(1000); ASSERT_EQ(txn_kv->init(), 0); @@ -3565,12 +3553,6 @@ TEST(RecycleOperationLogTest, RecycleOperationLogsSkippedBySnapshot) { // Test recycle_operation_logs with stats disabled (default) TEST(RecycleOperationLogTest, RecycleOperationLogsStatsDisabled) { - auto old_flag = config::enable_recycler_stats_metrics; - config::enable_recycler_stats_metrics = false; - DORIS_CLOUD_DEFER { - config::enable_recycler_stats_metrics = old_flag; - }; - auto txn_kv = std::make_shared(); txn_kv->update_commit_version(1000); ASSERT_EQ(txn_kv->init(), 0); @@ -3615,12 +3597,6 @@ TEST(RecycleOperationLogTest, RecycleOperationLogsStatsDisabled) { // Test recycle_operation_logs compaction log with stats: recycled_compaction tracking TEST(RecycleOperationLogTest, RecycleCompactionLogWithStats) { - auto old_flag = config::enable_recycler_stats_metrics; - config::enable_recycler_stats_metrics = true; - DORIS_CLOUD_DEFER { - config::enable_recycler_stats_metrics = old_flag; - }; - auto txn_kv = std::make_shared(); txn_kv->update_commit_version(1000); ASSERT_EQ(txn_kv->init(), 0); diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp index 467c4b57bb2e8d..f2dbf6906424fe 100644 --- a/cloud/test/recycler_test.cpp +++ b/cloud/test/recycler_test.cpp @@ -69,7 +69,22 @@ using namespace doris; std::string instance_id = "instance_id_recycle_test"; int64_t current_time = 0; static constexpr int64_t db_id = 1000; -static RecyclerMetricsContext ctx; + +// Lazily-constructed shared metrics context for tests. +// +// RecyclerMetricsContext's constructor publishes to bvars, which in turn logs +// via glog. A namespace-scope RecyclerMetricsContext would therefore run during +// static initialization, before main() sets up glog and the cross-TU bvar +// globals, causing a crash. This proxy has a trivial constructor and builds the +// real context on first use (inside a running test), avoiding the static +// initialization order fiasco while keeping call sites (`ctx`) unchanged. +struct GlobalRecyclerMetricsContext { + operator RecyclerMetricsContext&() { + static RecyclerMetricsContext ctx(instance_id, "test"); + return ctx; + } +}; +static GlobalRecyclerMetricsContext ctx; namespace { @@ -7378,7 +7393,7 @@ TEST(RecyclerTest, delete_partial_update_rowset_data_by_recycling_state) { ASSERT_EQ(accessor->put_file(segment, ""), 0); ASSERT_EQ(accessor->put_file(extra_segment, ""), 0); - RecyclerMetricsContext metrics_context; + RecyclerMetricsContext metrics_context(instance_id, "recycle_rowsets"); ASSERT_EQ(recycler.delete_rowset_data(rowsets, type, metrics_context), 0); EXPECT_EQ(accessor->exists(segment), 1); EXPECT_EQ(accessor->exists(extra_segment), @@ -7454,7 +7469,7 @@ TEST(RecyclerTest, delete_v2_inverted_index_with_segment_list) { std::map rowsets; rowsets.emplace(batch_rowset.rowset_id_v2(), batch_rowset); - RecyclerMetricsContext metrics_context; + RecyclerMetricsContext metrics_context(instance_id, "test"); ASSERT_EQ(recycler.delete_rowset_data(rowsets, RowsetRecyclingState::FORMAL_ROWSET, metrics_context), 0); @@ -8858,10 +8873,145 @@ TEST(RecyclerTest, recycle_tablet_with_empty_resource_id_and_no_segments) { std::make_shared(txn_kv)); EXPECT_EQ(recycler.init(), 0); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx("test_instance", "recycle_tablet"); EXPECT_EQ(recycler.recycle_tablet(0, ctx), 0); } +TEST(RecyclerTest, recycler_metrics_publish_last_round_and_reset_current_round) { + constexpr std::string_view metrics_instance_id = "metrics_last_round_test"; + constexpr std::string_view metrics_operation_type = "recycle_tablet"; + const auto total_num_before = g_bvar_recycler_instance_recycle_total_num_since_started.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}); + const auto total_bytes_before = g_bvar_recycler_instance_recycle_total_bytes_since_started.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}); + + { + RecyclerMetricsContext metrics_context {std::string(metrics_instance_id), + std::string(metrics_operation_type)}; + metrics_context.kv_scanned_num += 5; + metrics_context.kv_expired_num += 4; + metrics_context.kv_recycled_num += 3; + metrics_context.kv_recycled_bytes += 7; + metrics_context.update_metrics(); + + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_scanned.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 5); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_expired.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 4); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_recycled.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 3); + EXPECT_EQ(g_bvar_recycler_instance_current_round_recycled_object_bytes.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 7); + } + + { + RecyclerResourceMetricsContext unused_context {std::string(metrics_instance_id), + std::string(metrics_operation_type)}; + } + + EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_num.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 3); + EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_bytes.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 7); + EXPECT_EQ(g_bvar_recycler_instance_recycle_total_num_since_started.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + total_num_before + 3); + EXPECT_EQ(g_bvar_recycler_instance_recycle_total_bytes_since_started.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + total_bytes_before + 7); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_scanned.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_expired.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_recycled.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_current_round_recycled_object_bytes.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_current_round_task_elapsed_ms.get( + {std::string(metrics_instance_id), std::string(metrics_operation_type)}), + 0); +} + +TEST(RecyclerTest, recycler_resource_metrics_publish_each_round) { + const std::string instance_id = "resource_metrics_round_test"; + const std::string resource_type = "recycle_segment"; + const auto total_num_before = g_bvar_recycler_instance_recycle_total_num_since_started.get( + {instance_id, resource_type}); + const auto total_bytes_before = g_bvar_recycler_instance_recycle_total_bytes_since_started.get( + {instance_id, resource_type}); + + RecyclerResourceMetricsContext metrics_context(instance_id, resource_type); + metrics_context.start_round(); + metrics_context.recycled_num += 2; + metrics_context.recycled_bytes += 7; + metrics_context.update_metrics(); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_recycled.get( + {instance_id, resource_type}), + 2); + EXPECT_EQ(g_bvar_recycler_instance_current_round_recycled_object_bytes.get( + {instance_id, resource_type}), + 7); + + metrics_context.finish_round(); + EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_num.get({instance_id, resource_type}), + 2); + EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_bytes.get({instance_id, resource_type}), + 7); + EXPECT_EQ(g_bvar_recycler_instance_recycle_total_num_since_started.get( + {instance_id, resource_type}), + total_num_before + 2); + EXPECT_EQ(g_bvar_recycler_instance_recycle_total_bytes_since_started.get( + {instance_id, resource_type}), + total_bytes_before + 7); + metrics_context.update_metrics(); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_recycled.get( + {instance_id, resource_type}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_current_round_recycled_object_bytes.get( + {instance_id, resource_type}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_current_round_task_elapsed_ms.get( + {instance_id, resource_type}), + 0); + + metrics_context.start_round(); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_recycled.get( + {instance_id, resource_type}), + 0); + metrics_context.recycled_num += 1; + metrics_context.recycled_bytes += 2; + metrics_context.finish_round(); + EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_num.get({instance_id, resource_type}), + 1); + EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_bytes.get({instance_id, resource_type}), + 2); + EXPECT_EQ(g_bvar_recycler_instance_recycle_total_num_since_started.get( + {instance_id, resource_type}), + total_num_before + 3); + EXPECT_EQ(g_bvar_recycler_instance_recycle_total_bytes_since_started.get( + {instance_id, resource_type}), + total_bytes_before + 9); + EXPECT_EQ(g_bvar_recycler_instance_recycle_current_round_kv_recycled.get( + {instance_id, resource_type}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_current_round_recycled_object_bytes.get( + {instance_id, resource_type}), + 0); + EXPECT_EQ(g_bvar_recycler_instance_current_round_task_elapsed_ms.get( + {instance_id, resource_type}), + 0); +} + TEST(RecyclerTest, recycle_tablet_with_resource_id_and_no_segments) { auto* sp = SyncPoint::get_instance(); DORIS_CLOUD_DEFER { @@ -8895,7 +9045,7 @@ TEST(RecyclerTest, recycle_tablet_with_resource_id_and_no_segments) { recycler.TEST_add_accessor("resource_id", accessor); EXPECT_EQ(accessor->exists("data/1234/orphan.dat"), 0); - RecyclerMetricsContext ctx; + RecyclerMetricsContext ctx("test_instance", "recycle_tablet"); EXPECT_EQ(recycler.recycle_tablet(tablet_id, ctx), 0); EXPECT_EQ(accessor->exists("data/1234/orphan.dat"), 1); } diff --git a/cloud/test/table_stream_recycler_test.cpp b/cloud/test/table_stream_recycler_test.cpp index dff94a89fa0d27..9488825fde5012 100644 --- a/cloud/test/table_stream_recycler_test.cpp +++ b/cloud/test/table_stream_recycler_test.cpp @@ -152,8 +152,6 @@ TEST(TableStreamRecyclerTest, RecycleStreamDeletesOnlyOffsets) { txn->put(unrelated_key, "unrelated physical data"); ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); - const int recycle_round_before = g_bvar_recycler_instance_recycle_round.get( - {std::string(kInstanceId), "recycle_stream"}); const int64_t recycle_total_before = g_bvar_recycler_instance_recycle_total_num_since_started.get( {std::string(kInstanceId), "recycle_stream"}); @@ -172,12 +170,6 @@ TEST(TableStreamRecyclerTest, RecycleStreamDeletesOnlyOffsets) { kStreamDbId, kStreamId)), 0); EXPECT_TRUE(key_exists(txn_kv.get(), unrelated_key)); - EXPECT_EQ(g_bvar_recycler_instance_last_round_recycled_num.get( - {std::string(kInstanceId), "recycle_stream"}), - 6); - EXPECT_EQ(g_bvar_recycler_instance_recycle_round.get( - {std::string(kInstanceId), "recycle_stream"}), - recycle_round_before + 1); EXPECT_EQ(g_bvar_recycler_instance_recycle_total_num_since_started.get( {std::string(kInstanceId), "recycle_stream"}), recycle_total_before + 6); @@ -317,43 +309,6 @@ TEST(TableStreamRecyclerTest, RecycleStreamDeletesOffsetsAcrossRangePages) { 0); } -TEST(TableStreamRecyclerTest, StatisticsDispatchesStreamToOffsetScan) { - const bool old_force_immediate_recycle = config::force_immediate_recycle; - config::force_immediate_recycle = true; - DORIS_CLOUD_DEFER { - config::force_immediate_recycle = old_force_immediate_recycle; - }; - - auto txn_kv = std::make_shared(); - ASSERT_EQ(txn_kv->init(), 0); - std::unique_ptr txn; - ASSERT_EQ(txn_kv->create_txn(&txn), TxnErrorCode::TXN_OK); - put_offset(txn.get(), kPartitionId); - put_offset(txn.get(), kPartitionId + 1); - - RecycleIndexPB recycle_index; - recycle_index.set_db_id(kBaseDbId); - recycle_index.set_table_id(kBaseTableId); - recycle_index.set_creation_time(0); - recycle_index.set_expiration(0); - recycle_index.set_state(RecycleIndexPB::DROPPED); - recycle_index.set_object_type(IndexObjectTypePB::TABLE_STREAM); - recycle_index.set_stream_db_id(kStreamDbId); - txn->put(recycle_index_key({std::string(kInstanceId), kStreamId}), - recycle_index.SerializeAsString()); - ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK); - - InstanceRecycler recycler = make_recycler(txn_kv); - ASSERT_EQ(recycler.scan_and_statistics_indexes(), 0); - - EXPECT_EQ(g_bvar_recycler_instance_last_round_to_recycle_num.get( - {std::string(kInstanceId), "recycle_stream"}), - 6); - EXPECT_EQ(g_bvar_recycler_instance_last_round_to_recycle_num.get( - {std::string(kInstanceId), "recycle_indexes"}), - 0); -} - TEST(TableStreamRecyclerTest, RecyclePartitionDeletesOnlyThatPartitionOffsets) { const bool old_force_immediate_recycle = config::force_immediate_recycle; config::force_immediate_recycle = true; diff --git a/regression-test/plugins/cloud_recycler_plugin.groovy b/regression-test/plugins/cloud_recycler_plugin.groovy index 3f37bc50fc105e..bbca1719a9682b 100644 --- a/regression-test/plugins/cloud_recycler_plugin.groovy +++ b/regression-test/plugins/cloud_recycler_plugin.groovy @@ -510,92 +510,31 @@ Suite.metaClass.checkRecycleMetrics = { String recyclerHttpPort, String recycleJ int retryCount = 0 while (true) { - def metricDataBeforeRecycle = getRecyclerMetricsMethod.call( - recyclerHttpPort, - "recycler_instance_last_round_to_recycle_bytes", - recycleJobType - ) - - def metricDataAftereRecycle = getRecyclerMetricsMethod.call( + def recycledBytesMetric = getRecyclerMetricsMethod.call( recyclerHttpPort, "recycler_instance_last_round_recycled_bytes", recycleJobType ) - - // not all resource types have bytes metrics - def validResourceTypes = ["recycle_indexes", "recycle_partitions", "recycle_tmp_rowsets", "recycle_rowsets", "recycle_tablet", "recycle_segment"] - - boolean checkFlag1 = false - boolean checkFlag2 = false - - if (validResourceTypes.contains(recycleJobType)) { - checkFlag1 = true - } - - if (metricDataBeforeRecycle && metricDataAftereRecycle && !checkFlag1) { - if (metricDataBeforeRecycle.value == metricDataAftereRecycle.value) { - logger.info("--- Recycle Success ---") - logger.info("Metric Name: recycler_instance_last_round_recycled_bytes") - logger.info("Value: ${metricDataBeforeRecycle.value}") - logger.info("Resource Type: ${metricDataBeforeRecycle.labels?.resource_type}") - logger.info("--------------------------------------") - checkFlag1 = true - } else { - logger.info("--- Recycle failed ---") - logger.info("Metric Name: recycler_instance_last_round_to_recycle_bytes") - logger.info("Value: ${metricDataBeforeRecycle.value}") - logger.info("Resource Type: ${metricDataBeforeRecycle.labels?.resource_type}") - logger.info("--------------------------------------") - logger.info("Metric Name: recycler_instance_last_round_recycled_bytes") - logger.info("Value: ${metricDataAftereRecycle.value}") - logger.info("Resource Type: ${metricDataAftereRecycle.labels?.resource_type}") - logger.info("--------------------------------------") - } - } - - metricDataBeforeRecycle = getRecyclerMetricsMethod.call( - recyclerHttpPort, - "recycler_instance_last_round_to_recycle_num", - recycleJobType - ) - - metricDataAftereRecycle = getRecyclerMetricsMethod.call( + def recycledNumMetric = getRecyclerMetricsMethod.call( recyclerHttpPort, "recycler_instance_last_round_recycled_num", recycleJobType ) - if (metricDataBeforeRecycle && metricDataAftereRecycle && !checkFlag2) { - if (metricDataBeforeRecycle.value == metricDataAftereRecycle.value) { - logger.info("--- Recycle Success ---") - logger.info("Metric Name: recycler_instance_last_round_recycled_num") - logger.info("Value: ${metricDataBeforeRecycle.value}") - logger.info("Resource Type: ${metricDataBeforeRecycle.labels?.resource_type}") - logger.info("--------------------------------------") - checkFlag2 = true - } else { - logger.info("--- Recycle failed ---") - logger.info("Metric Name: recycler_instance_last_round_to_recycle_num") - logger.info("Value: ${metricDataBeforeRecycle.value}") - logger.info("Resource Type: ${metricDataBeforeRecycle.labels?.resource_type}") - logger.info("--------------------------------------") - logger.info("Metric Name: recycler_instance_last_round_recycled_num") - logger.info("Value: ${metricDataAftereRecycle.value}") - logger.info("Resource Type: ${metricDataAftereRecycle.labels?.resource_type}") - logger.info("--------------------------------------") - } - } - - if (checkFlag1 && checkFlag2) { - break; + if (recycledBytesMetric && recycledNumMetric) { + logger.info("Found recycler metrics for resource type '${recycleJobType}': " + + "recycled_bytes=${recycledBytesMetric.value}, recycled_num=${recycledNumMetric.value}") + break } retryCount++ if (retryCount > 10) { - logger.error("Failed to get metric 'recycler_instance_last_round_to_recycle_bytes' after 10 retries.") - return; + assertNotNull(recycledBytesMetric, + "Missing metric 'recycler_instance_last_round_recycled_bytes' for resource type '${recycleJobType}'") + assertNotNull(recycledNumMetric, + "Missing metric 'recycler_instance_last_round_recycled_num' for resource type '${recycleJobType}'") } sleep(5000) } } -logger.info("Added 'checkRecycleMetrics' function to Suite") \ No newline at end of file +logger.info("Added 'checkRecycleMetrics' function to Suite")