| // 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 "olap/task/engine_clone_task.h" |
| |
| #include <absl/strings/str_split.h> |
| #include <curl/curl.h> |
| #include <fcntl.h> |
| #include <fmt/format.h> |
| #include <gen_cpp/AgentService_types.h> |
| #include <gen_cpp/BackendService.h> |
| #include <gen_cpp/HeartbeatService_types.h> |
| #include <gen_cpp/MasterService_types.h> |
| #include <gen_cpp/Status_types.h> |
| #include <gen_cpp/Types_constants.h> |
| #include <sys/stat.h> |
| |
| #include <filesystem> |
| #include <memory> |
| #include <mutex> |
| #include <ostream> |
| #include <set> |
| #include <shared_mutex> |
| #include <system_error> |
| #include <unordered_map> |
| #include <unordered_set> |
| #include <utility> |
| #include <vector> |
| |
| #include "common/config.h" |
| #include "common/logging.h" |
| #include "http/http_client.h" |
| #include "http/utils.h" |
| #include "io/fs/file_system.h" |
| #include "io/fs/local_file_system.h" |
| #include "io/fs/path.h" |
| #include "olap/data_dir.h" |
| #include "olap/olap_common.h" |
| #include "olap/olap_define.h" |
| #include "olap/pb_helper.h" |
| #include "olap/rowset/rowset.h" |
| #include "olap/snapshot_manager.h" |
| #include "olap/storage_engine.h" |
| #include "olap/tablet.h" |
| #include "olap/tablet_manager.h" |
| #include "olap/tablet_meta.h" |
| #include "runtime/client_cache.h" |
| #include "runtime/memory/mem_tracker_limiter.h" |
| #include "runtime/thread_context.h" |
| #include "util/debug_points.h" |
| #include "util/defer_op.h" |
| #include "util/network_util.h" |
| #include "util/security.h" |
| #include "util/stopwatch.hpp" |
| #include "util/thrift_rpc_helper.h" |
| #include "util/trace.h" |
| |
| using std::set; |
| using std::stringstream; |
| |
| namespace doris { |
| using namespace ErrorCode; |
| |
| namespace { |
| /// if binlog file exist, then check if binlog file md5sum equal |
| /// if equal, then skip link file |
| /// if not equal, then return error |
| /// return value: if binlog file not exist, then return to binlog file path |
| Result<std::string> check_dest_binlog_valid(const std::string& tablet_dir, |
| const std::string& clone_dir, |
| const std::string& clone_file, bool* skip_link_file) { |
| std::string from, to; |
| std::string new_clone_file = clone_file; |
| if (clone_file.ends_with(".binlog")) { |
| // change clone_file suffix from .binlog to .dat |
| new_clone_file.replace(clone_file.size() - 7, 7, ".dat"); |
| } else if (clone_file.ends_with(".binlog-index")) { |
| // change clone_file suffix from .binlog-index to .idx |
| new_clone_file.replace(clone_file.size() - 13, 13, ".idx"); |
| } |
| from = fmt::format("{}/{}", clone_dir, clone_file); |
| to = fmt::format("{}/_binlog/{}", tablet_dir, new_clone_file); |
| |
| // check to to file exist |
| bool exists = true; |
| auto status = io::global_local_filesystem()->exists(to, &exists); |
| if (!status.ok()) { |
| return ResultError(std::move(status)); |
| } |
| |
| if (!exists) { |
| return to; |
| } |
| |
| LOG(WARNING) << "binlog file already exist. " |
| << "tablet_dir=" << tablet_dir << ", clone_file=" << from << ", to=" << to; |
| |
| std::string clone_file_md5sum; |
| status = io::global_local_filesystem()->md5sum(from, &clone_file_md5sum); |
| if (!status.ok()) { |
| return ResultError(std::move(status)); |
| } |
| std::string to_file_md5sum; |
| status = io::global_local_filesystem()->md5sum(to, &to_file_md5sum); |
| if (!status.ok()) { |
| return ResultError(std::move(status)); |
| } |
| |
| if (clone_file_md5sum == to_file_md5sum) { |
| // if md5sum equal, then skip link file |
| *skip_link_file = true; |
| return to; |
| } else { |
| auto err_msg = fmt::format( |
| "binlog file already exist, but md5sum not equal. " |
| "tablet_dir={}, clone_file={}", |
| tablet_dir, clone_file); |
| LOG(WARNING) << err_msg; |
| return ResultError(Status::InternalError(std::move(err_msg))); |
| } |
| } |
| } // namespace |
| |
| #define RETURN_IF_ERROR_(status, stmt) \ |
| do { \ |
| status = (stmt); \ |
| if (UNLIKELY(!status.ok())) { \ |
| return status; \ |
| } \ |
| } while (false) |
| |
| EngineCloneTask::EngineCloneTask(StorageEngine& engine, const TCloneReq& clone_req, |
| const ClusterInfo* cluster_info, int64_t signature, |
| std::vector<TTabletInfo>* tablet_infos) |
| : _engine(engine), |
| _clone_req(clone_req), |
| _tablet_infos(tablet_infos), |
| _signature(signature), |
| _cluster_info(cluster_info) { |
| _mem_tracker = MemTrackerLimiter::create_shared( |
| MemTrackerLimiter::Type::OTHER, |
| "EngineCloneTask#tabletId=" + std::to_string(_clone_req.tablet_id)); |
| } |
| |
| Status EngineCloneTask::execute() { |
| // register the tablet to avoid it is deleted by gc thread during clone process |
| Status st = _do_clone(); |
| _engine.tablet_manager()->update_partitions_visible_version( |
| {{_clone_req.partition_id, _clone_req.version}}); |
| return st; |
| } |
| |
| Status EngineCloneTask::_do_clone() { |
| DBUG_EXECUTE_IF("EngineCloneTask.wait_clone", { |
| auto duration = std::chrono::milliseconds(dp->param("duration", 10 * 1000)); |
| std::this_thread::sleep_for(duration); |
| }); |
| |
| DBUG_EXECUTE_IF("EngineCloneTask.failed_clone", { |
| LOG_WARNING("EngineCloneTask.failed_clone") |
| .tag("tablet_id", _clone_req.tablet_id) |
| .tag("replica_id", _clone_req.replica_id) |
| .tag("version", _clone_req.version); |
| return Status::InternalError( |
| "in debug point, EngineCloneTask.failed_clone tablet={}, replica={}, version={}", |
| _clone_req.tablet_id, _clone_req.replica_id, _clone_req.version); |
| }); |
| Status status = Status::OK(); |
| std::string src_file_path; |
| TBackend src_host; |
| RETURN_IF_ERROR( |
| _engine.tablet_manager()->register_transition_tablet(_clone_req.tablet_id, "clone")); |
| Defer defer {[&]() { |
| _engine.tablet_manager()->unregister_transition_tablet(_clone_req.tablet_id, "clone"); |
| }}; |
| |
| // Check local tablet exist or not |
| TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(_clone_req.tablet_id); |
| |
| // The status of a tablet is not ready, indicating that it is a residual tablet after a schema |
| // change failure. Clone a new tablet from remote be to overwrite it. This situation basically only |
| // occurs when the be_rebalancer_fuzzy_test configuration is enabled. |
| if (tablet && tablet->tablet_state() == TABLET_NOTREADY) { |
| LOG(WARNING) << "tablet state is not ready when clone, need to drop old tablet, tablet_id=" |
| << tablet->tablet_id(); |
| RETURN_IF_ERROR(_engine.tablet_manager()->drop_tablet(tablet->tablet_id(), |
| tablet->replica_id(), false)); |
| tablet.reset(); |
| } |
| _is_new_tablet = tablet == nullptr; |
| // try to incremental clone |
| Versions missed_versions; |
| // try to repair a tablet with missing version |
| if (tablet != nullptr) { |
| std::shared_lock migration_rlock(tablet->get_migration_lock(), std::try_to_lock); |
| if (!migration_rlock.owns_lock()) { |
| return Status::Error<TRY_LOCK_FAILED>( |
| "EngineCloneTask::_do_clone meet try lock failed"); |
| } |
| if (tablet->replica_id() < _clone_req.replica_id) { |
| // `tablet` may be a dropped replica in FE, e.g: |
| // BE1 migrates replica of tablet_1 to BE2, but before BE1 drop this replica, another new replica of tablet_1 is migrated to BE1. |
| // Clone can still continue in this case. But to keep `replica_id` consitent with FE, MUST reset `replica_id` with request `replica_id`. |
| tablet->tablet_meta()->set_replica_id(_clone_req.replica_id); |
| } |
| |
| // get download path |
| auto local_data_path = fmt::format("{}/{}", tablet->tablet_path(), CLONE_PREFIX); |
| bool allow_incremental_clone = false; |
| |
| int64_t specified_version = _clone_req.version; |
| if (tablet->enable_unique_key_merge_on_write()) { |
| int64_t min_pending_ver = _engine.get_pending_publish_min_version(tablet->tablet_id()); |
| if (min_pending_ver - 1 < specified_version) { |
| LOG(INFO) << "use min pending publish version for clone, min_pending_ver: " |
| << min_pending_ver << " visible_version: " << _clone_req.version; |
| specified_version = min_pending_ver - 1; |
| } |
| } |
| |
| missed_versions = tablet->get_missed_versions(specified_version); |
| |
| // if missed version size is 0, then it is useless to clone from remote be, it means local data is |
| // completed. Or remote be will just return header not the rowset files. clone will failed. |
| if (missed_versions.empty()) { |
| LOG(INFO) << "missed version size = 0, skip clone and return success. tablet_id=" |
| << _clone_req.tablet_id << " replica_id=" << _clone_req.replica_id; |
| RETURN_IF_ERROR(_set_tablet_info()); |
| return Status::OK(); |
| } |
| |
| LOG(INFO) << "clone to existed tablet. missed_versions_size=" << missed_versions.size() |
| << ", allow_incremental_clone=" << allow_incremental_clone |
| << ", signature=" << _signature << ", tablet_id=" << _clone_req.tablet_id |
| << ", visible_version=" << _clone_req.version |
| << ", replica_id=" << _clone_req.replica_id; |
| |
| // try to download missing version from src backend. |
| // if tablet on src backend does not contains missing version, it will download all versions, |
| // and set allow_incremental_clone to false |
| RETURN_IF_ERROR(_make_and_download_snapshots(*(tablet->data_dir()), local_data_path, |
| &src_host, &src_file_path, missed_versions, |
| &allow_incremental_clone)); |
| RETURN_IF_ERROR(_finish_clone(tablet.get(), local_data_path, specified_version, |
| allow_incremental_clone)); |
| } else { |
| LOG(INFO) << "clone tablet not exist, begin clone a new tablet from remote be. " |
| << "signature=" << _signature << ", tablet_id=" << _clone_req.tablet_id |
| << ", visible_version=" << _clone_req.version |
| << ", req replica=" << _clone_req.replica_id; |
| // create a new tablet in this be |
| // Get local disk from olap |
| std::string local_shard_root_path; |
| DataDir* store = nullptr; |
| RETURN_IF_ERROR(_engine.obtain_shard_path(_clone_req.storage_medium, |
| _clone_req.dest_path_hash, &local_shard_root_path, |
| &store, _clone_req.partition_id)); |
| auto tablet_dir = fmt::format("{}/{}/{}", local_shard_root_path, _clone_req.tablet_id, |
| _clone_req.schema_hash); |
| |
| Defer remove_useless_dir {[&] { |
| if (status.ok()) { |
| return; |
| } |
| LOG(INFO) << "clone failed. want to delete local dir: " << tablet_dir |
| << ". signature: " << _signature; |
| WARN_IF_ERROR(io::global_local_filesystem()->delete_directory(tablet_dir), |
| "failed to delete useless clone dir "); |
| WARN_IF_ERROR(DataDir::delete_tablet_parent_path_if_empty(tablet_dir), |
| "failed to delete parent dir"); |
| }}; |
| |
| bool exists = true; |
| Status exists_st = io::global_local_filesystem()->exists(tablet_dir, &exists); |
| if (!exists_st) { |
| LOG(WARNING) << "cant get path=" << tablet_dir << " state, st=" << exists_st; |
| return exists_st; |
| } |
| if (exists) { |
| LOG(WARNING) << "before clone dest path=" << tablet_dir << " exist, remove it first"; |
| RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(tablet_dir)); |
| } |
| |
| bool allow_incremental_clone = false; |
| RETURN_IF_ERROR_(status, |
| _make_and_download_snapshots(*store, tablet_dir, &src_host, &src_file_path, |
| missed_versions, &allow_incremental_clone)); |
| |
| LOG(INFO) << "clone copy done. src_host: " << src_host.host |
| << " src_file_path: " << src_file_path; |
| auto tablet_manager = _engine.tablet_manager(); |
| RETURN_IF_ERROR_(status, tablet_manager->load_tablet_from_dir(store, _clone_req.tablet_id, |
| _clone_req.schema_hash, |
| tablet_dir, false)); |
| auto nested_tablet = tablet_manager->get_tablet(_clone_req.tablet_id); |
| if (!nested_tablet) { |
| status = Status::NotFound("tablet not found, tablet_id={}", _clone_req.tablet_id); |
| return status; |
| } |
| // MUST reset `replica_id` to request `replica_id` to keep consistent with FE |
| nested_tablet->tablet_meta()->set_replica_id(_clone_req.replica_id); |
| // clone success, delete .hdr file because tablet meta is stored in rocksdb |
| std::string header_path = |
| TabletMeta::construct_header_file_path(tablet_dir, _clone_req.tablet_id); |
| RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(header_path)); |
| } |
| |
| return _set_tablet_info(); |
| } |
| |
| Status EngineCloneTask::_set_tablet_info() { |
| // Get clone tablet info |
| TTabletInfo tablet_info; |
| tablet_info.__set_tablet_id(_clone_req.tablet_id); |
| tablet_info.__set_replica_id(_clone_req.replica_id); |
| tablet_info.__set_schema_hash(_clone_req.schema_hash); |
| RETURN_IF_ERROR(_engine.tablet_manager()->report_tablet_info(&tablet_info)); |
| if (_clone_req.__isset.version && tablet_info.version < _clone_req.version) { |
| // if it is a new tablet and clone failed, then remove the tablet |
| // if it is incremental clone, then must not drop the tablet |
| if (_is_new_tablet) { |
| // we need to check if this cloned table's version is what we expect. |
| // if not, maybe this is a stale remaining table which is waiting for drop. |
| // we drop it. |
| LOG(WARNING) << "begin to drop the stale tablet. tablet_id:" << _clone_req.tablet_id |
| << ", replica_id:" << _clone_req.replica_id |
| << ", schema_hash:" << _clone_req.schema_hash |
| << ", signature:" << _signature << ", version:" << tablet_info.version |
| << ", expected_version: " << _clone_req.version; |
| WARN_IF_ERROR(_engine.tablet_manager()->drop_tablet(_clone_req.tablet_id, |
| _clone_req.replica_id, false), |
| "drop stale cloned table failed"); |
| } |
| return Status::InternalError("unexpected version. tablet version: {}, expected version: {}", |
| tablet_info.version, _clone_req.version); |
| } |
| LOG(INFO) << "clone get tablet info success. tablet_id:" << _clone_req.tablet_id |
| << ", schema_hash:" << _clone_req.schema_hash << ", signature:" << _signature |
| << ", replica id:" << _clone_req.replica_id << ", version:" << tablet_info.version; |
| _tablet_infos->push_back(tablet_info); |
| return Status::OK(); |
| } |
| |
| /// This method will do following things: |
| /// 1. Make snapshots on source BE. |
| /// 2. Download all snapshots to CLONE dir. |
| /// 3. Convert rowset ids of downloaded snapshots(would also change the replica id). |
| /// 4. Release the snapshots on source BE. |
| Status EngineCloneTask::_make_and_download_snapshots(DataDir& data_dir, |
| const std::string& local_data_path, |
| TBackend* src_host, std::string* snapshot_path, |
| const std::vector<Version>& missed_versions, |
| bool* allow_incremental_clone) { |
| Status status; |
| |
| const auto& token = _cluster_info->token; |
| |
| int timeout_s = 0; |
| if (_clone_req.__isset.timeout_s) { |
| timeout_s = _clone_req.timeout_s; |
| } |
| |
| for (auto&& src : _clone_req.src_backends) { |
| // Make snapshot in remote olap engine |
| *src_host = src; |
| // make snapshot |
| status = _make_snapshot(src.host, src.be_port, _clone_req.tablet_id, _clone_req.schema_hash, |
| timeout_s, missed_versions, snapshot_path, allow_incremental_clone); |
| if (!status.ok()) [[unlikely]] { |
| LOG_WARNING("failed to make snapshot in remote BE") |
| .tag("host", src.host) |
| .tag("port", src.be_port) |
| .tag("tablet", _clone_req.tablet_id) |
| .tag("signature", _signature) |
| .tag("missed_versions", missed_versions) |
| .error(status); |
| continue; // Try another BE |
| } |
| LOG_INFO("successfully make snapshot in remote BE") |
| .tag("host", src.host) |
| .tag("port", src.be_port) |
| .tag("tablet", _clone_req.tablet_id) |
| .tag("snapshot_path", *snapshot_path) |
| .tag("signature", _signature) |
| .tag("missed_versions", missed_versions); |
| Defer defer {[host = src.host, port = src.be_port, &snapshot_path = *snapshot_path, this] { |
| // TODO(plat1ko): Async release snapshot |
| auto st = _release_snapshot(host, port, snapshot_path); |
| if (!st.ok()) [[unlikely]] { |
| LOG_WARNING("failed to release snapshot in remote BE") |
| .tag("host", host) |
| .tag("port", port) |
| .tag("snapshot_path", snapshot_path) |
| .error(st); |
| } |
| }}; |
| |
| std::string remote_dir; |
| { |
| std::stringstream ss; |
| if (snapshot_path->back() == '/') { |
| ss << *snapshot_path << _clone_req.tablet_id << "/" << _clone_req.schema_hash |
| << "/"; |
| } else { |
| ss << *snapshot_path << "/" << _clone_req.tablet_id << "/" << _clone_req.schema_hash |
| << "/"; |
| } |
| remote_dir = ss.str(); |
| } |
| |
| std::string address = get_host_port(src.host, src.http_port); |
| if (config::enable_batch_download && is_support_batch_download(address).ok()) { |
| // download files via batch api. |
| LOG_INFO("remote BE supports batch download, use batch file download") |
| .tag("address", address) |
| .tag("remote_dir", remote_dir); |
| status = _batch_download_files(&data_dir, address, remote_dir, local_data_path); |
| if (!status.ok()) [[unlikely]] { |
| LOG_WARNING("failed to download snapshot from remote BE in batch") |
| .tag("address", address) |
| .tag("remote_dir", remote_dir) |
| .error(status); |
| continue; // Try another BE |
| } |
| } else { |
| if (config::enable_batch_download) { |
| LOG_INFO("remote BE does not support batch download, use single file download") |
| .tag("address", address) |
| .tag("remote_dir", remote_dir); |
| } else { |
| LOG_INFO("batch download is disabled, use single file download") |
| .tag("address", address) |
| .tag("remote_dir", remote_dir); |
| } |
| |
| std::string remote_url_prefix; |
| { |
| std::stringstream ss; |
| ss << "http://" << address << HTTP_REQUEST_PREFIX << HTTP_REQUEST_TOKEN_PARAM |
| << token << HTTP_REQUEST_FILE_PARAM << remote_dir; |
| remote_url_prefix = ss.str(); |
| } |
| |
| status = _download_files(&data_dir, remote_url_prefix, local_data_path); |
| if (!status.ok()) [[unlikely]] { |
| LOG_WARNING("failed to download snapshot from remote BE") |
| .tag("url", mask_token(remote_url_prefix)) |
| .error(status); |
| continue; // Try another BE |
| } |
| } |
| |
| // No need to try again with another BE |
| _pending_rs_guards = DORIS_TRY(_engine.snapshot_mgr()->convert_rowset_ids( |
| local_data_path, _clone_req.tablet_id, _clone_req.replica_id, _clone_req.table_id, |
| _clone_req.partition_id, _clone_req.schema_hash)); |
| break; |
| } // clone copy from one backend |
| return status; |
| } |
| |
| Status EngineCloneTask::_make_snapshot(const std::string& ip, int port, TTableId tablet_id, |
| TSchemaHash schema_hash, int timeout_s, |
| const std::vector<Version>& missed_versions, |
| std::string* snapshot_path, bool* allow_incremental_clone) { |
| TSnapshotRequest request; |
| request.__set_tablet_id(tablet_id); |
| request.__set_schema_hash(schema_hash); |
| request.__set_preferred_snapshot_version(g_Types_constants.TPREFER_SNAPSHOT_REQ_VERSION); |
| request.__set_version(_clone_req.version); |
| request.__set_is_copy_binlog(true); |
| // TODO: missing version composed of singleton delta. |
| // if not, this place should be rewrote. |
| // we make every TSnapshotRequest sent from be with __isset.missing_version = true |
| // then if one be received one req with __isset.missing_version = false it means |
| // this req is sent from FE(FE would never set this field) |
| request.__isset.missing_version = true; |
| for (auto& version : missed_versions) { |
| request.missing_version.push_back(version.first); |
| } |
| if (timeout_s > 0) { |
| request.__set_timeout(timeout_s); |
| } |
| |
| TAgentResult result; |
| RETURN_IF_ERROR(ThriftRpcHelper::rpc<BackendServiceClient>( |
| ip, port, [&request, &result](BackendServiceConnection& client) { |
| client->make_snapshot(result, request); |
| })); |
| if (result.status.status_code != TStatusCode::OK) { |
| return Status::create(result.status); |
| } |
| |
| if (!result.__isset.snapshot_path) { |
| return Status::InternalError("success snapshot request without snapshot path"); |
| } |
| *snapshot_path = result.snapshot_path; |
| if (snapshot_path->at(snapshot_path->length() - 1) != '/') { |
| snapshot_path->append("/"); |
| } |
| |
| if (result.__isset.allow_incremental_clone) { |
| // During upgrading, some BE nodes still be installed an old previous old. |
| // which incremental clone is not ready in those nodes. |
| // should add a symbol to indicate it. |
| *allow_incremental_clone = result.allow_incremental_clone; |
| } |
| return Status::OK(); |
| } |
| |
| Status EngineCloneTask::_release_snapshot(const std::string& ip, int port, |
| const std::string& snapshot_path) { |
| TAgentResult result; |
| RETURN_IF_ERROR(ThriftRpcHelper::rpc<BackendServiceClient>( |
| ip, port, [&snapshot_path, &result](BackendServiceConnection& client) { |
| client->release_snapshot(result, snapshot_path); |
| })); |
| return Status::create(result.status); |
| } |
| |
| Status EngineCloneTask::_download_files(DataDir* data_dir, const std::string& remote_url_prefix, |
| const std::string& local_path) { |
| // Check local path exist, if exist, remove it, then create the dir |
| // local_file_full_path = tabletid/clone， for a specific tablet, there should be only one folder |
| // if this folder exists, then should remove it |
| // for example, BE clone from BE 1 to download file 1 with version (2,2), but clone from BE 1 failed |
| // then it will try to clone from BE 2, but it will find the file 1 already exist, but file 1 with same |
| // name may have different versions. |
| RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(local_path)); |
| RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(local_path)); |
| |
| // Get remote dir file list |
| std::string file_list_str; |
| auto list_files_cb = [&remote_url_prefix, &file_list_str](HttpClient* client) { |
| RETURN_IF_ERROR(client->init(remote_url_prefix)); |
| client->set_timeout_ms(LIST_REMOTE_FILE_TIMEOUT * 1000); |
| return client->execute(&file_list_str); |
| }; |
| RETURN_IF_ERROR(HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, list_files_cb)); |
| std::vector<std::string> file_name_list = |
| absl::StrSplit(file_list_str, "\n", absl::SkipWhitespace()); |
| |
| // If the header file is not exist, the table couldn't loaded by olap engine. |
| // Avoid of data is not complete, we copy the header file at last. |
| // The header file's name is end of .hdr. |
| for (int i = 0; i + 1 < file_name_list.size(); ++i) { |
| if (file_name_list[i].ends_with(".hdr")) { |
| std::swap(file_name_list[i], file_name_list[file_name_list.size() - 1]); |
| break; |
| } |
| } |
| |
| // Get copy from remote |
| uint64_t total_file_size = 0; |
| MonotonicStopWatch watch; |
| watch.start(); |
| for (auto& file_name : file_name_list) { |
| auto remote_file_url = remote_url_prefix + file_name; |
| |
| // get file length |
| uint64_t file_size = 0; |
| auto get_file_size_cb = [&remote_file_url, &file_size](HttpClient* client) { |
| RETURN_IF_ERROR(client->init(remote_file_url)); |
| client->set_timeout_ms(GET_LENGTH_TIMEOUT * 1000); |
| RETURN_IF_ERROR(client->head()); |
| RETURN_IF_ERROR(client->get_content_length(&file_size)); |
| return Status::OK(); |
| }; |
| RETURN_IF_ERROR( |
| HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, get_file_size_cb)); |
| // check disk capacity |
| if (data_dir->reach_capacity_limit(file_size)) { |
| return Status::Error<EXCEEDED_LIMIT>( |
| "reach the capacity limit of path {}, file_size={}", data_dir->path(), |
| file_size); |
| } |
| |
| total_file_size += file_size; |
| uint64_t estimate_timeout = file_size / config::download_low_speed_limit_kbps / 1024; |
| if (estimate_timeout < config::download_low_speed_time) { |
| estimate_timeout = config::download_low_speed_time; |
| } |
| |
| std::string local_file_path = local_path + "/" + file_name; |
| |
| LOG(INFO) << "clone begin to download file from: " << mask_token(remote_file_url) |
| << " to: " << local_file_path << ". size(B): " << file_size |
| << ", timeout(s): " << estimate_timeout; |
| |
| auto download_cb = [&remote_file_url, estimate_timeout, &local_file_path, |
| file_size](HttpClient* client) { |
| RETURN_IF_ERROR(client->init(remote_file_url)); |
| client->set_timeout_ms(estimate_timeout * 1000); |
| RETURN_IF_ERROR(client->download(local_file_path)); |
| |
| std::error_code ec; |
| // Check file length |
| uint64_t local_file_size = std::filesystem::file_size(local_file_path, ec); |
| if (ec) { |
| LOG(WARNING) << "download file error" << ec.message(); |
| return Status::IOError("can't retrive file_size of {}, due to {}", local_file_path, |
| ec.message()); |
| } |
| if (local_file_size != file_size) { |
| LOG(WARNING) << "download file length error" |
| << ", remote_path=" << mask_token(remote_file_url) |
| << ", file_size=" << file_size |
| << ", local_file_size=" << local_file_size; |
| return Status::InternalError("downloaded file size is not equal"); |
| } |
| return io::global_local_filesystem()->permission(local_file_path, |
| io::LocalFileSystem::PERMS_OWNER_RW); |
| }; |
| RETURN_IF_ERROR(HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1, download_cb)); |
| } // Clone files from remote backend |
| |
| uint64_t total_time_ms = watch.elapsed_time() / 1000 / 1000; |
| total_time_ms = total_time_ms > 0 ? total_time_ms : 0; |
| double copy_rate = 0.0; |
| if (total_time_ms > 0) { |
| copy_rate = total_file_size / ((double)total_time_ms) / 1000; |
| } |
| _copy_size = (int64_t)total_file_size; |
| _copy_time_ms = (int64_t)total_time_ms; |
| LOG(INFO) << "succeed to copy tablet " << _signature |
| << ", total files: " << file_name_list.size() |
| << ", total file size: " << total_file_size << " B, cost: " << total_time_ms << " ms" |
| << ", rate: " << copy_rate << " MB/s"; |
| return Status::OK(); |
| } |
| |
| Status EngineCloneTask::_batch_download_files(DataDir* data_dir, const std::string& address, |
| const std::string& remote_dir, |
| const std::string& local_dir) { |
| constexpr size_t BATCH_FILE_SIZE = 64 << 20; // 64MB |
| constexpr size_t BATCH_FILE_NUM = 64; |
| |
| // Check local path exist, if exist, remove it, then create the dir |
| // local_file_full_path = tabletid/clone， for a specific tablet, there should be only one folder |
| // if this folder exists, then should remove it |
| // for example, BE clone from BE 1 to download file 1 with version (2,2), but clone from BE 1 failed |
| // then it will try to clone from BE 2, but it will find the file 1 already exist, but file 1 with same |
| // name may have different versions. |
| RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(local_dir)); |
| RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(local_dir)); |
| |
| const std::string& token = _cluster_info->token; |
| std::vector<std::pair<std::string, size_t>> file_info_list; |
| RETURN_IF_ERROR(list_remote_files_v2(address, token, remote_dir, &file_info_list)); |
| |
| // If the header file is not exist, the table couldn't loaded by olap engine. |
| // Avoid of data is not complete, we copy the header file at last. |
| // The header file's name is end of .hdr. |
| for (int i = 0; i + 1 < file_info_list.size(); ++i) { |
| if (file_info_list[i].first.ends_with(".hdr")) { |
| std::swap(file_info_list[i], file_info_list[file_info_list.size() - 1]); |
| break; |
| } |
| } |
| |
| MonotonicStopWatch watch; |
| watch.start(); |
| |
| size_t total_file_size = 0; |
| size_t total_files = file_info_list.size(); |
| std::vector<std::pair<std::string, size_t>> batch_files; |
| for (size_t i = 0; i < total_files;) { |
| size_t batch_file_size = 0; |
| for (size_t j = i; j < total_files; j++) { |
| // Split batchs by file number and file size, |
| if (BATCH_FILE_NUM <= batch_files.size() || BATCH_FILE_SIZE <= batch_file_size || |
| // ... or separate the last .hdr file into a single batch. |
| (j + 1 == total_files && !batch_files.empty())) { |
| break; |
| } |
| batch_files.push_back(file_info_list[j]); |
| batch_file_size += file_info_list[j].second; |
| } |
| |
| // check disk capacity |
| if (data_dir->reach_capacity_limit(batch_file_size)) { |
| return Status::Error<EXCEEDED_LIMIT>( |
| "reach the capacity limit of path {}, file_size={}", data_dir->path(), |
| batch_file_size); |
| } |
| |
| RETURN_IF_ERROR(download_files_v2(address, token, remote_dir, local_dir, batch_files)); |
| |
| total_file_size += batch_file_size; |
| i += batch_files.size(); |
| batch_files.clear(); |
| } |
| |
| uint64_t total_time_ms = watch.elapsed_time() / 1000 / 1000; |
| total_time_ms = total_time_ms > 0 ? total_time_ms : 0; |
| double copy_rate = 0.0; |
| if (total_time_ms > 0) { |
| copy_rate = total_file_size / ((double)total_time_ms) / 1000; |
| } |
| _copy_size = (int64_t)total_file_size; |
| _copy_time_ms = (int64_t)total_time_ms; |
| LOG(INFO) << "succeed to copy tablet " << _signature |
| << ", total files: " << file_info_list.size() |
| << ", total file size: " << total_file_size << " B, cost: " << total_time_ms << " ms" |
| << ", rate: " << copy_rate << " MB/s"; |
| |
| return Status::OK(); |
| } |
| |
| /// This method will only be called if tablet already exist in this BE when doing clone. |
| /// This method will do the following things: |
| /// 1. Link all files from CLONE dir to tablet dir if file does not exist in tablet dir |
| /// 2. Call _finish_xx_clone() to revise the tablet meta. |
| Status EngineCloneTask::_finish_clone(Tablet* tablet, const std::string& clone_dir, int64_t version, |
| bool is_incremental_clone) { |
| Defer remove_clone_dir {[&]() { |
| std::error_code ec; |
| std::filesystem::remove_all(clone_dir, ec); |
| if (ec) { |
| LOG(WARNING) << "failed to remove=" << clone_dir << " msg=" << ec.message(); |
| } |
| }}; |
| |
| // check clone dir existed |
| bool exists = true; |
| RETURN_IF_ERROR(io::global_local_filesystem()->exists(clone_dir, &exists)); |
| if (!exists) { |
| return Status::InternalError("clone dir not existed. clone_dir={}", clone_dir); |
| } |
| |
| // Load src header. |
| // The tablet meta info is downloaded from source BE as .hdr file. |
| // So we load it and generate cloned_tablet_meta. |
| auto cloned_tablet_meta_file = fmt::format("{}/{}.hdr", clone_dir, tablet->tablet_id()); |
| auto cloned_tablet_meta = std::make_shared<TabletMeta>(); |
| RETURN_IF_ERROR(cloned_tablet_meta->create_from_file(cloned_tablet_meta_file)); |
| |
| // remove the cloned meta file |
| RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(cloned_tablet_meta_file)); |
| |
| // remove rowset binlog metas |
| const auto& tablet_dir = tablet->tablet_path(); |
| auto binlog_metas_file = fmt::format("{}/rowset_binlog_metas.pb", clone_dir); |
| bool binlog_metas_file_exists = false; |
| auto file_exists_status = |
| io::global_local_filesystem()->exists(binlog_metas_file, &binlog_metas_file_exists); |
| if (!file_exists_status.ok()) { |
| return file_exists_status; |
| } |
| bool contain_binlog = false; |
| RowsetBinlogMetasPB rowset_binlog_metas_pb; |
| if (binlog_metas_file_exists) { |
| std::error_code ec; |
| auto binlog_meta_filesize = std::filesystem::file_size(binlog_metas_file, ec); |
| if (ec) { |
| LOG(WARNING) << "get file size error" << ec.message(); |
| return Status::IOError("can't retrive file_size of {}, due to {}", binlog_metas_file, |
| ec.message()); |
| } |
| if (binlog_meta_filesize > 0) { |
| contain_binlog = true; |
| RETURN_IF_ERROR(read_pb(binlog_metas_file, &rowset_binlog_metas_pb)); |
| } |
| RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(binlog_metas_file)); |
| } |
| if (contain_binlog) { |
| auto binlog_dir = fmt::format("{}/_binlog", tablet_dir); |
| RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(binlog_dir)); |
| } |
| |
| // check all files in /clone and /tablet |
| std::vector<io::FileInfo> clone_files; |
| RETURN_IF_ERROR(io::global_local_filesystem()->list(clone_dir, true, &clone_files, &exists)); |
| std::unordered_set<std::string> clone_file_names; |
| for (auto& file : clone_files) { |
| clone_file_names.insert(file.file_name); |
| } |
| |
| std::vector<io::FileInfo> local_files; |
| RETURN_IF_ERROR(io::global_local_filesystem()->list(tablet_dir, true, &local_files, &exists)); |
| std::unordered_set<std::string> local_file_names; |
| for (auto& file : local_files) { |
| local_file_names.insert(file.file_name); |
| } |
| |
| Status status; |
| std::vector<std::string> linked_success_files; |
| Defer remove_linked_files {[&]() { // clear linked files if errors happen |
| if (!status.ok()) { |
| std::vector<io::Path> paths; |
| for (auto& file : linked_success_files) { |
| paths.emplace_back(file); |
| } |
| static_cast<void>(io::global_local_filesystem()->batch_delete(paths)); |
| } |
| }}; |
| /// Traverse all downloaded clone files in CLONE dir. |
| /// If it does not exist in local tablet dir, link the file to local tablet dir |
| /// And save all linked files in linked_success_files. |
| for (const std::string& clone_file : clone_file_names) { |
| if (local_file_names.find(clone_file) != local_file_names.end()) { |
| VLOG_NOTICE << "find same file when clone, skip it. " |
| << "tablet=" << tablet->tablet_id() << ", clone_file=" << clone_file; |
| continue; |
| } |
| |
| /// if binlog exist in clone dir and md5sum equal, then skip link file |
| bool skip_link_file = false; |
| std::string to; |
| if (clone_file.ends_with(".binlog") || clone_file.ends_with(".binlog-index")) { |
| if (!contain_binlog) { |
| LOG(WARNING) << "clone binlog file, but not contain binlog metas. " |
| << "tablet=" << tablet->tablet_id() << ", clone_file=" << clone_file; |
| break; |
| } |
| |
| if (auto&& result = |
| check_dest_binlog_valid(tablet_dir, clone_dir, clone_file, &skip_link_file); |
| result) { |
| to = std::move(result.value()); |
| } else { |
| status = std::move(result.error()); |
| return status; |
| } |
| } else { |
| to = fmt::format("{}/{}", tablet_dir, clone_file); |
| } |
| |
| if (!skip_link_file) { |
| auto from = fmt::format("{}/{}", clone_dir, clone_file); |
| status = io::global_local_filesystem()->link_file(from, to); |
| if (!status.ok()) { |
| return status; |
| } |
| linked_success_files.emplace_back(std::move(to)); |
| } |
| } |
| if (contain_binlog) { |
| status = tablet->ingest_binlog_metas(&rowset_binlog_metas_pb); |
| if (!status.ok()) { |
| return status; |
| } |
| } |
| |
| // clone and compaction operation should be performed sequentially |
| std::lock_guard base_compaction_lock(tablet->get_base_compaction_lock()); |
| std::lock_guard cumulative_compaction_lock(tablet->get_cumulative_compaction_lock()); |
| std::lock_guard cold_compaction_lock(tablet->get_cold_compaction_lock()); |
| std::lock_guard build_inverted_index_lock(tablet->get_build_inverted_index_lock()); |
| std::lock_guard<std::mutex> push_lock(tablet->get_push_lock()); |
| std::lock_guard<std::mutex> rwlock(tablet->get_rowset_update_lock()); |
| std::lock_guard<std::shared_mutex> wrlock(tablet->get_header_lock()); |
| SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD); |
| if (is_incremental_clone) { |
| status = _finish_incremental_clone(tablet, cloned_tablet_meta, version); |
| } else { |
| status = _finish_full_clone(tablet, cloned_tablet_meta); |
| } |
| |
| // if full clone success, need to update cumulative layer point |
| if (!is_incremental_clone && status.ok()) { |
| tablet->set_cumulative_layer_point(Tablet::K_INVALID_CUMULATIVE_POINT); |
| } |
| |
| // clear clone dir |
| return status; |
| } |
| |
| /// This method will do: |
| /// 1. Get missing version from local tablet again and check if they exist in cloned tablet. |
| /// 2. Revise the local tablet meta to add all incremental cloned rowset's meta. |
| Status EngineCloneTask::_finish_incremental_clone(Tablet* tablet, |
| const TabletMetaSharedPtr& cloned_tablet_meta, |
| int64_t version) { |
| LOG(INFO) << "begin to finish incremental clone. tablet=" << tablet->tablet_id() |
| << ", visible_version=" << version |
| << ", cloned_tablet_replica_id=" << cloned_tablet_meta->replica_id(); |
| |
| /// Get missing versions again from local tablet. |
| /// We got it before outside the lock, so it has to be got again. |
| Versions missed_versions = tablet->get_missed_versions_unlocked(version); |
| VLOG_NOTICE << "get missed versions again when finish incremental clone. " |
| << "tablet=" << tablet->tablet_id() << ", clone version=" << version |
| << ", missed_versions_size=" << missed_versions.size(); |
| |
| // check missing versions exist in clone src |
| std::vector<RowsetSharedPtr> rowsets_to_clone; |
| for (Version nested_version : missed_versions) { |
| auto rs_meta = cloned_tablet_meta->acquire_rs_meta_by_version(nested_version); |
| if (rs_meta == nullptr) { |
| return Status::InternalError("missed version {} is not found in cloned tablet meta", |
| nested_version.to_string()); |
| } |
| RowsetSharedPtr rs; |
| RETURN_IF_ERROR(tablet->create_rowset(rs_meta, &rs)); |
| rowsets_to_clone.push_back(std::move(rs)); |
| } |
| |
| /// clone_data to tablet |
| /// For incremental clone, nothing will be deleted. |
| /// So versions_to_delete is empty. |
| return tablet->revise_tablet_meta(rowsets_to_clone, {}, true); |
| } |
| |
| /// This method will do: |
| /// 1. Compare the version of local tablet and cloned tablet to decide which version to keep |
| /// 2. Revise the local tablet meta |
| Status EngineCloneTask::_finish_full_clone(Tablet* tablet, |
| const TabletMetaSharedPtr& cloned_tablet_meta) { |
| Version cloned_max_version = cloned_tablet_meta->max_version(); |
| LOG(INFO) << "begin to finish full clone. tablet=" << tablet->tablet_id() |
| << ", cloned_max_version=" << cloned_max_version; |
| |
| // Compare the version of local tablet and cloned tablet. |
| // For example: |
| // clone version is 8 |
| // |
| // local tablet: [0-1] [2-5] [6-6] [7-7] [9-10] |
| // clone tablet: [0-1] [2-4] [5-6] [7-8] |
| // |
| // after compare, the version mark with "x" will be deleted |
| // |
| // local tablet: [0-1]x [2-5]x [6-6]x [7-7]x [9-10] |
| // clone tablet: [0-1] [2-4] [5-6] [7-8] |
| |
| std::vector<RowsetSharedPtr> to_delete; |
| std::vector<RowsetSharedPtr> to_add; |
| for (auto& [v, rs] : tablet->rowset_map()) { |
| // if local version cross src latest, clone failed |
| // if local version is : 0-0, 1-1, 2-10, 12-14, 15-15,16-16 |
| // cloned max version is 13-13, this clone is failed, because could not |
| // fill local data by using cloned data. |
| // It should not happen because if there is a hole, the following delta will not |
| // do compaction. |
| if (v.first <= cloned_max_version.second && v.second > cloned_max_version.second) { |
| return Status::InternalError( |
| "version cross src latest. cloned_max_version={}, local_version={}", |
| cloned_max_version.second, v.to_string()); |
| } |
| if (v.second <= cloned_max_version.second) { |
| to_delete.push_back(rs); |
| } else { |
| // cooldowned rowsets MUST be continuous, so rowsets whose version > missed version MUST be local rowset |
| DCHECK(rs->is_local()); |
| } |
| } |
| |
| to_add.reserve(cloned_tablet_meta->all_rs_metas().size()); |
| for (const auto& [_, rs_meta] : cloned_tablet_meta->all_rs_metas()) { |
| RowsetSharedPtr rs; |
| RETURN_IF_ERROR(tablet->create_rowset(rs_meta, &rs)); |
| to_add.push_back(std::move(rs)); |
| } |
| { |
| std::shared_lock cooldown_conf_rlock(tablet->get_cooldown_conf_lock()); |
| if (tablet->cooldown_conf_unlocked().cooldown_replica_id == tablet->replica_id()) { |
| // If this replica is cooldown replica, MUST generate a new `cooldown_meta_id` to avoid use `cooldown_meta_id` |
| // generated in old cooldown term which may lead to such situation: |
| // Replica A is cooldown replica, cooldown_meta_id=2, |
| // Replica B: cooldown_replica=A, cooldown_meta_id=1 |
| // Replica A: full clone Replica A, cooldown_meta_id=1, but remote cooldown_meta is still with cooldown_meta_id=2 |
| // After tablet report. FE finds all replicas' cooldowned data is consistent |
| // Replica A: confirm_unused_remote_files, delete some cooldowned data of cooldown_meta_id=2 |
| // Replica B: follow_cooldown_data, cooldown_meta_id=2, data lost |
| tablet->tablet_meta()->set_cooldown_meta_id(UniqueId::gen_uid()); |
| } else { |
| tablet->tablet_meta()->set_cooldown_meta_id(cloned_tablet_meta->cooldown_meta_id()); |
| } |
| } |
| if (tablet->enable_unique_key_merge_on_write()) { |
| tablet->tablet_meta()->delete_bitmap().merge(cloned_tablet_meta->delete_bitmap()); |
| } |
| return tablet->revise_tablet_meta(to_add, to_delete, false); |
| // TODO(plat1ko): write cooldown meta to remote if this replica is cooldown replica |
| } |
| } // namespace doris |
