Merge branch 'dev_4.2.0' into 'master'

Dev 4.2.0

Closes DIMA-2026012900113789954 and DIMA-2026012200113616145

See merge request oceanbase/ob-deploy!818

Author: gys-git

_cmd.py

@@ -872,6 +872,11 @@ def __init__(self):
         self.parser.add_option('-c', '--components', type='string', help="List the components. Multiple components are separated with commas. [oceanbase-ce,obproxy-ce,obagent,prometheus,grafana,ob-configserver]\nExample: \nstart oceanbase-ce: obd demo -c oceanbase-ce\n"
          + "start -c oceanbase-ce V3.2.3: obd perf -c oceanbase-ce --oceanbase-ce.version=3.2.3\n"
          + "start oceanbase-ce and obproxy-ce: obd perf -c oceanbase-ce,obproxy-ce", default='oceanbase-ce,obproxy-ce,obagent,prometheus,grafana')
+        self.parser.add_option('--server', type='string', help="The IP address of the server to deploy on.")
+        self.parser.add_option('--username', type='string', help="The username for SSH connection.", default=None)
+        self.parser.add_option('--password', type='string', help="The password for SSH connection.", default='')
+        self.parser.add_option('--port', type='int', help="The port for SSH connection.", default=22)
+        self.parser.add_option('--key-file', type='string', help="The path to the private key file for SSH connection.", default='')
         self.parser.allow_undefine = True
         self.parser.undefine_warn = False
 
@@ -1733,7 +1738,7 @@ def __init__(self):
         self.parser.add_option('--slb-host', type='string', help='The host of soft load balance.')
         self.parser.add_option('--exec-id', type='string', help='The unique execute id.')
         self.parser.add_option('--case-filter', type='string', help='The case filter file for mysqltest.')
-        self.parser.add_option('--psmall-test', type='string', help='The file maintain psmall cases.', default='./mysql_test/psmalltest.py')
+        self.parser.add_option('--psmall-test', type='string', help='The file maintain psmall cases.', default='mysqltest_config.yaml')
         self.parser.add_option('--psmall-source', type='string', help='The file maintain psmall source control.', default='./mysql_test/psmallsource.py')
         self.parser.add_option('--ps', action='store_true', help='Run in ps mode.', default=False)
         self.parser.add_option('--test-tags', type='string', help='The file maintain basic tags.', default='./mysql_test/test_tags.py')
@@ -2564,6 +2569,7 @@ def __init__(self):
         self.parser.add_option('--replicate-num', type='int', help="Number of copies", default=1)
         self.parser.add_option('-d', '--obproxy-deployname', '--odp', type='string', help='Obproxy deploy name')
         self.parser.add_option('-p', '--cdcro-password', type='string', help='Password of user `cdcro`.If user already been created manually, you need to enter the password here.')
+        self.parser.add_option('--rs', '--root-server-list', type='string', help="The list of machines where the rootservice resides, in the format 'server_ip:server_rpc_port:server_sql_port'")
 
     def init(self, cmd, args):
         super(BinlogCreateCommand, self).init(cmd, args)

_errno.py

@@ -261,7 +261,7 @@ class InitDirFailedErrorMessage(object):
 
 # obbinlog
 EC_OBBINLOG_DEPENDS_COMP_MIN_VERSION = OBDErrorCodeTemplate(4601, 'OBBinlog {obbinlog_version} needs to use {comp} with version {min_version} or above.')
-EC_OBBINLOG_TARGET_DEPLOY_NEED_CONFIGSERVER = OBDErrorCodeTemplate(4602, 'Deploy {target_oceanbase_deploy} need depends ob-configserver. You could use `obd cluster component add {target_oceanbase_deploy} -c <ob-configserver.config>` to add.')
+EC_OBBINLOG_TARGET_DEPLOY_NEED_CONFIGSERVER = OBDErrorCodeTemplate(4602, 'Deploy {target_oceanbase_deploy} need depends ob-configserver. You could use `obd cluster component add {target_oceanbase_deploy} -c <ob-configserver.config>` to add or set `--rs` paramater')
 EC_OBBINLOG_CE_WITH_OCENABASE_CE = OBDErrorCodeTemplate(4603, 'The Binlog service for the community version can only be used with the community version of the OceanBase database.')
 EC_OBBINLOG_WITH_OCENABASE = OBDErrorCodeTemplate(4604, 'The Binlog service for the enterprise version can only be used with the enterprise version of the OceanBase database.')
 

