hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Refactor object storage multipart upload interface to distinguish close, complete and abort operations

Browse Source
This commit is contained in:
obdev
2024-02-02 12:12:33 +00:00
committed by ob-robot
parent 9ddb3fb67b
commit 6c8cde0b84
28 changed files with 506 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tools/ob_admin/io_bench/ob_admin_io_adapter_bench.cpp
View File

@ -139,6 +139,8 @@ int ObAdminIOAdapterBenchmarkExecutor::parse_cmd_(int argc, char *argv[])
config_.type_ = BenchmarkTaskType::BENCHMARK_TASK_READ;
} else if (0 == STRCMP("del", optarg)) {
config_.type_ = BenchmarkTaskType::BENCHMARK_TASK_DEL;
} else if (0 == STRCMP("is_exist", optarg)) {
config_.type_ = BenchmarkTaskType::BENCHMARK_TASK_IS_EXIST;
} else {
ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "unknown test type", K((char *)optarg), K(ret));

69
tools/ob_admin/io_bench/task_executor.cpp
View File

@ -333,6 +333,10 @@ int init_task_executor(const char *base_uri,
if (OB_FAIL(alloc_executor<DelTaskExecutor>(executor, attr))) {
OB_LOG(WARN, "fail to alloc and construct executor", K(ret), K(config));
}
} else if (config.type_ == BenchmarkTaskType::BENCHMARK_TASK_IS_EXIST) {
if (OB_FAIL(alloc_executor<IsExistTaskExecutor>(executor, attr))) {
OB_LOG(WARN, "fail to alloc and construct executor", K(ret), K(config));
}
} else {
ret = OB_ERR_UNEXPECTED;
OB_LOG(WARN, "unknown benchmark task type", K(ret), K(config));
@ -627,18 +631,21 @@ int MultipartWriteTaskExecutor::execute()
if (OB_FAIL(device_handle->pwrite(fd, cur_offset, cur_part_size,
write_buf_, actual_write_size))) {
OB_LOG(WARN, "fail to upload part",
K(ret), K_(base_uri), K(cur_offset), K(cur_part_size));
K(ret), K_(base_uri), K(cur_offset), K(cur_part_size), K(fd));
} else {
cur_offset += cur_part_size;
}
}
const int64_t close_start_time_us = ObTimeUtility::current_time();
if (FAILEDx(adapter.close_device_and_fd(device_handle, fd))) {
OB_LOG(WARN, "fail to close device handle", K(ret), K_(base_uri));
} else {
metrics_.close_time_ms_map_.log_entry(close_start_time_us);
if (FAILEDx(device_handle->complete(fd))) {
OB_LOG(WARN, "fail to complete multipart writer", K(ret), K_(base_uri), K(fd));
}
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(adapter.close_device_and_fd(device_handle, fd))) {
OB_LOG(WARN, "fail to close device handle", K(ret), K(tmp_ret), K_(base_uri));
}
metrics_.close_time_ms_map_.log_entry(close_start_time_us);
if (OB_SUCC(ret)) {
metrics_.operation_num_++;
@ -781,5 +788,57 @@ int DelTaskExecutor::execute()
return ret;
}
/*--------------------------------Is Exist Task Executor--------------------------------*/
IsExistTaskExecutor::IsExistTaskExecutor()
: ITaskExecutor(), obj_num_(-1)
{
}
int IsExistTaskExecutor::init(const char *base_uri,
share::ObBackupStorageInfo *storage_info, const TaskConfig &config)
{
int ret = OB_SUCCESS;
if (OB_FAIL(ITaskExecutor::init(base_uri, storage_info, config))) {
OB_LOG(WARN, "fail to init ITaskExecutor", K(ret), K(base_uri), KPC(storage_info), K(config));
} else if (OB_UNLIKELY(config.obj_num_ <= 0)) {
ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "invalid arguments", K(ret), K(config));
} else {
obj_num_ = config.obj_num_;
is_inited_ = true;
}
if (OB_FAIL(ret)) {
reset();
}
return ret;
}
int IsExistTaskExecutor::execute()
{
int ret = OB_SUCCESS;
const int64_t object_id = ObRandom::rand(0, obj_num_ - 1);
const int64_t start_time_us = ObTimeUtility::current_time();
if (OB_UNLIKELY(!is_inited_)) {
ret = OB_NOT_INIT;
OB_LOG(WARN, "ReadTaskExecutor not init", K(ret), K_(base_uri));
} else if (OB_FAIL(prepare_(object_id))) {
OB_LOG(WARN, "fail to prepare", K(ret), K_(base_uri), K(object_id));
} else {
ObBackupIoAdapter adapter;
bool is_exist = false;
if (OB_FAIL(adapter.is_exist(base_uri_, storage_info_, is_exist))) {
OB_LOG(WARN, "fail to check is exist",
K(ret), K_(base_uri), KPC_(storage_info), K(object_id));
} else {
metrics_.operation_num_++;
metrics_.total_op_time_ms_map_.log_entry(start_time_us);
}
}
finish_(ret);
return ret;
}
} //tools
} //oceanbase

