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

Optimize the calling method of proxy in recovery ls service

Browse Source
This commit is contained in:
obdev
2023-06-06 11:12:39 +00:00
committed by ob-robot
parent f8ddffa9b9
commit 7eae7f33bb
2 changed files with 100 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/rootserver/ob_recovery_ls_service.cpp
View File

@ -43,7 +43,6 @@
#include "storage/tx/ob_tx_log.h" //ObTxLogHeader #include "storage/tx/ob_tx_log.h" //ObTxLogHeader
#include "storage/tx_storage/ob_ls_service.h" //ObLSService #include "storage/tx_storage/ob_ls_service.h" //ObLSService
#include "storage/tx_storage/ob_ls_handle.h" //ObLSHandle #include "storage/tx_storage/ob_ls_handle.h" //ObLSHandle
#include "share/ob_log_restore_proxy.h" // ObLogRestoreProxyUtil
#include "src/rootserver/ob_rs_event_history_table_operator.h" #include "src/rootserver/ob_rs_event_history_table_operator.h"
namespace oceanbase namespace oceanbase
@ -56,6 +55,21 @@ using namespace palf;
namespace rootserver namespace rootserver
{ {
#define RESTORE_EVENT_ADD \
int ret_code = OB_SUCCESS; \
switch (ret) { \
case -ER_ACCESS_DENIED_ERROR: \
ret_code = OB_PASSWORD_WRONG; \
case -ER_CONNECT_FAILED: \
ret_code = OB_CONNECT_ERROR; \
} \
ROOTSERVICE_EVENT_ADD("root_service", "update_primary_ip_list", \
"tenant_id", tenant_id_, K(ret), \
"ob_error_name", ob_error_name(ret_code), \
"ob_error_str", ob_strerror(ret_code), \
"primary_user_tenant", user_and_tenant.ptr(), \
"primary_ip_list", service_attr.addr_); \
int ObRecoveryLSService::init() int ObRecoveryLSService::init()
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
@ -83,6 +97,7 @@ void ObRecoveryLSService::destroy()
inited_ = false; inited_ = false;
tenant_id_ = OB_INVALID_TENANT_ID; tenant_id_ = OB_INVALID_TENANT_ID;
proxy_ = NULL; proxy_ = NULL;
restore_proxy_.destroy();
primary_is_avaliable_= true; primary_is_avaliable_= true;
} }
@ -873,14 +888,13 @@ void ObRecoveryLSService::try_update_primary_ip_list()
uint64_t primary_tenant_id; uint64_t primary_tenant_id;
char passwd[OB_MAX_PASSWORD_LENGTH + 1] = { 0 }; //unencrypted password char passwd[OB_MAX_PASSWORD_LENGTH + 1] = { 0 }; //unencrypted password
SMART_VAR(ObLogRestoreProxyUtil, proxy) {
if (OB_FAIL(restore_source_mgr.init(tenant_id_, proxy_))) { if (OB_FAIL(restore_source_mgr.init(tenant_id_, proxy_))) {
LOG_WARN("fail to init restore_source_mgr", K_(tenant_id)); LOG_WARN("fail to init restore_source_mgr", K_(tenant_id));
} else if (OB_FAIL(restore_source_mgr.get_source(item))) { } else if (OB_FAIL(restore_source_mgr.get_source(item))) {
LOG_WARN("get source failed", K_(tenant_id)); LOG_WARN("get source failed", K_(tenant_id));
} else if (!check_need_update_ip_list(item)) { } else if (!check_need_update_ip_list_(item)) {
LOG_INFO("there is no log restore source record or the log restore source type is not service" , K(item)); LOG_INFO("there is no log restore source record or the log restore source type is not service" , K(item));
} else if (OB_FAIL(get_restore_source_value(item, standby_source_value))) { } else if (OB_FAIL(get_restore_source_value_(item, standby_source_value))) {
LOG_WARN("fail to get service standby log_restore source value", K(item)); LOG_WARN("fail to get service standby log_restore source value", K(item));
} else if (OB_FAIL(service_attr.parse_service_attr_from_str(standby_source_value))) { } else if (OB_FAIL(service_attr.parse_service_attr_from_str(standby_source_value))) {
LOG_WARN("fail to parse service attr", K(item), K(standby_source_value)); LOG_WARN("fail to parse service attr", K(item), K(standby_source_value));
@ -891,38 +905,21 @@ void ObRecoveryLSService::try_update_primary_ip_list()
LOG_WARN("get servcie attr password failed", K(service_attr)); LOG_WARN("get servcie attr password failed", K(service_attr));
} else if (OB_FAIL(service_attr.get_user_str_(user_and_tenant))) { } else if (OB_FAIL(service_attr.get_user_str_(user_and_tenant))) {
LOG_WARN("get user str failed", K(service_attr.user_.user_name_), K(service_attr.user_.tenant_name_)); LOG_WARN("get user str failed", K(service_attr.user_.user_name_), K(service_attr.user_.tenant_name_));
} else if (OB_FAIL(proxy.init(tenant_id_/*standby*/, } else if (!primary_is_avaliable_ && restore_proxy_.is_inited()) {
LOG_WARN("primary is not avaliable, retry init restore proxy");
restore_proxy_.destroy();
} else if (!restore_proxy_.is_inited() && OB_FAIL(restore_proxy_.init(tenant_id_/*standby*/,
service_attr.addr_, service_attr.addr_,
user_and_tenant.ptr(), user_and_tenant.ptr(),
passwd, passwd,
service_attr.user_.mode_ == ObCompatibilityMode::MYSQL_MODE ? OB_SYS_DATABASE_NAME : OB_ORA_SYS_SCHEMA_NAME))) { service_attr.user_.mode_ == ObCompatibilityMode::MYSQL_MODE ? OB_SYS_DATABASE_NAME : OB_ORA_SYS_SCHEMA_NAME))) {
LOG_WARN("proxy fail to connect to primary", K_(tenant_id), K(service_attr.addr_), K(user_and_tenant)); LOG_WARN("restore_proxy_ fail to connect to primary", K_(tenant_id), K(service_attr.addr_), K(user_and_tenant));
bool cur_primary_state = false; } else if (OB_FAIL(restore_proxy_.get_cluster_id(service_attr.user_.tenant_id_, primary_cluster_id))) {
if (cur_primary_state != primary_is_avaliable_) { LOG_WARN("restore proxy fail to get primary cluster id", K(service_attr.user_.tenant_id_));
primary_is_avaliable_ = false; } else if (OB_FAIL(restore_proxy_.get_tenant_id(service_attr.user_.tenant_name_, primary_tenant_id))) {
LOG_WARN("standby recovery ls service state changed"); LOG_WARN("restore proxy fail to get primary tenant id", K(service_attr.user_.tenant_name_));
if (-ER_ACCESS_DENIED_ERROR == ret) { } else if (OB_FAIL(restore_proxy_.get_server_ip_list(service_attr.user_.tenant_id_, primary_addrs))) {
ROOTSERVICE_EVENT_ADD("root_service", "update_primary_ip_list", LOG_WARN("restore proxy fail to get primary server ip list", K(service_attr.user_.tenant_id_));
"tenant_id", tenant_id_, K(ret),
"ob_error_name", ob_error_name(OB_PASSWORD_WRONG),
"ob_error_str", ob_strerror(OB_PASSWORD_WRONG),
"primary_user_tenant", user_and_tenant.ptr(),
"primary_ip_list", service_attr.addr_);
} else if (-ER_CONNECT_FAILED == ret) {
ROOTSERVICE_EVENT_ADD("root_service", "update_primary_ip_list",
"tenant_id", tenant_id_, K(ret),
"ob_error_name", ob_error_name(OB_CONNECT_ERROR),
"ob_error_str", ob_strerror(OB_CONNECT_ERROR),
"primary_user_tenant", user_and_tenant.ptr(),
"primary_ip_list", service_attr.addr_);
}
}
} else if (OB_FAIL(proxy.get_cluster_id(service_attr.user_.tenant_id_, primary_cluster_id))) {
LOG_WARN("proxy fail to get primary cluster id", K(service_attr.user_.tenant_id_));
} else if (OB_FAIL(proxy.get_tenant_id(service_attr.user_.tenant_name_, primary_tenant_id))) {
LOG_WARN("proxy fail to get primary tenant id", K(service_attr.user_.tenant_name_));
} else if (OB_FAIL(proxy.get_server_ip_list(service_attr.user_.tenant_id_, primary_addrs))) {
LOG_WARN("fail to get primary server ip list", K(service_attr.user_.tenant_id_));
} else if (primary_addrs.empty()) { } else if (primary_addrs.empty()) {
ret = OB_INVALID_ARGUMENT; ret = OB_INVALID_ARGUMENT;
LOG_WARN("tenant ip list is empty", K(primary_addrs)); LOG_WARN("tenant ip list is empty", K(primary_addrs));
@ -943,7 +940,10 @@ void ObRecoveryLSService::try_update_primary_ip_list()
bool addr_is_same = false; bool addr_is_same = false;
addr_is_same = service_attr.compare_addr_(primary_addrs); addr_is_same = service_attr.compare_addr_(primary_addrs);
if (!addr_is_same) { if (!addr_is_same) {
common::ObArray<common::ObAddr> tmp_addr = service_attr.addr_; common::ObArray<common::ObAddr> tmp_addr;
if (OB_FAIL(tmp_addr.assign(service_attr.addr_))) {
LOG_WARN("fail to assign service attr addr", K(service_attr.addr_));
} else {
service_attr.addr_.reset(); service_attr.addr_.reset();
ARRAY_FOREACH_N(primary_addrs, idx, cnt) { ARRAY_FOREACH_N(primary_addrs, idx, cnt) {
if (OB_FAIL(service_attr.addr_.push_back(primary_addrs.at(idx)))) { if (OB_FAIL(service_attr.addr_.push_back(primary_addrs.at(idx)))) {
@ -953,26 +953,34 @@ void ObRecoveryLSService::try_update_primary_ip_list()
LOG_INFO("primary ip list has changed", K(addr_is_same), K(primary_addrs), K(service_attr.addr_)); LOG_INFO("primary ip list has changed", K(addr_is_same), K(primary_addrs), K(service_attr.addr_));
if (OB_FAIL(ret)) { if (OB_FAIL(ret)) {
LOG_WARN("fail to update primary ip list"); LOG_WARN("fail to update primary ip list");
} else if (OB_FAIL(do_update_restore_source(service_attr, restore_source_mgr))) { } else if (OB_FAIL(do_update_restore_source_(service_attr, restore_source_mgr))) {
LOG_WARN("fail to update restore source", K(service_attr)); LOG_WARN("fail to update restore source", K(service_attr));
} }
ROOTSERVICE_EVENT_ADD("root_service", "update_primary_ip_list", ROOTSERVICE_EVENT_ADD("root_service", "update_primary_ip_list",
"tenant_id", tenant_id_, "tenant_id", tenant_id_,
"info", "do update primary ip list", "info", "do update primary ip list",
"primary_user_tenant", user_and_tenant.ptr(), "primary_user_tenant", user_and_tenant.ptr(),
"new_primary_ip_list", tmp_addr, "new_primary_ip_list", primary_addrs,
"old_primary_ip_list", primary_addrs); "old_primary_ip_list", tmp_addr);
} }
} }
} }
if (OB_FAIL(ret)) {
bool cur_primary_state = false;
if (cur_primary_state != primary_is_avaliable_) {
primary_is_avaliable_ = false;
LOG_WARN("standby recovery ls service state changed");
RESTORE_EVENT_ADD;
}
}
} }
bool ObRecoveryLSService::check_need_update_ip_list(ObLogRestoreSourceItem &item) bool ObRecoveryLSService::check_need_update_ip_list_(ObLogRestoreSourceItem &item)
{ {
return item.is_valid() && ObLogRestoreSourceType::SERVICE == item.type_; return item.is_valid() && ObLogRestoreSourceType::SERVICE == item.type_;
} }
int ObRecoveryLSService::get_restore_source_value(ObLogRestoreSourceItem &item, ObSqlString &standby_source_value) int ObRecoveryLSService::get_restore_source_value_(ObLogRestoreSourceItem &item, ObSqlString &standby_source_value)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
ObSqlString value; ObSqlString value;
@ -985,7 +993,7 @@ int ObRecoveryLSService::get_restore_source_value(ObLogRestoreSourceItem &item,
return ret; return ret;
} }
int ObRecoveryLSService::do_update_restore_source( int ObRecoveryLSService::do_update_restore_source_(
ObRestoreSourceServiceAttr &old_attr, ObRestoreSourceServiceAttr &old_attr,
ObLogRestoreSourceMgr &restore_source_mgr) ObLogRestoreSourceMgr &restore_source_mgr)
{ {
@ -1005,7 +1013,7 @@ int ObRecoveryLSService::do_update_restore_source(
} else if (ObLogRestoreSourceType::SERVICE != tmp_item.type_) { } else if (ObLogRestoreSourceType::SERVICE != tmp_item.type_) {
ret = OB_EAGAIN; ret = OB_EAGAIN;
LOG_WARN("log restore source type is not service", K_(tenant_id), K(tmp_item.type_)); LOG_WARN("log restore source type is not service", K_(tenant_id), K(tmp_item.type_));
} else if (OB_FAIL(get_restore_source_value(tmp_item, tmp_standby_source_value))) { } else if (OB_FAIL(get_restore_source_value_(tmp_item, tmp_standby_source_value))) {
LOG_WARN("fail to get service standby log restore source value", K(tmp_item)); LOG_WARN("fail to get service standby log restore source value", K(tmp_item));
} else if (OB_FAIL(tmp_service_attr.parse_service_attr_from_str(tmp_standby_source_value))) { } else if (OB_FAIL(tmp_service_attr.parse_service_attr_from_str(tmp_standby_source_value))) {
LOG_WARN("fail to parse service attr", K(tmp_item), K(tmp_standby_source_value)); LOG_WARN("fail to parse service attr", K(tmp_item), K(tmp_standby_source_value));
@ -1018,7 +1026,7 @@ int ObRecoveryLSService::do_update_restore_source(
} else if (!(0 == STRCMP(tmp_service_attr.encrypt_passwd_, old_attr.encrypt_passwd_))) { } else if (!(0 == STRCMP(tmp_service_attr.encrypt_passwd_, old_attr.encrypt_passwd_))) {
ret = OB_EAGAIN; ret = OB_EAGAIN;
LOG_WARN("log restore source password may be modified", K_(tenant_id)); LOG_WARN("log restore source password may be modified", K_(tenant_id));
} else if (OB_FAIL(update_source_inner_table(updated_value_str, sizeof(updated_value_str), trans, tmp_item))) { } else if (OB_FAIL(update_source_inner_table_(updated_value_str, sizeof(updated_value_str), trans, tmp_item))) {
LOG_WARN("fail to add service source", K(updated_value_str), K(tmp_item.until_scn_)); LOG_WARN("fail to add service source", K(updated_value_str), K(tmp_item.until_scn_));
} }
@ -1032,7 +1040,10 @@ int ObRecoveryLSService::do_update_restore_source(
return ret; return ret;
} }
int ObRecoveryLSService::update_source_inner_table(char *buf, const int64_t buf_size, ObMySQLTransaction &trans, const ObLogRestoreSourceItem &item) int ObRecoveryLSService::update_source_inner_table_(char *buf,
const int64_t buf_size,
ObMySQLTransaction &trans,
const ObLogRestoreSourceItem &item)
{ {
int ret = OB_SUCCESS; int ret = OB_SUCCESS;
int64_t affected_rows = 0; int64_t affected_rows = 0;

15
src/rootserver/ob_recovery_ls_service.h
View File

@ -22,6 +22,7 @@
#include "src/share/restore/ob_log_restore_source.h" //ObLogRestoreSourceItem #include "src/share/restore/ob_log_restore_source.h" //ObLogRestoreSourceItem
#include "src/share/backup/ob_backup_struct.h" //ObRestoreSourceServiceAttr #include "src/share/backup/ob_backup_struct.h" //ObRestoreSourceServiceAttr
#include "share/restore/ob_log_restore_source_mgr.h" //ObLogRestoreSourceMgr #include "share/restore/ob_log_restore_source_mgr.h" //ObLogRestoreSourceMgr
#include "share/ob_log_restore_proxy.h" // ObLogRestoreProxyUtil
namespace oceanbase namespace oceanbase
{ {
@ -67,8 +68,8 @@ namespace rootserver
class ObRecoveryLSService : public ObTenantThreadHelper class ObRecoveryLSService : public ObTenantThreadHelper
{ {
public: public:
ObRecoveryLSService() : inited_(false), ObRecoveryLSService() : inited_(false), tenant_id_(OB_INVALID_TENANT_ID), proxy_(NULL),
tenant_id_(OB_INVALID_TENANT_ID), proxy_(NULL), primary_is_avaliable_(true) {} restore_proxy_(), primary_is_avaliable_(true) {}
virtual ~ObRecoveryLSService() {} virtual ~ObRecoveryLSService() {}
int init(); int init();
void destroy(); void destroy();
@ -104,15 +105,17 @@ private:
int get_min_data_version_(uint64_t &compatible); int get_min_data_version_(uint64_t &compatible);
int process_ls_operator_in_trans_(const share::ObLSAttr &ls_attr, int process_ls_operator_in_trans_(const share::ObLSAttr &ls_attr,
const share::SCN &sync_scn, common::ObMySQLTransaction &trans); const share::SCN &sync_scn, common::ObMySQLTransaction &trans);
int reset_restore_proxy_(ObRestoreSourceServiceAttr &service_attr);
void try_update_primary_ip_list(); void try_update_primary_ip_list();
bool check_need_update_ip_list(share::ObLogRestoreSourceItem &item); bool check_need_update_ip_list_(share::ObLogRestoreSourceItem &item);
int get_restore_source_value(ObLogRestoreSourceItem &item, ObSqlString &standby_source_value); int get_restore_source_value_(ObLogRestoreSourceItem &item, ObSqlString &standby_source_value);
int do_update_restore_source(ObRestoreSourceServiceAttr &old_attr, ObLogRestoreSourceMgr &restore_source_mgr); int do_update_restore_source_(ObRestoreSourceServiceAttr &old_attr, ObLogRestoreSourceMgr &restore_source_mgr);
int update_source_inner_table(char *buf, const int64_t buf_size, ObMySQLTransaction &trans, const ObLogRestoreSourceItem &item); int update_source_inner_table_(char *buf, const int64_t buf_size, ObMySQLTransaction &trans, const ObLogRestoreSourceItem &item);
private: private:
bool inited_; bool inited_;
uint64_t tenant_id_; uint64_t tenant_id_;
common::ObMySQLProxy *proxy_; common::ObMySQLProxy *proxy_;
ObLogRestoreProxyUtil restore_proxy_;
bool primary_is_avaliable_; bool primary_is_avaliable_;
}; };
} }