hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
develop
oceanbase/mittest/palf_cluster/logservice/role_coordinator.cpp

748 lines
30 KiB
C++
Raw Permalink Blame History

/**
* Copyright (c) 2021 OceanBase
* OceanBase CE is licensed under Mulan PubL v2.
* You can use this software according to the terms and conditions of the Mulan PubL v2.
* You may obtain a copy of Mulan PubL v2 at:
* http://license.coscl.org.cn/MulanPubL-2.0
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PubL v2 for more details.
*/
#include "role_coordinator.h"
#include "common/ob_role.h"
#include "lib/ob_define.h"
#include "lib/ob_errno.h"
#include "lib/time/ob_time_utility.h"
#include "lib/utility/ob_macro_utils.h"
#include "lib/thread/thread_mgr.h"
#include "logservice/palf/log_define.h"
#include "share/ob_errno.h"
#include "share/ob_ls_id.h"
#include "share/ob_thread_define.h"
#include "share/rc/ob_tenant_base.h"
#include "storage/tx_storage/ob_ls_service.h"
#include "storage/tx_storage/ob_ls_handle.h"
#include "share/ob_occam_time_guard.h"
namespace oceanbase
{
using namespace common;
using namespace palf;
using namespace logservice;
using namespace share;
namespace palfcluster
{
RoleChangeEvent::RoleChangeEvent(const RoleChangeEventType &event_type,
const share::ObLSID &ls_id) : event_type_(event_type),
ls_id_(ls_id)
{
}
RoleChangeEvent::RoleChangeEvent(const RoleChangeEventType &event_type,
const share::ObLSID &ls_id,
const common::ObAddr &dst_addr) : event_type_(event_type),
ls_id_(ls_id),
dst_addr_(dst_addr)
{
}
bool RoleChangeEvent::is_valid() const
{
return RoleChangeEventType::INVALID_RC_EVENT_TYPE != event_type_
&& false != ls_id_.is_valid();
}
void RoleChangeEvent::reset()
{
event_type_ = RoleChangeEventType::INVALID_RC_EVENT_TYPE;
ls_id_.reset();
dst_addr_.reset();
}
bool RoleChangeEvent::operator==(const RoleChangeEvent &rhs) const
{
// for change leader event, we just check 'ls_id'.
return event_type_ == rhs.event_type_ && ls_id_ == rhs.ls_id_;
}
RoleChangeEventSet::RoleChangeEventSet()
{}
RoleChangeEventSet::~RoleChangeEventSet()
{}
int RoleChangeEventSet::insert(const RoleChangeEvent &event)
{
int ret = OB_SUCCESS;
int64_t free_idx = -1;
ObSpinLockGuard guard(lock_);
for (int64_t i = 0; i < MAX_ARRAY_SIZE; i++) {
if (event == events_[i]) {
ret = OB_ENTRY_EXIST;
}
}
for (int64_t i = 0; i < MAX_ARRAY_SIZE && -1 == free_idx && OB_SUCC(ret); i++) {
if (false == events_[i].is_valid()) {
free_idx = i;
}
}
if (OB_ENTRY_EXIST == ret) {
} else if (-1 != free_idx) {
events_[free_idx] = event;
} else {
ret = OB_SIZE_OVERFLOW;
}
CLOG_LOG(INFO, "insert event into set success", K(ret), K(event), K(free_idx));
return ret;
}
int RoleChangeEventSet::remove(const RoleChangeEvent &event)
{
int ret = OB_SUCCESS;
int64_t delete_idx = -1;
ObSpinLockGuard guard(lock_);
for (int64_t i = 0; i < MAX_ARRAY_SIZE && -1 == delete_idx; i++) {
if (event == events_[i]) {
delete_idx = i;
}
};
if (-1 != delete_idx) {
events_[delete_idx].reset();
} else {
ret = OB_ENTRY_NOT_EXIST;
}
CLOG_LOG(INFO, "remove slog from set success", K(ret), K(delete_idx), K(event));
return ret;
}
RoleCoordinator::RoleCoordinator() : log_client_map_(NULL),
apply_service_(NULL),
tg_id_(-1),
is_inited_(false)
{
}
RoleCoordinator::~RoleCoordinator()
{
if (IS_INIT) {
destroy();
}
}
int RoleCoordinator::init(palfcluster::LogClientMap *log_client_map,
logservice::ObLogApplyService *apply_service)
{
int ret = OB_SUCCESS;
const int tg_id = lib::TGDefIDs::RCService;
if (IS_INIT) {
ret = OB_INIT_TWICE;
} else if (OB_ISNULL(log_client_map) || OB_ISNULL(apply_service)) {
ret = OB_INVALID_ARGUMENT;
CLOG_LOG(WARN, "invalid argument", K(ret), KP(apply_service));
} else if (OB_FAIL(TG_CREATE_TENANT(tg_id, tg_id_))) {
CLOG_LOG(WARN, "RoleCoordinator TG_CREATE failed", K(ret));
} else {
apply_service_ = apply_service;
log_client_map_ = log_client_map;
is_inited_ = true;
CLOG_LOG(INFO, "RoleCoordinator init success", K(ret), K(tg_id_), KP(apply_service));
}
return ret;
}
int RoleCoordinator::start()
{
int ret = OB_SUCCESS;
if (IS_NOT_INIT) {
ret = OB_NOT_INIT;
} else if (OB_FAIL(TG_SET_HANDLER_AND_START(tg_id_, *this))) {
CLOG_LOG(WARN, "RoleCoordinator start failed", K(ret), K(tg_id_));
} else {
CLOG_LOG(INFO, "RoleCoordinator start success", K(ret), K(tg_id_));
}
return ret;
}
void RoleCoordinator::wait()
{
if (IS_INIT) {
TG_STOP(tg_id_);
TG_WAIT(tg_id_);
}
CLOG_LOG(INFO, "RoleCoordinator wait finish", K(tg_id_));
}
void RoleCoordinator::stop()
{
if (IS_INIT) {
TG_STOP(tg_id_);
}
CLOG_LOG(INFO, "RoleCoordinator stop finish", K(tg_id_));
}
void RoleCoordinator::destroy()
{
if (IS_INIT) {
(void)stop();
(void)wait();
TG_DESTROY(tg_id_);
is_inited_ = false;
tg_id_ = -1;
apply_service_ = NULL;
CLOG_LOG(INFO, "RoleCoordinator destroy success");
}
}
void RoleCoordinator::handle(void *task)
{
int ret = OB_SUCCESS;
// When role chage service hang exceeds 30 seconds, we think there is dead lock in 'handle_role_change_event_',
// TIMEGUARD will pring lbt().
TIMEGUARD_INIT(CLOG, 30_s, 30_s);
RoleChangeEvent *event = reinterpret_cast<RoleChangeEvent*>(task);
const int64_t ls_id = event->ls_id_.id();
const int64_t start_ts = ObTimeUtility::current_time();
RetrySubmitRoleChangeEventCtx retry_ctx;
CLOG_LOG(INFO, "begin handle_role_change_event_", "sequence:", start_ts, KPC(event));
if (NULL == event) {
CLOG_LOG(WARN, "unexpected error, task is nullptr", KP(event));
} else if (OB_FAIL(handle_role_change_event_(*event, retry_ctx))) {
CLOG_LOG(WARN, "handle_role_change_event_ failed", K(ret), KPC(event), K(retry_ctx));
} else {
CLOG_LOG(INFO, "end handle_role_change_event_", "sequence:", start_ts, KPC(event));
}
if (NULL != event) {
OB_DELETE(RoleChangeEvent, "RCService", event);
}
if (retry_ctx.need_retry() && OB_FAIL(on_role_change(ls_id))) {
CLOG_LOG(WARN, "retry submit role change event failed", K(ls_id), K(retry_ctx));
}
}
int RoleCoordinator::on_role_change(const int64_t id)
{
int ret = OB_SUCCESS;
share::ObLSID ls_id(id);
RoleChangeEvent event(RoleChangeEventType::ROLE_CHANGE_CB_EVENT_TYPE, ls_id);
// TODO by runlin: if task queue has been full, push task will be failed, the role change event
// will be lost.
if (OB_FAIL(submit_role_change_event_(event))) {
CLOG_LOG(WARN, "submit_role_change_event_ failed", K(ret), K(event));
} else {
CLOG_LOG(INFO, "on_role_change success", K(ret), K(event));
}
return ret;
}
int RoleCoordinator::on_need_change_leader(const int64_t ls_id, const common::ObAddr &dst_addr)
{
int ret = OB_SUCCESS;
RoleChangeEvent event(RoleChangeEventType::CHANGE_LEADER_EVENT_TYPE, share::ObLSID(ls_id), dst_addr);
if (OB_FAIL(submit_role_change_event_(event))) {
CLOG_LOG(WARN, "submit_role_change_event_ failed", K(ret), K(event));
} else {
CLOG_LOG(INFO, "change_leader success", K(ret), K(event));
}
return ret;
}
int RoleCoordinator::submit_role_change_event_(const RoleChangeEvent &event)
{
int ret = OB_SUCCESS;
if(OB_FAIL(rc_set_.insert(event)) && OB_ENTRY_EXIST != ret) {
CLOG_LOG(ERROR, "insert into rc_set failed", K(ret), K(event));
} else if (OB_ENTRY_EXIST == ret) {
CLOG_LOG(INFO, "repeat role change event, filter it", K(ret), K(event));
ret = OB_SUCCESS;
} else if (OB_FAIL(push_event_into_queue_(event))) {
CLOG_LOG(WARN, "push_event_into_queue_ failed", K(ret), K(event));
} else {
CLOG_LOG(INFO, "submit_role_change_event_ success", K(ret), K(event));
}
return ret;
}
// TODO: use poll to avoid alloc memory failed.
int RoleCoordinator::push_event_into_queue_(const RoleChangeEvent &event)
{
int ret = OB_SUCCESS;
RoleChangeEvent *rc_event = NULL;
int64_t warn_time = OB_INVALID_TIMESTAMP;
do {
if (NULL == (rc_event =
MTL_NEW(RoleChangeEvent, "RCService", event.event_type_, event.ls_id_, event.dst_addr_))) {
ret = OB_ALLOCATE_MEMORY_FAILED;
if (palf_reach_time_interval(1 * 1000 * 1000, warn_time)) {
CLOG_LOG(WARN, "allocate memory failed", K(ret), K(event));
}
usleep(1 * 1000);
} else {
ret = OB_SUCCESS;
}
} while(OB_FAIL(ret));
if (OB_FAIL(TG_PUSH_TASK(tg_id_, rc_event))) {
CLOG_LOG(WARN, "ObRoleChangeTask push task failed", K(ret), K(event));
}
if (OB_FAIL(ret) && NULL != rc_event) {
MTL_DELETE(RoleChangeEvent, "RCService", rc_event);
}
return ret;
}
int RoleCoordinator::handle_role_change_event_(const RoleChangeEvent &event,
RetrySubmitRoleChangeEventCtx &retry_ctx)
{
int ret = OB_SUCCESS;
ObLogClient *ls = nullptr;
AccessMode curr_access_mode;
int64_t unused_mode_version;
OB_ASSERT(OB_SUCCESS == rc_set_.remove(event));
if (false == event.is_valid()) {
ret = OB_INVALID_ARGUMENT;
CLOG_LOG(WARN, "invalid argument", K(event));
} else if (OB_FAIL(log_client_map_->get(event.ls_id_, ls) || OB_ISNULL(ls))) {
CLOG_LOG(ERROR, "not exist loghandler", K(ret), K(event.ls_id_), KP(ls));
ret = OB_ERR_UNEXPECTED;
} else if (OB_FAIL(ls->get_log_handler()->get_access_mode(unused_mode_version, curr_access_mode))) {
CLOG_LOG(WARN, "ObLogHandler get_access_mode failed", K(ret));
} else {
switch (event.event_type_) {
case RoleChangeEventType::CHANGE_LEADER_EVENT_TYPE:
CLOG_LOG(INFO, "begin change leader", K(curr_access_mode), K(event), KPC(ls));
if (is_append_mode(curr_access_mode)
&& OB_FAIL(handle_change_leader_event_for_log_handler_(event.dst_addr_, ls))) {
CLOG_LOG(WARN, "ObLogHandler change leader failed", K(ret), K(event), KPC(ls));
}
CLOG_LOG(INFO, "end change leader", K(ret), K(curr_access_mode), K(event), KPC(ls));
break;
case RoleChangeEventType::ROLE_CHANGE_CB_EVENT_TYPE:
CLOG_LOG(INFO, "begin log handler role change", K(curr_access_mode), K(event), KPC(ls));
if (OB_FAIL(handle_role_change_cb_event_for_log_handler_(curr_access_mode, ls, retry_ctx))) {
CLOG_LOG(WARN, "handle_role_change_cb_event_for_log_handler_ failed", K(ret),
K(curr_access_mode), KPC(ls));
}
CLOG_LOG(INFO, "end log handler role change", K(ret), K(curr_access_mode), K(event), KPC(ls), K(retry_ctx));
break;
default:
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(WARN, "unexpected role change event type", K(ret));
}
}
return ret;
}
int RoleCoordinator::handle_role_change_cb_event_for_log_handler_(
const AccessMode &curr_access_mode,
ObLogClient*ls,
RetrySubmitRoleChangeEventCtx &retry_ctx)
{
int ret = OB_SUCCESS;
const bool log_handler_is_offline = ls->get_log_handler()->is_offline();
// If log handler is offline, need execute LEADER_2_FOLLOWER or FOLLOWER_2_FOLLOWER
//
// when access mode is APPEND, log_handler need execute leader to follower or
// follower to leader. otherwise, only need execute leader to follower or follower
// to follower, therefore, we set 'need_transform_by_access_mode' to false when
// 'curr_access_mode' is APPEND.
const bool only_need_change_to_follower = !is_append_mode(curr_access_mode) || log_handler_is_offline;
RoleChangeOptType opt_type;
ObRole curr_role = ObRole::INVALID_ROLE;
ObRole new_role = ObRole::INVALID_ROLE;
bool is_pending_state = false;
int64_t curr_proposal_id = -1;
int64_t new_proposal_id = -1;
if (OB_FAIL(ls->get_log_handler()->prepare_switch_role(curr_role,
curr_proposal_id, new_role, new_proposal_id, is_pending_state))) {
CLOG_LOG(WARN, "ObLogHandler prepare_switch_role failed", K(ret), K(curr_role), K(curr_proposal_id),
K(new_role), K(new_proposal_id));
} else if (false == need_execute_role_change(curr_proposal_id, curr_role, new_proposal_id,
new_role, is_pending_state, log_handler_is_offline)) {
CLOG_LOG(INFO, "no need change role", K(ret), K(is_pending_state), K(curr_role), K(curr_proposal_id),
K(new_role), K(new_proposal_id), K(is_pending_state), K(log_handler_is_offline));
} else if (FALSE_IT(opt_type = get_role_change_opt_type_(curr_role, new_role, only_need_change_to_follower))) {
} else {
switch (opt_type) {
// leader -> follower
case RoleChangeOptType::LEADER_2_FOLLOWER:
if (OB_FAIL(switch_leader_to_follower_forcedly_(new_proposal_id, ls))) {
CLOG_LOG(WARN, "switch_leader_to_follower_forcedly_ failed", K(ret), K(curr_role),
K(curr_proposal_id), K(new_role), K(curr_access_mode), K(new_proposal_id));
}
break;
// follower -> follower
case RoleChangeOptType::FOLLOWER_2_LEADER:
if (OB_FAIL(switch_follower_to_leader_(new_proposal_id, ls, retry_ctx))) {
CLOG_LOG(WARN, "switch_follower_to_leader_ failed", K(ret), K(curr_role),
K(curr_proposal_id), K(new_role), K(curr_access_mode), K(new_proposal_id));
}
break;
// leader -> leader
case RoleChangeOptType::LEADER_2_LEADER:
if (OB_FAIL(switch_leader_to_leader_(new_proposal_id, curr_proposal_id, ls, retry_ctx))) {
CLOG_LOG(WARN, "switch_leader_to_leader_ failed", K(ret), K(curr_role),
K(curr_proposal_id), K(new_role), K(curr_access_mode), K(new_proposal_id));
}
break;
// follower -> follower
case RoleChangeOptType::FOLLOWER_2_FOLLOWER:
if (OB_FAIL(switch_follower_to_follower_(new_proposal_id, ls))) {
CLOG_LOG(WARN, "switch_follower_to_follower_ failed", K(ret), K(curr_role),
K(curr_proposal_id), K(new_role), K(curr_access_mode), K(new_proposal_id));
}
break;
default:
ret = OB_ERR_UNEXPECTED;
CLOG_LOG(ERROR, "unexpected error, can not handle role change", K(ret), K(curr_role),
K(curr_proposal_id), K(new_role), K(new_proposal_id), KPC(ls));
}
}
return ret;
}
int RoleCoordinator::handle_change_leader_event_for_log_handler_(
const common::ObAddr &dst_addr,
ObLogClient*ls)
{
int ret = OB_SUCCESS;
ObRole curr_role = ObRole::INVALID_ROLE;
ObRole new_role = ObRole::INVALID_ROLE;
bool is_pending_state = false;
int64_t curr_proposal_id = -1;
int64_t new_proposal_id = -1;
if (OB_FAIL(ls->get_log_handler()->prepare_switch_role(curr_role,
curr_proposal_id, new_role, new_proposal_id, is_pending_state))) {
CLOG_LOG(WARN, "ObLogHandler prepare_switch_role failed", K(ret), K(curr_role), K(curr_proposal_id),
K(new_role), K(new_proposal_id));
} else if (true == is_pending_state
|| curr_proposal_id != new_proposal_id || LEADER != curr_role || LEADER != new_role) {
// when log handler is not LEDAER, we also need execute change_leader_to, otherwise, the leader can not be changed by election.
ls->get_log_handler()->change_leader_to(dst_addr);
CLOG_LOG(INFO, "no need execute switch_leader_to_follower_gracefully, change leader directlly",
K(ret), K(is_pending_state), K(curr_proposal_id), K(new_proposal_id), K(curr_role), K(new_role));
} else if (OB_FAIL(switch_leader_to_follower_gracefully_(new_proposal_id, curr_proposal_id,
dst_addr, ls))) {
CLOG_LOG(WARN, "switch_leader_to_follower_gracefully_ failed", K(ret), KPC(ls),
K(curr_role), K(curr_proposal_id), K(new_role), K(new_proposal_id));
} else {
CLOG_LOG(INFO, "handle_change_leader_event_for_log_handler_ success", K(ret), K(curr_role),
K(curr_proposal_id), K(new_role), K(new_proposal_id), K(dst_addr));
}
return ret;
}
int RoleCoordinator::switch_follower_to_leader_(
const int64_t new_proposal_id,
ObLogClient*ls,
RetrySubmitRoleChangeEventCtx &retry_ctx)
{
int ret = OB_SUCCESS;
const ObRole new_role = LEADER;
const share::ObLSID &ls_id = ls->get_ls_id();
palf::LSN end_lsn;
ObTimeGuard time_guard("switch_to_leader", EACH_ROLE_CHANGE_COST_MAX_TIME);
ObLogHandler *log_handler = ls->get_log_handler();
// ObRoleChangeHandler *role_change_handler = ls->get_role_change_handler();
if (OB_FAIL(log_handler->get_end_lsn(end_lsn))) {
CLOG_LOG(WARN, "get_end_lsn failed", K(ret), KPC(ls));
// NB: order is vital!!!
// We must guarantee that 'replay_service_' has replayed complete data, and before
// stop 'replay_service_', other components can not submit log.
} else if (FALSE_IT(time_guard.click("wait_replay_service_apply_done_"))
|| OB_FAIL(wait_replay_service_replay_done_(ls_id, end_lsn, WAIT_REPLAY_DONE_TIMEOUT_US))) {
if (need_retry_submit_role_change_event_(ret)) {
retry_ctx.set_retry_reason(RetrySubmitRoleChangeEventReason::WAIT_REPLAY_DONE_TIMEOUT);
} else {
CLOG_LOG(WARN, "wait_replay_service_replay_done_ failed", K(ret), K(end_lsn));
}
} else if (FALSE_IT(time_guard.click("apply_service->switch_to_leader"))
|| OB_FAIL(apply_service_->switch_to_leader(ls_id, new_proposal_id))) {
CLOG_LOG(WARN, "apply_service_ switch_to_leader failed", K(ret), K(new_role), K(new_proposal_id));
// } else if (FALSE_IT(time_guard.click("replay_service->switch_to_leader"))
// || OB_FAIL(replay_service_->switch_to_leader(ls_id))) {
} else if (FALSE_IT(log_handler->switch_role(new_role, new_proposal_id))) {
CLOG_LOG(WARN, "ObLogHandler switch role failed", K(ret), K(new_role), K(new_proposal_id));
// } else if (FALSE_IT(time_guard.click("role_change_handler->switch_to_leader"))
// || OB_FAIL(role_change_handler->switch_to_leader())) {
// CLOG_LOG(WARN, "ObRoleChangeHandler switch_to_leader failed", K(ret), KPC(ls));
} else {
CLOG_LOG(INFO, "switch_follower_to_leader_ success", K(ret), KPC(ls));
}
if (OB_FAIL(ret) && !retry_ctx.need_retry()) {
log_handler->change_leader_to(GCTX.self_addr());
CLOG_LOG(WARN, "switch_follower_to_leader_ failed", K(ret), KPC(ls));
}
return ret;
}
int RoleCoordinator::switch_leader_to_follower_forcedly_(
const int64_t new_proposal_id,
ObLogClient*ls)
{
int ret = OB_SUCCESS;
const ObRole new_role = FOLLOWER;
const share::ObLSID &ls_id = ls->get_ls_id();
ObLogHandler *log_handler = ls->get_log_handler();
// ObRoleChangeHandler *role_change_handler = ls->get_role_change_handler();
palf::LSN end_lsn;
ObTimeGuard time_guard("switch_leader_to_follower_forcedly_", EACH_ROLE_CHANGE_COST_MAX_TIME);
// Why need wait_apply_sync?
//
// when we can execute 'switch_to_follower_forcedly', means that there is no possibility to submit log via log handler successfully.
// however, the flying callback may have not been pushed into apply service, and then, 'switch_to_follower' will be executed, for trans,
// if the callback be executed after 'switch_to_follower', will cause abort.
if (OB_FAIL(apply_service_->wait_append_sync(ls_id))) {
CLOG_LOG(WARN, "wait_apply_sync failed", K(ret), K(ls_id));
} else if (FALSE_IT(time_guard.click("apply_service->wait_apply_sync"))
|| OB_FAIL(apply_service_->switch_to_follower(ls_id))) {
CLOG_LOG(WARN, "apply_service_ switch_to_follower failed", K(ret), K(new_role), K(new_proposal_id));
} else if (FALSE_IT(time_guard.click("apply_service->switch_to_follower"))
|| OB_FAIL(wait_apply_service_apply_done_(ls_id, end_lsn))) {
CLOG_LOG(WARN, "wait_apply_service_apply_done_ failed", K(ret), K(end_lsn));
} else {
time_guard.click("wait_apply_service_apply_done_");
// role_change_handler->switch_to_follower_forcedly();
// NB: order is vital
// We must guarantee that this replica will not submit any logs after switch_role.
log_handler->switch_role(new_role, new_proposal_id);
// NB: in case of leader reovke, do we no need retry.
// (void)replay_service_->switch_to_follower(ls_id, end_lsn);
CLOG_LOG(INFO, "switch_leader_to_follower_forcedly_ success", K(ret), KPC(ls));
}
if (OB_FAIL(ret)) {
log_handler->change_leader_to(GCTX.self_addr());
CLOG_LOG(WARN, "switch_leader_to_follower_forcedly_ failed", K(ret), K(new_proposal_id), K(new_role));
}
return ret;
}
int RoleCoordinator::switch_leader_to_follower_gracefully_(
const int64_t new_proposal_id,
const int64_t curr_proposal_id,
const common::ObAddr &dst_addr,
ObLogClient*ls)
{
int ret = OB_SUCCESS;
int tmp_ret = OB_SUCCESS;
const ObRole new_role = FOLLOWER;
const share::ObLSID &ls_id = ls->get_ls_id();
ObLogHandler *log_handler = ls->get_log_handler();
// ObRoleChangeHandler *role_change_handler = ls->get_role_change_handler();
LSN end_lsn;
ObTimeGuard time_guard("switch_leader_to_follower_gracefully_", EACH_ROLE_CHANGE_COST_MAX_TIME);
// 1. OB_SUCCESS means execute transaction successfully, we need execute follow steps.
// 2. OB_LS_NEED_REVOKE means the transaction execute failed, and can't been rollback, need revoke LS.
// 3. OTHERS, switch_to_follower_gracefully failed, and 'role_change_handler' has rollback success,
// no need to execute follow steps.
// if (FALSE_IT(time_guard.click("role_change_handler->switch_to_follower_gracefully"))
// || OB_SUCCESS != (tmp_ret = role_change_handler->switch_to_follower_gracefully())) {
// CLOG_LOG(WARN, "switch_to_follower_gracefully failed, need revoke leader", K(tmp_ret),
// K(new_role), K(new_proposal_id), K(dst_addr));
// NB: order is vital!!!
// we must ensure that the 'end_lsn' provid by 'apply_service_' is correctly.
// just switch_role to follower firstly, avoid sync log failed because palf has changed leader.
if (FALSE_IT(log_handler->switch_role(new_role, curr_proposal_id))) {
// apply service will not update end_lsn after switch_to_follower, so wait apply done first here
} else if (FALSE_IT(time_guard.click("wait_apply_service_apply_done_when_change_leader_"))
|| OB_FAIL(wait_apply_service_apply_done_when_change_leader_(log_handler, curr_proposal_id, ls_id, end_lsn))) {
CLOG_LOG(WARN, "wait_apply_service_apply_done_when_change_leader_ failed", K(ret),
K(new_role), K(new_proposal_id), K(dst_addr));
// wait apply service done my fail, we need :
// 1. switch log handler to origin status.
// 2. resume role change handler
log_handler->switch_role(LEADER, curr_proposal_id);
// if (OB_FAIL(role_change_handler->resume_to_leader())) {
// CLOG_LOG(WARN, "resume to leader failed", K(ret), KPC(ls));
// }
// NB: the following steps mustn't be failed.
} else if (FALSE_IT(time_guard.click("apply_service->switch_to_follower"))
|| OB_FAIL(apply_service_->switch_to_follower(ls_id))) {
CLOG_LOG(WARN, "apply_service_ switch_to_follower failed", K(ret), K(new_role), K(new_proposal_id), K(dst_addr));
// } else if (FALSE_IT(time_guard.click("replay_service->switch_to_follower"))
// || OB_FAIL(replay_service_->switch_to_follower(ls_id, end_lsn))) {
// CLOG_LOG(WARN, "replay_service_ switch_to_follower failed", K(ret), KPC(ls), K(new_role), K(new_proposal_id));
// NB: execute 'change_leader_to' lastly, can make 'wait_apply_service_apply_done_when_change_leader_' finish quickly.
} else if (OB_FAIL(log_handler->change_leader_to(dst_addr))) {
CLOG_LOG(WARN, "ObLogHandler change_leader failed", K(ret), K(new_role), K(new_proposal_id), K(dst_addr));
} else {
CLOG_LOG(INFO, "switch_to_follower_gracefully success", K(ret), K(new_role), K(new_proposal_id), K(dst_addr));
}
if (OB_FAIL(ret) || OB_LS_NEED_REVOKE == tmp_ret) {
log_handler->change_leader_to(GCTX.self_addr());
CLOG_LOG(WARN, "switch_leader_to_follower_gracefully failed, revoke leader", K(ret), K(tmp_ret), K(dst_addr),
K(new_role), K(new_proposal_id));
ret = (OB_SUCCESS == ret ? tmp_ret : ret);
}
return ret;
}
int RoleCoordinator::switch_leader_to_leader_(
const int64_t new_proposal_id,
const int64_t curr_proposal_id,
ObLogClient*ls,
RetrySubmitRoleChangeEventCtx &retry_ctx)
{
int ret = OB_SUCCESS;
ObTimeGuard time_guard("switch_leader_to_leader", EACH_ROLE_CHANGE_COST_MAX_TIME);
if (FALSE_IT(time_guard.click("switch_leader_to_follower_forcedly_"))
|| OB_FAIL(switch_leader_to_follower_forcedly_(curr_proposal_id, ls))) {
CLOG_LOG(WARN, "switch_leader_to_leader_, switch leader to follower failed", K(ret), KPC(ls));
} else if (FALSE_IT(time_guard.click("switch_follower_to_leader_"))
|| OB_FAIL(switch_follower_to_leader_(new_proposal_id, ls, retry_ctx))) {
CLOG_LOG(WARN, "switch_follower_to_leader_ failed", K(ret), K(new_proposal_id));
} else {
CLOG_LOG(INFO, "switch_leader_to_leader_ success", K(ret), KPC(ls));
}
return ret;
}
int RoleCoordinator::switch_follower_to_follower_(const int64_t new_proposal_id, ObLogClient*ls)
{
int ret = OB_SUCCESS;
// need update proposal_id
const share::ObLSID &ls_id = ls->get_ls_id();
ObLogHandler *log_handler = ls->get_log_handler();
(void) log_handler->switch_role(common::ObRole::FOLLOWER, new_proposal_id);
CLOG_LOG(INFO, "switch_follower_to_follower_");
return ret;
}
int RoleCoordinator::wait_replay_service_replay_done_(
const share::ObLSID &ls_id,
const palf::LSN &end_lsn,
const int64_t timeout_us)
{
int ret = OB_SUCCESS;
return ret;
}
int RoleCoordinator::wait_apply_service_apply_done_(
const share::ObLSID &ls_id,
palf::LSN &end_lsn)
{
int ret = OB_SUCCESS;
bool is_done = false;
const int64_t start_ts = ObTimeUtility::current_time();
while (OB_SUCC(ret) && false == is_done) {
if (OB_FAIL(apply_service_->is_apply_done(ls_id, is_done, end_lsn))) {
CLOG_LOG(WARN, "apply_service_ is_apply_done failed", K(ret), K(is_done), K(end_lsn));
} else if (false == is_done) {
ob_usleep(5*1000);
CLOG_LOG(WARN, "wait apply done return false, need retry", K(ls_id), K(is_done), K(end_lsn), K(start_ts));
} else {
}
}
return ret;
}
int RoleCoordinator::wait_apply_service_apply_done_when_change_leader_(
const ObLogHandler *log_handler,
const int64_t proposal_id,
const share::ObLSID &ls_id,
palf::LSN &end_lsn)
{
int ret = OB_SUCCESS;
bool is_done = false;
palf::LSN max_lsn;
common::ObRole unused_curr_role;
int64_t unused_curr_proposal_id;
common::ObRole new_role;
int64_t new_proposal_id;
bool is_pending_state = false;
while (OB_SUCC(ret) && (false == is_done || end_lsn != max_lsn)) {
if (OB_FAIL(apply_service_->is_apply_done(ls_id, is_done, end_lsn))) {
CLOG_LOG(WARN, "apply_service_ is_apply_done failed", K(ret), K(is_done), K(end_lsn));
// NB: ApplyService execute on_failure only when it's FOLLOWER, therefore ApplyService my not return apply done
// when it's LEADER, we need check the role of palf when has changed.
} else if (OB_FAIL(log_handler->get_max_lsn(max_lsn))) {
CLOG_LOG(WARN, "get_end_lsn from palf failed", K(ret), K(ls_id), K(end_lsn));
} else if (OB_FAIL(log_handler->prepare_switch_role(unused_curr_role, unused_curr_proposal_id,
new_role, new_proposal_id, is_pending_state))) {
CLOG_LOG(WARN, "failed prepare_switch_role", K(ret), K(new_role), K(proposal_id), K(ls_id));
// if palf has changed role, return OB_STATE_NOT_MATCH, change leader failed.
} else if (LEADER != new_role || proposal_id != new_proposal_id) {
ret = OB_STATE_NOT_MATCH;
CLOG_LOG(WARN, "palf has changed leader, wait_apply_service_apply_done_when_change_leader_ failed", K(ret), K(proposal_id),
K(new_proposal_id));
} else if (false == is_done || end_lsn != max_lsn) {
CLOG_LOG(INFO, "wait apply done return false, need retry", K(ls_id), K(is_done),
K(end_lsn), K(max_lsn));
ob_usleep(5*1000);
} else {
}
}
return ret;
}
RoleCoordinator::RoleChangeOptType RoleCoordinator::get_role_change_opt_type_(
const ObRole &old_role,
const ObRole &new_role,
const bool need_transform_by_access_mode) const
{
RoleChangeOptType rc_opt_type = RoleChangeOptType::INVALID_RC_OPT_TYPE;
if (FOLLOWER == old_role && LEADER == new_role) {
rc_opt_type = RoleChangeOptType::FOLLOWER_2_LEADER;
} else if (LEADER == old_role && FOLLOWER == new_role) {
rc_opt_type = RoleChangeOptType::LEADER_2_FOLLOWER;
} else if (FOLLOWER == old_role && FOLLOWER == new_role) {
rc_opt_type = RoleChangeOptType::FOLLOWER_2_FOLLOWER;
} else if (LEADER == old_role && LEADER == new_role) {
rc_opt_type = RoleChangeOptType::LEADER_2_LEADER;
} else {
}
if (true == need_transform_by_access_mode) {
if (LEADER == old_role) {
rc_opt_type = RoleChangeOptType::LEADER_2_FOLLOWER;
} else {
rc_opt_type = RoleChangeOptType::FOLLOWER_2_FOLLOWER;
}
}
return rc_opt_type;
}
// NB:
// 1. when log handler is offline, need execute role change;
// 2. when palf is not in pending:
// 1. proposal_id is not same or
// 2. role is not same.(If there are no pending logs in sliding window,
// leader to follower will not advance proposal_id)
bool RoleCoordinator::need_execute_role_change(const int64_t curr_proposal_id,
const common::ObRole curr_role,
const int64_t new_proposal_id,
const common::ObRole new_role,
const bool is_pending_state,
const bool is_offline) const
{
return is_offline
|| (!is_pending_state
&& (curr_proposal_id != new_proposal_id || curr_role != new_role));
}
bool RoleCoordinator::is_append_mode(const AccessMode &mode) const
{
return (AccessMode::APPEND == mode);
}
bool RoleCoordinator::is_raw_write_or_flashback_mode(const AccessMode &mode) const
{
return (AccessMode::RAW_WRITE == mode || AccessMode::FLASHBACK == mode ||
AccessMode::PREPARE_FLASHBACK == mode);
}
bool RoleCoordinator::need_retry_submit_role_change_event_(int ret) const
{
bool bool_ret = false;
if (OB_TIMEOUT == ret) {
bool_ret = true;
}
return bool_ret;
}
} // end namespace palfcluster
} // end namespace oceanbase
Reference in New Issue View Git Blame Copy Permalink