core.py

@@ -1798,11 +1798,38 @@ def demo(self, name='demo'):
         if not components:
             self._call_stdio('error', 'Use `-c/--components` to set in the components to be deployed')
             return
-                
+
+        configs = OrderedDict()
         global_key = 'global'
         home_path_key = 'home_path'
-        global_config = {home_path_key: os.getenv('HOME')}
+        username = getattr(self.options, 'username', None)
+        if not username:
+            home_path_prefix = os.getenv('HOME')
+        elif username == 'root':
+            home_path_prefix = '/root'
+        else:
+            home_path_prefix = '/home/{}'.format(username)
+        global_config = {home_path_key: home_path_prefix}
         opt_config = {}
+        target_ip = getattr(self.options, 'server', None)
+        if target_ip:
+            configs['user'] = {}
+            user_config = configs['user']
+            target_ip_list = [target_ip]
+            if username:
+                user_config['username'] = username
+            password = getattr(self.options, 'password', None)
+            if password is not None:
+                user_config['password'] = password
+            port = getattr(self.options, 'port', 22)
+            if port:
+                user_config['port'] = port
+            key_file = getattr(self.options, 'key_file', '')
+            if key_file:
+                user_config['key_file'] = key_file
+        else:
+            target_ip_list = ['127.0.0.1']
+
         for key in self.options.__dict__:
             tmp = key.split('.', 1)
             if len(tmp) == 1:
@@ -1820,10 +1847,9 @@ def demo(self, name='demo'):
                     _config = opt_config[component_name][global_key]
                 _config[tmp[1]] = self.options.__dict__[key]
 
-        configs = OrderedDict()
         for component_name in components:
             configs[component_name] = {
-                'servers': ['127.0.0.1'],
+                'servers': deepcopy(target_ip_list),
                 global_key: deepcopy(global_config)
             }
             configs[component_name][global_key][home_path_key] = os.path.join(configs[component_name][global_key][home_path_key], component_name)

plugins/mysqltest/3.1.0/check_test.py

@@ -18,6 +18,7 @@
 import os
 import sys
 import re
+import yaml
 from glob import glob
 
 import tool
@@ -86,8 +87,13 @@ def check_test(plugin_context, env, *args, **kwargs):
     if 'all' in opt and opt['all'] and os.path.isdir(os.path.realpath(opt['suite_dir'])):
         opt['suite'] = ','.join(os.listdir(os.path.realpath(opt['suite_dir'])))
     if 'psmall' in opt and opt['psmall']:
-        test_set = get_variable_from_python_file(
-            opt.get('psmall_test'), 'psmall_test', default_file='psmalltest.py', default_value=[], stdio=stdio)
+        if os.path.exists(opt.get('psmall_test')):
+            with open(opt.get('psmall_test'), 'r') as f:
+                config = yaml.safe_load(f)
+                test_set = config.get('runtime_configs', {}).get('psmall', {}).get('test-set', [])
+                opt['reboot_after_run_count'] = config.get('runtime_configs', {}).get('psmall', {}).get('reboot-after-run-count', 0)
+        else:
+            test_set = []
         opt['source_limit'] = get_variable_from_python_file(
             opt.get('psmall_source'), 'psmall_source', default_file='psmallsource.py', default_value={}, stdio=stdio)
         has_test_point = True

plugins/mysqltest/3.1.0/run_test.py

@@ -201,6 +201,7 @@ def return_true(**kw):
     auto_retry = env.get('auto_retry')
     is_retry = env.get('is_retry', False)
     reboot_cases = env.get('reboot_cases', [])
+    reboot_after_run_count = env.get('reboot_after_run_count', 0)
     need_reboot = env.get('need_reboot', False)
     collect_all = env.get('collect_all', False)
     collect_log = False
@@ -444,6 +445,8 @@ def return_true(**kw):
             case_results.append(result)
             index += 1
             is_retry = False
+            if reboot_after_run_count > 0 and index % reboot_after_run_count == 0:
+                need_reboot = True
         elif is_retry or not auto_retry:
             # failed and no chance to retry
             case_results.append(result)

plugins/obbinlog-ce/4.0.1/create_binlog.py

