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

This commit is contained in:
obdev 2024-02-10 04:27:55 +00:00 committed by ob-robot
parent bef3ebb26f
commit 7da0974a40
28 changed files with 506 additions and 144 deletions

View File

@ -330,6 +330,8 @@ public:
//file/dir interfaces //file/dir interfaces
virtual int open(const char *pathname, const int flags, const mode_t mode, virtual int open(const char *pathname, const int flags, const mode_t mode,
ObIOFd &fd, ObIODOpts *opts= NULL) = 0; ObIOFd &fd, ObIODOpts *opts= NULL) = 0;
virtual int complete(const ObIOFd &fd) = 0;
virtual int abort(const ObIOFd &fd) = 0;
virtual int close(const ObIOFd &fd) = 0; virtual int close(const ObIOFd &fd) = 0;
virtual int mkdir(const char *pathname, mode_t mode) = 0; virtual int mkdir(const char *pathname, mode_t mode) = 0;
virtual int rmdir(const char *pathname) = 0; virtual int rmdir(const char *pathname) = 0;

View File

@ -99,9 +99,11 @@ static void convert_io_error(cos_status_t *cos_ret, int &ob_errcode)
} else { } else {
ob_errcode = OB_COS_ERROR; ob_errcode = OB_COS_ERROR;
} }
break;
} }
default: { default: {
ob_errcode = OB_COS_ERROR; ob_errcode = OB_COS_ERROR;
break;
} }
} }
} }

View File

@ -288,7 +288,8 @@ int ObAppendableFragmentMeta::parse_from(ObString &fragment_name)
ret = OB_INVALID_ARGUMENT; ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "invalid fragment suffix", K(ret), K(fragment_suffix)); OB_LOG(WARN, "invalid fragment suffix", K(ret), K(fragment_suffix));
} else { } else {
STRCPY(suffix_, fragment_suffix); MEMCPY(suffix_, fragment_suffix, strlen(fragment_suffix));
suffix_[strlen(fragment_suffix)] = '\0';
} }
} }
} }

View File

@ -276,6 +276,18 @@ public:
virtual bool is_opened() const = 0; virtual bool is_opened() const = 0;
}; };
class ObIStorageMultiPartWriter
{
public:
virtual int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info) = 0;
virtual int write(const char *buf, const int64_t size) = 0;
virtual int pwrite(const char *buf, const int64_t size, const int64_t offset) = 0;
virtual int complete() = 0;
virtual int abort() = 0;
virtual int close() = 0;
virtual int64_t get_length() const = 0;
virtual bool is_opened() const = 0;
};
}//common }//common
}//oceanbase }//oceanbase

View File

@ -405,6 +405,60 @@ int ObObjectDevice::open(const char *pathname, const int flags, const mode_t mod
return ret; return ret;
} }
int ObObjectDevice::complete(const ObIOFd &fd)
{
int ret = OB_SUCCESS;
int flag = -1;
void *ctx = nullptr;
fd_mng_.get_fd_flag(fd, flag);
if (!fd_mng_.validate_fd(fd, true)) {
ret = OB_NOT_INIT;
OB_LOG(WARN, "fd is not init!", K(fd.first_id_), K(fd.second_id_));
} else if (OB_FAIL(fd_mng_.fd_to_ctx(fd, ctx))) {
OB_LOG(WARN, "fail to get ctx accroding fd!", K(ret), K(fd));
} else if (OB_ISNULL(ctx)) {
ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "fd ctx is null!", K(flag), K(ret), K(fd));
} else if (flag == OB_STORAGE_ACCESS_MULTIPART_WRITER) {
ObStorageMultiPartWriter *multipart_writer = static_cast<ObStorageMultiPartWriter*>(ctx);
if (OB_FAIL(multipart_writer->complete())) {
OB_LOG(WARN, "fail to complete!", K(ret), K(flag));
}
} else {
ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "unknow access type, not a multipart writer fd!", K(flag), K(ret));
}
return ret;
}
int ObObjectDevice::abort(const ObIOFd &fd)
{
int ret = OB_SUCCESS;
int flag = -1;
void *ctx = nullptr;
fd_mng_.get_fd_flag(fd, flag);
if (!fd_mng_.validate_fd(fd, true)) {
ret = OB_NOT_INIT;
OB_LOG(WARN, "fd is not init!", K(fd.first_id_), K(fd.second_id_));
} else if (OB_FAIL(fd_mng_.fd_to_ctx(fd, ctx))) {
OB_LOG(WARN, "fail to get ctx accroding fd!", K(ret), K(fd));
} else if (OB_ISNULL(ctx)) {
ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "fd ctx is null!", K(flag), K(ret), K(fd));
} else if (flag == OB_STORAGE_ACCESS_MULTIPART_WRITER) {
ObStorageMultiPartWriter *multipart_writer = static_cast<ObStorageMultiPartWriter*>(ctx);
if (OB_FAIL(multipart_writer->abort())) {
OB_LOG(WARN, "fail to abort!", K(ret), K(flag));
}
} else {
ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "unknow access type, not a multipart writer fd!", K(flag), K(ret));
}
return ret;
}
int ObObjectDevice::close(const ObIOFd &fd) int ObObjectDevice::close(const ObIOFd &fd)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -416,9 +470,9 @@ int ObObjectDevice::close(const ObIOFd &fd)
ret = OB_NOT_INIT; ret = OB_NOT_INIT;
OB_LOG(WARN, "fail to close fd. since fd is invalid!", K(ret) ,K(fd.first_id_), K(fd.second_id_)); OB_LOG(WARN, "fail to close fd. since fd is invalid!", K(ret) ,K(fd.first_id_), K(fd.second_id_));
} else if (OB_FAIL(fd_mng_.fd_to_ctx(fd, ctx))) { } else if (OB_FAIL(fd_mng_.fd_to_ctx(fd, ctx))) {
OB_LOG(WARN, "fail to get ctx accroding fd!", K(ret)); OB_LOG(WARN, "fail to get ctx accroding fd!", K(ret), K(fd));
} else if (OB_FAIL(release_res(ctx, fd, (ObStorageAccessType)flag))) { } else if (OB_FAIL(release_res(ctx, fd, (ObStorageAccessType)flag))) {
OB_LOG(WARN, "fail to release the resource!", K(ret)); OB_LOG(WARN, "fail to release the resource!", K(ret), K(flag));
} }
return ret; return ret;
} }
@ -434,7 +488,7 @@ int ObObjectDevice::seal_for_adaptive(const ObIOFd &fd)
ret = OB_NOT_INIT; ret = OB_NOT_INIT;
OB_LOG(WARN, "fd is not init!", K(fd.first_id_), K(fd.second_id_)); OB_LOG(WARN, "fd is not init!", K(fd.first_id_), K(fd.second_id_));
} else if (OB_FAIL(fd_mng_.fd_to_ctx(fd, ctx))) { } else if (OB_FAIL(fd_mng_.fd_to_ctx(fd, ctx))) {
OB_LOG(WARN, "fail to get ctx accroding fd!", K(ret)); OB_LOG(WARN, "fail to get ctx accroding fd!", K(ret), K(fd));
} else if (OB_ISNULL(ctx)) { } else if (OB_ISNULL(ctx)) {
ret = OB_INVALID_ARGUMENT; ret = OB_INVALID_ARGUMENT;
OB_LOG(WARN, "fd ctx is null!", K(flag), K(ret)); OB_LOG(WARN, "fd ctx is null!", K(flag), K(ret));

View File

@ -57,6 +57,8 @@ public:
//file/dir interfaces //file/dir interfaces
virtual int open(const char *pathname, const int flags, const mode_t mode, virtual int open(const char *pathname, const int flags, const mode_t mode,
ObIOFd &fd, ObIODOpts *opts= NULL) override; ObIOFd &fd, ObIODOpts *opts= NULL) override;
virtual int complete(const ObIOFd &fd) override;
virtual int abort(const ObIOFd &fd) override;
virtual int close(const ObIOFd &fd) override; virtual int close(const ObIOFd &fd) override;
virtual int mkdir(const char *pathname, mode_t mode) override; virtual int mkdir(const char *pathname, mode_t mode) override;
virtual int rmdir(const char *pathname) override; virtual int rmdir(const char *pathname) override;

View File

@ -2238,6 +2238,44 @@ int64_t ObStorageMultiPartWriter::get_length()
return ret_int; return ret_int;
} }
int ObStorageMultiPartWriter::complete()
{
int ret = OB_SUCCESS;
const int64_t start_ts = ObTimeUtility::current_time();
if (OB_FAIL(ret)) {
} else if (ObStorageGlobalIns::get_instance().is_io_prohibited()) {
ret = OB_BACKUP_IO_PROHIBITED;
STORAGE_LOG(WARN, "current observer backup io is prohibited", K(ret));
} else if (OB_ISNULL(multipart_writer_)) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "multipart writer not opened", K(ret));
} else if (OB_FAIL(multipart_writer_->complete())) {
STORAGE_LOG(WARN, "failed to complete", K(ret));
}
print_access_storage_log("ObStorageMultiPartWriter::complete", uri_, start_ts, 0);
return ret;
}
int ObStorageMultiPartWriter::abort()
{
int ret = OB_SUCCESS;
const int64_t start_ts = ObTimeUtility::current_time();
if (OB_FAIL(ret)) {
} else if (ObStorageGlobalIns::get_instance().is_io_prohibited()) {
ret = OB_BACKUP_IO_PROHIBITED;
STORAGE_LOG(WARN, "current observer backup io is prohibited");
} else if (OB_ISNULL(multipart_writer_)) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "multipart writer not opened", K(ret));
} else if (OB_FAIL(multipart_writer_->abort())) {
STORAGE_LOG(WARN, "failed to abort", K(ret));
}
print_access_storage_log("ObStorageMultiPartWriter::abort", uri_, start_ts, 0);
return ret;
}
int ObStorageMultiPartWriter::close() int ObStorageMultiPartWriter::close()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;

View File

@ -387,13 +387,15 @@ public:
virtual int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info); virtual int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info);
int write(const char *buf, const int64_t size); int write(const char *buf, const int64_t size);
int pwrite(const char *buf, const int64_t size, const int64_t offset); int pwrite(const char *buf, const int64_t size, const int64_t offset);
int complete();
int abort();
int close(); int close();
bool is_opened() const {return is_opened_;} bool is_opened() const {return is_opened_;}
int64_t get_length(); int64_t get_length();
TO_STRING_KV(KP_(multipart_writer), K_(start_ts), K_(is_opened), KCSTRING_(uri)); TO_STRING_KV(KP_(multipart_writer), K_(start_ts), K_(is_opened), KCSTRING_(uri));
protected: protected:
ObIStorageWriter *multipart_writer_; ObIStorageMultiPartWriter *multipart_writer_;
ObStorageFileMultiPartWriter file_multipart_writer_; ObStorageFileMultiPartWriter file_multipart_writer_;
ObStorageCosMultiPartWriter cos_multipart_writer_; ObStorageCosMultiPartWriter cos_multipart_writer_;
ObStorageOssMultiPartWriter oss_multipart_writer_; ObStorageOssMultiPartWriter oss_multipart_writer_;

View File

@ -1089,7 +1089,6 @@ void ObStorageCosMultiPartWriter::reuse()
int ObStorageCosMultiPartWriter::open(const ObString &uri, common::ObObjectStorageInfo *storage_info) int ObStorageCosMultiPartWriter::open(const ObString &uri, common::ObObjectStorageInfo *storage_info)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;
if (OB_UNLIKELY(is_opened_)) { if (OB_UNLIKELY(is_opened_)) {
ret = OB_COS_ERROR; ret = OB_COS_ERROR;
@ -1122,10 +1121,6 @@ int ObStorageCosMultiPartWriter::open(const ObString &uri, common::ObObjectStora
base_buf_pos_ = 0; base_buf_pos_ = 0;
file_length_ = 0; file_length_ = 0;
} }
if (OB_FAIL(ret) && OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), KP_(upload_id));
}
} }
} }
return ret; return ret;
@ -1154,10 +1149,6 @@ int ObStorageCosMultiPartWriter::write(const char * buf, const int64_t size)
if (base_buf_pos_ == COS_MULTIPART_UPLOAD_BUF_SIZE) { if (base_buf_pos_ == COS_MULTIPART_UPLOAD_BUF_SIZE) {
if (OB_FAIL(write_single_part())) { if (OB_FAIL(write_single_part())) {
OB_LOG(WARN, "fail to write part into cos", K(ret)); OB_LOG(WARN, "fail to write part into cos", K(ret));
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
} else { } else {
base_buf_pos_ = 0; base_buf_pos_ = 0;
} }
@ -1206,22 +1197,17 @@ int ObStorageCosMultiPartWriter::write_single_part()
return ret; return ret;
} }
int ObStorageCosMultiPartWriter::close() int ObStorageCosMultiPartWriter::complete()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;
const int64_t start_time = ObTimeUtility::current_time(); const int64_t start_time = ObTimeUtility::current_time();
if (OB_UNLIKELY(!is_opened_)) { if (OB_UNLIKELY(!is_opened_)) {
ret = OB_COS_ERROR; ret = OB_COS_ERROR;
OB_LOG(WARN, "cos writer cannot close before it is opened", K(ret)); OB_LOG(WARN, "cos multipart writer cannot close before it is opened", K(ret));
} else if (0 != base_buf_pos_) { } else if (0 != base_buf_pos_) {
if(OB_SUCCESS != (ret = write_single_part())) { if (OB_FAIL(write_single_part())) {
OB_LOG(WARN, "fail to write the last size to cos", K(ret), K_(base_buf_pos)); OB_LOG(WARN, "fail to write the last size to cos", K(ret), K_(base_buf_pos));
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
ret = OB_COS_ERROR;
} else { } else {
base_buf_pos_ = 0; base_buf_pos_ = 0;
} }
@ -1238,27 +1224,31 @@ int ObStorageCosMultiPartWriter::close()
if (OB_FAIL(qcloud_cos::ObCosWrapper::complete_multipart_upload(handle_.get_ptr(), bucket_name, if (OB_FAIL(qcloud_cos::ObCosWrapper::complete_multipart_upload(handle_.get_ptr(), bucket_name,
object_name, upload_id_str))) { object_name, upload_id_str))) {
OB_LOG(WARN, "fail to complete multipart upload", K(ret), K_(upload_id)); OB_LOG(WARN, "fail to complete multipart upload", K(ret), K_(upload_id));
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
} }
} }
reuse();
const int64_t total_cost_time = ObTimeUtility::current_time() - start_time; const int64_t total_cost_time = ObTimeUtility::current_time() - start_time;
if (total_cost_time > 3 * 1000 * 1000) { if (total_cost_time > 3 * 1000 * 1000) {
OB_LOG_RET(WARN, OB_ERR_TOO_MUCH_TIME, "cos writer close cost too much time", K(total_cost_time), K(ret)); OB_LOG_RET(WARN, OB_ERR_TOO_MUCH_TIME, "cos multipart writer complete cost too much time",
K(total_cost_time), K(ret));
} }
return ret; return ret;
} }
int ObStorageCosMultiPartWriter::cleanup() int ObStorageCosMultiPartWriter::close()
{
int ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard;
reuse();
return ret;
}
int ObStorageCosMultiPartWriter::abort()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_opened_)) { if (OB_UNLIKELY(!is_opened_)) {
ret = OB_COS_ERROR; ret = OB_COS_ERROR;
OB_LOG(WARN, "cos multipart writer cannot cleanup before it is opened", K(ret)); OB_LOG(WARN, "cos multipart writer cannot abort before it is opened", K(ret));
} else { } else {
qcloud_cos::CosStringBuffer bucket_name = qcloud_cos::CosStringBuffer( qcloud_cos::CosStringBuffer bucket_name = qcloud_cos::CosStringBuffer(
handle_.get_bucket_name().ptr(), handle_.get_bucket_name().length()); handle_.get_bucket_name().ptr(), handle_.get_bucket_name().length());

View File

@ -164,7 +164,7 @@ private:
// part size is in [1MB, 5GB], exclude the last part // part size is in [1MB, 5GB], exclude the last part
// max part num 10000 // max part num 10000
class ObStorageCosMultiPartWriter: public ObStorageCosBase, public ObIStorageWriter class ObStorageCosMultiPartWriter: public ObStorageCosBase, public ObIStorageMultiPartWriter
{ {
public: public:
ObStorageCosMultiPartWriter(); ObStorageCosMultiPartWriter();
@ -172,8 +172,9 @@ public:
int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info); int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info);
int write(const char *buf, const int64_t size); int write(const char *buf, const int64_t size);
int pwrite(const char *buf, const int64_t size, const int64_t offset); int pwrite(const char *buf, const int64_t size, const int64_t offset);
virtual int complete() override;
virtual int abort() override;
int close(); int close();
int cleanup();
int64_t get_length() const { return file_length_; } int64_t get_length() const { return file_length_; }
virtual bool is_opened() const { return is_opened_; } virtual bool is_opened() const { return is_opened_; }

View File

@ -1173,5 +1173,65 @@ int ObStorageFileMultiPartWriter::pwrite(const char *buf, const int64_t size, co
return write(buf, size); return write(buf, size);
} }
int ObStorageFileMultiPartWriter::complete()
{
int ret = OB_SUCCESS;
char errno_buf[OB_MAX_ERROR_MSG_LEN] = "";
if (!is_opened_) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "not opened", K(ret), K(fd_));
}
if (OB_SUCC(ret)) {
#ifdef ERRSIM
ret = OB_E(EventTable::EN_FILE_SYSTEM_RENAME_ERROR) OB_SUCCESS;
if (OB_FAIL(ret)) {
has_error_ = true;
}
#endif
if (has_error_) {
STORAGE_LOG(WARN, "multipart writer has error, skip complete",
KCSTRING(path_), KCSTRING(real_path_));
} else if (0 != ::rename(path_, real_path_)) {
convert_io_error(errno, ret);
STORAGE_LOG(WARN, "failed to complete", K(ret), KCSTRING(real_path_),
KCSTRING(path_), K(errno), "errno", strerror_r(errno, errno_buf, sizeof(errno_buf)));
} else {
STORAGE_LOG(INFO, "succeed to rename file after complete",
KCSTRING(path_), KCSTRING(real_path_));
}
}
return ret;
}
int ObStorageFileMultiPartWriter::abort()
{
int ret = OB_SUCCESS;
char errno_buf[OB_MAX_ERROR_MSG_LEN] = "";
if (!is_opened_) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "not opened", K(ret), K(fd_));
} else if (0 != ::remove(path_)) {
convert_io_error(errno, ret);
STORAGE_LOG(WARN, "failed to abort(remove tmp file)", K(ret), KCSTRING(path_),
K(errno), "errno", strerror_r(errno, errno_buf, sizeof(errno_buf)));
}
return ret;
}
int ObStorageFileMultiPartWriter::close()
{
int ret = OB_SUCCESS;
if (!is_opened_) {
ret = OB_NOT_INIT;
STORAGE_LOG(WARN, "not opened", K(ret), K_(fd));
} else if (OB_FAIL(ObStorageFileBaseWriter::close())) {
STORAGE_LOG(WARN, "failed to do close", K(ret), KCSTRING(path_), KCSTRING(real_path_));
}
return ret;
}
}//common }//common
}//oceanbase }//oceanbase

