289 lines
18 KiB
Python
Executable File
289 lines
18 KiB
Python
Executable File
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from my_error import MyError
|
|
from actions import BaseEachTenantDDLAction
|
|
from actions import reflect_action_cls_list
|
|
from actions import fetch_observer_version
|
|
from actions import QueryCursor
|
|
import logging
|
|
import time
|
|
import my_utils
|
|
import actions
|
|
|
|
'''
|
|
添加一条each tenant ddl的方法:
|
|
|
|
在本文件中，添加一个类名以"EachTenantDDLActionPost"开头并且继承自BaseEachTenantDDLAction的类，
|
|
然后在这个类中实现以下成员函数，并且每个函数执行出错都要抛错：
|
|
(1)@staticmethod get_seq_num():
|
|
返回一个代表着执行顺序的序列号，该序列号在本文件中不允许重复，若有重复则会报错。
|
|
(2)dump_before_do_action(self):
|
|
执行action sql之前把一些相关数据dump到日志中。
|
|
(3)check_before_do_action(self):
|
|
执行action sql之前的检查。
|
|
(4)dump_before_do_each_tenant_action(self, tenant_id):
|
|
执行用参数tenant_id拼成的这条action sql之前把一些相关数据dump到日志中。
|
|
(5)check_before_do_each_tenant_action(self, tenant_id):
|
|
执行用参数tenant_id拼成的这条action sql之前的检查。
|
|
(6)@staticmethod get_each_tenant_action_ddl(tenant_id):
|
|
返回用参数tenant_id拼成的一条action sql，并且该sql必须为ddl。
|
|
(7)@staticmethod get_each_tenant_rollback_sql(tenant_id):
|
|
返回一条sql，用于回滚get_each_tenant_action_ddl(tenant_id)返回的sql。
|
|
(8)dump_after_do_each_tenant_action(self, tenant_id):
|
|
执行用参数tenant_id拼成的这条action sql之后把一些相关数据dump到日志中。
|
|
(9)check_after_do_each_tenant_action(self, tenant_id):
|
|
执行用参数tenant_id拼成的这条action sql之后的检查。
|
|
(10)dump_after_do_action(self):
|
|
执行action sql之后把一些相关数据dump到日志中。
|
|
(11)check_after_do_action(self):
|
|
执行action sql之后的检查。
|
|
(12)skip_pre_check(self):
|
|
check if check_before_do_action() can be skipped
|
|
(13)skip_each_tenant_action(self):
|
|
check if check_before_do_each_tenant_action() and do_each_tenant_action() can be skipped
|
|
|
|
举例: 以下为schema拆分后加租户级系统表的示例
|
|
class EachTenantDDLActionPostCreateAllTenantBackupBackupLogArchiveStatus(BaseEachTenantDDLAction):
|
|
table_name = '__all_tenant_backup_backup_log_archive_status'
|
|
@staticmethod
|
|
def get_seq_num():
|
|
return 24
|
|
def dump_before_do_action(self):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '{1}'""".format(self.get_all_table_name(), self.table_name))
|
|
def skip_pre_check(self):
|
|
return True
|
|
def skip_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '{1}' and tenant_id = {2}""".format(self.get_all_table_name(), self.table_name, tenant_id))
|
|
return (1 == len(results))
|
|
def check_before_do_action(self):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '{1}'""".format(self.get_all_table_name(), self.table_name))
|
|
if len(results) > 0:
|
|
raise MyError("""{0} already created""".format(self.table_name))
|
|
def dump_before_do_each_tenant_action(self, tenant_id):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '{1}' and tenant_id = {2}""".format(self.get_all_table_name(), self.table_name, tenant_id))
|
|
def check_before_do_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '{1}' and tenant_id = {2}""".format(self.get_all_table_name(), self.table_name, tenant_id))
|
|
if len(results) > 0:
|
|
raise MyError("""tenant_id:{0} has already create table {1}""".format(tenant_id, self.table_name))
|
|
@staticmethod
|
|
def get_each_tenant_action_ddl(tenant_id):
|
|
pure_table_id = 303
|
|
table_id = (tenant_id << 40) | pure_table_id
|
|
return """CREATE TABLE `__all_tenant_backup_backup_log_archive_status` (
|
|
`gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
|
|
`gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
|
|
`tenant_id` bigint(20) NOT NULL,
|
|
`incarnation` bigint(20) NOT NULL,
|
|
`log_archive_round` bigint(20) NOT NULL,
|
|
`copy_id` bigint(20) NOT NULL,
|
|
`min_first_time` timestamp(6) NOT NULL,
|
|
`max_next_time` timestamp(6) NOT NULL,
|
|
`input_bytes` bigint(20) NOT NULL DEFAULT '0',
|
|
`output_bytes` bigint(20) NOT NULL DEFAULT '0',
|
|
`deleted_input_bytes` bigint(20) NOT NULL DEFAULT '0',
|
|
`deleted_output_bytes` bigint(20) NOT NULL DEFAULT '0',
|
|
`pg_count` bigint(20) NOT NULL DEFAULT '0',
|
|
`status` varchar(64) NOT NULL DEFAULT '',
|
|
PRIMARY KEY (`tenant_id`, `incarnation`, `log_archive_round`, `copy_id`)
|
|
) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = DYNAMIC COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase'
|
|
""".format(table_id)
|
|
@staticmethod
|
|
def get_each_tenant_rollback_sql(tenant_id):
|
|
return """select 1"""
|
|
def dump_after_do_each_tenant_action(self, tenant_id):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '{1}' and tenant_id = {2}""".format(self.get_all_table_name(), self.table_name, tenant_id))
|
|
def check_after_do_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '{1}' and tenant_id = {2}""".format(self.get_all_table_name(), self.table_name, tenant_id))
|
|
if len(results) != 1:
|
|
raise MyError("""tenant_id:{0} create table {1} failed""".format(tenant_id, self.table_name))
|
|
def dump_after_do_action(self):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '{1}'""".format(self.get_all_table_name(), self.table_name))
|
|
def check_after_do_action(self):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '{1}'""".format(self.get_all_table_name(), self.table_name))
|
|
if len(results) != len(self.get_tenant_id_list()):
|
|
raise MyError("""there should be {0} rows in {1} whose table_name is {2}, but there has {3} rows like that""".format(len(self.get_tenant_id_list()), self.get_all_table_name(), self.table_name, len(results)))
|
|
'''
|
|
|
|
|
|
#升级语句对应的action要写在下面的actions begin和actions end这两行之间，
|
|
#因为基准版本更新的时候会调用reset_upgrade_scripts.py来清空actions begin和actions end
|
|
#这两行之间的这些action，如果不写在这两行之间的话会导致清空不掉相应的action。
|
|
|
|
####========******####======== actions begin ========####******========####
|
|
class EachTenantDDLActionPostCreateAllKvTTLTasks(BaseEachTenantDDLAction):
|
|
@staticmethod
|
|
def get_seq_num():
|
|
return 0
|
|
def dump_before_do_action(self):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
|
|
def check_before_do_action(self):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
|
|
if len(results) > 0:
|
|
raise MyError('__all_kv_ttl_task already created')
|
|
def dump_before_do_each_tenant_action(self, tenant_id):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
def skip_pre_check(self):
|
|
return True
|
|
def skip_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
return (1 == len(results))
|
|
def check_before_do_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
if len(results) > 0:
|
|
raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task'.format(tenant_id))
|
|
@staticmethod
|
|
def get_each_tenant_action_ddl(tenant_id):
|
|
pure_table_id = 410
|
|
table_id = (tenant_id << 40) | pure_table_id
|
|
return """CREATE TABLE `__all_kv_ttl_task` (
|
|
`gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
|
|
`gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
|
|
`tenant_id` bigint(20) NOT NULL,
|
|
`task_id` bigint(20) NOT NULL,
|
|
`table_id` bigint(20) NOT NULL,
|
|
`partition_id` bigint(20) NOT NULL,
|
|
`task_start_time` bigint(20) NOT NULL,
|
|
`task_update_time` bigint(20) NOT NULL,
|
|
`trigger_type` bigint(20) NOT NULL,
|
|
`status` bigint(20) NOT NULL,
|
|
`ttl_del_cnt` bigint(20) NOT NULL,
|
|
`max_version_del_cnt` bigint(20) NOT NULL,
|
|
`scan_cnt` bigint(20) NOT NULL,
|
|
`row_key` varbinary(2048) NOT NULL,
|
|
`ret_code` varchar(512) NOT NULL,
|
|
PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
|
|
) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
|
|
@staticmethod
|
|
def get_each_tenant_rollback_sql(tenant_id):
|
|
return """select 1"""
|
|
def dump_after_do_each_tenant_action(self, tenant_id):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
def check_after_do_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
if len(results) != 1:
|
|
raise MyError('tenant_id:{0} create table __all_kv_ttl_task failed'.format(tenant_id))
|
|
def dump_after_do_action(self):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
|
|
def check_after_do_action(self):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task'""".format(self.get_all_table_name()))
|
|
if len(results) != len(self.get_tenant_id_list()):
|
|
raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))
|
|
|
|
class EachTenantDDLActionPostCreateAllKvTTLTaskHistory(BaseEachTenantDDLAction):
|
|
@staticmethod
|
|
def get_seq_num():
|
|
return 1
|
|
def dump_before_do_action(self):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
|
|
def skip_pre_check(self):
|
|
return True
|
|
def skip_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
return (1 == len(results))
|
|
def check_before_do_action(self):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
|
|
if len(results) > 0:
|
|
raise MyError('__all_kv_ttl_task_history already created')
|
|
def dump_before_do_each_tenant_action(self, tenant_id):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
def check_before_do_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
if len(results) > 0:
|
|
raise MyError('tenant_id:{0} has already create table __all_kv_ttl_task_history'.format(tenant_id))
|
|
@staticmethod
|
|
def get_each_tenant_action_ddl(tenant_id):
|
|
pure_table_id = 411
|
|
table_id = (tenant_id << 40) | pure_table_id
|
|
return """CREATE TABLE `__all_kv_ttl_task_history` (
|
|
`gmt_create` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6),
|
|
`gmt_modified` timestamp(6) NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
|
|
`tenant_id` bigint(20) NOT NULL,
|
|
`task_id` bigint(20) NOT NULL,
|
|
`table_id` bigint(20) NOT NULL,
|
|
`partition_id` bigint(20) NOT NULL,
|
|
`task_start_time` bigint(20) NOT NULL,
|
|
`task_update_time` bigint(20) NOT NULL,
|
|
`trigger_type` bigint(20) NOT NULL,
|
|
`status` bigint(20) NOT NULL,
|
|
`ttl_del_cnt` bigint(20) NOT NULL,
|
|
`max_version_del_cnt` bigint(20) NOT NULL,
|
|
`scan_cnt` bigint(20) NOT NULL,
|
|
`row_key` varbinary(2048) NOT NULL,
|
|
`ret_code` varchar(512) NOT NULL,
|
|
PRIMARY KEY (`tenant_id`, `task_id`, `table_id`, `partition_id`)
|
|
) TABLE_ID={0} DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase' """.format(table_id)
|
|
@staticmethod
|
|
def get_each_tenant_rollback_sql(tenant_id):
|
|
return """select 1"""
|
|
def dump_after_do_each_tenant_action(self, tenant_id):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
def check_after_do_each_tenant_action(self, tenant_id):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history' and tenant_id = {1}""".format(self.get_all_table_name(), tenant_id))
|
|
if len(results) != 1:
|
|
raise MyError('tenant_id:{0} create table __all_kv_ttl_task_history failed'.format(tenant_id))
|
|
def dump_after_do_action(self):
|
|
my_utils.query_and_dump_results(self._query_cursor, """select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
|
|
def check_after_do_action(self):
|
|
(desc, results) = self._query_cursor.exec_query("""select tenant_id, table_id, table_name from {0} where table_name = '__all_kv_ttl_task_history'""".format(self.get_all_table_name()))
|
|
if len(results) != len(self.get_tenant_id_list()):
|
|
raise MyError('there should be {0} rows in {1} whose table_name is __all_kv_ttl_task_history, but there has {2} rows like that'.format(len(self.get_tenant_id_list()), self.get_all_table_name(), len(results)))
|
|
####========******####========= actions end =========####******========####
|
|
|
|
def do_each_tenant_ddl_actions(cur, tenant_id_list):
|
|
import each_tenant_ddl_actions_post
|
|
# 组户级系统表没法通过虚拟表暴露，需要根据版本决定查哪张实体表
|
|
query_cur = QueryCursor(cur)
|
|
version = fetch_observer_version(query_cur)
|
|
all_table_name = "__all_table"
|
|
if (cmp(version, "2.2.60") >= 0) :
|
|
all_table_name = "__all_table_v2"
|
|
|
|
cls_list = reflect_action_cls_list(each_tenant_ddl_actions_post, 'EachTenantDDLActionPost')
|
|
|
|
# set parameter
|
|
if len(cls_list) > 0:
|
|
actions.set_parameter(cur, 'enable_sys_table_ddl' , 'True')
|
|
ori_enable_ddl = actions.get_ori_enable_ddl(cur)
|
|
if ori_enable_ddl == 0:
|
|
actions.set_parameter(cur, 'enable_ddl', 'True')
|
|
|
|
for cls in cls_list:
|
|
logging.info('do each tenant ddl acion, seq_num: %d', cls.get_seq_num())
|
|
action = cls(cur, tenant_id_list)
|
|
action.set_all_table_name(all_table_name)
|
|
action.dump_before_do_action()
|
|
if False == action.skip_pre_check():
|
|
action.check_before_do_action()
|
|
else:
|
|
logging.info("skip pre check. seq_num: %d", cls.get_seq_num())
|
|
# 系统租户组户级系统表创建成功会覆盖普通租户系统表，所以系统租户要最后建表
|
|
for tenant_id in action.get_tenant_id_list():
|
|
action.dump_before_do_each_tenant_action(tenant_id)
|
|
if False == action.skip_each_tenant_action(tenant_id):
|
|
action.check_before_do_each_tenant_action(tenant_id)
|
|
action.do_each_tenant_action(tenant_id)
|
|
else:
|
|
logging.info("skip each tenant ddl action, seq_num: %d, tenant_id: %d", cls.get_seq_num(), tenant_id)
|
|
action.dump_after_do_each_tenant_action(tenant_id)
|
|
action.check_after_do_each_tenant_action(tenant_id)
|
|
action.dump_after_do_action()
|
|
action.check_after_do_action()
|
|
|
|
# reset parameter
|
|
if len(cls_list) > 0:
|
|
if ori_enable_ddl == 0:
|
|
actions.set_parameter(cur, 'enable_ddl' , 'False')
|
|
actions.set_parameter(cur, 'enable_sys_table_ddl' , 'False')
|
|
|
|
def get_each_tenant_ddl_actions_sqls_str(tenant_id_list):
|
|
import each_tenant_ddl_actions_post
|
|
ret_str = ''
|
|
cls_list = reflect_action_cls_list(each_tenant_ddl_actions_post, 'EachTenantDDLActionPost')
|
|
for i in range(0, len(cls_list)):
|
|
for j in range(0, len(tenant_id_list)):
|
|
if i > 0 or j > 0:
|
|
ret_str += '\n'
|
|
ret_str += cls_list[i].get_each_tenant_action_ddl(tenant_id_list[j]) + ';'
|
|
return ret_str
|