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

[Election] fix arb server compare priority bug

Browse Source
This commit is contained in:
fengdeyiji
2023-06-05 08:50:59 +00:00
committed by ob-robot
parent 61b129e849
commit a795853f8e
13 changed files with 334 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
unittest/logservice/test_ob_election_message_compat.cpp
View File

@ -101,7 +101,26 @@ protected:
OB_SERIALIZE_MEMBER_TEMP(inline, ElectionMsgBase, id_, sender_, receiver_,
restart_counter_, ballot_number_, msg_type_, debug_ts_);
class ElectionPrepareRequestMsgMiddleOld : public oceanbase::palf::election::ElectionMsgBase
ElectionMsgBase::ElectionMsgBase() :
id_(INVALID_VALUE),
restart_counter_(INVALID_VALUE),
ballot_number_(INVALID_VALUE),
msg_type_(static_cast<int64_t>(ElectionMsgType::INVALID_TYPE)) {}
ElectionMsgBase::ElectionMsgBase(const int64_t id,
const common::ObAddr &self_addr,
const int64_t restart_counter,
const int64_t ballot_number,
const ElectionMsgType msg_type) :
id_(id),
sender_(self_addr),
restart_counter_(restart_counter),
ballot_number_(ballot_number),
msg_type_(static_cast<int64_t>(msg_type)) {
debug_ts_.src_construct_ts_ = ObClockGenerator::getRealClock();
}
class ElectionPrepareRequestMsgMiddleOld : public oceanbase::unittest::ElectionMsgBase
{
OB_UNIS_VERSION(1);
public:
@ -189,13 +208,13 @@ public:
int deserialize(const char* buf, const int64_t data_len, int64_t& pos) {
int ret = ElectionPrepareRequestMsgMiddleOld::deserialize(buf, data_len, pos);
debug_ts_.dest_deserialize_ts_ = ObClockGenerator::getRealClock();
print_debug_ts_if_reach_warn_threshold(*this, MSG_DELAY_WARN_THRESHOLD);
// print_debug_ts_if_reach_warn_threshold(*this, MSG_DELAY_WARN_THRESHOLD);
return ret;
}
int64_t get_serialize_size() const {
if (debug_ts_.src_serialize_ts_ == 0) {// cause get_serialize_size maybe call more than once
const_cast<int64_t&>(debug_ts_.src_serialize_ts_) = ObClockGenerator::getRealClock();
print_debug_ts_if_reach_warn_threshold(*this, MSG_DELAY_WARN_THRESHOLD);
// print_debug_ts_if_reach_warn_threshold(*this, MSG_DELAY_WARN_THRESHOLD);
}
return ElectionPrepareRequestMsgMiddleOld::get_serialize_size();
}
@ -206,7 +225,7 @@ TEST_F(TestElectionMsgCompat, old_new_msg_serialize) {
config_version.generate(1, 1);
ElectionPrepareRequestMsgOld prepare_msg_old1(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, config_version);
ElectionPrepareRequestMsg prepare_msg_new1;
ElectionPrepareRequestMsg prepare_msg_new2(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, (1 << 10), config_version);
ElectionPrepareRequestMsg prepare_msg_new2(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), (1 << 10), config_version);
ElectionPrepareRequestMsgOld prepare_msg_old2;
constexpr int64_t buffer_size = 2_KB;
char buffer[buffer_size];

12
unittest/logservice/test_ob_election_message_compat2.cpp
View File

@ -764,7 +764,7 @@ TEST_F(TestElectionMsgCompat2, old_to_new) {
LogConfigVersion config_version;
config_version.generate(1, 1);
int64_t pos = 0;
unittest::ElectionAcceptRequestMsg msg_request_old(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, 123, 234, config_version);
unittest::ElectionAcceptRequestMsg msg_request_old(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), 123, 234, config_version);
palf::election::ElectionAcceptRequestMsg msg_request_new;
ASSERT_EQ(msg_request_old.serialize(buffer, BUFFER_SIZE, pos), OB_SUCCESS);
int64_t pos2 = 0;
@ -773,7 +773,7 @@ TEST_F(TestElectionMsgCompat2, old_to_new) {
ASSERT_EQ(msg_request_new.flag_not_less_than_4_2_, false);
ElectionPriorityImpl priority;
unittest::ElectionAcceptResponseMsg msg_response_old(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, config_version, msg_request_old);
unittest::ElectionAcceptResponseMsg msg_response_old(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, config_version, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), msg_request_old);
palf::election::ElectionAcceptResponseMsg msg_response_new;
ASSERT_EQ(msg_response_old.set_accepted(1, &priority), OB_SUCCESS);
pos = 0;
@ -794,7 +794,7 @@ TEST_F(TestElectionMsgCompat2, new_to_old_fake_new) {
int64_t pos = 0;
ASSERT_EQ(observer::ObServer::get_instance().is_arbitration_mode(), false);
ASSERT_EQ(GET_MIN_CLUSTER_VERSION() < CLUSTER_VERSION_4_2_0_0, true);
palf::election::ElectionAcceptRequestMsg msg_request_new(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, 123, 234, config_version);
palf::election::ElectionAcceptRequestMsg msg_request_new(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), 123, 234, config_version);
ASSERT_EQ(msg_request_new.flag_not_less_than_4_2_, false);
unittest::ElectionAcceptRequestMsg msg_request_old;
int64_t serialize_size = msg_request_new.get_serialize_size();
@ -805,7 +805,7 @@ TEST_F(TestElectionMsgCompat2, new_to_old_fake_new) {
ASSERT_EQ(pos, pos2);
ElectionPriorityImpl priority;
palf::election::ElectionAcceptResponseMsg msg_response_new(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, config_version, msg_request_old);
palf::election::ElectionAcceptResponseMsg msg_response_new(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, config_version, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), msg_request_old);
ASSERT_EQ(msg_response_new.flag_not_less_than_4_2_, false);
unittest::ElectionAcceptResponseMsg msg_response_old;
ASSERT_EQ(msg_response_new.set_accepted(1, &priority), OB_SUCCESS);
@ -828,7 +828,7 @@ TEST_F(TestElectionMsgCompat2, new_to_new_real_new) {
int64_t pos = 0;
ASSERT_EQ(observer::ObServer::get_instance().is_arbitration_mode(), false);
ASSERT_EQ(GET_MIN_CLUSTER_VERSION() < CLUSTER_VERSION_4_2_0_0, false);
palf::election::ElectionAcceptRequestMsg msg_request_src(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, 123, 234, config_version);
palf::election::ElectionAcceptRequestMsg msg_request_src(1, ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, 1, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), 123, 234, config_version);
ASSERT_EQ(msg_request_src.flag_not_less_than_4_2_, true);
palf::election::ElectionAcceptRequestMsg msg_request_dst;
int64_t serialize_size = msg_request_src.get_serialize_size();
@ -840,7 +840,7 @@ TEST_F(TestElectionMsgCompat2, new_to_new_real_new) {
ASSERT_EQ(msg_request_dst.flag_not_less_than_4_2_, true);
ElectionPriorityImpl priority;
palf::election::ElectionAcceptResponseMsg msg_response_src(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, config_version, msg_request_dst);
palf::election::ElectionAcceptResponseMsg msg_response_src(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", 1), 1, config_version, LsBiggestMinClusterVersionEverSeen(CLUSTER_VERSION_4_1_0_0), msg_request_dst);
ASSERT_EQ(msg_response_src.flag_not_less_than_4_2_, true);
palf::election::ElectionAcceptResponseMsg msg_response_dst;
ASSERT_EQ(msg_response_src.set_accepted(1, &priority), OB_SUCCESS);

158
unittest/logservice/test_ob_election_with_priority.cpp
View File

@ -28,6 +28,7 @@
#include "share/ob_occam_timer.h"
#include "share/rc/ob_tenant_base.h"
#include "mock_logservice_container/mock_election_user.h"
#include "observer/ob_server.h"
#include <iostream>
#include <vector>
@ -55,6 +56,10 @@ namespace logservice
{
namespace coordinator
{
int PriorityV0::refresh_(const share::ObLSID &)
{
return OB_SUCCESS;
}
int PriorityV1::refresh_(const share::ObLSID &)
{
return OB_SUCCESS;
@ -113,7 +118,7 @@ public:
};
template <typename TAKEOVER_OP>
vector<ElectionImpl *> create_election_group(vector<ElectionPriorityImpl> &v_pri, TAKEOVER_OP &&op)
vector<ElectionImpl *> create_election_group(vector<ElectionPriorityImpl> &v_pri, TAKEOVER_OP &&op, const vector<int> &v_port = {})
{
vector<ElectionImpl *> v;
int election_num = v_pri.size();
@ -121,8 +126,12 @@ vector<ElectionImpl *> create_election_group(vector<ElectionPriorityImpl> &v_pri
MemberList member_list;
ObArray<ObAddr> addr_list;
static int port = 1;
for (int i = 0; i < election_num; ++i)
addr_list.push_back(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", port + i));
if (v_port.empty())
for (int i = 0; i < election_num; ++i)
addr_list.push_back(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", port + i));
else
for (int port : v_port)
addr_list.push_back(ObAddr(ObAddr::VER::IPV4, "127.0.0.1", port));
palf::LogConfigVersion version;
version.proposal_id_ = 1;
version.config_seq_ = 1;
@ -131,7 +140,7 @@ vector<ElectionImpl *> create_election_group(vector<ElectionPriorityImpl> &v_pri
int ret = OB_SUCCESS;
for (int i = 0; i < election_num; ++i) {
ElectionImpl *election = new ElectionImpl();
election->self_addr_ = ObAddr(ObAddr::VER::IPV4, "127.0.0.1", port + i);
election->self_addr_ = ObAddr(ObAddr::VER::IPV4, "127.0.0.1", addr_list[i].port_);
v.push_back(election);
}
for (auto &election_1 : v) {
@ -145,7 +154,7 @@ vector<ElectionImpl *> create_election_group(vector<ElectionPriorityImpl> &v_pri
1,
&timer,
&GlobalNetService,
ObAddr(ObAddr::VER::IPV4, "127.0.0.1", port + index),
election->self_addr_,
true,
1,
[election](int64_t, const ObAddr &dest_addr) {
@ -357,9 +366,148 @@ TEST_F(TestElectionWithPriority, meet_fatal_failure)
delete election;
ASSERT_EQ(change_leader_from_prepare_change_leader_cb, false);
}
}
}
namespace oceanbase
{
namespace palf
{
namespace election
{
uint64_t ElectionImpl::get_ls_biggest_min_cluster_version_ever_seen_() const// 让port=5555的副本认为自己是A副本
{
#define PRINT_WRAPPER K(*this)
int ret = OB_SUCCESS;
uint64_t ls_biggest_min_cluster_version_ever_seen = 0;
//if (observer::ObServer::get_instance().is_arbitration_mode()) {
if (observer::ObServer::get_instance().is_arbitration_mode() || self_addr_.port_ == 5555) {
if (CLUSTER_CURRENT_VERSION < ls_biggest_min_cluster_version_ever_seen_.version_) {
ret = OB_ERR_UNEXPECTED;
LOG_NONE(ERROR, "ls_biggest_min_cluster_version_ever_seen_ less than arb binary version");
} else if (ls_biggest_min_cluster_version_ever_seen_.version_ == 0) {
LOG_NONE(WARN, "ls_biggest_min_cluster_version_ever_seen_ not setted yet");
}
ls_biggest_min_cluster_version_ever_seen = ls_biggest_min_cluster_version_ever_seen_.version_;
} else {
ls_biggest_min_cluster_version_ever_seen = std::max(GET_MIN_CLUSTER_VERSION(),
ls_biggest_min_cluster_version_ever_seen_.version_);
}
return ls_biggest_min_cluster_version_ever_seen;
#undef PRINT_WRAPPER
}
}
}
namespace unittest
{
TEST_F(TestElectionWithPriority, arb_server_split_vote_cause_not_set_priority)// 复现仲裁bug场景
{
// oceanbase::common::ObClusterVersion::get_instance().cluster_version_ = CLUSTER_VERSION_3_2_3_0;// 此时采用V0版本的优先级逻辑比较，投给IP
vector<ElectionPriorityImpl> v_pri(4);
for (auto &pri : v_pri)
init_pri(pri);
v_pri[0].priority_tuple_.element<1>().zone_priority_ = 3;//F
v_pri[1].priority_tuple_.element<1>().zone_priority_ = 2;//F
v_pri[2].priority_tuple_.element<1>().zone_priority_ = 1;//F
v_pri[3].priority_tuple_.element<1>().zone_priority_ = 0;//A
auto election_group = create_election_group(v_pri, [](){}, {1,2,3,5555/*仲裁*/});
election_group[0]->stop();// kill掉一个，还有2F1A
election_group[3]->set_inner_priority_seed(0ULL | static_cast<uint64_t>(palf::election::PRIORITY_SEED_BIT::SEED_NOT_NORMOL_REPLICA_BIT));
election_group[3]->reset_priority();// 移除优先级，模拟A副本，会投票给IP最小的副本
this_thread::sleep_for(chrono::seconds(5));// 等待选出第一任Leader
ASSERT_EQ(election_group[1]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(election_group[2]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(election_group[3]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(leader_takeover_times, 0);
ASSERT_EQ(leader_revoke_times, 0);
ASSERT_EQ(devote_to_be_leader_count, 0);
ASSERT_EQ(lease_expired_to_be_follower_count, 0);
ASSERT_EQ(change_leader_to_be_leader_count, 0);
ASSERT_EQ(change_leader_to_be_follower_count, 0);
ASSERT_EQ(stop_to_be_follower_count, 0);
for (auto &election : election_group)
election->stop();
this_thread::sleep_for(chrono::seconds(1));
for (auto &election : election_group)
delete election;
}
TEST_F(TestElectionWithPriority, arb_server_won_t_split_vote_cause_set_priority)// 测试修复后的行为
{
oceanbase::common::ObClusterVersion::get_instance().cluster_version_ = CLUSTER_VERSION_3_2_3_0;// 此时采用V0版本的优先级逻辑比较，投给port_number_较大的副本
vector<ElectionPriorityImpl> v_pri(4);
for (auto &pri : v_pri)
init_pri(pri);
// 优先级V0
v_pri[0].priority_tuple_.element<0>().port_number_ = 3;//F
v_pri[1].priority_tuple_.element<0>().port_number_ = 2;//F
v_pri[2].priority_tuple_.element<0>().port_number_ = 1;//F
v_pri[3].priority_tuple_.element<0>().port_number_ = 0;//A
// 优先级V1
v_pri[0].priority_tuple_.element<1>().zone_priority_ = 3;//F
v_pri[1].priority_tuple_.element<1>().zone_priority_ = 2;//F
v_pri[2].priority_tuple_.element<1>().zone_priority_ = 1;//F
v_pri[3].priority_tuple_.element<1>().zone_priority_ = 0;//A
auto election_group = create_election_group(v_pri, [](){}, {1,2,3,5555/*仲裁*/});
election_group[0]->stop();// kill掉一个，还有2F1A
election_group[3]->set_inner_priority_seed(0ULL | static_cast<uint64_t>(palf::election::PRIORITY_SEED_BIT::SEED_NOT_NORMOL_REPLICA_BIT));
this_thread::sleep_for(chrono::seconds(5));// 等待选出第一任Leader
ASSERT_EQ(election_group[1]->proposer_.role_, ObRole::LEADER);
ASSERT_EQ(election_group[2]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(election_group[3]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(leader_takeover_times, 1);
ASSERT_EQ(leader_revoke_times, 0);
ASSERT_EQ(devote_to_be_leader_count, 1);
ASSERT_EQ(lease_expired_to_be_follower_count, 0);
ASSERT_EQ(change_leader_to_be_leader_count, 0);
ASSERT_EQ(change_leader_to_be_follower_count, 0);
ASSERT_EQ(stop_to_be_follower_count, 0);
// 升级版本号
oceanbase::common::ObClusterVersion::get_instance().cluster_version_ = CLUSTER_VERSION_4_2_0_0;// 此时将根据V1版本的优先级，将leader切换至election 2
this_thread::sleep_for(chrono::seconds(2));// 等待执行切主
ASSERT_EQ(election_group[1]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(election_group[2]->proposer_.role_, ObRole::LEADER);
ASSERT_EQ(election_group[3]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(leader_takeover_times, 2);
ASSERT_EQ(leader_revoke_times, 1);
ASSERT_EQ(devote_to_be_leader_count, 1);
ASSERT_EQ(lease_expired_to_be_follower_count, 0);
ASSERT_EQ(change_leader_to_be_leader_count, 1);
ASSERT_EQ(change_leader_to_be_follower_count, 1);
ASSERT_EQ(stop_to_be_follower_count, 0);
// 给leader断网，触发无主选举
GlobalNetService.disconnect_two_side(election_group[2], election_group[1]);
GlobalNetService.disconnect_two_side(election_group[2], election_group[3]);
this_thread::sleep_for(chrono::seconds(5));// 等待leader卸任
// 恢复leader的网络，预期无主时，就算是仲裁副本也正确采用了V1版本的优先级
GlobalNetService.connect_two_side(election_group[2], election_group[1]);
GlobalNetService.connect_two_side(election_group[2], election_group[3]);
this_thread::sleep_for(chrono::seconds(5));// 等待无主选举
ASSERT_EQ(election_group[1]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(election_group[2]->proposer_.role_, ObRole::LEADER);
ASSERT_EQ(election_group[3]->proposer_.role_, ObRole::FOLLOWER);
ASSERT_EQ(leader_takeover_times, 3);
ASSERT_EQ(leader_revoke_times, 2);
ASSERT_EQ(devote_to_be_leader_count, 2);
ASSERT_EQ(lease_expired_to_be_follower_count, 1);
ASSERT_EQ(change_leader_to_be_leader_count, 1);
ASSERT_EQ(change_leader_to_be_follower_count, 1);
ASSERT_EQ(stop_to_be_follower_count, 0);
for (auto &election : election_group)
election->stop();
this_thread::sleep_for(chrono::seconds(1));
for (auto &election : election_group)
delete election;
}
}
}
int main(int argc, char **argv)
{