@@ -35,7 +35,11 @@ def create_binlog(plugin_context, ob_deploy, tenant_name,  ob_cluster_repositori
     cluster_name = plugin_context.get_variable('cluster_name')
     obconfig_url = plugin_context.get_variable('obconfig_url')
     replicate_num = getattr(plugin_context.options, 'replicate_num')
-    sql = f"CREATE BINLOG FOR TENANT `{cluster_name}`.`{tenant_name}` TO USER `cdcro` PASSWORD `{ob_global_config.get('cdcro_password')}` WITH CLUSTER URL `{obconfig_url}`, REPLICATE NUM {replicate_num};"
+    if obconfig_url:
+        sql = f"CREATE BINLOG FOR TENANT `{cluster_name}`.`{tenant_name}` TO USER `cdcro` PASSWORD `{ob_global_config.get('cdcro_password')}` WITH CLUSTER URL `{obconfig_url}`, REPLICATE NUM {replicate_num};"
+    else:
+        root_service_list = plugin_context.get_variable('root_service_list')
+        sql = f"CREATE BINLOG FOR TENANT `{cluster_name}`.`{tenant_name}` TO USER `cdcro` PASSWORD `{ob_global_config.get('cdcro_password')}` WITH ROOTSERVER_LIST '{root_service_list}', REPLICATE NUM {replicate_num};"
     try:
         cursor.execute(sql, raise_exception=True)
     except Exception as e:

plugins/obbinlog-ce/4.0.1/create_env_check.py

@@ -94,13 +94,28 @@ def create_env_check(plugin_context, ob_deploy, ob_cluster_repositories, *args,
                 return plugin_context.return_false()
             ob_cluster_config.update_global_conf('cdcro_password', getattr(plugin_context.options, 'cdcro_password'), True)
     try:
+        rs_lists = []
+        rs_list = ''
+        for comp in const.COMPS_OB:
+            if comp in cluster_config.depends:
+                ob_servers = cluster_config.get_depend_servers(comp)
+                for ob_server in ob_servers:
+                    ob_server_conf = cluster_config.get_depend_config(comp, ob_server)
+                    rs_lists.append('{}:{}:{}'.format(ob_server.ip, ob_server_conf['rpc_port'], ob_server_conf['mysql_port']))
+        if rs_lists:
+            rs_list = ';'.join(rs_lists)
+
+        rs_list = rs_list if rs_list else (getattr(plugin_context.options, 'root_server_list', None) or getattr(plugin_context.options, 'rs', None))
         ret = cursor.fetchone("SHOW PARAMETERS LIKE 'obconfig_url';")
         obconfig_url = ret['value']
-        if not obconfig_url:
+        if not obconfig_url and not rs_list:
             stdio.error(EC_OBBINLOG_TARGET_DEPLOY_NEED_CONFIGSERVER.format(target_oceanbase_deploy=ob_deploy.name))
             stdio.stop_loading('fail')
             return plugin_context.return_false()
-        plugin_context.set_variable('obconfig_url', obconfig_url)
+        if obconfig_url:
+            plugin_context.set_variable('obconfig_url', obconfig_url)
+        else:
+            plugin_context.set_variable('root_service_list', rs_list)
     except Exception as e:
         stdio.error(e)
     stdio.stop_loading('succeed')

ssh.py

@@ -547,6 +547,9 @@ def remote_transporter(self):
         self.stdio.verbose("current remote_transporter {}".format(self._remote_transporter))
         return self._remote_transporter
 
+    def set_remote_transporter(self, transporter, stdio=None):
+        self._remote_transporter = transporter
+
     def put_file(self, local_path, remote_path, stdio=None):
         if not os.path.isfile(local_path):
             stdio.error('path: %s is not file' % local_path)
@@ -555,7 +558,14 @@ def put_file(self, local_path, remote_path, stdio=None):
             return LocalClient.put_file(local_path, remote_path, stdio=stdio)
         if not self._open_sftp(stdio=stdio):
             return False
-        return self._put_file(local_path, remote_path, stdio=stdio)
+        ret = self._put_file(local_path, remote_path, stdio=stdio)
+        if not ret:
+            if self.remote_transporter == RemoteTransporter.RSYNC:
+                COMMAND_ENV.set(ENV_DISABLE_RSYNC, '0', save=True)
+                self.set_remote_transporter(RemoteTransporter.CLIENT)
+                return self._put_file(local_path, remote_path, stdio=stdio)
+        else:
+            return True
 
     def write_file(self, content, file_path, mode='w', stdio=None):
         if self._is_local: