You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pegasus.apache.org by wa...@apache.org on 2022/08/30 13:31:33 UTC
[incubator-pegasus] branch backup_restore-dev updated: feat(backup): 6. meta handle backup_response during checkpointing (#1129)
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch backup_restore-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
The following commit(s) were added to refs/heads/backup_restore-dev by this push:
new c06628b95 feat(backup): 6. meta handle backup_response during checkpointing (#1129)
c06628b95 is described below
commit c06628b95e2fdb5441e75b064ebaf3202ba25782
Author: HeYuchen <he...@xiaomi.com>
AuthorDate: Tue Aug 30 21:31:28 2022 +0800
feat(backup): 6. meta handle backup_response during checkpointing (#1129)
---
src/rdsn/src/meta/meta_backup_engine.cpp | 166 ++++++++++++---------
src/rdsn/src/meta/meta_backup_engine.h | 22 ++-
src/rdsn/src/meta/test/meta_backup_engine_test.cpp | 153 +++++++++++++++++++
3 files changed, 263 insertions(+), 78 deletions(-)
diff --git a/src/rdsn/src/meta/meta_backup_engine.cpp b/src/rdsn/src/meta/meta_backup_engine.cpp
index 03bd56b6a..f2fbaaed7 100644
--- a/src/rdsn/src/meta/meta_backup_engine.cpp
+++ b/src/rdsn/src/meta/meta_backup_engine.cpp
@@ -179,6 +179,86 @@ void meta_backup_engine::backup_app_partition(const gpid &pid)
});
}
+// ThreadPool: THREAD_POOL_DEFAULT
+void meta_backup_engine::on_backup_reply(const error_code err,
+ const backup_response &response,
+ const gpid &pid,
+ const rpc_address &primary)
+{
+ {
+ zauto_read_lock l(_lock);
+ if (_is_backup_failed) {
+ derror_f("partition[{}] handle backup failed", pid);
+ return;
+ }
+
+ // TODO(heyuchen): check if backup canceled
+ }
+
+ auto rep_error = err == ERR_OK ? response.err : err;
+ if (rep_error != ERR_OK) {
+ derror_f(
+ "backup_id({}): receive backup response for partition {} from server {}, error = {}",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ primary.to_string(),
+ rep_error);
+ handle_replica_backup_failed(pid.get_app_id());
+ return;
+ }
+
+ if (response.backup_id != _cur_backup.backup_id) {
+ dwarn_f("backup_id({}): receive outdated backup response(backup_id={}) for partition {} "
+ "from server {}, ignore it",
+ _cur_backup.backup_id,
+ response.backup_id,
+ pid.to_string(),
+ primary.to_string());
+ retry_backup(pid);
+ return;
+ }
+
+ if (response.__isset.checkpoint_upload_err) {
+ auto type = response.status == backup_status::UPLOADING ? "upload" : "checkpoint";
+ derror_f("backup_id({}): receive backup response for partition {} from server {}, meet {} "
+ "error = {}",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ primary.to_string(),
+ type,
+ response.checkpoint_upload_err);
+ handle_replica_backup_failed(pid.get_app_id());
+ retry_backup(pid);
+ return;
+ }
+
+ if (response.status == backup_status::CHECKPOINTED) {
+ ddebug_f("backup_id({}): backup for partition {} from server {} finish checkpoint",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ primary.to_string());
+ {
+ zauto_write_lock l(_lock);
+ _backup_status[pid.get_partition_index()] = backup_status::UPLOADING;
+ }
+ if (check_partition_backup_status(backup_status::UPLOADING)) {
+ update_backup_item_on_remote_storage(backup_status::UPLOADING);
+ }
+ }
+
+ // TODO(heyuchen): handle other status
+
+ // backup is not finished, meta polling to send request
+ ddebug_f("backup_id({}): receive backup response for partition {} from server {}, "
+ "backup_status = {}, retry to send backup request.",
+ _cur_backup.backup_id,
+ pid.to_string(),
+ primary.to_string(),
+ dsn::enum_to_string(response.status));
+
+ retry_backup(pid);
+}
+
// ThreadPool: THREAD_POOL_DEFAULT
void meta_backup_engine::update_backup_item_on_remote_storage(backup_status::type new_status,
int64_t end_time)
@@ -204,26 +284,23 @@ void meta_backup_engine::update_backup_item_on_remote_storage(backup_status::typ
});
}
-// TODO(heyuchen): update following functions
-
-inline void meta_backup_engine::handle_replica_backup_failed(const backup_response &response,
- const gpid pid)
+// ThreadPool: THREAD_POOL_DEFAULT
+void meta_backup_engine::handle_replica_backup_failed(int32_t app_id)
{
- dcheck_eq(response.pid, pid);
- dcheck_eq(response.backup_id, _cur_backup.backup_id);
-
- derror_f("backup_id({}): backup for partition {} failed, response.err: {}",
- _cur_backup.backup_id,
- pid.to_string(),
- response.err.to_string());
zauto_write_lock l(_lock);
- // if one partition fail, the whole backup plan fail.
+ // if one partition fail, the whole backup process fail.
_is_backup_failed = true;
- _backup_status[pid.get_partition_index()] = backup_status::FAILED;
+ for (auto i = 0; i < _backup_status.size(); i++) {
+ _backup_status[i] = backup_status::FAILED;
+ retry_backup(gpid(app_id, i));
+ }
+ update_backup_item_on_remote_storage(backup_status::FAILED, dsn_now_ms());
}
-inline void meta_backup_engine::retry_backup(const dsn::gpid pid)
+// ThreadPool: THREAD_POOL_DEFAULT
+inline void meta_backup_engine::retry_backup(const gpid &pid)
{
+ FAIL_POINT_INJECT_F("meta_retry_backup", [](dsn::string_view) {});
tasking::enqueue(LPC_DEFAULT_CALLBACK,
&_tracker,
[this, pid]() { backup_app_partition(pid); },
@@ -231,66 +308,7 @@ inline void meta_backup_engine::retry_backup(const dsn::gpid pid)
std::chrono::seconds(1));
}
-void meta_backup_engine::on_backup_reply(const error_code err,
- const backup_response &response,
- const gpid pid,
- const rpc_address &primary)
-{
- {
- zauto_read_lock l(_lock);
- // if backup of some partition failed, we would not handle response from other partitions.
- if (_is_backup_failed) {
- return;
- }
- }
-
- // if backup completed, receive ERR_OK and
- // resp.progress=backup_constant::PROGRESS_FINISHED;
- // if backup failed, receive ERR_LOCAL_APP_FAILURE;
- // backup not completed in other cases.
- // see replica::on_cold_backup() for details.
-
- auto rep_error = err == ERR_OK ? response.err : err;
-
- if (rep_error == ERR_LOCAL_APP_FAILURE) {
- handle_replica_backup_failed(response, pid);
- return;
- }
-
- if (rep_error != ERR_OK) {
- derror_f("backup_id({}): backup request to server {} failed, error: {}, retry to "
- "send backup request.",
- _cur_backup.backup_id,
- primary.to_string(),
- rep_error.to_string());
- retry_backup(pid);
- return;
- };
-
- if (response.upload_progress == backup_constant::PROGRESS_FINISHED) {
- dcheck_eq(response.pid, pid);
- dcheck_eq(response.backup_id, _cur_backup.backup_id);
- ddebug_f("backup_id({}): backup for partition {} completed.",
- _cur_backup.backup_id,
- pid.to_string());
- {
- zauto_write_lock l(_lock);
- _backup_status[pid.get_partition_index()] = backup_status::SUCCEED;
- }
- complete_current_backup();
- return;
- }
-
- // backup is not finished, meta polling to send request
- ddebug_f("backup_id({}): receive backup response for partition {} from server {}, now "
- "progress {}, retry to send backup request.",
- _cur_backup.backup_id,
- pid.to_string(),
- primary.to_string(),
- response.upload_progress);
-
- retry_backup(pid);
-}
+// TODO(heyuchen): update following functions
void meta_backup_engine::write_backup_info()
{
diff --git a/src/rdsn/src/meta/meta_backup_engine.h b/src/rdsn/src/meta/meta_backup_engine.h
index 278726f2b..c6ae9fda1 100644
--- a/src/rdsn/src/meta/meta_backup_engine.h
+++ b/src/rdsn/src/meta/meta_backup_engine.h
@@ -73,9 +73,14 @@ public:
explicit meta_backup_engine(meta_service *meta_svc, bool is_periodic);
~meta_backup_engine();
- int64_t get_current_backup_id() const { return _cur_backup.backup_id; }
int32_t get_backup_app_id() const { return _cur_backup.app_id; }
+ int64_t get_backup_id() const
+ {
+ zauto_read_lock l(_lock);
+ return _cur_backup.backup_id;
+ }
+
backup_status::type get_backup_status() const
{
zauto_read_lock l(_lock);
@@ -106,10 +111,10 @@ private:
void backup_app_partition(const gpid &pid);
void on_backup_reply(error_code err,
const backup_response &response,
- gpid pid,
+ const gpid &pid,
const rpc_address &primary);
- void retry_backup(const dsn::gpid pid);
- void handle_replica_backup_failed(const backup_response &response, const gpid pid);
+ void retry_backup(const gpid &pid);
+ void handle_replica_backup_failed(int32_t app_id);
error_code write_backup_file(const std::string &remote_dir,
const std::string &file_name,
@@ -119,6 +124,15 @@ private:
void update_backup_item_on_remote_storage(backup_status::type new_status, int64_t end_time = 0);
+ bool check_partition_backup_status(backup_status::type expected_status) const
+ {
+ zauto_read_lock l(_lock);
+ return std::all_of(
+ _backup_status.begin(),
+ _backup_status.end(),
+ [expected_status](backup_status::type status) { return expected_status == status; });
+ }
+
std::string get_remote_storage_root() const
{
return meta_options::concat_path_unix_style(_meta_svc->get_cluster_root(), "backup");
diff --git a/src/rdsn/src/meta/test/meta_backup_engine_test.cpp b/src/rdsn/src/meta/test/meta_backup_engine_test.cpp
index 6feda94b1..de5b48f9b 100644
--- a/src/rdsn/src/meta/test/meta_backup_engine_test.cpp
+++ b/src/rdsn/src/meta/test/meta_backup_engine_test.cpp
@@ -71,12 +71,67 @@ public:
return _backup_engine->get_backup_status();
}
+ backup_status::type
+ on_backup_reply(const backup_response &resp, backup_status::type meta_status, bool mock_all)
+ {
+ mock_backup_context(resp.status, meta_status, mock_all);
+ _backup_engine->on_backup_reply(ERR_OK, resp, gpid(_app_id, INDEX), PRIMARY);
+ return _backup_engine->get_backup_status();
+ }
+
+ void mock_backup_context(backup_status::type resp_status,
+ backup_status::type old_status,
+ bool mock_all)
+ {
+ auto other_status = backup_status::UNINITIALIZED;
+ if (mock_all) {
+ if (resp_status == backup_status::CHECKPOINTED) {
+ other_status = backup_status::UPLOADING;
+ }
+ if (resp_status == backup_status::SUCCEED) {
+ other_status = backup_status::SUCCEED;
+ }
+ }
+ for (auto i = 0; i < PARTITION_COUNT; i++) {
+ if (i == INDEX) {
+ _backup_engine->_backup_status[i] = old_status;
+ } else {
+ _backup_engine->_backup_status[i] = other_status;
+ }
+ }
+ }
+
+ backup_response mock_backup_response(backup_status::type status,
+ int64_t backup_id,
+ error_code resp_err,
+ error_code checkpoint_err,
+ error_code upload_err,
+ int32_t upload_progress)
+ {
+ backup_response resp;
+ resp.err = resp_err;
+ resp.pid = gpid(_app_id, INDEX);
+ resp.backup_id = backup_id;
+ resp.status = status;
+ if (checkpoint_err != ERR_OK) {
+ resp.__set_checkpoint_upload_err(checkpoint_err);
+ }
+ if (upload_err != ERR_OK) {
+ resp.__set_checkpoint_upload_err(upload_err);
+ }
+ if (upload_progress > 0) {
+ resp.__set_upload_progress(upload_progress);
+ }
+ return resp;
+ }
+
protected:
const std::string APP_NAME = "backup_test";
const int32_t PARTITION_COUNT = 8;
const int32_t INDEX = 0;
const std::string PROVIDER = "local_service";
const std::string PATH = "unit_test";
+ const rpc_address PRIMARY = rpc_address("127.0.0.1", 10086);
int32_t _app_id;
std::shared_ptr<meta_backup_engine> _backup_engine;
};
@@ -103,5 +158,103 @@ TEST_F(meta_backup_engine_test, start_test)
}
}
+TEST_F(meta_backup_engine_test, on_backup_reply_succeed_test)
+{
+ fail::cfg("meta_retry_backup", "return()");
+ fail::cfg("meta_update_backup_item", "return()");
+
+ struct on_backup_reply_test
+ {
+ backup_status::type resp_status;
+ backup_status::type old_status;
+ bool mock_all;
+ backup_status::type expected_status;
+ bool in_progress;
+ } tests[] = {
+ {backup_status::CHECKPOINTING,
+ backup_status::CHECKPOINTING,
+ false,
+ backup_status::CHECKPOINTING,
+ true},
+ {backup_status::CHECKPOINTED,
+ backup_status::CHECKPOINTING,
+ false,
+ backup_status::CHECKPOINTING,
+ true},
+ {backup_status::CHECKPOINTED,
+ backup_status::CHECKPOINTING,
+ true,
+ backup_status::UPLOADING,
+ true},
+ // TODO(heyuchen): add other status cases
+ };
+
+ for (const auto &test : tests) {
+ init_backup(_app_id, test.old_status);
+ auto resp = mock_backup_response(
+ test.resp_status, _backup_engine->get_backup_id(), ERR_OK, ERR_OK, ERR_OK, 0);
+ auto status = on_backup_reply(resp, test.old_status, test.mock_all);
+ ASSERT_EQ(status, test.expected_status);
+ ASSERT_EQ(_backup_engine->is_in_progress(), test.in_progress);
+ }
+}
+
+TEST_F(meta_backup_engine_test, on_backup_reply_failed_test)
+{
+ fail::cfg("meta_retry_backup", "return()");
+ fail::cfg("meta_update_backup_item", "return()");
+
+ // case1. response error
+ {
+ init_backup(_app_id, backup_status::CHECKPOINTING);
+ auto resp = mock_backup_response(backup_status::CHECKPOINTED,
+ _backup_engine->get_backup_id(),
+ ERR_INVALID_STATE,
+ ERR_OK,
+ ERR_OK,
+ 0);
+ auto status = on_backup_reply(resp, backup_status::CHECKPOINTING, false);
+ ASSERT_EQ(status, backup_status::FAILED);
+ ASSERT_FALSE(_backup_engine->is_in_progress());
+ }
+
+ // case2. old backup_id
+ {
+ init_backup(_app_id, backup_status::UPLOADING);
+ auto resp = mock_backup_response(backup_status::UPLOADING, 0, ERR_OK, ERR_OK, ERR_OK, 0);
+ auto status = on_backup_reply(resp, backup_status::UPLOADING, false);
+ ASSERT_EQ(status, backup_status::UPLOADING);
+ ASSERT_TRUE(_backup_engine->is_in_progress());
+ }
+
+ // case3. checkpoint error
+ {
+ init_backup(_app_id, backup_status::CHECKPOINTING);
+ auto resp = mock_backup_response(backup_status::CHECKPOINTING,
+ _backup_engine->get_backup_id(),
+ ERR_OK,
+ ERR_FILE_OPERATION_FAILED,
+ ERR_OK,
+ 0);
+ auto status = on_backup_reply(resp, backup_status::CHECKPOINTING, true);
+ ASSERT_EQ(status, backup_status::FAILED);
+ ASSERT_FALSE(_backup_engine->is_in_progress());
+ }
+
+ // case4. upload error
+ {
+ init_backup(_app_id, backup_status::UPLOADING);
+ auto resp = mock_backup_response(backup_status::UPLOADING,
+ _backup_engine->get_backup_id(),
+ ERR_OK,
+ ERR_OK,
+ ERR_FS_INTERNAL,
+ 0);
+ auto status = on_backup_reply(resp, backup_status::UPLOADING, false);
+ ASSERT_EQ(status, backup_status::FAILED);
+ ASSERT_FALSE(_backup_engine->is_in_progress());
+ }
+}
+
} // namespace replication
} // namespace dsn
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org