Limit the disk usage to avoid running out of disk capacity

be/src/agent/status.h

@@ -41,6 +41,7 @@ enum AgentStatus {
  DORIS_PUSH_HAD_LOADED = -504,
  DORIS_TIMEOUT = -901,
  DORIS_INTERNAL_ERROR = -902,
+ DORIS_DISK_REACH_CAPACITY_LIMIT = -903,
 };
 } // namespace doris
 #endif // DORIS_BE_SRC_AGENT_STATUS_H

be/src/agent/task_worker_pool.cpp

@@ -433,21 +433,35 @@ void* TaskWorkerPool::_create_tablet_worker_thread_callback(void* arg_this) {
  vector<string> error_msgs;
  TStatus task_status;
 
-OLAPStatus create_status =
- worker_pool_this->_env->storage_engine()->create_tablet(create_tablet_req);
+std::vector<TTabletInfo> finish_tablet_infos;
+OLAPStatus create_status = worker_pool_this->_env->storage_engine()->create_tablet(create_tablet_req);
  if (create_status != OLAPStatus::OLAP_SUCCESS) {
  OLAP_LOG_WARNING("create table failed. status: %d, signature: %ld",
  create_status, agent_task_req.signature);
  // TODO liutao09 distinguish the OLAPStatus
  status_code = TStatusCode::RUNTIME_ERROR;
  } else {
  ++_s_report_version;
+ // get path hash of the created tablet
+ TabletSharedPtr tablet = StorageEngine::instance()->tablet_manager()->get_tablet(
+ create_tablet_req.tablet_id, create_tablet_req.tablet_schema.schema_hash);
+ DCHECK(tablet != nullptr);
+ TTabletInfo tablet_info;
+ tablet_info.tablet_id = tablet->table_id(); 
+ tablet_info.schema_hash = tablet->schema_hash();
+ tablet_info.version = create_tablet_req.version;
+ tablet_info.version_hash = create_tablet_req.version_hash;
+ tablet_info.row_count = 0;
+ tablet_info.data_size = 0;
+ tablet_info.__set_path_hash(tablet->data_dir()->path_hash());
+ finish_tablet_infos.push_back(tablet_info);
  }
 
  task_status.__set_status_code(status_code);
  task_status.__set_error_msgs(error_msgs);
 
  TFinishTaskRequest finish_task_request;
+ finish_task_request.__set_finish_tablet_infos(finish_tablet_infos);
  finish_task_request.__set_backend(worker_pool_this->_backend);
  finish_task_request.__set_report_version(_s_report_version);
  finish_task_request.__set_task_type(agent_task_req.task_type);
@@ -1252,7 +1266,7 @@ void* TaskWorkerPool::_report_disk_state_worker_thread_callback(void* arg_this)
  }
 #endif
  vector<DataDirInfo> data_dir_infos;