View File

@ -103,8 +103,9 @@ public:
virtual ~ObStorageFileWriter(); virtual ~ObStorageFileWriter();
virtual int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info = NULL); virtual int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info = NULL);
virtual int close() override; virtual int close() override;
private: protected:
char real_path_[OB_MAX_URI_LENGTH]; char real_path_[OB_MAX_URI_LENGTH];
private:
DISALLOW_COPY_AND_ASSIGN(ObStorageFileWriter); DISALLOW_COPY_AND_ASSIGN(ObStorageFileWriter);
}; };
@ -127,12 +128,33 @@ private:
DISALLOW_COPY_AND_ASSIGN(ObStorageFileAppender); DISALLOW_COPY_AND_ASSIGN(ObStorageFileAppender);
}; };
class ObStorageFileMultiPartWriter : public ObStorageFileWriter class ObStorageFileMultiPartWriter : public ObStorageFileWriter, public ObIStorageMultiPartWriter
{ {
public: public:
ObStorageFileMultiPartWriter() {} ObStorageFileMultiPartWriter() {}
virtual ~ObStorageFileMultiPartWriter() {} virtual ~ObStorageFileMultiPartWriter() {}
virtual int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info) override
{
return ObStorageFileWriter::open(uri, storage_info);
}
virtual int write(const char *buf, const int64_t size) override
{
return ObStorageFileWriter::write(buf, size);
}
virtual int64_t get_length() const override
{
return ObStorageFileWriter::get_length();
}
virtual bool is_opened() const override
{
return ObStorageFileWriter::is_opened();
}
virtual int pwrite(const char *buf, const int64_t size, const int64_t offset) override; virtual int pwrite(const char *buf, const int64_t size, const int64_t offset) override;
virtual int complete() override;
virtual int abort() override;
virtual int close() override;
private: private:
DISALLOW_COPY_AND_ASSIGN(ObStorageFileMultiPartWriter); DISALLOW_COPY_AND_ASSIGN(ObStorageFileMultiPartWriter);

View File

@ -213,39 +213,40 @@ static void convert_io_error(aos_status_t *aos_ret, int &ob_errcode)
ob_errcode = OB_OSS_ERROR; ob_errcode = OB_OSS_ERROR;
} else if (!aos_status_is_ok(aos_ret)) { } else if (!aos_status_is_ok(aos_ret)) {
switch (aos_ret->code) { switch (aos_ret->code) {
case OSS_PERMISSION_DENIED: {
case OSS_PERMISSION_DENIED: {
ob_errcode = OB_BACKUP_PERMISSION_DENIED; ob_errcode = OB_BACKUP_PERMISSION_DENIED;
break; break;
}
case OSS_OBJECT_NOT_EXIST: {
ob_errcode = OB_BACKUP_FILE_NOT_EXIST;
break;
}
case OSS_OBJECT_PWRITE_OFFSET_NOT_MATH: {
ob_errcode = OB_BACKUP_PWRITE_OFFSET_NOT_MATCH;
break;
}
case OSS_LIMIT_EXCEEDED: {
ob_errcode = OB_IO_LIMIT;
break;
}
case OSS_BAD_REQUEST: {
if (0 == STRCMP("InvalidDigest", aos_ret->error_code)) {
ob_errcode = OB_CHECKSUM_ERROR;
} else {
ob_errcode = OB_OSS_ERROR;
} }
}
default: { case OSS_OBJECT_NOT_EXIST: {
ob_errcode = OB_OSS_ERROR; ob_errcode = OB_BACKUP_FILE_NOT_EXIST;
} break;
} // end swtich }
case OSS_OBJECT_PWRITE_OFFSET_NOT_MATH: {
ob_errcode = OB_BACKUP_PWRITE_OFFSET_NOT_MATCH;
break;
}
case OSS_LIMIT_EXCEEDED: {
ob_errcode = OB_IO_LIMIT;
break;
}
case OSS_BAD_REQUEST: {
if (0 == STRCMP("InvalidDigest", aos_ret->error_code)) {
ob_errcode = OB_CHECKSUM_ERROR;
} else {
ob_errcode = OB_OSS_ERROR;
}
break;
}
default: {
ob_errcode = OB_OSS_ERROR;
break;
}
} // end swtich
} }
} }
@ -762,11 +763,6 @@ int ObStorageOssMultiPartWriter::write(const char * buf,const int64_t size)
if (base_buf_pos_ == OSS_BASE_BUFFER_SIZE) { if (base_buf_pos_ == OSS_BASE_BUFFER_SIZE) {
if (OB_FAIL(write_single_part())) { if (OB_FAIL(write_single_part())) {
OB_LOG(WARN, "write file error", K(bucket_), K(object_), K(ret)); OB_LOG(WARN, "write file error", K(bucket_), K(object_), K(ret));
int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K(upload_id_.data));
}
} else { } else {
base_buf_pos_ = 0; base_buf_pos_ = 0;
} }
@ -883,10 +879,9 @@ int ObStorageOssMultiPartWriter::write_single_part()
return ret; return ret;
} }
int ObStorageOssMultiPartWriter::close() int ObStorageOssMultiPartWriter::complete()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;
const int64_t start_time = ObTimeUtility::current_time(); const int64_t start_time = ObTimeUtility::current_time();
if (OB_UNLIKELY(!is_inited())) { if (OB_UNLIKELY(!is_inited())) {
@ -894,22 +889,17 @@ int ObStorageOssMultiPartWriter::close()
OB_LOG(WARN, "oss client not inited", K(ret)); OB_LOG(WARN, "oss client not inited", K(ret));
} else if (OB_UNLIKELY(!is_opened_)) { } else if (OB_UNLIKELY(!is_opened_)) {
ret = OB_OSS_ERROR; ret = OB_OSS_ERROR;
OB_LOG(WARN, "oss writer cannot close before it is opened", K(ret)); OB_LOG(WARN, "oss multipart writer cannot close before it is opened", K(ret));
} else if (0 != base_buf_pos_) {//base_buf has data } else if (0 != base_buf_pos_) {//base_buf has data
if(OB_SUCCESS != (ret = write_single_part())) { if (OB_FAIL(write_single_part())) {
OB_LOG(WARN, "write the last size to oss error", OB_LOG(WARN, "write the last size to oss error",
K_(base_buf_pos), K_(bucket), K_(object), K(ret)); K_(base_buf_pos), K_(bucket), K_(object), K(ret));
ret = OB_OSS_ERROR;
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K(upload_id_.data));
}
} else { } else {
base_buf_pos_ = 0; base_buf_pos_ = 0;
} }
} }
if (OB_SUCCESS == ret) { if (OB_SUCC(ret)) {
aos_string_t bucket; aos_string_t bucket;
aos_string_t object; aos_string_t object;
aos_str_set(&bucket, bucket_.ptr()); aos_str_set(&bucket, bucket_.ptr());
@ -987,30 +977,34 @@ int ObStorageOssMultiPartWriter::close()
} }
} }
if (OB_FAIL(ret) && OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K(upload_id_.data));
}
} }
reset();
const int64_t total_cost_time = ObTimeUtility::current_time() - start_time; const int64_t total_cost_time = ObTimeUtility::current_time() - start_time;
if (total_cost_time > 3 * 1000 * 1000) { if (total_cost_time > 3 * 1000 * 1000) {
OB_LOG_RET(WARN, OB_ERR_TOO_MUCH_TIME, "oss writer close cost too much time", K(total_cost_time), K(ret)); OB_LOG_RET(WARN, OB_ERR_TOO_MUCH_TIME, "oss writer complete cost too much time",
K(total_cost_time), K(ret));
} }
return ret; return ret;
} }
int ObStorageOssMultiPartWriter::cleanup() int ObStorageOssMultiPartWriter::close()
{
int ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard;
reset();
return ret;
}
int ObStorageOssMultiPartWriter::abort()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
if (OB_UNLIKELY(!is_inited())) { if (OB_UNLIKELY(!is_inited())) {
ret = OB_NOT_INIT; ret = OB_NOT_INIT;
OB_LOG(WARN, "oss client not inited", K(ret)); OB_LOG(WARN, "oss client not inited", K(ret));
} else if(OB_UNLIKELY(!is_opened_)) { } else if (OB_UNLIKELY(!is_opened_)) {
ret = OB_OSS_ERROR; ret = OB_OSS_ERROR;
OB_LOG(WARN, "oss writer cannot cleanup before it is opened", K(ret)); OB_LOG(WARN, "oss multipart writer cannot abort before it is opened", K(ret));
} else { } else {
aos_string_t bucket; aos_string_t bucket;
aos_string_t object; aos_string_t object;
@ -1022,7 +1016,7 @@ int ObStorageOssMultiPartWriter::cleanup()
if (OB_ISNULL(aos_ret = oss_abort_multipart_upload(oss_option_, &bucket, &object, if (OB_ISNULL(aos_ret = oss_abort_multipart_upload(oss_option_, &bucket, &object,
&upload_id_, &resp_headers)) || !aos_status_is_ok(aos_ret)) { &upload_id_, &resp_headers)) || !aos_status_is_ok(aos_ret)) {
convert_io_error(aos_ret, ret); convert_io_error(aos_ret, ret);
OB_LOG(WARN, "Abort the multipart error", K(bucket_), K(object_), K(ret)); OB_LOG(WARN, "Abort the multipart error", K_(bucket), K_(object), K(ret));
} }
} }
return ret; return ret;

