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

Squashed commit of the following: #2

Merged
merged 1 commit into from
Jun 20, 2022
Merged

Squashed commit of the following: #2

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
2 changes: 2 additions & 0 deletions be/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -534,6 +534,7 @@ set(KRB5_LIBS
set(AWS_LIBS
aws-sdk-s3
aws-sdk-core
aws-sdk-transfer
aws-checksums
aws-c-io
aws-c-event-stream
Expand Down Expand Up @@ -727,6 +728,7 @@ add_subdirectory(${SRC_DIR}/geo)
add_subdirectory(${SRC_DIR}/io)
add_subdirectory(${SRC_DIR}/gutil)
add_subdirectory(${SRC_DIR}/http)
add_subdirectory(${SRC_DIR}/io)
add_subdirectory(${SRC_DIR}/olap)
add_subdirectory(${SRC_DIR}/runtime)
add_subdirectory(${SRC_DIR}/service)
Expand Down
9 changes: 5 additions & 4 deletions be/src/agent/agent_server.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ AgentServer::AgentServer(ExecEnv* exec_env, const TMasterInfo& master_info)
: _exec_env(exec_env), _master_info(master_info), _topic_subscriber(new TopicSubscriber()) {
for (auto& path : exec_env->store_paths()) {
try {
string dpp_download_path_str = path.path + DPP_PREFIX;
string dpp_download_path_str = path.path + "/" + DPP_PREFIX;
std::filesystem::path dpp_download_path(dpp_download_path_str);
if (std::filesystem::exists(dpp_download_path)) {
std::filesystem::remove_all(dpp_download_path);
Expand Down Expand Up @@ -90,7 +90,8 @@ AgentServer::AgentServer(ExecEnv* exec_env, const TMasterInfo& master_info)
CREATE_AND_START_THREAD(REPORT_DISK_STATE, _report_disk_state_workers);
CREATE_AND_START_THREAD(REPORT_OLAP_TABLE, _report_tablet_workers);
CREATE_AND_START_POOL(SUBMIT_TABLE_COMPACTION, _submit_table_compaction_workers);
CREATE_AND_START_POOL(STORAGE_MEDIUM_MIGRATE_V2, _storage_medium_migrate_v2_workers);
CREATE_AND_START_POOL(REFRESH_STORAGE_POLICY, _storage_refresh_policy_workers);
CREATE_AND_START_POOL(UPDATE_STORAGE_POLICY, _storage_update_policy_workers);
#undef CREATE_AND_START_POOL
#undef CREATE_AND_START_THREAD

Expand Down Expand Up @@ -153,8 +154,8 @@ void AgentServer::submit_tasks(TAgentResult& agent_result,
HANDLE_TYPE(TTaskType::UPDATE_TABLET_META_INFO, _update_tablet_meta_info_workers,
update_tablet_meta_info_req);
HANDLE_TYPE(TTaskType::COMPACTION, _submit_table_compaction_workers, compaction_req);
HANDLE_TYPE(TTaskType::STORAGE_MEDIUM_MIGRATE_V2, _storage_medium_migrate_v2_workers,
storage_migration_req_v2);
HANDLE_TYPE(TTaskType::NOTIFY_UPDATE_STORAGE_POLICY, _storage_update_policy_workers,
update_policy);

case TTaskType::REALTIME_PUSH:
case TTaskType::PUSH:
Expand Down
4 changes: 2 additions & 2 deletions be/src/agent/agent_server.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,8 +88,8 @@ class AgentServer {

std::unique_ptr<TaskWorkerPool> _submit_table_compaction_workers;

std::unique_ptr<TaskWorkerPool> _storage_medium_migrate_v2_workers;

std::unique_ptr<TaskWorkerPool> _storage_refresh_policy_workers;
std::unique_ptr<TaskWorkerPool> _storage_update_policy_workers;
std::unique_ptr<TopicSubscriber> _topic_subscriber;
};

Expand Down
224 changes: 107 additions & 117 deletions be/src/agent/task_worker_pool.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,14 +35,14 @@
#include "olap/olap_common.h"
#include "olap/snapshot_manager.h"
#include "olap/storage_engine.h"
#include "olap/storage_policy_mgr.h"
#include "olap/tablet.h"
#include "olap/task/engine_alter_tablet_task.h"
#include "olap/task/engine_batch_load_task.h"
#include "olap/task/engine_checksum_task.h"
#include "olap/task/engine_clone_task.h"
#include "olap/task/engine_publish_version_task.h"
#include "olap/task/engine_storage_migration_task.h"
#include "olap/task/engine_storage_migration_task_v2.h"
#include "olap/utils.h"
#include "runtime/exec_env.h"
#include "runtime/snapshot_loader.h"
Expand Down Expand Up @@ -193,9 +193,14 @@ void TaskWorkerPool::start() {
cb = std::bind<void>(&TaskWorkerPool::_submit_table_compaction_worker_thread_callback,
this);
break;
case TaskWorkerType::STORAGE_MEDIUM_MIGRATE_V2:
case TaskWorkerType::REFRESH_STORAGE_POLICY:
_worker_count = 1;
cb = std::bind<void>(&TaskWorkerPool::_storage_medium_migrate_v2_worker_thread_callback,
cb = std::bind<void>(
&TaskWorkerPool::_storage_refresh_storage_policy_worker_thread_callback, this);
break;
case TaskWorkerType::UPDATE_STORAGE_POLICY:
_worker_count = 1;
cb = std::bind<void>(&TaskWorkerPool::_storage_update_storage_policy_worker_thread_callback,
this);
break;
default:
Expand Down Expand Up @@ -361,6 +366,7 @@ void TaskWorkerPool::_create_tablet_worker_thread_callback() {
TStatus task_status;

std::vector<TTabletInfo> finish_tablet_infos;
LOG(INFO) << "create tablet: " << create_tablet_req;
Status create_status = _env->storage_engine()->create_tablet(create_tablet_req);
if (!create_status.ok()) {
LOG(WARNING) << "create table failed. status: " << create_status
Expand Down Expand Up @@ -437,6 +443,11 @@ void TaskWorkerPool::_drop_tablet_worker_thread_callback() {
StorageEngine::instance()->txn_manager()->force_rollback_tablet_related_txns(
dropped_tablet->data_dir()->get_meta(), drop_tablet_req.tablet_id,
drop_tablet_req.schema_hash, dropped_tablet->tablet_uid());
// We remove remote rowset directly.
// TODO(cyx): do remove in background
if (drop_tablet_req.is_drop_table_or_partition) {
dropped_tablet->remove_all_remote_rowsets();
}
} else {
status_code = TStatusCode::NOT_FOUND;
error_msgs.push_back(err);
Expand Down Expand Up @@ -854,8 +865,15 @@ void TaskWorkerPool::_update_tablet_meta_worker_thread_callback() {
tablet->set_partition_id(tablet_meta_info.partition_id);
break;
case TTabletMetaType::INMEMORY:
tablet->tablet_meta()->mutable_tablet_schema()->set_is_in_memory(
tablet_meta_info.is_in_memory);
if (tablet_meta_info.storage_policy.empty()) {
tablet->tablet_meta()->mutable_tablet_schema()->set_is_in_memory(
tablet_meta_info.is_in_memory);
} else {
LOG(INFO) << "set tablet cooldown resource "
<< tablet_meta_info.storage_policy;
tablet->tablet_meta()->set_cooldown_resource(
tablet_meta_info.storage_policy);
}
break;
}
}
Expand Down Expand Up @@ -1036,8 +1054,8 @@ Status TaskWorkerPool::_check_migrate_request(const TStorageMediumMigrateReq& re
return Status::OLAPInternalError(OLAP_REQUEST_FAILED);
}

// check disk capacity
int64_t tablet_size = tablet->tablet_footprint();
// check local disk capacity
int64_t tablet_size = tablet->tablet_local_size();
if ((*dest_store)->reach_capacity_limit(tablet_size)) {
LOG(WARNING) << "reach the capacity limit of path: " << (*dest_store)->path()
<< ", tablet size: " << tablet_size;
Expand Down Expand Up @@ -1171,14 +1189,15 @@ void TaskWorkerPool::_report_disk_state_worker_thread_callback() {
map<string, TDisk> disks;
for (auto& root_path_info : data_dir_infos) {
TDisk disk;
disk.__set_root_path(root_path_info.path_desc.filepath);
disk.__set_root_path(root_path_info.path);
disk.__set_path_hash(root_path_info.path_hash);
disk.__set_storage_medium(root_path_info.storage_medium);
disk.__set_disk_total_capacity(root_path_info.disk_capacity);
disk.__set_data_used_capacity(root_path_info.data_used_capacity);
disk.__set_data_used_capacity(root_path_info.local_used_capacity);
disk.__set_remote_used_capacity(root_path_info.remote_used_capacity);
disk.__set_disk_available_capacity(root_path_info.available);
disk.__set_used(root_path_info.is_used);
disks[root_path_info.path_desc.filepath] = disk;
disks[root_path_info.path] = disk;
}
request.__set_disks(disks);
_handle_report(request, ReportType::DISK);
Expand Down Expand Up @@ -1565,7 +1584,6 @@ Status TaskWorkerPool::_move_dir(const TTabletId tablet_id, const std::string& s
return Status::InvalidArgument("Could not find tablet");
}

std::string dest_tablet_dir = tablet->tablet_path_desc().filepath;
SnapshotLoader loader(_env, job_id, tablet_id);
Status status = loader.move(src, tablet, overwrite);

Expand Down Expand Up @@ -1683,9 +1701,64 @@ void TaskWorkerPool::_submit_table_compaction_worker_thread_callback() {
}
}

void TaskWorkerPool::_storage_medium_migrate_v2_worker_thread_callback() {
void TaskWorkerPool::_storage_refresh_storage_policy_worker_thread_callback() {
while (_is_work) {
_is_doing_work = false;
// wait at most report_task_interval_seconds, or being notified
std::unique_lock<std::mutex> worker_thread_lock(_worker_thread_lock);
_worker_thread_condition_variable.wait_for(
worker_thread_lock,
std::chrono::seconds(config::storage_refresh_storage_policy_task_interval_seconds));
if (!_is_work) {
break;
}

if (_master_info.network_address.port == 0) {
// port == 0 means not received heartbeat yet
// sleep a short time and try again
LOG(INFO)
<< "waiting to receive first heartbeat from frontend before doing task report";
continue;
}

_is_doing_work = true;

TGetStoragePolicyResult result;
Status status = _master_client->refresh_storage_policy(&result);
if (!status.ok()) {
LOG(WARNING) << "refresh storage policy status not ok. ";
} else if (result.status.status_code != TStatusCode::OK) {
LOG(WARNING) << "refresh storage policy result status status_code not ok. ";
} else {
// update storage policy mgr.
StoragePolicyMgr* spm = ExecEnv::GetInstance()->storage_policy_mgr();
for (const auto& iter : result.result_entrys) {
shared_ptr<StoragePolicy> policy_ptr = make_shared<StoragePolicy>();
policy_ptr->storage_policy_name = iter.policy_name;
policy_ptr->cooldown_datetime = iter.cooldown_datetime;
policy_ptr->cooldown_ttl = iter.cooldown_ttl;
policy_ptr->s3_endpoint = iter.s3_storage_param.s3_endpoint;
policy_ptr->s3_region = iter.s3_storage_param.s3_region;
policy_ptr->s3_ak = iter.s3_storage_param.s3_ak;
policy_ptr->s3_sk = iter.s3_storage_param.s3_sk;
policy_ptr->root_path = iter.s3_storage_param.root_path;
policy_ptr->bucket = iter.s3_storage_param.bucket;
policy_ptr->s3_conn_timeout_ms = iter.s3_storage_param.s3_conn_timeout_ms;
policy_ptr->s3_max_conn = iter.s3_storage_param.s3_max_conn;
policy_ptr->s3_request_timeout_ms = iter.s3_storage_param.s3_request_timeout_ms;
policy_ptr->md5_sum = iter.md5_checksum;

LOG(INFO) << "refresh storage policy task, policy " << *policy_ptr;
spm->periodic_put(iter.policy_name, std::move(policy_ptr));
}
}
}
}

void TaskWorkerPool::_storage_update_storage_policy_worker_thread_callback() {
while (_is_work) {
TAgentTaskRequest agent_task_req;
TGetStoragePolicy get_storage_policy_req;
{
std::unique_lock<std::mutex> worker_thread_lock(_worker_thread_lock);
while (_is_work && _tasks.empty()) {
Expand All @@ -1696,115 +1769,32 @@ void TaskWorkerPool::_storage_medium_migrate_v2_worker_thread_callback() {
}

agent_task_req = _tasks.front();
get_storage_policy_req = agent_task_req.update_policy;
_tasks.pop_front();
}
int64_t signature = agent_task_req.signature;
LOG(INFO) << "get migration table v2 task, signature: " << agent_task_req.signature;
bool is_task_timeout = false;
if (agent_task_req.__isset.recv_time) {
int64_t time_elapsed = time(nullptr) - agent_task_req.recv_time;
if (time_elapsed > config::report_task_interval_seconds * 20) {
LOG(INFO) << "task elapsed " << time_elapsed
<< " seconds since it is inserted to queue, it is timeout";
is_task_timeout = true;
}
}
if (!is_task_timeout) {
TFinishTaskRequest finish_task_request;
TTaskType::type task_type = agent_task_req.task_type;
switch (task_type) {
case TTaskType::STORAGE_MEDIUM_MIGRATE_V2:
_storage_medium_migrate_v2(agent_task_req, signature, task_type,
&finish_task_request);
break;
default:
// pass
break;
}
_finish_task(finish_task_request);
}
_remove_task_info(agent_task_req.task_type, agent_task_req.signature);
}
}

void TaskWorkerPool::_storage_medium_migrate_v2(const TAgentTaskRequest& agent_task_req,
int64_t signature, const TTaskType::type task_type,
TFinishTaskRequest* finish_task_request) {
Status status = Status::OK();
TStatus task_status;
std::vector<string> error_msgs;

string process_name;
switch (task_type) {
case TTaskType::STORAGE_MEDIUM_MIGRATE_V2:
process_name = "StorageMediumMigrationV2";
break;
default:
std::string task_name;
EnumToString(TTaskType, task_type, task_name);
LOG(WARNING) << "Storage medium migration v2 type invalid. type: " << task_name
<< ", signature: " << signature;
status = Status::NotSupported("Storage medium migration v2 type invalid");
break;
}

// Check last storage medium migration v2 status, if failed delete tablet file
// Do not need to adjust delete success or not
// Because if delete failed task will failed
TTabletId new_tablet_id;
TSchemaHash new_schema_hash = 0;
if (status.ok()) {
new_tablet_id = agent_task_req.storage_migration_req_v2.new_tablet_id;
new_schema_hash = agent_task_req.storage_migration_req_v2.new_schema_hash;
EngineStorageMigrationTaskV2 engine_task(agent_task_req.storage_migration_req_v2);
Status sc_status = _env->storage_engine()->execute_task(&engine_task);
if (!sc_status.ok()) {
if (sc_status.precise_code() == OLAP_ERR_DATA_QUALITY_ERR) {
error_msgs.push_back("The data quality does not satisfy, please check your data. ");
}
status = sc_status;
} else {
status = Status::OK();
}
}

if (status.ok()) {
++_s_report_version;
LOG(INFO) << process_name << " finished. signature: " << signature;
}

// Return result to fe
finish_task_request->__set_backend(_backend);
finish_task_request->__set_report_version(_s_report_version);
finish_task_request->__set_task_type(task_type);
finish_task_request->__set_signature(signature);

std::vector<TTabletInfo> finish_tablet_infos;
if (status.ok()) {
TTabletInfo tablet_info;
status = _get_tablet_info(new_tablet_id, new_schema_hash, signature, &tablet_info);

if (!status.ok()) {
LOG(WARNING) << process_name << " success, but get new tablet info failed."
<< "tablet_id: " << new_tablet_id << ", schema_hash: " << new_schema_hash
<< ", signature: " << signature;
} else {
finish_tablet_infos.push_back(tablet_info);
}
}

if (status.ok()) {
finish_task_request->__set_finish_tablet_infos(finish_tablet_infos);
LOG(INFO) << process_name << " success. signature: " << signature;
error_msgs.push_back(process_name + " success");
} else {
LOG(WARNING) << process_name << " failed. signature: " << signature;
error_msgs.push_back(process_name + " failed");
error_msgs.push_back("status: " + status.to_string());
StoragePolicyMgr* spm = ExecEnv::GetInstance()->storage_policy_mgr();
shared_ptr<StoragePolicy> policy_ptr = make_shared<StoragePolicy>();
policy_ptr->storage_policy_name = get_storage_policy_req.policy_name;
policy_ptr->cooldown_datetime = get_storage_policy_req.cooldown_datetime;
policy_ptr->cooldown_ttl = get_storage_policy_req.cooldown_ttl;
policy_ptr->s3_endpoint = get_storage_policy_req.s3_storage_param.s3_endpoint;
policy_ptr->s3_region = get_storage_policy_req.s3_storage_param.s3_region;
policy_ptr->s3_ak = get_storage_policy_req.s3_storage_param.s3_ak;
policy_ptr->s3_sk = get_storage_policy_req.s3_storage_param.s3_sk;
policy_ptr->root_path = get_storage_policy_req.s3_storage_param.root_path;
policy_ptr->bucket = get_storage_policy_req.s3_storage_param.bucket;
policy_ptr->s3_conn_timeout_ms = get_storage_policy_req.s3_storage_param.s3_conn_timeout_ms;
policy_ptr->s3_max_conn = get_storage_policy_req.s3_storage_param.s3_max_conn;
policy_ptr->s3_request_timeout_ms =
get_storage_policy_req.s3_storage_param.s3_request_timeout_ms;
policy_ptr->md5_sum = get_storage_policy_req.md5_checksum;

LOG(INFO) << "get storage update policy task, update policy " << *policy_ptr;

spm->update(get_storage_policy_req.policy_name, std::move(policy_ptr));
_remove_task_info(agent_task_req.task_type, agent_task_req.signature);
}
task_status.__set_status_code(status.code());
task_status.__set_error_msgs(error_msgs);
finish_task_request->__set_task_status(task_status);
}

} // namespace doris
Loading