Merge branch 'dev_3.3.0' into 'master'

Dev 3.3.0

See merge request oceanbase/ob-deploy!489

Author: gys-git

_cmd.py

@@ -1225,6 +1225,10 @@ def __init__(self):
         self.parser.add_option('-t', '-n', '--tenant-name', type='string', help="The standby tenant name. The default tenant name is consistent with the primary tenant name.", default='')
         self.parser.add_option('--standbyro-password', type='string', help="standbyro user password.")
         self.parser.add_option('-p', '--tenant-root-password', type='string', help="tenant root password,for crate standby user.")
+        self.parser.add_option('--type', type='string', help="Standby tenant data sync mode. Supports 'SERVICE' and 'LOCATION' modes. Defaults: 'SERVICE'.", default='SERVICE')
+        self.parser.add_option('-d', '--data_backup_uri', type='string',help='The path to the directory where the backups are stored.')
+        self.parser.add_option('-a', '--archive_log_uri', type='string',help='The Path to the directory where archive logs are stored.')
+        self.parser.add_option('-D', '--decryption', type='string', help='The decryption password for all backups. example: key1,key2,key3')
 
 
     def init(self, cmd, args):
@@ -1409,7 +1413,7 @@ def __init__(self):
         self.parser.add_option('--replica_type ', type='string', help='The replica type of the tenant.')
         self.parser.add_option('-p', '--primary_zone', type='string', help="The primary zone of the tenant to be restored.")
         self.parser.add_option('-T', '--timestamp', type='string', help='The timestamp to restore to.')
-        self.parser.add_option('-S', '--scn', type='int', help="The SCN to restore to.. Default: 0.")
+        self.parser.add_option('-S', '--scn', type='int', help="The SCN to restore to. Default: 0.")
         self.parser.add_option('-s', '--ha_high_thread_score', type='int', help='The high thread score for HA. Range: [0, 100]')
         self.parser.add_option('-c', '--concurrency', type='int', help='The number of threads to use for the restore operation.')
         self.parser.add_option('-D', '--decryption', type='string', help='The decryption password for all backups. example: key1,key2,key3')
@@ -1500,6 +1504,43 @@ def _do_command(self, obd):
         else:
             return self._show_help()
 
+class ClusterTenantRecoverCommand(ClusterMirrorCommand):
+    def __init__(self):
+        super(ClusterTenantRecoverCommand, self).__init__('recover', 'Modify the recovery target of the new standby tenant')
+        self.parser.add_option('-T', '--timestamp', type='string', help='The timestamp to restore to.')
+        self.parser.add_option('-S', '--scn', type='int', help="The SCN to restore to.")
+        self.parser.add_option('-u', '--unlimited', action='store_true', help="Continuously replay archived source logs")
+
+    def init(self, cmd, args):
+        super(ClusterTenantRecoverCommand, self).init(cmd, args)
+        self.parser.set_usage('%s <deploy name> <tenant name>' % self.prev_cmd)
+        return self
+
+    def _do_command(self, obd):
+        if len(self.cmds) == 2:
+            return obd.log_recover(self.cmds[0], self.cmds[1])
+        else:
+            return self._show_help()
+
+class ClusterTenantSwitchLogSourceCommand(ClusterMirrorCommand):
+    def __init__(self):
+        super(ClusterTenantSwitchLogSourceCommand, self).__init__('switch-log-source', 'Switch standby tenant sync mode')
+        self.parser.add_option('--type', type='string', help="Standby tenant data sync mode. Supports 'SERVICE' and 'LOCATION' modes. Defaults: 'SERVICE'.", default='SERVICE')
+        self.parser.add_option('-a', '--archive_log_uri', type='string',help='The Path to the directory where archive logs are stored.')
+        self.parser.add_option('-p', '--tenant-root-password', type='string', help="tenant root password,for crate standby user.")
+        self.parser.add_option('--standbyro-password', type='string', help="standbyro user password.")
+
+    def init(self, cmd, args):
+        super(ClusterTenantSwitchLogSourceCommand, self).init(cmd, args)
+        self.parser.set_usage('%s <deploy name> <tenant name>' % self.prev_cmd)
+        return self
+
+    def _do_command(self, obd):
+        if len(self.cmds) == 2:
+            return obd.switch_log_source(self.cmds[0], self.cmds[1])
+        else:
+            return self._show_help()
+
 
 class ClusterTenantCommand(MajorCommand):
 
@@ -1520,6 +1561,8 @@ def __init__(self):
         self.register_command(ClusterTenantQueryRestoreTaskCommand())
         self.register_command(ClusterTenantCancelBackupTaskCommand())
         self.register_command(ClusterTenantCancelRestoreTaskCommand())