View File

@ -162,7 +162,7 @@ private:
DISALLOW_COPY_AND_ASSIGN(ObStorageOssWriter); DISALLOW_COPY_AND_ASSIGN(ObStorageOssWriter);
}; };
class ObStorageOssMultiPartWriter: public ObStorageOssBase, public ObIStorageWriter class ObStorageOssMultiPartWriter: public ObStorageOssBase, public ObIStorageMultiPartWriter
{ {
public: public:
@ -171,8 +171,9 @@ public:
int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info); int open(const common::ObString &uri, common::ObObjectStorageInfo *storage_info);
int write(const char *buf,const int64_t size); int write(const char *buf,const int64_t size);
int pwrite(const char *buf, const int64_t size, const int64_t offset); int pwrite(const char *buf, const int64_t size, const int64_t offset);
virtual int complete() override;
virtual int abort() override;
int close(); int close();
int cleanup();
int64_t get_length() const { return file_length_; } int64_t get_length() const { return file_length_; }
virtual bool is_opened() const { return is_opened_; } virtual bool is_opened() const { return is_opened_; }

View File

@ -599,6 +599,7 @@ static void convert_http_error(const Aws::S3::S3Error &s3_err, int &ob_errcode)
} }
default: { default: {
ob_errcode = OB_S3_ERROR; ob_errcode = OB_S3_ERROR;
break;
} }
} }
} }
@ -624,6 +625,7 @@ static void convert_io_error(const Aws::S3::S3Error &s3_err, int &ob_errcode)
} }
default: { default: {
convert_http_error(s3_err, ob_errcode); convert_http_error(s3_err, ob_errcode);
break;
} }
} }
} }
@ -1838,7 +1840,6 @@ void ObStorageS3MultiPartWriter::reset()
int ObStorageS3MultiPartWriter::open_(const ObString &uri, ObObjectStorageInfo *storage_info) int ObStorageS3MultiPartWriter::open_(const ObString &uri, ObObjectStorageInfo *storage_info)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;
if (OB_UNLIKELY(is_opened_)) { if (OB_UNLIKELY(is_opened_)) {
ret = OB_S3_ERROR; ret = OB_S3_ERROR;
@ -1876,10 +1877,6 @@ int ObStorageS3MultiPartWriter::open_(const ObString &uri, ObObjectStorageInfo *
file_length_ = 0; file_length_ = 0;
is_opened_ = true; is_opened_ = true;
} }
if (OB_FAIL(ret) && OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
} }
} }
@ -1892,7 +1889,6 @@ int ObStorageS3MultiPartWriter::open_(const ObString &uri, ObObjectStorageInfo *
int ObStorageS3MultiPartWriter::write_(const char *buf, const int64_t size) int ObStorageS3MultiPartWriter::write_(const char *buf, const int64_t size)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
int64_t fill_size = 0; int64_t fill_size = 0;
int64_t buf_pos = 0; int64_t buf_pos = 0;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;
@ -1912,10 +1908,6 @@ int ObStorageS3MultiPartWriter::write_(const char *buf, const int64_t size)
if (base_buf_pos_ == S3_MULTIPART_UPLOAD_BUFFER_SIZE) { if (base_buf_pos_ == S3_MULTIPART_UPLOAD_BUFFER_SIZE) {
if (OB_FAIL(write_single_part_())) { if (OB_FAIL(write_single_part_())) {
OB_LOG(WARN, "failed to write single s3 part", K(ret), K_(bucket), K_(object)); OB_LOG(WARN, "failed to write single s3 part", K(ret), K_(bucket), K_(object));
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
} else { } else {
base_buf_pos_ = 0; base_buf_pos_ = 0;
} }
@ -1934,21 +1926,16 @@ int ObStorageS3MultiPartWriter::pwrite_(const char *buf, const int64_t size, con
return write(buf, size); return write(buf, size);
} }
int ObStorageS3MultiPartWriter::close_() int ObStorageS3MultiPartWriter::complete_()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;
if (OB_UNLIKELY(!is_opened_)) { if (OB_UNLIKELY(!is_opened_)) {
ret = OB_NOT_INIT; ret = OB_NOT_INIT;
OB_LOG(WARN, "s3 multipart writer cannot close before it is opened", K(ret)); OB_LOG(WARN, "s3 multipart writer cannot compelete before it is opened", K(ret));
} else if (base_buf_pos_ > 0) { } else if (base_buf_pos_ > 0) {
if (OB_FAIL(write_single_part_())) { if (OB_FAIL(write_single_part_())) {
OB_LOG(WARN, "failed to upload last part into s3", K(ret), K_(base_buf_pos)); OB_LOG(WARN, "failed to upload last part into s3", K(ret), K_(base_buf_pos));
if (OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
} else { } else {
base_buf_pos_ = 0; base_buf_pos_ = 0;
} }
@ -2002,16 +1989,19 @@ int ObStorageS3MultiPartWriter::close_()
K(ret), K_(bucket), K_(object)); K(ret), K_(bucket), K_(object));
} }
if (OB_FAIL(ret) && OB_TMP_FAIL(cleanup())) {
OB_LOG(WARN, "fail to abort multiupload", K(ret), K(tmp_ret), K_(upload_id));
}
} }
return ret;
}
int ObStorageS3MultiPartWriter::close_()
{
int ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard;
reset(); reset();
return ret; return ret;
} }
int ObStorageS3MultiPartWriter::cleanup() int ObStorageS3MultiPartWriter::abort_()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObExternalIOCounterGuard io_guard; ObExternalIOCounterGuard io_guard;