14
tools/ob_admin/io_bench/task_executor.h
View File

@ -31,6 +31,7 @@ enum BenchmarkTaskType
BENCHMARK_TASK_MULTIPART_WRITE = 2,
BENCHMARK_TASK_READ = 3,
BENCHMARK_TASK_DEL = 4,
BENCHMARK_TASK_IS_EXIST = 5,
BENCHMARK_TASK_MAX_TYPE
};
@ -193,6 +194,19 @@ private:
bool is_adaptive_;
};
class IsExistTaskExecutor : public ITaskExecutor
{
public:
IsExistTaskExecutor();
virtual ~IsExistTaskExecutor() {}
virtual int init(const char *base_uri,
share::ObBackupStorageInfo *storage_info, const TaskConfig &config) override;
virtual int execute() override;
private:
int64_t obj_num_;
};
} //namespace tools
} //namespace oceanbase

73
tools/ob_admin/io_device/ob_admin_test_io_device_executor.cpp
View File

@ -268,6 +268,7 @@ int ObAdminTestIODeviceExecutor::test_appendable_check_file_(const char* check_f
int ObAdminTestIODeviceExecutor::test_multipart_upload_check_file_(const char* check_file_dir)
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObBackupIoAdapter util;
share::ObBackupStorageInfo storage_info;
ObIOFd fd;
@ -296,26 +297,39 @@ int ObAdminTestIODeviceExecutor::test_multipart_upload_check_file_(const char* c
STORAGE_LOG(WARN, "device handle is NULL", K(ret), K(device_handle), K_(backup_path));
} else if (OB_FAIL(device_handle->write(fd, check_file_content, real_len, write_size))) {
STORAGE_LOG(WARN, "failed to write check file", K(ret), K(check_file_content), K(real_len));
} else if (OB_FAIL(util.close_device_and_fd(device_handle, fd))) {
STORAGE_LOG(WARN, "fail to close device and fd", K(ret), KP(device_handle), K(fd));
} else if (OB_FAIL(util.is_exist(check_file_path, &storage_info, is_exist))) {
STORAGE_LOG(WARN, "failed to check if normal check file is exist", K(ret), K(check_file_path));
} else if (!is_exist) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "normal check file does not exist!", K(ret), K(check_file_path));
} else if (OB_FAIL(util.get_file_length(check_file_path, &storage_info, check_file_len))) {
STORAGE_LOG(WARN, "failed to get check file length", K(ret), K(check_file_path));
} else if (real_len != check_file_len) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "get check file length does not match real length", K(ret), K(check_file_path), K(real_len), K(check_file_len));
} else if (OB_ISNULL(read_file_buf = reinterpret_cast<char*>(allocator_.alloc(check_file_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
STORAGE_LOG(WARN, "failed to allocate buf", K(ret), K(check_file_path), K(check_file_len));
} else if (OB_FAIL(util.read_single_file(check_file_path, &storage_info, read_file_buf, check_file_len, read_size))) {
STORAGE_LOG(WARN, "failed to read check file", K(ret), K(check_file_path), K(check_file_len));
} else if (read_size != check_file_len) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "read check file length does not match real length", K(ret), K(check_file_path), K(read_size), K(check_file_len));
} else if (OB_FAIL(device_handle->complete(fd))) {
STORAGE_LOG(WARN, "fail to complete multipart upload", K(ret), K(device_handle), K(fd));
}
if (OB_SUCCESS != ret) {
if (OB_TMP_FAIL(device_handle->abort(fd))) {
ret = COVER_SUCC(tmp_ret);
STORAGE_LOG(WARN, "fail to abort multipart upload", K(ret), K(tmp_ret), K(device_handle), K(fd));
}
}
if (OB_TMP_FAIL(util.close_device_and_fd(device_handle, fd))) {
ret = COVER_SUCC(tmp_ret);
STORAGE_LOG(WARN, "fail to close device and fd", K(ret), K(tmp_ret), K(device_handle), K(fd));
}
if (OB_SUCC(ret)) {
if (OB_FAIL(util.is_exist(check_file_path, &storage_info, is_exist))) {
STORAGE_LOG(WARN, "failed to check if normal check file is exist", K(ret), K(check_file_path));
} else if (!is_exist) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "normal check file does not exist!", K(ret), K(check_file_path));
} else if (OB_FAIL(util.get_file_length(check_file_path, &storage_info, check_file_len))) {
STORAGE_LOG(WARN, "failed to get check file length", K(ret), K(check_file_path));
} else if (real_len != check_file_len) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "get check file length does not match real length", K(ret), K(check_file_path), K(real_len), K(check_file_len));
} else if (OB_ISNULL(read_file_buf = reinterpret_cast<char*>(allocator_.alloc(check_file_len)))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
STORAGE_LOG(WARN, "failed to allocate buf", K(ret), K(check_file_path), K(check_file_len));
} else if (OB_FAIL(util.read_single_file(check_file_path, &storage_info, read_file_buf, check_file_len, read_size))) {
STORAGE_LOG(WARN, "failed to read check file", K(ret), K(check_file_path), K(check_file_len));
} else if (read_size != check_file_len) {
ret = OB_ERR_UNEXPECTED;
STORAGE_LOG(WARN, "read check file length does not match real length", K(ret), K(check_file_path), K(read_size), K(check_file_len));
}
}
return ret;
}
@ -406,6 +420,7 @@ int ObAdminTestIODeviceExecutor::test_clean_backup_file_()
int ObAdminTestIODeviceExecutor::test_backup_data_()
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObBackupIoAdapter util;
share::ObBackupStorageInfo storage_info;
ObIOFd fd;
@ -455,10 +470,22 @@ int ObAdminTestIODeviceExecutor::test_backup_data_()
total_size += buf_size;
}
}
if (OB_SUCC(ret)) {
if (OB_FAIL(device_handle->complete(fd))) {
STORAGE_LOG(WARN, "fail to complete multipart upload", K(ret), K(tmp_ret), K(device_handle), K(fd));
}
} else {
if (OB_TMP_FAIL(device_handle->abort(fd))) {
ret = COVER_SUCC(tmp_ret);
STORAGE_LOG(WARN, "fail to abort multipart upload", K(ret), K(tmp_ret), K(device_handle), K(fd));
}
}
if (OB_TMP_FAIL(util.close_device_and_fd(device_handle, fd))) {
ret = COVER_SUCC(tmp_ret);
STORAGE_LOG(WARN, "fail to close device and fd", K(ret), K(device_handle), K(fd));
}
if(OB_SUCC(ret)) {
if (OB_FAIL(util.close_device_and_fd(device_handle, fd))) {
STORAGE_LOG(WARN, "fail to close multipart writer and release device!", K(ret), K(data_file_path));
} else if (OB_FAIL(util.get_file_length(data_file_path, &storage_info, read_size))) {
if (OB_FAIL(util.get_file_length(data_file_path, &storage_info, read_size))) {
STORAGE_LOG(WARN, "fail to get multipart upload file length", K(ret), K(data_file_path));
} else if (read_size != total_size) {
ret = OB_ERR_UNEXPECTED;