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

Fix compatible config related problems

Browse Source
This commit is contained in:
tino247
2023-01-10 04:11:44 +00:00
committed by ob-robot
parent 57adfb8442
commit 64e170f343
10 changed files with 85 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/observer/omt/ob_tenant_config.cpp
View File

@ -34,7 +34,7 @@ ObTenantConfig::ObTenantConfig() : ObTenantConfig(OB_INVALID_TENANT_ID)
}
ObTenantConfig::ObTenantConfig(uint64_t tenant_id)
: tenant_id_(tenant_id), current_version_(1),
: tenant_id_(tenant_id), current_version_(INITIAL_TENANT_CONF_VERSION),
mutex_(),
update_task_(), system_config_(), config_mgr_(nullptr),
lock_(ObLatchIds::CONFIG_LOCK), is_deleting_(false)

2
src/observer/omt/ob_tenant_config.h
View File

@ -33,6 +33,8 @@ class ObTenantConfigMgr;
class ObTenantConfig : public ObCommonConfig
{
public:
static const int64_t INITIAL_TENANT_CONF_VERSION = 1;
public:
class TenantConfigUpdateTask : public common::ObTimerTask
{

15
src/observer/omt/ob_tenant_config_mgr.cpp
View File

@ -514,6 +514,7 @@ void ObTenantConfigMgr::get_lease_request(share::ObLeaseRequest &lease_request)
} // for
}
// for __all_virtual_tenant_parameter_info
int ObTenantConfigMgr::get_all_tenant_config_info(common::ObArray<TenantConfigInfo> &all_config)
{
int ret = OB_SUCCESS;
@ -525,10 +526,18 @@ int ObTenantConfigMgr::get_all_tenant_config_info(common::ObArray<TenantConfigIn
for (ObConfigContainer::const_iterator iter = tenant_config->get_container().begin();
iter != tenant_config->get_container().end(); iter++) {
TenantConfigInfo config_info(tenant_id);
if (OB_FAIL(config_info.set_name(iter->first.str()))) {
if (0 == ObString("compatible").case_compare(iter->first.str())
&& !iter->second->value_updated()) {
if (OB_FAIL(config_info.set_value("0.0.0.0"))) {
LOG_WARN("set value fail", K(iter->second->str()), K(ret));
}
} else {
if (OB_FAIL(config_info.set_value(iter->second->str()))) {
LOG_WARN("set value fail", K(iter->second->str()), K(ret));
}
}
if (FAILEDx(config_info.set_name(iter->first.str()))) {
LOG_WARN("set name fail", K(iter->first.str()), K(ret));
} else if (OB_FAIL(config_info.set_value(iter->second->str()))) {
LOG_WARN("set value fail", K(iter->second->str()), K(ret));
} else if (OB_FAIL(config_info.set_info(iter->second->info()))) {
LOG_WARN("set info fail", K(iter->second->info()), K(ret));
} else if (OB_FAIL(config_info.set_section(iter->second->section()))) {

2
src/rootserver/ob_ddl_operator.cpp
View File

@ -5598,7 +5598,7 @@ int ObDDLOperator::init_tenant_config_(
ObConfigItem *item = NULL;
char svr_ip[OB_MAX_SERVER_ADDR_SIZE] = "ANY";
int64_t svr_port = 0;
int64_t config_version = 1; // hard code init version
int64_t config_version = omt::ObTenantConfig::INITIAL_TENANT_CONF_VERSION + 1;
FOREACH_X(config, tenant_config.get_configs(), OB_SUCC(ret)) {
const ObConfigStringKey key(config->key_.ptr());
if (OB_ISNULL(hard_code_config->get_container().get(key))

36
tools/upgrade/actions.py
View File

@ -109,11 +109,11 @@ def set_parameter(cur, parameter, value, timeout = 0):
sql = """alter system set {0} = '{1}'""".format(parameter, value)
logging.info(sql)
cur.execute(sql)
wait_parameter_sync(cur, parameter, value, timeout)
wait_parameter_sync(cur, False, parameter, value, timeout)
def get_ori_enable_ddl(cur, timeout):
ori_value_str = fetch_ori_enable_ddl(cur)
wait_parameter_sync(cur, 'enable_ddl', ori_value_str, timeout)
wait_parameter_sync(cur, False, 'enable_ddl', ori_value_str, timeout)
ori_value = (ori_value_str == 'True')
return ori_value
@ -179,9 +179,10 @@ def check_server_version_by_cluster(cur):
else:
logging.info("check server version success")
def check_parameter(cur, key, value):
sql = """select * from oceanbase.GV$OB_PARAMETERS
where name = '{0}' and value = '{1}'""".format(key, value)
def check_parameter(cur, is_tenant_config, key, value):
table_name = "GV$OB_PARAMETERS" if not is_tenant_config else "__all_virtual_tenant_parameter_info"
sql = """select * from oceanbase.{0}
where name = '{1}' and value = '{2}'""".format(table_name, key, value)
logging.info(sql)
cur.execute(sql)
result = cur.fetchall()
@ -192,9 +193,10 @@ def check_parameter(cur, key, value):
bret = False
return bret
def wait_parameter_sync(cur, key, value, timeout):
sql = """select count(*) as cnt from oceanbase.GV$OB_PARAMETERS
where name = '{0}' and value != '{1}'""".format(key, value)
def wait_parameter_sync(cur, is_tenant_config, key, value, timeout):
table_name = "GV$OB_PARAMETERS" if not is_tenant_config else "__all_virtual_tenant_parameter_info"
sql = """select count(*) as cnt from oceanbase.{0}
where name = '{1}' and value != '{2}'""".format(table_name, key, value)
times = (timeout if timeout > 0 else 60) / 5
while times >= 0:
logging.info(sql)
@ -219,11 +221,11 @@ def do_begin_upgrade(cur, timeout):
action_sql = "alter system begin upgrade"
rollback_sql = "alter system end upgrade"
if not check_parameter(cur, "enable_upgrade_mode", "True"):
if not check_parameter(cur, False, "enable_upgrade_mode", "True"):
logging.info(action_sql)
cur.execute(action_sql)
wait_parameter_sync(cur, "enable_upgrade_mode", "True", timeout)
wait_parameter_sync(cur, False, "enable_upgrade_mode", "True", timeout)
global g_succ_sql_list
g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -232,11 +234,11 @@ def do_begin_rolling_upgrade(cur, timeout):
action_sql = "alter system begin rolling upgrade"
rollback_sql = "alter system end upgrade"
if not check_parameter(cur, "_upgrade_stage", "DBUPGRADE"):
if not check_parameter(cur, False, "_upgrade_stage", "DBUPGRADE"):
logging.info(action_sql)
cur.execute(action_sql)
wait_parameter_sync(cur, "_upgrade_stage", "DBUPGRADE", timeout)
wait_parameter_sync(cur, False, "_upgrade_stage", "DBUPGRADE", timeout)
global g_succ_sql_list
g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -247,12 +249,12 @@ def do_end_rolling_upgrade(cur, timeout):
action_sql = "alter system end rolling upgrade"
rollback_sql = "alter system end upgrade"
if not check_parameter(cur, "_upgrade_stage", "POSTUPGRADE") or not check_parameter(cur, "min_observer_version", current_cluster_version):
if not check_parameter(cur, False, "_upgrade_stage", "POSTUPGRADE") or not check_parameter(cur, False, "min_observer_version", current_cluster_version):
logging.info(action_sql)
cur.execute(action_sql)
wait_parameter_sync(cur, "min_observer_version", current_data_version, timeout)
wait_parameter_sync(cur, "_upgrade_stage", "POSTUPGRADE", timeout)
wait_parameter_sync(cur, False, "min_observer_version", current_data_version, timeout)
wait_parameter_sync(cur, False, "_upgrade_stage", "POSTUPGRADE", timeout)
global g_succ_sql_list
g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -261,11 +263,11 @@ def do_end_upgrade(cur, timeout):
action_sql = "alter system end upgrade"
rollback_sql = ""
if not check_parameter(cur, "enable_upgrade_mode", "False"):
if not check_parameter(cur, False, "enable_upgrade_mode", "False"):
logging.info(action_sql)
cur.execute(action_sql)
wait_parameter_sync(cur, "enable_upgrade_mode", "False", timeout)
wait_parameter_sync(cur, False, "enable_upgrade_mode", "False", timeout)
global g_succ_sql_list
g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))

2
tools/upgrade/tenant_upgrade_action.py
View File

@ -83,7 +83,7 @@ def upgrade_across_version(cur):
# 2. check if compatible match with current_data_version
if not across_version:
sql = "select count(*) from oceanbase.GV$OB_PARAMETERS where name = 'compatible' and value != '{0}'".format(current_data_version)
sql = "select count(*) from oceanbase.__all_virtual_tenant_parameter_info where name = 'compatible' and value != '{0}'".format(current_data_version)
results = query(cur, sql)
if len(results) < 1 or len(results[0]) < 1:
logging.warn("row/column cnt not match")

2
tools/upgrade/upgrade_checker.py
View File

@ -332,7 +332,7 @@ def check_data_version(query_cur):
data_version_str = ''
data_version = 0
# check compatible is same
sql = """select distinct value from GV$OB_PARAMETERS where name='compatible'"""
sql = """select distinct value from oceanbase.__all_virtual_tenant_parameter_info where name='compatible'"""
(desc, results) = query_cur.exec_query(sql)
if len(results) != 1:
fail_list.append('compatible is not sync')

44
tools/upgrade/upgrade_post.py
View File

@ -117,11 +117,11 @@
# sql = """alter system set {0} = '{1}'""".format(parameter, value)
# logging.info(sql)
# cur.execute(sql)
# wait_parameter_sync(cur, parameter, value, timeout)
# wait_parameter_sync(cur, False, parameter, value, timeout)
#
#def get_ori_enable_ddl(cur, timeout):
# ori_value_str = fetch_ori_enable_ddl(cur)
# wait_parameter_sync(cur, 'enable_ddl', ori_value_str, timeout)
# wait_parameter_sync(cur, False, 'enable_ddl', ori_value_str, timeout)
# ori_value = (ori_value_str == 'True')
# return ori_value
#
@ -187,9 +187,10 @@
# else:
# logging.info("check server version success")
#
#def check_parameter(cur, key, value):
# sql = """select * from oceanbase.GV$OB_PARAMETERS
# where name = '{0}' and value = '{1}'""".format(key, value)
#def check_parameter(cur, is_tenant_config, key, value):
# table_name = "GV$OB_PARAMETERS" if not is_tenant_config else "__all_virtual_tenant_parameter_info"
# sql = """select * from oceanbase.{0}
# where name = '{1}' and value = '{2}'""".format(table_name, key, value)
# logging.info(sql)
# cur.execute(sql)
# result = cur.fetchall()
@ -200,9 +201,10 @@
# bret = False
# return bret
#
#def wait_parameter_sync(cur, key, value, timeout):
# sql = """select count(*) as cnt from oceanbase.GV$OB_PARAMETERS
# where name = '{0}' and value != '{1}'""".format(key, value)
#def wait_parameter_sync(cur, is_tenant_config, key, value, timeout):
# table_name = "GV$OB_PARAMETERS" if not is_tenant_config else "__all_virtual_tenant_parameter_info"
# sql = """select count(*) as cnt from oceanbase.{0}
# where name = '{1}' and value != '{2}'""".format(table_name, key, value)
# times = (timeout if timeout > 0 else 60) / 5
# while times >= 0:
# logging.info(sql)
@ -227,11 +229,11 @@
# action_sql = "alter system begin upgrade"
# rollback_sql = "alter system end upgrade"
#
# if not check_parameter(cur, "enable_upgrade_mode", "True"):
# if not check_parameter(cur, False, "enable_upgrade_mode", "True"):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "enable_upgrade_mode", "True", timeout)
# wait_parameter_sync(cur, False, "enable_upgrade_mode", "True", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -240,11 +242,11 @@
# action_sql = "alter system begin rolling upgrade"
# rollback_sql = "alter system end upgrade"
#
# if not check_parameter(cur, "_upgrade_stage", "DBUPGRADE"):
# if not check_parameter(cur, False, "_upgrade_stage", "DBUPGRADE"):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "_upgrade_stage", "DBUPGRADE", timeout)
# wait_parameter_sync(cur, False, "_upgrade_stage", "DBUPGRADE", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -255,12 +257,12 @@
# action_sql = "alter system end rolling upgrade"
# rollback_sql = "alter system end upgrade"
#
# if not check_parameter(cur, "_upgrade_stage", "POSTUPGRADE") or not check_parameter(cur, "min_observer_version", current_cluster_version):
# if not check_parameter(cur, False, "_upgrade_stage", "POSTUPGRADE") or not check_parameter(cur, False, "min_observer_version", current_cluster_version):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "min_observer_version", current_data_version, timeout)
# wait_parameter_sync(cur, "_upgrade_stage", "POSTUPGRADE", timeout)
# wait_parameter_sync(cur, False, "min_observer_version", current_data_version, timeout)
# wait_parameter_sync(cur, False, "_upgrade_stage", "POSTUPGRADE", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -269,11 +271,11 @@
# action_sql = "alter system end upgrade"
# rollback_sql = ""
#
# if not check_parameter(cur, "enable_upgrade_mode", "False"):
# if not check_parameter(cur, False, "enable_upgrade_mode", "False"):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "enable_upgrade_mode", "False", timeout)
# wait_parameter_sync(cur, False, "enable_upgrade_mode", "False", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -1319,7 +1321,7 @@
#
# # 2. check if compatible match with current_data_version
# if not across_version:
# sql = "select count(*) from oceanbase.GV$OB_PARAMETERS where name = 'compatible' and value != '{0}'".format(current_data_version)
# sql = "select count(*) from oceanbase.__all_virtual_tenant_parameter_info where name = 'compatible' and value != '{0}'".format(current_data_version)
# results = query(cur, sql)
# if len(results) < 1 or len(results[0]) < 1:
# logging.warn("row/column cnt not match")
@ -1808,7 +1810,7 @@
# data_version_str = ''
# data_version = 0
# # check compatible is same
# sql = """select distinct value from GV$OB_PARAMETERS where name='compatible'"""
# sql = """select distinct value from oceanbase.__all_virtual_tenant_parameter_info where name='compatible'"""
# (desc, results) = query_cur.exec_query(sql)
# if len(results) != 1:
# fail_list.append('compatible is not sync')
@ -2410,13 +2412,13 @@
## 1 检查版本号
#def check_cluster_version(cur, timeout):
# current_cluster_version = actions.get_current_cluster_version()
# actions.wait_parameter_sync(cur, "min_observer_version", current_cluster_version, timeout)
# actions.wait_parameter_sync(cur, False, "min_observer_version", current_cluster_version, timeout)
#
## 2 检查租户版本号
#def check_data_version(cur, query_cur, timeout):
# # check compatible
# current_data_version = actions.get_current_data_version()
# actions.wait_parameter_sync(cur, "compatible", current_data_version, 10)
# actions.wait_parameter_sync(cur, True, "compatible", current_data_version, 10)
#
# # check target_data_version/current_data_version except standby tenant
# sql = "select tenant_id from oceanbase.__all_tenant except select tenant_id from oceanbase.__all_virtual_tenant_info where tenant_role = 'STANDBY'"

4
tools/upgrade/upgrade_post_checker.py
View File

@ -14,13 +14,13 @@ import actions
# 1 检查版本号
def check_cluster_version(cur, timeout):
current_cluster_version = actions.get_current_cluster_version()
actions.wait_parameter_sync(cur, "min_observer_version", current_cluster_version, timeout)
actions.wait_parameter_sync(cur, False, "min_observer_version", current_cluster_version, timeout)
# 2 检查租户版本号
def check_data_version(cur, query_cur, timeout):
# check compatible
current_data_version = actions.get_current_data_version()
actions.wait_parameter_sync(cur, "compatible", current_data_version, 10)
actions.wait_parameter_sync(cur, True, "compatible", current_data_version, 10)
# check target_data_version/current_data_version except standby tenant
sql = "select tenant_id from oceanbase.__all_tenant except select tenant_id from oceanbase.__all_virtual_tenant_info where tenant_role = 'STANDBY'"

44
tools/upgrade/upgrade_pre.py
View File

@ -117,11 +117,11 @@
# sql = """alter system set {0} = '{1}'""".format(parameter, value)
# logging.info(sql)
# cur.execute(sql)
# wait_parameter_sync(cur, parameter, value, timeout)
# wait_parameter_sync(cur, False, parameter, value, timeout)
#
#def get_ori_enable_ddl(cur, timeout):
# ori_value_str = fetch_ori_enable_ddl(cur)
# wait_parameter_sync(cur, 'enable_ddl', ori_value_str, timeout)
# wait_parameter_sync(cur, False, 'enable_ddl', ori_value_str, timeout)
# ori_value = (ori_value_str == 'True')
# return ori_value
#
@ -187,9 +187,10 @@
# else:
# logging.info("check server version success")
#
#def check_parameter(cur, key, value):
# sql = """select * from oceanbase.GV$OB_PARAMETERS
# where name = '{0}' and value = '{1}'""".format(key, value)
#def check_parameter(cur, is_tenant_config, key, value):
# table_name = "GV$OB_PARAMETERS" if not is_tenant_config else "__all_virtual_tenant_parameter_info"
# sql = """select * from oceanbase.{0}
# where name = '{1}' and value = '{2}'""".format(table_name, key, value)
# logging.info(sql)
# cur.execute(sql)
# result = cur.fetchall()
@ -200,9 +201,10 @@
# bret = False
# return bret
#
#def wait_parameter_sync(cur, key, value, timeout):
# sql = """select count(*) as cnt from oceanbase.GV$OB_PARAMETERS
# where name = '{0}' and value != '{1}'""".format(key, value)
#def wait_parameter_sync(cur, is_tenant_config, key, value, timeout):
# table_name = "GV$OB_PARAMETERS" if not is_tenant_config else "__all_virtual_tenant_parameter_info"
# sql = """select count(*) as cnt from oceanbase.{0}
# where name = '{1}' and value != '{2}'""".format(table_name, key, value)
# times = (timeout if timeout > 0 else 60) / 5
# while times >= 0:
# logging.info(sql)
@ -227,11 +229,11 @@
# action_sql = "alter system begin upgrade"
# rollback_sql = "alter system end upgrade"
#
# if not check_parameter(cur, "enable_upgrade_mode", "True"):
# if not check_parameter(cur, False, "enable_upgrade_mode", "True"):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "enable_upgrade_mode", "True", timeout)
# wait_parameter_sync(cur, False, "enable_upgrade_mode", "True", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -240,11 +242,11 @@
# action_sql = "alter system begin rolling upgrade"
# rollback_sql = "alter system end upgrade"
#
# if not check_parameter(cur, "_upgrade_stage", "DBUPGRADE"):
# if not check_parameter(cur, False, "_upgrade_stage", "DBUPGRADE"):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "_upgrade_stage", "DBUPGRADE", timeout)
# wait_parameter_sync(cur, False, "_upgrade_stage", "DBUPGRADE", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -255,12 +257,12 @@
# action_sql = "alter system end rolling upgrade"
# rollback_sql = "alter system end upgrade"
#
# if not check_parameter(cur, "_upgrade_stage", "POSTUPGRADE") or not check_parameter(cur, "min_observer_version", current_cluster_version):
# if not check_parameter(cur, False, "_upgrade_stage", "POSTUPGRADE") or not check_parameter(cur, False, "min_observer_version", current_cluster_version):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "min_observer_version", current_data_version, timeout)
# wait_parameter_sync(cur, "_upgrade_stage", "POSTUPGRADE", timeout)
# wait_parameter_sync(cur, False, "min_observer_version", current_data_version, timeout)
# wait_parameter_sync(cur, False, "_upgrade_stage", "POSTUPGRADE", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -269,11 +271,11 @@
# action_sql = "alter system end upgrade"
# rollback_sql = ""
#
# if not check_parameter(cur, "enable_upgrade_mode", "False"):
# if not check_parameter(cur, False, "enable_upgrade_mode", "False"):
# logging.info(action_sql)
# cur.execute(action_sql)
#
# wait_parameter_sync(cur, "enable_upgrade_mode", "False", timeout)
# wait_parameter_sync(cur, False, "enable_upgrade_mode", "False", timeout)
#
# global g_succ_sql_list
# g_succ_sql_list.append(SqlItem(action_sql, rollback_sql))
@ -1319,7 +1321,7 @@
#
# # 2. check if compatible match with current_data_version
# if not across_version:
# sql = "select count(*) from oceanbase.GV$OB_PARAMETERS where name = 'compatible' and value != '{0}'".format(current_data_version)
# sql = "select count(*) from oceanbase.__all_virtual_tenant_parameter_info where name = 'compatible' and value != '{0}'".format(current_data_version)
# results = query(cur, sql)
# if len(results) < 1 or len(results[0]) < 1:
# logging.warn("row/column cnt not match")
@ -1808,7 +1810,7 @@
# data_version_str = ''
# data_version = 0
# # check compatible is same
# sql = """select distinct value from GV$OB_PARAMETERS where name='compatible'"""
# sql = """select distinct value from oceanbase.__all_virtual_tenant_parameter_info where name='compatible'"""
# (desc, results) = query_cur.exec_query(sql)
# if len(results) != 1:
# fail_list.append('compatible is not sync')
@ -2410,13 +2412,13 @@
## 1 检查版本号
#def check_cluster_version(cur, timeout):
# current_cluster_version = actions.get_current_cluster_version()
# actions.wait_parameter_sync(cur, "min_observer_version", current_cluster_version, timeout)
# actions.wait_parameter_sync(cur, False, "min_observer_version", current_cluster_version, timeout)
#
## 2 检查租户版本号
#def check_data_version(cur, query_cur, timeout):
# # check compatible
# current_data_version = actions.get_current_data_version()
# actions.wait_parameter_sync(cur, "compatible", current_data_version, 10)
# actions.wait_parameter_sync(cur, True, "compatible", current_data_version, 10)
#
# # check target_data_version/current_data_version except standby tenant
# sql = "select tenant_id from oceanbase.__all_tenant except select tenant_id from oceanbase.__all_virtual_tenant_info where tenant_role = 'STANDBY'"