View File

@ -499,7 +499,7 @@ private:
DISALLOW_COPY_AND_ASSIGN(ObStorageS3AppendWriter); DISALLOW_COPY_AND_ASSIGN(ObStorageS3AppendWriter);
}; };
class ObStorageS3MultiPartWriter : public ObStorageS3Base, public ObIStorageWriter class ObStorageS3MultiPartWriter : public ObStorageS3Base, public ObIStorageMultiPartWriter
{ {
public: public:
ObStorageS3MultiPartWriter(); ObStorageS3MultiPartWriter();
@ -518,6 +518,14 @@ public:
{ {
return do_safely(&ObStorageS3MultiPartWriter::pwrite_, this, buf, size, offset); return do_safely(&ObStorageS3MultiPartWriter::pwrite_, this, buf, size, offset);
} }
virtual int complete() override
{
return do_safely(&ObStorageS3MultiPartWriter::complete_, this);
}
virtual int abort() override
{
return do_safely(&ObStorageS3MultiPartWriter::abort_, this);
}
virtual int close() override virtual int close() override
{ {
return do_safely(&ObStorageS3MultiPartWriter::close_, this); return do_safely(&ObStorageS3MultiPartWriter::close_, this);
@ -525,12 +533,12 @@ public:
virtual int64_t get_length() const override { return file_length_; } virtual int64_t get_length() const override { return file_length_; }
virtual bool is_opened() const override { return is_opened_; } virtual bool is_opened() const override { return is_opened_; }
int cleanup();
private: private:
int open_(const ObString &uri, ObObjectStorageInfo *storage_info); int open_(const ObString &uri, ObObjectStorageInfo *storage_info);
int write_(const char *buf, const int64_t size); int write_(const char *buf, const int64_t size);
int pwrite_(const char *buf, const int64_t size, const int64_t offset); int pwrite_(const char *buf, const int64_t size, const int64_t offset);
int complete_();
int abort_();
int close_(); int close_();
int write_single_part_(); int write_single_part_();

View File

@ -548,6 +548,10 @@ int TestCommonStorage::check_basic_multipartwriter()
} }
} }
if (OB_SUCC(ret) && OB_FAIL(multiwriter.complete())) {
OB_LOG(WARN, "fail to complete", K(ret));
}
int tmp_ret = OB_SUCCESS; int tmp_ret = OB_SUCCESS;
if (OB_TMP_FAIL(multiwriter.close())) { if (OB_TMP_FAIL(multiwriter.close())) {
ret = (OB_SUCCESS != ret) ? ret : tmp_ret; ret = (OB_SUCCESS != ret) ? ret : tmp_ret;

View File

@ -1346,11 +1346,22 @@ TEST_F(TestObjectStorage, test_multipart_write)
write_buf[content_size - 1] = '\0'; write_buf[content_size - 1] = '\0';
ASSERT_EQ(OB_SUCCESS, databuff_printf(uri, sizeof(uri), "%s/test_multipart", dir_uri)); ASSERT_EQ(OB_SUCCESS, databuff_printf(uri, sizeof(uri), "%s/test_multipart", dir_uri));
{
// test abort
ObStorageMultiPartWriter writer;
ASSERT_EQ(OB_SUCCESS, writer.open(uri, &info_base));
ASSERT_EQ(OB_SUCCESS, writer.abort());
ASSERT_NE(OB_SUCCESS, writer.complete());
ASSERT_EQ(OB_SUCCESS, writer.close());
}
ObStorageMultiPartWriter writer; ObStorageMultiPartWriter writer;
// ObStorageWriter writer; // ObStorageWriter writer;
ASSERT_EQ(OB_SUCCESS, writer.open(uri, &info_base)); ASSERT_EQ(OB_SUCCESS, writer.open(uri, &info_base));
ASSERT_EQ(OB_SUCCESS, writer.write(write_buf, content_size)); ASSERT_EQ(OB_SUCCESS, writer.write(write_buf, content_size));
ASSERT_EQ(content_size, writer.get_length()); ASSERT_EQ(content_size, writer.get_length());
ASSERT_EQ(OB_SUCCESS, writer.complete());
ASSERT_EQ(OB_SUCCESS, writer.close()); ASSERT_EQ(OB_SUCCESS, writer.close());
OB_LOG(INFO, "-----------------------------------------------------------------------------"); OB_LOG(INFO, "-----------------------------------------------------------------------------");

View File

@ -303,6 +303,18 @@ int ObLocalDevice::open(const char *pathname, const int flags, const mode_t mode
return ret; return ret;
} }
int ObLocalDevice::complete(const ObIOFd &fd)
{
UNUSED(fd);
return OB_NOT_SUPPORTED;
}
int ObLocalDevice::abort(const ObIOFd &fd)
{
UNUSED(fd);
return OB_NOT_SUPPORTED;
}
int ObLocalDevice::close(const ObIOFd &fd) int ObLocalDevice::close(const ObIOFd &fd)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;

View File

@ -70,6 +70,8 @@ public:
//file/dir interfaces //file/dir interfaces
virtual int open(const char *pathname, const int flags, const mode_t mode, common::ObIOFd &fd, common::ObIODOpts *opts = NULL) override; virtual int open(const char *pathname, const int flags, const mode_t mode, common::ObIOFd &fd, common::ObIODOpts *opts = NULL) override;
virtual int complete(const ObIOFd &fd) override;
virtual int abort(const ObIOFd &fd) override;
virtual int close(const common::ObIOFd &fd) override; virtual int close(const common::ObIOFd &fd) override;
virtual int mkdir(const char *pathname, mode_t mode) override; virtual int mkdir(const char *pathname, mode_t mode) override;
virtual int rmdir(const char *pathname) override; virtual int rmdir(const char *pathname) override;

View File

@ -402,6 +402,18 @@ int ObBackupDataCtx::close()
} else if (OB_FAIL(file_write_ctx_.close())) { } else if (OB_FAIL(file_write_ctx_.close())) {
LOG_WARN("failed to close file writer", K(ret)); LOG_WARN("failed to close file writer", K(ret));
} }
if (OB_SUCC(ret)) {
if (OB_FAIL(dev_handle_->complete(io_fd_))) {
LOG_WARN("fail to complete multipart upload", K(ret), K_(dev_handle), K_(io_fd));
}
} else {
if (OB_TMP_FAIL(dev_handle_->abort(io_fd_))) {
ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to abort multipart upload", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
}
}
if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) { if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
ret = COVER_SUCC(tmp_ret); ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd)); LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));

