merge to ddb65b69f9c788e359e191889cb31f15279c41ec

be/CMakeLists.txt

@@ -252,14 +252,13 @@ execute_process(
 #  -fno-omit-frame-pointers: Keep frame pointer for functions in register
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wall -Wno-sign-compare -Wno-unknown-pragmas -pthread")
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -fno-strict-aliasing -fno-omit-frame-pointer")
-set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -std=gnu++11")
+set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -std=gnu++11 -D__STDC_FORMAT_MACROS")
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -Wno-deprecated -Wno-vla")
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG")
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -DBOOST_SYSTEM_NO_DEPRECATED")
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -msse4.2")
 set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -DLLVM_ON_UNIX")
 
-# for bprc
 if (CMAKE_CXX_COMPILER_VERSION VERSION_GREATER 7.0)
     set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} -faligned-new")
 endif()
@@ -350,7 +349,6 @@ endif()
 # optimizations will be less effective if the initial code is also optimized.
 set(CLANG_IR_CXX_FLAGS "-gcc-toolchain" ${GCC_HOME})
 set(CLANG_IR_CXX_FLAGS ${CLANG_IR_CXX_FLAGS} "-std=gnu++11" "-c" "-emit-llvm" "-D__STDC_CONSTANT_MACROS" "-D__STDC_FORMAT_MACROS" "-D__STDC_LIMIT_MACROS" "-DIR_COMPILE" "-DNDEBUG" "-DHAVE_INTTYPES_H" "-DHAVE_NETINET_IN_H" "-DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG" "-D__GLIBCXX_BITSIZE_INT_N_0=128" "-D__GLIBCXX_TYPE_INT_N_0=__int128" "-U_GLIBCXX_USE_FLOAT128" "-DLLVM_ON_UNIX")
-
 if (CMAKE_CXX_COMPILER_VERSION VERSION_GREATER 7.0)
     # for support float128
     set(CLANG_IR_CXX_FLAGS ${CLANG_IR_CXX_FLAGS} "-D__STRICT_ANSI__")

be/src/agent/agent_server.cpp

@@ -83,51 +83,71 @@ AgentServer::AgentServer(ExecEnv* exec_env,
     // init task worker pool
     _create_table_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::CREATE_TABLE,
+            _exec_env,
             master_info);
     _drop_table_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::DROP_TABLE,
+            _exec_env,
             master_info);
     _push_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::PUSH,
+            _exec_env,
             master_info);
     _delete_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::DELETE,
+            _exec_env,
             master_info);
     _alter_table_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::ALTER_TABLE,
+            _exec_env,
             master_info);
     _clone_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::CLONE,
+            _exec_env,
             master_info);
     _storage_medium_migrate_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::STORAGE_MEDIUM_MIGRATE,
+            _exec_env,
             master_info);
     _cancel_delete_data_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::CANCEL_DELETE_DATA,
+            _exec_env,
             master_info);
     _check_consistency_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::CHECK_CONSISTENCY,
+            _exec_env,
             master_info);
     _report_task_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::REPORT_TASK,
+            _exec_env,
             master_info);
     _report_disk_state_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::REPORT_DISK_STATE,
+            _exec_env,
             master_info);
     _report_olap_table_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::REPORT_OLAP_TABLE,
+            _exec_env,
             master_info);
     _upload_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::UPLOAD,
+            _exec_env,
             master_info);
-    _restore_workers = new TaskWorkerPool(
-            TaskWorkerPool::TaskWorkerType::RESTORE,
+    _download_workers = new TaskWorkerPool(
+            TaskWorkerPool::TaskWorkerType::DOWNLOAD,
+            _exec_env,
             master_info);
     _make_snapshot_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::MAKE_SNAPSHOT,
+            _exec_env,
             master_info);
     _release_snapshot_workers = new TaskWorkerPool(
             TaskWorkerPool::TaskWorkerType::RELEASE_SNAPSHOT,
+            _exec_env,
+            master_info);
+    _move_dir_workers= new TaskWorkerPool(
+            TaskWorkerPool::TaskWorkerType::MOVE,
+            _exec_env,
             master_info);
 #ifndef BE_TEST
     _create_table_workers->start();
