| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| #include "cloud/cloud_cluster_info.h" |
| |
| #include <glog/logging.h> |
| |
| #include "cloud/cloud_meta_mgr.h" |
| #include "cloud/cloud_storage_engine.h" |
| #include "cloud/cloud_tablet.h" |
| #include "cloud/config.h" |
| #include "gen_cpp/cloud.pb.h" |
| #include "runtime/exec_env.h" |
| #include "util/time.h" |
| |
| namespace doris { |
| |
| CloudClusterInfo::~CloudClusterInfo() { |
| stop_bg_worker(); |
| } |
| |
| void CloudClusterInfo::start_bg_worker() { |
| bool expected = true; |
| if (!_bg_worker_stopped.compare_exchange_strong(expected, false)) { |
| // Already running |
| return; |
| } |
| |
| _bg_worker = std::thread(&CloudClusterInfo::_bg_worker_func, this); |
| LOG(INFO) << "CloudClusterInfo background worker started, " |
| << "refresh_interval=" << config::cluster_status_cache_refresh_interval_sec << "s"; |
| } |
| |
| void CloudClusterInfo::stop_bg_worker() { |
| bool expected = false; |
| if (!_bg_worker_stopped.compare_exchange_strong(expected, true)) { |
| // Already stopped |
| return; |
| } |
| |
| { |
| std::lock_guard lock(_bg_worker_mutex); |
| _bg_worker_cv.notify_all(); |
| } |
| |
| if (_bg_worker.joinable()) { |
| _bg_worker.join(); |
| } |
| |
| LOG(INFO) << "CloudClusterInfo background worker stopped"; |
| } |
| |
| void CloudClusterInfo::_bg_worker_func() { |
| LOG(INFO) << "CloudClusterInfo background worker thread running"; |
| |
| while (!_bg_worker_stopped.load()) { |
| _refresh_cluster_status(); |
| |
| std::unique_lock lock(_bg_worker_mutex); |
| _bg_worker_cv.wait_for( |
| lock, std::chrono::seconds(config::cluster_status_cache_refresh_interval_sec), |
| [this] { return _bg_worker_stopped.load(); }); |
| } |
| } |
| |
| void CloudClusterInfo::_refresh_cluster_status() { |
| if (!config::is_cloud_mode()) { |
| return; |
| } |
| auto* cloud_engine = |
| dynamic_cast<CloudStorageEngine*>(&ExecEnv::GetInstance()->storage_engine()); |
| if (!cloud_engine) { |
| return; |
| } |
| |
| std::unordered_map<std::string, std::pair<int32_t, int64_t>> cluster_status; |
| std::string resolved_cluster_id; |
| Status st = cloud_engine->meta_mgr().get_cluster_status( |
| &cluster_status, my_cluster_id().empty() ? &resolved_cluster_id : nullptr); |
| if (!st.ok()) { |
| LOG(WARNING) << "Failed to refresh cluster status: " << st; |
| return; |
| } |
| |
| // Update cache |
| { |
| std::unique_lock lock(_mutex); |
| _cluster_status_cache.clear(); |
| for (const auto& [cluster_id, status_pair] : cluster_status) { |
| _cluster_status_cache[cluster_id] = {status_pair.first, status_pair.second}; |
| } |
| } |
| |
| VLOG_DEBUG << "Refreshed cluster status cache, " << cluster_status.size() << " clusters"; |
| |
| // Set our own cluster_id if resolved from the response |
| if (my_cluster_id().empty() && !resolved_cluster_id.empty()) { |
| set_my_cluster_id(resolved_cluster_id); |
| LOG(INFO) << "Resolved my cluster_id: " << resolved_cluster_id; |
| } |
| } |
| |
| bool CloudClusterInfo::should_skip_compaction(CloudTablet* tablet) const { |
| if (!config::enable_compaction_rw_separation) { |
| return false; |
| } |
| |
| std::string last_active_cluster = tablet->last_active_cluster_id(); |
| std::string my_cluster = my_cluster_id(); |
| int64_t tablet_id = tablet->tablet_id(); |
| |
| // Case 1: No active cluster record, any cluster can compact |
| if (last_active_cluster.empty()) { |
| VLOG_DEBUG << "tablet " << tablet_id << " has no last_active_cluster record, " |
| << "my_cluster=" << my_cluster << ", allow compaction"; |
| return false; |
| } |
| |
| // Case 2: This is the active cluster, allow compaction |
| if (last_active_cluster == my_cluster) { |
| VLOG_DEBUG << "tablet " << tablet_id << " last_active_cluster=" << last_active_cluster |
| << " equals my_cluster=" << my_cluster << ", allow compaction"; |
| return false; |
| } |
| |
| // Case 3: Check if the last active cluster is available |
| ClusterStatusCache cache; |
| if (!get_cluster_status(last_active_cluster, &cache)) { |
| // Cluster not found in cache, might be deleted, allow takeover |
| LOG(INFO) << "compaction_rw_separation: tablet " << tablet_id |
| << " last_active_cluster=" << last_active_cluster |
| << " not found in cache (maybe deleted), my_cluster=" << my_cluster |
| << ", allow takeover"; |
| return false; |
| } |
| |
| // Force compaction if tablet has too many rowsets (>80% of max_tablet_version_num), |
| // even on read clusters, to prevent version count from growing unbounded |
| // when the write cluster can't keep up or has compaction disabled. |
| int64_t num_rowsets = tablet->fetch_add_approximate_num_rowsets(0); |
| auto threshold = static_cast<int64_t>(tablet->max_version_config() * |
| config::compaction_rw_separation_version_threshold_ratio); |
| if (num_rowsets > threshold) { |
| LOG(INFO) << "compaction_rw_separation: force compaction on tablet " << tablet_id |
| << ", num_rowsets=" << num_rowsets << " > threshold=" << threshold << " (80% of " |
| << tablet->max_version_config() << ")" |
| << ", my_cluster=" << my_cluster; |
| return false; |
| } |
| |
| auto status = static_cast<cloud::ClusterStatus>(cache.status); |
| int64_t status_mtime = cache.mtime_ms; |
| int64_t now = UnixMillis(); |
| int64_t elapsed = now - status_mtime; |
| int64_t timeout = config::compaction_cluster_takeover_timeout_ms; |
| |
| // Case 4: Original cluster is NORMAL (still active), cannot takeover |
| if (status == cloud::ClusterStatus::NORMAL) { |
| LOG_EVERY_N(INFO, 100) << "compaction_rw_separation: skip tablet " << tablet_id |
| << ", last_active_cluster=" << last_active_cluster |
| << " is NORMAL (active), my_cluster=" << my_cluster; |
| return true; |
| } |
| |
| // Case 5: Original cluster is unavailable (SUSPENDED/MANUAL_SHUTDOWN/deleted) |
| if (elapsed > timeout) { |
| // Takeover successful |
| LOG(INFO) << "compaction_rw_separation: takeover tablet " << tablet_id |
| << ", last_active_cluster=" << last_active_cluster |
| << " status=" << cloud::ClusterStatus_Name(status) |
| << " status_mtime=" << status_mtime << " elapsed=" << elapsed |
| << "ms > timeout=" << timeout << "ms" |
| << ", my_cluster=" << my_cluster; |
| return false; |
| } else { |
| // Timeout not reached yet, waiting |
| LOG_EVERY_N(INFO, 100) << "compaction_rw_separation: skip tablet " << tablet_id |
| << ", last_active_cluster=" << last_active_cluster |
| << " status=" << cloud::ClusterStatus_Name(status) |
| << " status_mtime=" << status_mtime << " elapsed=" << elapsed |
| << "ms <= timeout=" << timeout << "ms" |
| << ", my_cluster=" << my_cluster << ", waiting for takeover"; |
| return true; |
| } |
| } |
| |
| } // namespace doris |