View File

@ -443,6 +443,17 @@ int ObExternTabletMetaWriter::close()
} else if (OB_FAIL(file_write_ctx_.close())) { } else if (OB_FAIL(file_write_ctx_.close())) {
LOG_WARN("failed to close file writer", K(ret)); LOG_WARN("failed to close file writer", K(ret));
} }
if (OB_SUCC(ret)) {
if (OB_FAIL(dev_handle_->complete(io_fd_))) {
LOG_WARN("fail to complete multipart upload", K(ret), K_(dev_handle), K_(io_fd));
}
} else {
if (OB_TMP_FAIL(dev_handle_->abort(io_fd_))) {
ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to abort multipart upload", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
}
}
if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) { if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
ret = COVER_SUCC(tmp_ret); ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd)); LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));

View File

@ -827,6 +827,16 @@ int ObBackupMacroBlockIndexMerger::merge_index()
if (OB_NOT_NULL(fuser)) { if (OB_NOT_NULL(fuser)) {
ObLSBackupFactory::free(fuser); ObLSBackupFactory::free(fuser);
} }
if (OB_SUCC(ret)) {
if (OB_FAIL(dev_handle_->complete(io_fd_))) {
LOG_WARN("fail to complete multipart upload", K(ret), K_(dev_handle), K_(io_fd));
}
} else {
if (OB_TMP_FAIL(dev_handle_->abort(io_fd_))) {
ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to abort multipart upload", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
}
}
if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) { if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
ret = COVER_SUCC(tmp_ret); ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd)); LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
@ -1338,6 +1348,16 @@ int ObBackupMetaIndexMerger::merge_index()
} }
} }
} }
if (OB_SUCC(ret)) {
if (OB_FAIL(dev_handle_->complete(io_fd_))) {
LOG_WARN("fail to complete multipart upload", K(ret), K_(dev_handle), K_(io_fd));
}
} else {
if (OB_TMP_FAIL(dev_handle_->abort(io_fd_))) {
ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to abort multipart upload", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));
}
}
if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) { if (OB_TMP_FAIL(util.close_device_and_fd(dev_handle_, io_fd_))) {
ret = COVER_SUCC(tmp_ret); ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd)); LOG_WARN("fail to close device or fd", K(ret), K(tmp_ret), K_(dev_handle), K_(io_fd));

View File

