Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Feature] Add compaction grafana metric (backport #30446) #30824

Closed
wants to merge 1 commit into from
Closed

[Feature] Add compaction grafana metric (backport #30446) #30824

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 12 additions & 4 deletions be/src/http/action/compaction_action.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,8 +117,12 @@ Status CompactionAction::do_compaction(uint64_t tablet_id, const string& compact

auto* mem_tracker = GlobalEnv::GetInstance()->compaction_mem_tracker();
if (tablet->updates() != nullptr) {
StarRocksMetrics::instance()->update_compaction_request_total.increment(1);
StarRocksMetrics::instance()->running_update_compaction_task_num.increment(1);
DeferOp op([&] { StarRocksMetrics::instance()->running_update_compaction_task_num.increment(-1); });
Status res;
if (rowset_ids_string.empty()) {
RETURN_IF_ERROR(tablet->updates()->compaction(mem_tracker));
res = tablet->updates()->compaction(mem_tracker);
} else {
vector<string> id_str_list = strings::Split(rowset_ids_string, ",", strings::SkipEmpty());
vector<uint32_t> rowset_ids;
Expand All @@ -138,7 +142,13 @@ Status CompactionAction::do_compaction(uint64_t tablet_id, const string& compact
if (rowset_ids.empty()) {
return Status::InvalidArgument(fmt::format("empty argument. rowset_ids:{}", rowset_ids_string));
}
RETURN_IF_ERROR(tablet->updates()->compaction(mem_tracker, rowset_ids));
res = tablet->updates()->compaction(mem_tracker, rowset_ids);
}
if (!res.ok()) {
StarRocksMetrics::instance()->update_compaction_request_failed.increment(1);
LOG(WARNING) << "failed to perform update compaction. res=" << res.get_error_msg()
<< ", tablet=" << tablet->full_name();
return res;
}
return Status::OK();
}
Expand All @@ -149,7 +159,6 @@ Status CompactionAction::do_compaction(uint64_t tablet_id, const string& compact
}

if (compaction_type == to_string(CompactionType::CUMULATIVE_COMPACTION)) {
StarRocksMetrics::instance()->cumulative_compaction_request_total.increment(1);
if (config::enable_size_tiered_compaction_strategy) {
if (tablet->need_compaction()) {
auto compaction_task = tablet->create_compaction_task();
Expand Down Expand Up @@ -183,7 +192,6 @@ Status CompactionAction::do_compaction(uint64_t tablet_id, const string& compact
}
tablet->set_last_cumu_compaction_failure_time(0);
} else if (compaction_type == to_string(CompactionType::BASE_COMPACTION)) {
StarRocksMetrics::instance()->base_compaction_request_total.increment(1);
if (config::enable_size_tiered_compaction_strategy) {
if (tablet->force_base_compaction()) {
auto compaction_task = tablet->create_compaction_task();
Expand Down
13 changes: 12 additions & 1 deletion be/src/storage/base_compaction.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,13 @@ Status BaseCompaction::compact() {
return Status::InvalidArgument("base compaction input parameter error.");
}

StarRocksMetrics::instance()->base_compaction_request_total.increment(1);
StarRocksMetrics::instance()->running_base_compaction_task_num.increment(1);
std::unique_lock lock(_tablet->get_base_lock(), std::try_to_lock);
if (!lock.owns_lock()) {
return Status::OK();
}
int64_t start_time = UnixMillis();
TRACE("got base compaction lock");

// 1. pick rowsets to compact
Expand All @@ -44,7 +47,10 @@ Status BaseCompaction::compact() {
TRACE_COUNTER_INCREMENT("input_rowsets_count", _input_rowsets.size());

MemTracker* prev_tracker = tls_thread_status.set_mem_tracker(_mem_tracker);
DeferOp op([&] { tls_thread_status.set_mem_tracker(prev_tracker); });
DeferOp op([&] {
tls_thread_status.set_mem_tracker(prev_tracker);
StarRocksMetrics::instance()->running_base_compaction_task_num.increment(-1);
});

// 2. do base compaction, merge rowsets
RETURN_IF_ERROR(do_compaction());
Expand All @@ -54,8 +60,13 @@ Status BaseCompaction::compact() {
_state = CompactionState::SUCCESS;

// 4. add metric to base compaction
int64_t end_time = UnixMillis();
int64_t cost_time = end_time - start_time;
StarRocksMetrics::instance()->base_compaction_deltas_total.increment(_input_rowsets.size());
StarRocksMetrics::instance()->base_compaction_bytes_total.increment(_input_rowsets_size);
StarRocksMetrics::instance()->base_compaction_task_cost_time_ms.set_value(cost_time);
StarRocksMetrics::instance()->base_compaction_task_byte_per_second.set_value(_input_rowsets_size /
(cost_time / 1000.0 + 1));
TRACE("save base compaction metrics");

return Status::OK();
Expand Down
26 changes: 26 additions & 0 deletions be/src/storage/compaction_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,11 @@ void CompactionManager::update_candidates(std::vector<CompactionCandidate> candi
bool has_erase = false;
for (auto& candidate : candidates) {
if (candidate.tablet->tablet_id() == iter->tablet->tablet_id()) {
if (iter->type == CompactionType::BASE_COMPACTION) {
StarRocksMetrics::instance()->wait_base_compaction_task_num.increment(-1);
} else {
StarRocksMetrics::instance()->wait_cumulative_compaction_task_num.increment(-1);
}
iter = _compaction_candidates.erase(iter);
has_erase = true;
break;
Expand All @@ -159,6 +164,11 @@ void CompactionManager::update_candidates(std::vector<CompactionCandidate> candi
if (candidate.tablet->enable_compaction()) {
VLOG(1) << "update candidate " << candidate.tablet->tablet_id() << " type "
<< starrocks::to_string(candidate.type) << " score " << candidate.score;
if (candidate.type == CompactionType::BASE_COMPACTION) {
StarRocksMetrics::instance()->wait_base_compaction_task_num.increment(1);
} else {
StarRocksMetrics::instance()->wait_cumulative_compaction_task_num.increment(1);
}
_compaction_candidates.emplace(std::move(candidate));
}
}
Expand All @@ -176,6 +186,11 @@ void CompactionManager::remove_candidate(int64_t tablet_id) {
std::lock_guard lg(_candidates_mutex);
for (auto iter = _compaction_candidates.begin(); iter != _compaction_candidates.end();) {
if (tablet_id == iter->tablet->tablet_id()) {
if (iter->type == CompactionType::BASE_COMPACTION) {
StarRocksMetrics::instance()->wait_base_compaction_task_num.increment(-1);
} else {
StarRocksMetrics::instance()->wait_cumulative_compaction_task_num.increment(-1);
}
iter = _compaction_candidates.erase(iter);
break;
} else {
Expand Down Expand Up @@ -266,6 +281,11 @@ bool CompactionManager::pick_candidate(CompactionCandidate* candidate) {
*candidate = *iter;
_compaction_candidates.erase(iter);
_last_score = candidate->score;
if (candidate->type == CompactionType::BASE_COMPACTION) {
StarRocksMetrics::instance()->wait_base_compaction_task_num.increment(-1);
} else {
StarRocksMetrics::instance()->wait_cumulative_compaction_task_num.increment(-1);
}
return true;
}
iter++;
Expand Down Expand Up @@ -356,9 +376,13 @@ bool CompactionManager::register_task(CompactionTask* compaction_task) {
if (compaction_task->compaction_type() == CUMULATIVE_COMPACTION) {
_data_dir_to_cumulative_task_num_map[data_dir]++;
_cumulative_compaction_concurrency++;
StarRocksMetrics::instance()->cumulative_compaction_request_total.increment(1);
StarRocksMetrics::instance()->running_cumulative_compaction_task_num.increment(1);
} else {
_data_dir_to_base_task_num_map[data_dir]++;
_base_compaction_concurrency++;
StarRocksMetrics::instance()->base_compaction_request_total.increment(1);
StarRocksMetrics::instance()->running_base_compaction_task_num.increment(1);
}
return true;
}
Expand All @@ -378,9 +402,11 @@ void CompactionManager::unregister_task(CompactionTask* compaction_task) {
if (compaction_task->compaction_type() == CUMULATIVE_COMPACTION) {
_data_dir_to_cumulative_task_num_map[data_dir]--;
_cumulative_compaction_concurrency--;
StarRocksMetrics::instance()->running_cumulative_compaction_task_num.increment(-1);
} else {
_data_dir_to_base_task_num_map[data_dir]--;
_base_compaction_concurrency--;
StarRocksMetrics::instance()->running_base_compaction_task_num.increment(-1);
}
}
if (iter->second.empty()) {
Expand Down
9 changes: 9 additions & 0 deletions be/src/storage/compaction_task.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,7 @@ bool CompactionTask::should_stop() const {
void CompactionTask::_success_callback() {
set_compaction_task_state(COMPACTION_SUCCESS);
// for compatible, update compaction time
int64_t cost_time = UnixMillis() - _task_info.start_time;
if (_task_info.compaction_type == CUMULATIVE_COMPACTION) {
_tablet->set_last_cumu_compaction_success_time(UnixMillis());
_tablet->set_last_cumu_compaction_failure_status(TStatusCode::OK);
Expand All @@ -158,9 +159,15 @@ void CompactionTask::_success_callback() {
if (_task_info.compaction_type == CUMULATIVE_COMPACTION) {
StarRocksMetrics::instance()->cumulative_compaction_deltas_total.increment(_input_rowsets.size());
StarRocksMetrics::instance()->cumulative_compaction_bytes_total.increment(_task_info.input_rowsets_size);
StarRocksMetrics::instance()->cumulative_compaction_task_cost_time_ms.set_value(cost_time);
StarRocksMetrics::instance()->cumulative_compaction_task_byte_per_second.set_value(
_task_info.input_rowsets_size / (cost_time / 1000.0 + 1));
} else {
StarRocksMetrics::instance()->base_compaction_deltas_total.increment(_input_rowsets.size());
StarRocksMetrics::instance()->base_compaction_bytes_total.increment(_task_info.input_rowsets_size);
StarRocksMetrics::instance()->base_compaction_task_cost_time_ms.set_value(cost_time);
StarRocksMetrics::instance()->base_compaction_task_byte_per_second.set_value(_task_info.input_rowsets_size /
(cost_time / 1000.0 + 1));
}

// preload the rowset
Expand All @@ -178,8 +185,10 @@ void CompactionTask::_failure_callback(const Status& st) {
if (_task_info.compaction_type == CUMULATIVE_COMPACTION) {
_tablet->set_last_cumu_compaction_failure_time(UnixMillis());
_tablet->set_last_cumu_compaction_failure_status(st.code());
StarRocksMetrics::instance()->cumulative_compaction_request_failed.increment(1);
} else {
_tablet->set_last_base_compaction_failure_time(UnixMillis());
StarRocksMetrics::instance()->base_compaction_request_failed.increment(1);
}
LOG(WARNING) << "compaction task:" << _task_info.task_id << ", tablet:" << _task_info.tablet_id << " failed.";
}
Expand Down
9 changes: 9 additions & 0 deletions be/src/storage/cumulative_compaction.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,14 @@ Status CumulativeCompaction::compact() {
return Status::InvalidArgument("cumulative compaction input parameter error.");
}

StarRocksMetrics::instance()->cumulative_compaction_request_total.increment(1);
StarRocksMetrics::instance()->running_cumulative_compaction_task_num.increment(1);
DeferOp op([&] { StarRocksMetrics::instance()->running_cumulative_compaction_task_num.increment(-1); });
std::unique_lock lock(_tablet->get_cumulative_lock(), std::try_to_lock);
if (!lock.owns_lock()) {
return Status::OK();
}
int64_t start_time = UnixMillis();
TRACE("got cumulative compaction lock");

// 1.calculate cumulative point
Expand All @@ -62,8 +66,13 @@ Status CumulativeCompaction::compact() {
}

// 6. add metric to cumulative compaction
int64_t end_time = UnixMillis();
int64_t cost_time = end_time - start_time;
StarRocksMetrics::instance()->cumulative_compaction_deltas_total.increment(_input_rowsets.size());
StarRocksMetrics::instance()->cumulative_compaction_bytes_total.increment(_input_rowsets_size);
StarRocksMetrics::instance()->cumulative_compaction_task_cost_time_ms.set_value(cost_time);
StarRocksMetrics::instance()->cumulative_compaction_task_byte_per_second.set_value(_input_rowsets_size /
(cost_time / 1000.0 + 1));
TRACE("save cumulative compaction metrics");

return Status::OK();
Expand Down
6 changes: 6 additions & 0 deletions be/src/storage/rowset_merger.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,12 @@ class RowsetMergerImpl : public RowsetMerger {
&total_rows, &total_chunk, &stats);
}
timer.stop();

// update compaction metric
float divided = 1000 * 1000 * 1000;
StarRocksMetrics::instance()->update_compaction_task_cost_time_ns.set_value(timer.elapsed_time());
StarRocksMetrics::instance()->update_compaction_task_byte_per_second.set_value(
total_input_size / (timer.elapsed_time() / divided + 1));
StarRocksMetrics::instance()->update_compaction_deltas_total.increment(rowsets.size());
StarRocksMetrics::instance()->update_compaction_bytes_total.increment(total_input_size);
StarRocksMetrics::instance()->update_compaction_outputs_total.increment(1);
Expand Down
6 changes: 2 additions & 4 deletions be/src/storage/storage_engine.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -898,8 +898,6 @@ Status StorageEngine::_perform_cumulative_compaction(DataDir* data_dir,
}
TRACE("found best tablet $0", best_tablet->get_tablet_info().tablet_id);

StarRocksMetrics::instance()->cumulative_compaction_request_total.increment(1);

std::unique_ptr<MemTracker> mem_tracker =
std::make_unique<MemTracker>(MemTracker::COMPACTION, -1, "", _options.compaction_mem_tracker);
CumulativeCompaction cumulative_compaction(mem_tracker.get(), best_tablet);
Expand Down Expand Up @@ -941,8 +939,6 @@ Status StorageEngine::_perform_base_compaction(DataDir* data_dir, std::pair<int3
}
TRACE("found best tablet $0", best_tablet->get_tablet_info().tablet_id);

StarRocksMetrics::instance()->base_compaction_request_total.increment(1);

std::unique_ptr<MemTracker> mem_tracker =
std::make_unique<MemTracker>(MemTracker::COMPACTION, -1, "", _options.compaction_mem_tracker);
BaseCompaction base_compaction(mem_tracker.get(), best_tablet);
Expand Down Expand Up @@ -1003,13 +999,15 @@ Status StorageEngine::_perform_update_compaction(DataDir* data_dir) {
int64_t duration_ns = 0;
{
StarRocksMetrics::instance()->update_compaction_request_total.increment(1);
StarRocksMetrics::instance()->running_update_compaction_task_num.increment(1);
SCOPED_RAW_TIMER(&duration_ns);

std::unique_ptr<MemTracker> mem_tracker =
std::make_unique<MemTracker>(MemTracker::COMPACTION, -1, "", _options.compaction_mem_tracker);
res = best_tablet->updates()->compaction(mem_tracker.get());
}
StarRocksMetrics::instance()->update_compaction_duration_us.increment(duration_ns / 1000);
StarRocksMetrics::instance()->running_update_compaction_task_num.increment(-1);
if (!res.ok()) {
StarRocksMetrics::instance()->update_compaction_request_failed.increment(1);
LOG(WARNING) << "failed to perform update compaction. res=" << res.to_string()
Expand Down
13 changes: 13 additions & 0 deletions be/src/util/starrocks_metrics.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,8 @@ StarRocksMetrics::StarRocksMetrics() : _metrics(_s_registry_name) {
REGISTER_ENGINE_REQUEST_METRIC(base_compaction, failed, base_compaction_request_failed);
REGISTER_ENGINE_REQUEST_METRIC(cumulative_compaction, total, cumulative_compaction_request_total);
REGISTER_ENGINE_REQUEST_METRIC(cumulative_compaction, failed, cumulative_compaction_request_failed);
REGISTER_ENGINE_REQUEST_METRIC(update_compaction, total, update_compaction_request_total);
REGISTER_ENGINE_REQUEST_METRIC(update_compaction, failed, update_compaction_request_failed);

REGISTER_ENGINE_REQUEST_METRIC(publish, total, publish_task_request_total);
REGISTER_ENGINE_REQUEST_METRIC(publish, failed, publish_task_failed_total);
Expand Down Expand Up @@ -196,6 +198,17 @@ StarRocksMetrics::StarRocksMetrics() : _metrics(_s_registry_name) {
REGISTER_STARROCKS_METRIC(tablet_cumulative_max_compaction_score);
REGISTER_STARROCKS_METRIC(tablet_base_max_compaction_score);
REGISTER_STARROCKS_METRIC(tablet_update_max_compaction_score);
REGISTER_STARROCKS_METRIC(wait_cumulative_compaction_task_num);
REGISTER_STARROCKS_METRIC(wait_base_compaction_task_num);
REGISTER_STARROCKS_METRIC(running_cumulative_compaction_task_num);
REGISTER_STARROCKS_METRIC(running_base_compaction_task_num);
REGISTER_STARROCKS_METRIC(running_update_compaction_task_num);
REGISTER_STARROCKS_METRIC(cumulative_compaction_task_cost_time_ms);
REGISTER_STARROCKS_METRIC(base_compaction_task_cost_time_ms);
REGISTER_STARROCKS_METRIC(update_compaction_task_cost_time_ns);
REGISTER_STARROCKS_METRIC(base_compaction_task_byte_per_second);
REGISTER_STARROCKS_METRIC(cumulative_compaction_task_byte_per_second);
REGISTER_STARROCKS_METRIC(update_compaction_task_byte_per_second);

REGISTER_STARROCKS_METRIC(max_tablet_rowset_num);

Expand Down
Loading
Loading