@@ -143,9 +163,10 @@ AgentServer::AgentServer(ExecEnv* exec_env,
     _report_disk_state_workers->start();
     _report_olap_table_workers->start();
     _upload_workers->start();
-    _restore_workers->start();
+    _download_workers->start();
     _make_snapshot_workers->start();
     _release_snapshot_workers->start();
+    _move_dir_workers->start();
     // Add subscriber here and register listeners
     TopicListener* user_resource_listener = new UserResourceListener(exec_env, master_info);
     LOG(INFO) << "Register user resource listener";
@@ -196,12 +217,15 @@ AgentServer::~AgentServer() {
     if (_upload_workers != NULL) {
         delete _upload_workers;
     }
-    if (_restore_workers != NULL) {
-        delete _restore_workers;
+    if (_download_workers != NULL) {
+        delete _download_workers;
     }
     if (_make_snapshot_workers != NULL) {
         delete _make_snapshot_workers;
     }
+    if (_move_dir_workers!= NULL) {
+        delete _move_dir_workers;
+    }
     if (_release_snapshot_workers != NULL) {
         delete _release_snapshot_workers;
     }
@@ -303,9 +327,9 @@ void AgentServer::submit_tasks(
                 status_code = TStatusCode::ANALYSIS_ERROR;
             }
             break;
-        case TTaskType::RESTORE:
-            if (task.__isset.restore_req) {
-                _restore_workers->submit_task(task);
+        case TTaskType::DOWNLOAD:
+            if (task.__isset.download_req) {
+                _download_workers->submit_task(task);
             } else {
                 status_code = TStatusCode::ANALYSIS_ERROR;
             }
@@ -324,6 +348,13 @@ void AgentServer::submit_tasks(
                 status_code = TStatusCode::ANALYSIS_ERROR;
             }
             break;
+        case TTaskType::MOVE:
+            if (task.__isset.move_dir_req) {
+                _move_dir_workers->submit_task(task);
+            } else {
+                status_code = TStatusCode::ANALYSIS_ERROR;
+            }
+            break;
         default:
             status_code = TStatusCode::ANALYSIS_ERROR;
             break;
@@ -391,7 +422,6 @@ void AgentServer::release_snapshot(TAgentResult& return_value, const std::string
 void AgentServer::publish_cluster_state(TAgentResult& _return, const TAgentPublishRequest& request) {
     vector<string> error_msgs;
     _topic_subscriber->handle_updates(request);
-    OLAP_LOG_INFO("AgentService receive contains %d publish updates", request.updates.size());
     _return.status.__set_status_code(TStatusCode::OK);
 }
 

be/src/agent/agent_server.h

@@ -106,9 +106,10 @@ class AgentServer {
     TaskWorkerPool* _report_disk_state_workers;
     TaskWorkerPool* _report_olap_table_workers;
     TaskWorkerPool* _upload_workers;
-    TaskWorkerPool* _restore_workers;
+    TaskWorkerPool* _download_workers;
     TaskWorkerPool* _make_snapshot_workers;
     TaskWorkerPool* _release_snapshot_workers;
+    TaskWorkerPool* _move_dir_workers;
 
     DISALLOW_COPY_AND_ASSIGN(AgentServer);
 

be/src/agent/cgroups_mgr.cpp

@@ -62,11 +62,8 @@ CgroupsMgr::~CgroupsMgr() {
 
 AgentStatus CgroupsMgr::update_local_cgroups(const TFetchResourceResult&  new_fetched_resource) {
 
-    LOG(INFO) << "Current resource version is " << _cur_version
-        << ". Resource version is " << new_fetched_resource.resourceVersion; 
     std::lock_guard<std::mutex> lck(_update_cgroups_mtx);
     if (!_is_cgroups_init_success) {
-        LOG(WARNING) << "Cgroups manager initialized failed, will not update local cgroups!";
         return AgentStatus::PALO_ERROR;
     }
 

be/src/agent/heartbeat_server.cpp

@@ -48,10 +48,12 @@ void HeartbeatServer::heartbeat(
     TStatusCode::type status_code = TStatusCode::OK;
     vector<string> error_msgs;
     TStatus heartbeat_status;
-    OLAP_LOG_INFO("get heartbeat, host: %s, port: %d, cluster id: %d",
-                  master_info.network_address.hostname.c_str(),
-                  master_info.network_address.port,
-                  master_info.cluster_id);
+    //print heartbeat in every minute
+    LOG_EVERY_N(INFO, 12) << "get heartbeat from FE."
+        << "host:" << master_info.network_address.hostname << ", "
+        << "port:" << master_info.network_address.port << ", "
+        << "cluster id:" << master_info.cluster_id << ", "
+        << "counter:" << google::COUNTER;
 
     // Check cluster id
     if (_master_info->cluster_id == -1) {