github-actions[bot] commented on code in PR #68005:
URL: https://github.com/apache/doris/pull/68005#discussion_r4059150311
##########
cloud/src/recycler/recycler.h:
##########
@@ -153,105 +154,136 @@ struct RowsetDeleteTask {
class RecyclerMetricsContext {
public:
- RecyclerMetricsContext() = default;
+ enum class MetricType {
+ SCANNED_NUM,
+ EXPIRED_NUM,
+ RECYCLED_NUM,
+ RECYCLED_BYTES,
+ };
- RecyclerMetricsContext(std::string instance_id, std::string operation_type)
- : operation_type(std::move(operation_type)),
instance_id(std::move(instance_id)) {
- start();
- }
+ class MetricValue {
+ public:
+ // Concurrent workers only update atomics. Batch boundaries publish
their snapshots.
+ MetricValue& operator+=(uint64_t delta) {
+ value_.fetch_add(delta, std::memory_order_relaxed);
+ return *this;
+ }
+
+ MetricValue& 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); }
- std::atomic_ullong total_recycled_data_size = 0;
- std::atomic_ullong total_recycled_num = 0;
+ void set(uint64_t v) { value_.store(v, std::memory_order_relaxed); }
- std::string operation_type;
- std::string instance_id;
+ uint64_t value() const { return
value_.load(std::memory_order_relaxed); }
+
+ private:
+ std::atomic_ullong value_ = 0;
+ };
- double start_time = 0;
+ RecyclerMetricsContext() = delete;
- void start() {
- start_time = duration_cast<std::chrono::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
+ explicit RecyclerMetricsContext(std::string instance_id, std::string
operation_type)
+ : operation_type(std::move(operation_type)),
+ instance_id(std::move(instance_id)),
+ start_time_(std::chrono::steady_clock::now()) {
+ reset();
}
- double duration() const {
- return duration_cast<std::chrono::milliseconds>(
- std::chrono::system_clock::now().time_since_epoch())
- .count() -
- start_time;
+ // Each context has one publisher; workers may update its MetricValues
concurrently.
+ void update_metrics() {
+ auto cost =
duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() -
+ start_time_)
+ .count();
+
g_bvar_recycler_instance_current_round_task_elapsed_ms.put({instance_id,
operation_type},
+ cost);
+ put(MetricType::SCANNED_NUM, kv_scanned_num.value());
+ put(MetricType::EXPIRED_NUM, kv_expired_num.value());
+ put(MetricType::RECYCLED_NUM, kv_recycled_num.value());
+ put(MetricType::RECYCLED_BYTES, kv_recycled_bytes.value());
}
+ ~RecyclerMetricsContext() { finish(); }
+
+ MetricValue kv_scanned_num;
+ MetricValue kv_expired_num;
+ MetricValue kv_recycled_num;
+ MetricValue kv_recycled_bytes;
+
+ std::string operation_type;
+ std::string instance_id;
+
+private:
+ std::chrono::steady_clock::time_point start_time_;
+
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::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
+ start_time_ = std::chrono::steady_clock::now();
+ kv_scanned_num.reset();
+ kv_expired_num.reset();
+ kv_recycled_num.reset();
+ kv_recycled_bytes.reset();
+ put(MetricType::SCANNED_NUM, 0);
+ put(MetricType::EXPIRED_NUM, 0);
+ put(MetricType::RECYCLED_NUM, 0);
+ put(MetricType::RECYCLED_BYTES, 0);
+
g_bvar_recycler_instance_current_round_task_elapsed_ms.put({instance_id,
operation_type},
+ 0);
}
- 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());
+ void finish() {
+ update_metrics();
+ if (auto num = kv_recycled_num.value(); num > 0) {
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);
- }
+ {instance_id, operation_type}, static_cast<int64_t>(num));
+ }
+ if (auto bytes = kv_recycled_bytes.value(); bytes > 0) {
+ g_bvar_recycler_instance_recycle_total_bytes_since_started.put(
+ {instance_id, operation_type},
static_cast<int64_t>(bytes));
}
+ g_bvar_recycler_instance_last_round_recycled_num.put(
+ {instance_id, operation_type},
static_cast<int64_t>(kv_recycled_num.value()));
+ g_bvar_recycler_instance_last_round_recycled_bytes.put(
+ {instance_id, operation_type},
static_cast<int64_t>(kv_recycled_bytes.value()));
+ reset();
Review Comment:
[P2] Publish current metrics before final reset. The stream contexts and the
member tablet/segment contexts update these atomics but never call
update_metrics() in production. Their only publication is this finalizer, where
update_metrics() is immediately followed by reset(), so a scrape during a long
round sees zero and cannot observe the work at all. Please publish at joined
batch boundaries (and advance/reset the instance current elapsed gauge) rather
than relying on destruction.
##########
cloud/src/recycler/recycler.h:
##########
@@ -153,105 +154,136 @@ struct RowsetDeleteTask {
class RecyclerMetricsContext {
public:
- RecyclerMetricsContext() = default;
+ enum class MetricType {
+ SCANNED_NUM,
+ EXPIRED_NUM,
+ RECYCLED_NUM,
+ RECYCLED_BYTES,
+ };
- RecyclerMetricsContext(std::string instance_id, std::string operation_type)
- : operation_type(std::move(operation_type)),
instance_id(std::move(instance_id)) {
- start();
- }
+ class MetricValue {
+ public:
+ // Concurrent workers only update atomics. Batch boundaries publish
their snapshots.
+ MetricValue& operator+=(uint64_t delta) {
+ value_.fetch_add(delta, std::memory_order_relaxed);
+ return *this;
+ }
+
+ MetricValue& 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); }
- std::atomic_ullong total_recycled_data_size = 0;
- std::atomic_ullong total_recycled_num = 0;
+ void set(uint64_t v) { value_.store(v, std::memory_order_relaxed); }
- std::string operation_type;
- std::string instance_id;
+ uint64_t value() const { return
value_.load(std::memory_order_relaxed); }
+
+ private:
+ std::atomic_ullong value_ = 0;
+ };
- double start_time = 0;
+ RecyclerMetricsContext() = delete;
- void start() {
- start_time = duration_cast<std::chrono::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
+ explicit RecyclerMetricsContext(std::string instance_id, std::string
operation_type)
+ : operation_type(std::move(operation_type)),
+ instance_id(std::move(instance_id)),
+ start_time_(std::chrono::steady_clock::now()) {
+ reset();
}
- double duration() const {
- return duration_cast<std::chrono::milliseconds>(
- std::chrono::system_clock::now().time_since_epoch())
- .count() -
- start_time;
+ // Each context has one publisher; workers may update its MetricValues
concurrently.
+ void update_metrics() {
+ auto cost =
duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() -
+ start_time_)
+ .count();
+
g_bvar_recycler_instance_current_round_task_elapsed_ms.put({instance_id,
operation_type},
+ cost);
+ put(MetricType::SCANNED_NUM, kv_scanned_num.value());
+ put(MetricType::EXPIRED_NUM, kv_expired_num.value());
+ put(MetricType::RECYCLED_NUM, kv_recycled_num.value());
+ put(MetricType::RECYCLED_BYTES, kv_recycled_bytes.value());
}
+ ~RecyclerMetricsContext() { finish(); }
+
+ MetricValue kv_scanned_num;
+ MetricValue kv_expired_num;
+ MetricValue kv_recycled_num;
+ MetricValue kv_recycled_bytes;
+
+ std::string operation_type;
+ std::string instance_id;
+
+private:
+ std::chrono::steady_clock::time_point start_time_;
+
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::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
+ start_time_ = std::chrono::steady_clock::now();
+ kv_scanned_num.reset();
+ kv_expired_num.reset();
+ kv_recycled_num.reset();
+ kv_recycled_bytes.reset();
+ put(MetricType::SCANNED_NUM, 0);
+ put(MetricType::EXPIRED_NUM, 0);
+ put(MetricType::RECYCLED_NUM, 0);
Review Comment:
[P2] Do not reset a shared series from an uncoordinated publisher. The
manual recycle_copy_jobs endpoint can construct a second InstanceRecycler for
an instance already in recycling_instance_map_ because it uses a separate
s_worker guard. That constructor resets the scheduled round's tablet/segment
series, and its destructor later overwrites their last-round values with zero
even though the manual worker never ran those tasks. Please share the
scheduler's per-instance exclusion or make publication generation-aware.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5732,7 +5628,9 @@ 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_.kv_recycled_num += 1;
+ segment_metrics_context_.kv_recycled_num += recycle_segments_number;
Review Comment:
[P2] Aggregate only physical deletion results, exactly once.
recycle_segments_number includes ref-count>1 rowsets whose files are
deliberately retained, while a unique batch-deleted rowset has already
incremented segment_metrics_context_ inside delete_rowset_data(), so this line
counts it a second time. The pre-aggregated tablet bytes above have the same
retained-rowset problem. Please collect per-rowset delete outcomes after
classification for the tablet/segment totals.
##########
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) {
Review Comment:
[P2] Exercise the new inline counters here. This condition only checks that
two legacy last-round records exist; every RecyclerMetricsContext destructor
creates them even when their values are zero, so the test passes with the
missing publishers and wrong task accounting introduced in this change. Please
assert expected values/deltas for workloads with recyclable data and cover at
least one failed/partial path.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -7822,8 +7762,9 @@ int InstanceRecycler::recycle_expired_stage_objects() {
ret = -1;
continue;
}
- metrics_context.total_recycled_num++;
- metrics_context.report();
+ ++num_expired;
Review Comment:
[P2] Do not publish this object-store operation as a recycled TxnKV. This
loop scans instance stages and delete_all() removes S3 objects; the following
comment even says no KV is recycled, but these locals are copied into
kv_expired_num/kv_recycled_num. A success therefore reports one nonexistent KV
deletion, while a storage failure also hides the eligible stage. Use an
object-operation/byte metric here (and mark eligibility before deletion if an
expired counter is retained).
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5968,6 +5866,12 @@ 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);
Review Comment:
[P2] Record the physical deletion for PREPARE rowsets. On this prefix path
the worker can delete the objects and recycle KV successfully, but it only
increments num_recycled; the known total_disk_size and num_segments never reach
the task or segment metrics. Carry those values into the completion callback
and update them only after the object deletion succeeds, including the
abort/recheck and versioned variants.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -2791,8 +2796,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();
Review Comment:
[P2] Advance expired at the eligibility boundary, not after commit. If
txn_remove fails for a page of N scanned offsets, this code publishes
scanned=N, expired=0, recycled=0 even though all N are still eligible for
retry. Increment expired while collecting the offsets and keep recycled here
after the successful removal so the two gauges distinguish backlog from
completed work.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -3417,6 +3428,7 @@ int InstanceRecycler::recycle_versions() {
if (iter->has_next()) { // Table is useful, should not recycle table
and partition versions
return 0;
}
+ ++num_expired;
Review Comment:
[P2] Keep all three counters in the same unit. For a dropped table with N
partition-version keys, this branch makes scanned=N and recycled=N but expired
remains 1 because only the first key increments it. The range delete can also
remove keys beyond the materialized page, so recycled can undercount the
committed removal. Count exact eligible/committed KVs, or expose these as
table-level metrics under different names.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -2881,9 +2883,8 @@ int
InstanceRecycler::recycle_partition_table_stream_offsets(
int64_t db_id, int64_t table_id, int64_t partition_id,
const google::protobuf::RepeatedPtrField<TableStreamIdentityPB>&
table_streams) {
RecyclerMetricsContext metrics_context(instance_id_,
"recycle_stream_partition_offsets");
- DORIS_CLOUD_DEFER {
- metrics_context.finish_report();
- };
+ metrics_context.kv_scanned_num += table_streams.size();
Review Comment:
[P2] Do not pre-count bindings as scanned/expired KVs before a transaction
exists. If create_txn() fails, this publishes scanned=N and expired=N without
touching TxnKV at all. Even on success no KV is scanned and each binding
removes two keyspaces. Account only completed work and either measure actual
KVs or expose this as a logical-binding metric.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5345,17 +5254,11 @@ int InstanceRecycler::recycle_tablet(int64_t tablet_id,
RecyclerMetricsContext&
return ret;
}
- tablet_metrics_context_.total_recycled_data_size +=
+ tablet_metrics_context_.kv_recycled_bytes +=
Review Comment:
[P2] Preserve bytes from successful vault deletions on partial failure. A
tablet may span storage vaults, and these directory deletes run independently;
if vault A succeeds and vault B fails, the aggregate return above skips this
update even though A's objects are already gone. A retry then attributes those
bytes to a later idempotent delete. Track rowset bytes per resource and publish
each successful task's physical deletion before reducing the overall tablet
status.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -5391,6 +5294,8 @@ int InstanceRecycler::recycle_tablet(int64_t tablet_id,
RecyclerMetricsContext&
if (ret == 0) {
// All object files under tablet have been deleted
+ tablet_metrics_context_.kv_recycled_num += 1;
Review Comment:
[P2] Count a versioned tablet only once. recycle_versioned_tablet() already
increments tablet_metrics_context_.kv_recycled_num after its metadata commit,
then this wrapper unconditionally runs the legacy tail and increments the same
counter again. The mode predicate uses the InstanceRecycler's fixed
instance_info_ snapshot, so the comment about a mid-call mode change does not
prevent the normal 2-per-tablet result. Give one layer sole ownership of the
tablet count.
##########
cloud/src/recycler/recycler.h:
##########
@@ -153,105 +154,136 @@ struct RowsetDeleteTask {
class RecyclerMetricsContext {
public:
- RecyclerMetricsContext() = default;
+ enum class MetricType {
+ SCANNED_NUM,
+ EXPIRED_NUM,
+ RECYCLED_NUM,
+ RECYCLED_BYTES,
+ };
- RecyclerMetricsContext(std::string instance_id, std::string operation_type)
- : operation_type(std::move(operation_type)),
instance_id(std::move(instance_id)) {
- start();
- }
+ class MetricValue {
+ public:
+ // Concurrent workers only update atomics. Batch boundaries publish
their snapshots.
+ MetricValue& operator+=(uint64_t delta) {
+ value_.fetch_add(delta, std::memory_order_relaxed);
+ return *this;
+ }
+
+ MetricValue& 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); }
- std::atomic_ullong total_recycled_data_size = 0;
- std::atomic_ullong total_recycled_num = 0;
+ void set(uint64_t v) { value_.store(v, std::memory_order_relaxed); }
- std::string operation_type;
- std::string instance_id;
+ uint64_t value() const { return
value_.load(std::memory_order_relaxed); }
+
+ private:
+ std::atomic_ullong value_ = 0;
+ };
- double start_time = 0;
+ RecyclerMetricsContext() = delete;
- void start() {
- start_time = duration_cast<std::chrono::milliseconds>(
-
std::chrono::system_clock::now().time_since_epoch())
- .count();
+ explicit RecyclerMetricsContext(std::string instance_id, std::string
operation_type)
+ : operation_type(std::move(operation_type)),
+ instance_id(std::move(instance_id)),
+ start_time_(std::chrono::steady_clock::now()) {
+ reset();
}
- double duration() const {
- return duration_cast<std::chrono::milliseconds>(
- std::chrono::system_clock::now().time_since_epoch())
- .count() -
- start_time;
+ // Each context has one publisher; workers may update its MetricValues
concurrently.
+ void update_metrics() {
+ auto cost =
duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() -
+ start_time_)
+ .count();
+
g_bvar_recycler_instance_current_round_task_elapsed_ms.put({instance_id,
operation_type},
+ cost);
+ put(MetricType::SCANNED_NUM, kv_scanned_num.value());
+ put(MetricType::EXPIRED_NUM, kv_expired_num.value());
+ put(MetricType::RECYCLED_NUM, kv_recycled_num.value());
+ put(MetricType::RECYCLED_BYTES, kv_recycled_bytes.value());
}
+ ~RecyclerMetricsContext() { finish(); }
Review Comment:
[P2] Do not finalize a context that never started. InstanceRecycler
constructs the tablet/segment members before init() and recycle-job
preparation; if either preflight step fails, destruction reaches this call and
writes zero last-round values even though do_recycle() never began. That erases
the last completed round for ordinary init/lease failures. Activate/finalize
these contexts only after the round starts, or guard finish() with explicit
started/used state.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -6610,10 +6517,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);
Review Comment:
[P2] Count the primary KV once its removal commits. Both tmp-rowset workers
first delete the objects and primary tmp-rowset KV, then attempt ref-count
cleanup in a second txn and return before incrementing num_recycled if that
cleanup fails. The worker error is not propagated and the primary key is gone,
so this round can report recycled=0 permanently. Combine the removals or
advance the primary count before the secondary cleanup and surface that failure
separately.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -7499,8 +7414,8 @@ int InstanceRecycler::recycle_copy_jobs() {
if (!config::force_immediate_recycle && current_time <=
copy_job.timeout_time_ms()) {
return 0;
}
- ++num_expired;
}
+ ++num_expired;
Review Comment:
[P2] Account for the internal job before returning from this branch.
BatchObjStoreAccessor later deletes its objects and commits removal of the
copy-file and copy-job KVs, but this early return bypasses both num_expired and
num_recycled, and consume() has no outcome channel back to them. Successful
internal jobs therefore remain scanned-only. Please record eligibility when
queued and report recycled only after each batch KV commit succeeds.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -6577,6 +6483,7 @@ 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;
Review Comment:
[P2] Preserve the physical-delete result across the metadata retry. If
object deletion succeeds but this commit conflicts, the loop starts again with
recycled_bytes=0; a later shared-ref path can return success while reporting no
deleted bytes. A non-conflict commit error also drops the already completed
deletion. Publish the byte result once at object-delete success (or retain it
across retries) without allowing a retry to double-add it.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -4730,13 +4761,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 +=
Review Comment:
[P2] Do not charge object bytes again on a metadata-only retry. This line
adds total_disk_size whenever delete_files() succeeds, but
cleanup_rowset_metadata() can then fail its separate transaction and
deliberately leave the recycle key for the next round. Deleting already-absent
files is also treated as success, so every retry re-adds the same metadata size
to last-round and cumulative bytes. Persist an object-deleted/accounted phase
or use a delete result that distinguishes files actually removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