- worker_pool_this->_env->storage_engine()->get_all_data_dir_info(&data_dir_infos);
+ worker_pool_this->_env->storage_engine()->get_all_data_dir_info(&data_dir_infos, true /* update */);
 
  map<string, TDisk> disks;
  for (auto& root_path_info : data_dir_infos) {

be/src/common/config.h

@@ -212,12 +212,11 @@ namespace config {
  // inc_rowset expired interval
  CONF_Int32(inc_rowset_expired_sec, "1800");
  // garbage sweep policy
- CONF_Int32(max_garbage_sweep_interval, "14400");
+ CONF_Int32(max_garbage_sweep_interval, "3600");
  CONF_Int32(min_garbage_sweep_interval, "180");
  CONF_Int32(snapshot_expire_time_sec, "172800");
  // 仅仅是建议值，当磁盘空间不足时，trash下的文件保存期可不遵守这个参数
  CONF_Int32(trash_file_expire_time_sec, "259200");
- CONF_Int32(disk_capacity_insufficient_percentage, "90");
  // check row nums for BE/CE and schema change. true is open, false is closed.
  CONF_Bool(row_nums_check, "true")
  //file descriptors cache, by default, cache 30720 descriptors
@@ -439,6 +438,13 @@ namespace config {
  CONF_Int32(path_gc_check_step, "1000");
  CONF_Int32(path_gc_check_step_interval_ms, "10");
  CONF_Int32(path_scan_interval_second, "86400");
+
+ // The following 2 configs limit the max usage of disk capacity of a data dir.
+ // If both of these 2 threshold reached, no more data can be writen into that data dir.
+ // The percent of max used capacity of a data dir
+ CONF_Int32(storage_flood_stage_usage_percent, "95"); // 95%
+ // The min bytes that should be left of a data dir
+ CONF_Int64(storage_flood_stage_left_capacity_bytes, "1073741824") // 1GB
 } // namespace config
 
 } // namespace doris

be/src/olap/compaction.h

@@ -32,6 +32,7 @@
 
 namespace doris {
 
+class DataDir;
 class Merger;
 
 // This class is a base class for compaction.

be/src/olap/data_dir.cpp

@@ -59,14 +59,14 @@ DataDir::DataDir(const std::string& path, int64_t capacity_bytes,
  TabletManager* tablet_manager, TxnManager* txn_manager)
  : _path(path),
  _capacity_bytes(capacity_bytes),
+ _available_bytes(0),
+ _disk_capacity_bytes(0),
+ _is_used(false),
  _tablet_manager(tablet_manager),
  _txn_manager(txn_manager),
  _cluster_id(-1),
- _available_bytes(0),
- _used_bytes(0),
- _current_shard(0),
- _is_used(false),
  _to_be_deleted(false),
+ _current_shard(0),
  _test_file_read_buf(nullptr),
  _test_file_write_buf(nullptr),
  _meta(nullptr) {
@@ -100,6 +100,7 @@ Status DataDir::init() {
  return Status::InternalError("invalid root path: ");
  }
 
+ RETURN_IF_ERROR(update_capacity());
  RETURN_IF_ERROR(_init_cluster_id());
  RETURN_IF_ERROR(_init_extension_and_capacity());
  RETURN_IF_ERROR(_init_file_system());
@@ -1057,4 +1058,35 @@ void DataDir::_remove_check_paths_no_lock(const std::set<std::string>& paths) {
  }
 }
 
+Status DataDir::update_capacity() {
+ try {
+ boost::filesystem::path path_name(_path);
+ boost::filesystem::space_info path_info = boost::filesystem::space(path_name);
+ _available_bytes = path_info.available;
+ if (_disk_capacity_bytes == 0) {
+ // disk capacity only need to be set once
+ _disk_capacity_bytes = path_info.capacity;
+ }
+ } catch (boost::filesystem::filesystem_error& e) {
+ LOG(WARNING) << "get space info failed. path: " << _path << " erro:" << e.what();
+ return Status::InternalError("get path available capacity failed");
+ }
+ LOG(INFO) << "path: " << _path << " total capacity: " << _disk_capacity_bytes
+ << ", available capacity: " << _available_bytes;
+
+ return Status::OK();
+}
+
+bool DataDir::reach_capacity_limit(int64_t incoming_data_size) {
+ double used_pct = (_available_bytes + incoming_data_size) / (double) _disk_capacity_bytes;
+ int64_t left_bytes = _disk_capacity_bytes - _available_bytes - incoming_data_size;
+
+ if (used_pct >= config::storage_flood_stage_usage_percent / 100.0
+ && left_bytes <= config::storage_flood_stage_left_capacity_bytes) {
+ LOG(WARNING) << "reach capacity limit. used pct: " << used_pct << ", left bytes: " << left_bytes
+ << ", path: " << _path;
+ return true;
+ }
+ return false;
+}
 } // namespace doris

be/src/olap/data_dir.h

@@ -48,12 +48,15 @@ class DataDir {
  bool is_used() const { return _is_used; }
  void set_is_used(bool is_used) { _is_used = is_used; }
  int32_t cluster_id() const { return _cluster_id; }
+
  DataDirInfo get_dir_info() {
  DataDirInfo info;
  info.path = _path;
  info.path_hash = _path_hash;
- info.is_used = _is_used;
  info.capacity = _capacity_bytes;
+ info.available = _available_bytes;
+ info.is_used = _is_used;
+ info.storage_medium = _storage_medium;
  return info;
  }
 
@@ -121,6 +124,16 @@ class DataDir {
 
  OLAPStatus set_convert_finished();
 
+ // check if the capacity reach the limit after adding the incoming data
+ // return true if limit reached, otherwise, return false.
+ // TODO(cmy): for now we can not precisely calculate the capacity Doris used,
+ // so in order to avoid running out of disk capacity, we currenty use the actual
+ // disk avaiable capacity and total capacity to do the calculation.
+ // So that the capacity Doris actually used may exceeds the user specified capacity.
+ bool reach_capacity_limit(int64_t incoming_data_size);
+
+ Status update_capacity();
+
 private:
  std::string _cluster_id_path() const { return _path + CLUSTER_ID_PREFIX; }
  Status _init_cluster_id();
@@ -146,23 +159,30 @@ class DataDir {
 
 private:
  std::string _path;
- size_t _path_hash;
+ int64_t _path_hash;
+ // user specified capacity
+ int64_t _capacity_bytes;
+ // the actual avaiable capacity of the disk of this data dir
+ // NOTICE that _available_byte smay be larger than _capacity_bytes, if capacity is set
+ // by user, not the disk's actual capacity
+ int64_t _available_bytes;
+ // the actual capacity of the disk of this data dir
+ int64_t _disk_capacity_bytes;
+ TStorageMedium::type _storage_medium;
+ bool _is_used;
+
  uint32_t _rand_seed;
 
  std::string _file_system;
- int64_t _capacity_bytes;
  TabletManager* _tablet_manager;
  TxnManager* _txn_manager;
  int32_t _cluster_id;
- int64_t _available_bytes;
- int64_t _used_bytes;
- uint64_t _current_shard;
- bool _is_used;
  // This flag will be set true if this store was not in root path when reloading
  bool _to_be_deleted;
 
+ // used to protect _current_shard and _tablet_set
  std::mutex _mutex;
-TStorageMedium::type _storage_medium; // 存储介质类型：SSD|HDD
+uint64_t _current_shard;
  std::set<TabletInfo> _tablet_set;
 
  static const size_t TEST_FILE_BUF_SIZE = 4096;

be/src/olap/delta_writer.cpp

@@ -137,10 +137,7 @@ OLAPStatus DeltaWriter::init() {
 
  // TODO: new RowsetBuilder according to tablet storage type
  _rowset_writer.reset(new AlphaRowsetWriter());
- status = _rowset_writer->init(writer_context);
- if (status != OLAP_SUCCESS) {
- return OLAP_ERR_ROWSET_WRITER_INIT;
- }
+ RETURN_NOT_OK(_rowset_writer->init(writer_context));
 
  const std::vector<SlotDescriptor*>& slots = _req.tuple_desc->slots();
  const TabletSchema& schema = _tablet->tablet_schema();

be/src/olap/olap_define.h

@@ -164,6 +164,7 @@ enum OLAPStatus {
  OLAP_ERR_TRANSACTION_ALREADY_VISIBLE = -229,
  OLAP_ERR_VERSION_ALREADY_MERGED = -230,
  OLAP_ERR_LZO_DISABLED = -231,
+ OLAP_ERR_DISK_REACH_CAPACITY_LIMIT = -232,
 
  // CommandExecutor
  // [-300, -400)

be/src/olap/olap_server.cpp

@@ -157,7 +157,9 @@ void* StorageEngine::_base_compaction_thread_callback(void* arg, DataDir* data_d
  // cgroup is not initialized at this time
  // add tid to cgroup
  CgroupsMgr::apply_system_cgroup();
- perform_base_compaction(data_dir);
+ if (!data_dir->reach_capacity_limit(0)) {
+ perform_base_compaction(data_dir);
+ }
 
  usleep(interval * 1000000);
  }
@@ -249,7 +251,9 @@ void* StorageEngine::_cumulative_compaction_thread_callback(void* arg, DataDir*
  // cgroup is not initialized at this time
  // add tid to cgroup
  CgroupsMgr::apply_system_cgroup();
- perform_cumulative_compaction(data_dir);
+ if (!data_dir->reach_capacity_limit(0)) {
+ perform_cumulative_compaction(data_dir);
+ }
  usleep(interval * 1000000);
  }
 

be/src/olap/rowset/column_data_writer.cpp

@@ -280,7 +280,7 @@ OLAPStatus ColumnDataWriter::_flush_segment_with_verfication() {
  OLAPStatus res = _finalize_segment();
  if (OLAP_SUCCESS != res) {
  OLAP_LOG_WARNING("fail to finalize segment. [res=%d]", res);
- return OLAP_ERR_WRITER_DATA_WRITE_ERROR;
+ return res;
  }
 
  _new_segment_created = false;
@@ -292,12 +292,12 @@ OLAPStatus ColumnDataWriter::_finalize_segment() {
  OLAPStatus res = OLAP_SUCCESS;
  uint32_t data_segment_size;
 
- if (OLAP_SUCCESS != _segment_writer->finalize(&data_segment_size)) {
+ if ((res = _segment_writer->finalize(&data_segment_size)) != OLAP_SUCCESS) {
  OLAP_LOG_WARNING("fail to finish segment from olap_data.");
- return OLAP_ERR_WRITER_DATA_WRITE_ERROR;
+ return res;
  }
 
- if (OLAP_SUCCESS != _segment_group->finalize_segment(data_segment_size, _num_rows)) {
+ if ((res != _segment_group->finalize_segment(data_segment_size, _num_rows)) != OLAP_SUCCESS) {
  OLAP_LOG_WARNING("fail to finish segment from olap_index.");
  return OLAP_ERR_WRITER_INDEX_WRITE_ERROR;
  }

be/src/olap/rowset/segment_writer.cpp

@@ -213,19 +213,25 @@ OLAPStatus SegmentWriter::finalize(uint32_t* segment_file_size) {
  boost::filesystem::path data_dir_path = tablet_path.parent_path().parent_path().parent_path().parent_path();
  std::string data_dir_string = data_dir_path.string();
  DataDir* data_dir = StorageEngine::instance()->get_store(data_dir_string);
- data_dir->add_pending_ids(ROWSET_ID_PREFIX + std::to_string(_segment_group->rowset_id()));
- if (OLAP_SUCCESS != (res = file_handle.open_with_mode(
- _file_name, O_CREAT | O_EXCL | O_WRONLY , S_IRUSR | S_IWUSR))) {
- LOG(WARNING) << "fail to open file. [file_name=" << _file_name << "]";
- return res;
- }
 
  res = _make_file_header(file_header.mutable_message());
  if (OLAP_SUCCESS != res) {
  OLAP_LOG_WARNING("fail to make file header. [res=%d]", res);
  return res;
  }
 
+ // check disk capacity
+ if (data_dir->reach_capacity_limit((int64_t) file_header.file_length())) {
+ return OLAP_ERR_DISK_REACH_CAPACITY_LIMIT;
+ }
+
+ data_dir->add_pending_ids(ROWSET_ID_PREFIX + std::to_string(_segment_group->rowset_id()));
+ if (OLAP_SUCCESS != (res = file_handle.open_with_mode(
+ _file_name, O_CREAT | O_EXCL | O_WRONLY , S_IRUSR | S_IWUSR))) {
+ LOG(WARNING) << "fail to open file. [file_name=" << _file_name << "]";
+ return res;
+ }
+
  res = file_header.prepare(&file_handle);
  if (OLAP_SUCCESS != res) {
  OLAP_LOG_WARNING("write file header error. [err=%m]");

be/src/olap/schema_change.cpp

@@ -1892,7 +1892,6 @@ OLAPStatus SchemaChangeHandler::_convert_historical_rowsets(const SchemaChangePa
  // set status for monitor
  // 只要有一个new_table为running，ref table就设置为running
  // NOTE 如果第一个sub_table先fail，这里会继续按正常走
-
  RowsetId rowset_id = 0;
  TabletSharedPtr new_tablet = sc_params.new_tablet;
  res = sc_params.new_tablet->next_rowset_id(&rowset_id);