+        self.register_command(ClusterTenantRecoverCommand())
+        self.register_command(ClusterTenantSwitchLogSourceCommand())
 
 
 class ClusterMajorCommand(MajorCommand):

_errno.py

@@ -172,6 +172,9 @@ class InitDirFailedErrorMessage(object):
 EC_OBSERVER_UNKONE_SCENARIO = OBDErrorCodeTemplate(2013, 'Unknown scenario: {scenario}')
 EC_CPU_NOT_SUPPORT_AVX = OBDErrorCodeTemplate(2014, "{server}'s cpu does not support avx")
 EC_OBSERVER_DISABLE_AUTOSTART = OBDErrorCodeTemplate(2015, "{server}: Failed to modify the configuration of the automatic startup. Please check whether the current user has sudo permissions")
+EC_OBSERVER_LOG_INCOMPLETE = OBDErrorCodeTemplate(2016, "Primary tenant {primary_tenant} have not full log, not support create standby cluster, rerun with '--type=LOCATION' if you want to create standby tenant.")
+EC_OBSERVER_LOG_RECOVER = OBDErrorCodeTemplate(2017, "Continuous log synchronization is not enabled. Please execute the command `obd cluster tenant recover {cluster_name} {tenant_name} --unlimited` to enable continuous log synchronization")
+EC_OBSERVER_LOCATION_CREATE_STANDBY = OBDErrorCodeTemplate(2018, "For the standby tenant created by {primary_tenant} in log archive mode, it is prohibited to create a network-mode standby tenant for this standby tenant.")
 
 WC_OBSERVER_SYS_MEM_TOO_LARGE = OBDErrorCodeTemplate(2010, '({server}): system_memory too large. system_memory should be less than {factor} * memory_limit/memory_limit_percentage.')
 

_stdio.py

@@ -745,7 +745,17 @@ def _print(self, msg_lv, msg, *args, **kwargs):
             del kwargs['_disable_log']
         else:
             enaable_log = True
-        
+
+        print_msg = str(print_msg)
+        if "PASSWORD" in print_msg and "IP_LIST=" not in print_msg:
+            print_msg = self._format(print_msg, *args)
+            pk_regex = r'(?i)(password([:|=]))(?! \S)(.*?)(,|\s)'
+            pattern = re.compile(pk_regex)
+            is_match = pattern.search(print_msg)
+            if is_match:
+                password_length = len(is_match.group(3))
+                replacement = r"\1" + '*' * password_length + r"\4"
+                print_msg = pattern.sub(replacement, print_msg)
         kwargs['file'] and print(self._format(print_msg, *args), **kwargs)
         del kwargs['file']
         enaable_log and self.log(msg_lv, msg, *args, **kwargs)
@@ -816,6 +826,17 @@ def table_log_masking(self, msg):
             if match:
                 masked_password = "*"*len(match.group(3))
                 str_msg = pattern.sub(rf"\1{masked_password}\4", str_msg)
+        elif 'access_' in str_msg:
+            access_regex = r"(access_id=)[^&]*|(access_key=)[^&,| ]*"
+            access_pattern = re.compile(access_regex)
+            def replace_with_stars(match):
+                full_match = match.group(0)  # Full match including prefix
+                key, value = full_match.split('=', 1)  # Split into key and value
+                masked_value = '*' * len(value)  # Create the mask
+                return f"{key}={masked_value}"
+            match = re.search(access_pattern, str_msg)
+            if match:
+                str_msg = re.sub(access_pattern, replace_with_stars, str_msg)
         return str_msg
 
     @staticmethod
@@ -900,7 +921,7 @@ def log_masking(self, msg):
             if is_match:
                 return msg
             if "access_id" in msg or "access_key" in msg:
-                access_regex = r'(access_id=|access_key=)([^&\']+)'
+                access_regex = r'(access_id=|access_key=)([^&\',]+)'
                 access_pattern = re.compile(access_regex)
                 if access_pattern.search(msg):
                     return access_pattern.sub(r'\1******', msg)
@@ -936,6 +957,7 @@ def verbose(self, msg, *args, **kwargs):
         if self.level > self.VERBOSE_LEVEL:
             self.log(MsgLevel.VERBOSE, '%s %s' % (self._verbose_prefix, msg), *args, **kwargs)
             return
+        msg = self.log_masking(msg)
         self._print(MsgLevel.VERBOSE, '%s %s' % (self._verbose_prefix, msg), *args, **kwargs)
 
     if sys.version_info.major == 2:

const.py

@@ -134,3 +134,6 @@
 #standalone
 INTERACTIVE_INSTALL = 'INTERACTIVE_INSTALL'
 
+SERVICE_MODE = 'SERVICE'
+LOCATION_MODE = 'LOCATION'
+