@ -5559,7 +5559,16 @@ int ObLSBackupComplementLogTask::transfer_clog_file_(const ObBackupPath &src_pat
break; break;
} }
} }
if (OB_SUCC(ret)) {
if (OB_FAIL(device_handle->complete(fd))) {
LOG_WARN("fail to complete multipart upload", K(ret), K(device_handle), K(fd));
}
} else {
if (OB_TMP_FAIL(device_handle->abort(fd))) {
ret = COVER_SUCC(tmp_ret);
LOG_WARN("fail to abort multipart upload", K(ret), K(tmp_ret), K(device_handle), K(fd));
}
}
if (OB_SUCCESS != (tmp_ret = util.close_device_and_fd(device_handle, fd))) { if (OB_SUCCESS != (tmp_ret = util.close_device_and_fd(device_handle, fd))) {
LOG_WARN("fail to close file", K(ret), K_(backup_dest), K(dst_path)); LOG_WARN("fail to close file", K(ret), K_(backup_dest), K(dst_path));
ret = OB_SUCCESS == ret ? tmp_ret : ret; ret = OB_SUCCESS == ret ? tmp_ret : ret;

View File

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

View File

@ -333,6 +333,10 @@ int init_task_executor(const char *base_uri,
if (OB_FAIL(alloc_executor<DelTaskExecutor>(executor, attr))) { if (OB_FAIL(alloc_executor<DelTaskExecutor>(executor, attr))) {
OB_LOG(WARN, "fail to alloc and construct executor", K(ret), K(config)); 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 { } else {
ret = OB_ERR_UNEXPECTED; ret = OB_ERR_UNEXPECTED;
OB_LOG(WARN, "unknown benchmark task type", K(ret), K(config)); 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, if (OB_FAIL(device_handle->pwrite(fd, cur_offset, cur_part_size,
write_buf_, actual_write_size))) { write_buf_, actual_write_size))) {
OB_LOG(WARN, "fail to upload part", 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 { } else {
cur_offset += cur_part_size; cur_offset += cur_part_size;
} }
} }
const int64_t close_start_time_us = ObTimeUtility::current_time(); const int64_t close_start_time_us = ObTimeUtility::current_time();
if (FAILEDx(adapter.close_device_and_fd(device_handle, fd))) { if (FAILEDx(device_handle->complete(fd))) {
OB_LOG(WARN, "fail to close device handle", K(ret), K_(base_uri)); OB_LOG(WARN, "fail to complete multipart writer", K(ret), K_(base_uri), K(fd));
} else {
metrics_.close_time_ms_map_.log_entry(close_start_time_us);
} }
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)) { if (OB_SUCC(ret)) {
metrics_.operation_num_++; metrics_.operation_num_++;
@ -781,5 +788,57 @@ int DelTaskExecutor::execute()
return ret; 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 } //tools
} //oceanbase } //oceanbase

View File

@ -31,6 +31,7 @@ enum BenchmarkTaskType
BENCHMARK_TASK_MULTIPART_WRITE = 2, BENCHMARK_TASK_MULTIPART_WRITE = 2,
BENCHMARK_TASK_READ = 3, BENCHMARK_TASK_READ = 3,
BENCHMARK_TASK_DEL = 4, BENCHMARK_TASK_DEL = 4,
BENCHMARK_TASK_IS_EXIST = 5,
BENCHMARK_TASK_MAX_TYPE BENCHMARK_TASK_MAX_TYPE
}; };
@ -193,6 +194,19 @@ private:
bool is_adaptive_; 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 tools
} //namespace oceanbase } //namespace oceanbase

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 ObAdminTestIODeviceExecutor::test_multipart_upload_check_file_(const char* check_file_dir)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObBackupIoAdapter util; ObBackupIoAdapter util;
share::ObBackupStorageInfo storage_info; share::ObBackupStorageInfo storage_info;
ObIOFd fd; 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)); 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))) { } 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)); 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))) { } else if (OB_FAIL(device_handle->complete(fd))) {
STORAGE_LOG(WARN, "fail to close device and fd", K(ret), KP(device_handle), K(fd)); STORAGE_LOG(WARN, "fail to complete multipart upload", K(ret), K(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)); if (OB_SUCCESS != ret) {
} else if (!is_exist) { if (OB_TMP_FAIL(device_handle->abort(fd))) {
ret = OB_ERR_UNEXPECTED; ret = COVER_SUCC(tmp_ret);
STORAGE_LOG(WARN, "normal check file does not exist!", K(ret), K(check_file_path)); STORAGE_LOG(WARN, "fail to abort multipart upload", K(ret), K(tmp_ret), K(device_handle), K(fd));
} 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) { if (OB_TMP_FAIL(util.close_device_and_fd(device_handle, fd))) {
ret = OB_ERR_UNEXPECTED; ret = COVER_SUCC(tmp_ret);
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)); STORAGE_LOG(WARN, "fail to close device and fd", K(ret), K(tmp_ret), K(device_handle), K(fd));
} else if (OB_ISNULL(read_file_buf = reinterpret_cast<char*>(allocator_.alloc(check_file_len)))) { }
ret = OB_ALLOCATE_MEMORY_FAILED; if (OB_SUCC(ret)) {
STORAGE_LOG(WARN, "failed to allocate buf", K(ret), K(check_file_path), K(check_file_len)); if (OB_FAIL(util.is_exist(check_file_path, &storage_info, is_exist))) {
} 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 check if normal check file is exist", K(ret), K(check_file_path));
STORAGE_LOG(WARN, "failed to read check file", K(ret), K(check_file_path), K(check_file_len)); } else if (!is_exist) {
} else if (read_size != check_file_len) { ret = OB_ERR_UNEXPECTED;
ret = OB_ERR_UNEXPECTED; STORAGE_LOG(WARN, "normal check file does not exist!", K(ret), K(check_file_path));
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(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; return ret;
} }
@ -406,6 +420,7 @@ int ObAdminTestIODeviceExecutor::test_clean_backup_file_()
int ObAdminTestIODeviceExecutor::test_backup_data_() int ObAdminTestIODeviceExecutor::test_backup_data_()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
ObBackupIoAdapter util; ObBackupIoAdapter util;
share::ObBackupStorageInfo storage_info; share::ObBackupStorageInfo storage_info;
ObIOFd fd; ObIOFd fd;
@ -455,10 +470,22 @@ int ObAdminTestIODeviceExecutor::test_backup_data_()
total_size += buf_size; 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_SUCC(ret)) {
if (OB_FAIL(util.close_device_and_fd(device_handle, fd))) { if (OB_FAIL(util.get_file_length(data_file_path, &storage_info, read_size))) {
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))) {
STORAGE_LOG(WARN, "fail to get multipart upload file length", K(ret), K(data_file_path)); STORAGE_LOG(WARN, "fail to get multipart upload file length", K(ret), K(data_file_path));
} else if (read_size != total_size) { } else if (read_size != total_size) {
ret = OB_ERR_UNEXPECTED; ret = OB_ERR_UNEXPECTED;