2.4.0 io update and support remote_client_sudo (#372)

* update io: add stderr

* update io: add stderr

* add --inner_config
add print_type

* support remote_client_sudo

* build test package

* fix --inner_config

* fix io stream

* test

* test

* test

* test

* test

* 取消测试分支打包

common/command.py

@@ -48,37 +48,6 @@ def run_get_stderr(self, cmd):
             self.stdio.error("run cmd = [{0}] on localhost".format(cmd))
 
 
-#
-# class SshClient(object):
-#     def __init__(self, stdio=None):
-#         self.stdio = stdio
-#
-#     def run(self, ssh_helper, cmd):
-#         try:
-#             self.stdio.verbose("[remote host {0}] excute cmd = [{1}]".format(ssh_helper.get_name(), cmd))
-#             stdout = ssh_helper.ssh_exec_cmd(cmd)
-#             self.stdio.verbose("[remote host {0}] excute cmd = [{1}] complete, stdout=[{2}]".format(ssh_helper.get_name(), cmd, stdout))
-#             return stdout
-#         except Exception as e:
-#             self.stdio.error("[remote host {0}] excute cmd = [{1}] except: [{2}]".format(ssh_helper.get_name(), cmd, e))
-#
-#     def run_get_stderr(self, ssh_helper, cmd):
-#         try:
-#             self.stdio.verbose("[remote host {0}] run cmd = [{1}] start ...".format(ssh_helper.get_name(), cmd))
-#             std = ssh_helper.ssh_exec_cmd_get_stderr(cmd)
-#             return std
-#         except Exception as e:
-#             self.stdio.error("[remote host {0}] run ssh cmd = [{1}] except: {2}".format(ssh_helper.get_name(), cmd, e))
-#
-#     def run_ignore_err(self, ssh_helper, cmd):
-#         try:
-#             self.stdio.verbose("[remote host {0}] run cmd = [{1}] start ...".format(ssh_helper.get_name(), cmd))
-#             std = ssh_helper.ssh_exec_cmd_ignore_err(cmd)
-#             return std
-#         except SSHException as e:
-#             self.stdio.error("[remote host {0}] run ssh cmd = [{1}] except: {2}".format(ssh_helper.get_name(), cmd, e))
-
-
 def download_file(ssh_client, remote_path, local_path, stdio=None):
     """
     download file
@@ -220,7 +189,7 @@ def zip_dir(ssh_client, father_dir, zip_dir, stdio=None):
     Compress files through zip
     :return:
     """
-    cmd = "cd {father_dir} && zip {zip_dir}.zip -rm {zip_dir}".format(father_dir=father_dir, zip_dir=zip_dir)
+    cmd = "zip  {father_dir}/{zip_dir}.zip -rm {father_dir}/{zip_dir}".format(father_dir=father_dir, zip_dir=zip_dir)
     ssh_client.exec_cmd(cmd)
 
 
@@ -229,7 +198,7 @@ def zip_encrypt_dir(ssh_client, zip_password, father_dir, zip_dir, stdio=None):
     Compress files by encryption
     :return:
     """
-    cmd = "cd {father_dir} && zip --password {zip_password} {zip_dir}.zip -rm {zip_dir}".format(zip_password=zip_password, father_dir=father_dir, zip_dir=zip_dir)
+    cmd = "zip --password {zip_password} {father_dir}/{zip_dir}.zip -rm {father_dir}/{zip_dir}".format(zip_password=zip_password, father_dir=father_dir, zip_dir=zip_dir)
     ssh_client.exec_cmd(cmd)
 
 

common/ssh_client/remote_client.py

@@ -50,8 +50,12 @@ def __init__(self, context, node):
         self.key_file = os.path.expanduser(self.key_file)
         self._ssh_fd = None
         self._sftp_client = None
+        # remote_client_sudo
+        self.remote_client_sudo = bool(self.context.inner_config.get("obdiag").get("ssh_client").get("remote_client_sudo"))
+        # remote_client_disable_rsa_algorithms
         DISABLED_ALGORITHMS = dict(pubkeys=["rsa-sha2-512", "rsa-sha2-256"])
-        if ENV_DISABLE_RSA_ALGORITHMS == 1:
+        remote_client_disable_rsa_algorithms = bool(self.context.inner_config.get("obdiag").get("basic").get("dis_rsa_algorithms"))
+        if remote_client_disable_rsa_algorithms:
             self._disabled_rsa_algorithms = DISABLED_ALGORITHMS
         self.ssh_type = "remote"
         if len(self.key_file) > 0:
@@ -75,6 +79,15 @@ def __init__(self, context, node):
 
     def exec_cmd(self, cmd):
         try:
+            if self.remote_client_sudo:
+                # check sudo without password
+                self.stdio.verbose("use remote_client_sudo")
+                stdin, stdout, stderr = self._ssh_fd.exec_command("sudo -n true")
+                if stderr:
+                    if len(stderr.read().decode('utf-8').strip()) > 0:
+                        raise Exception(stderr.read().decode('utf-8'))
+                cmd = "sudo {0}".format(cmd)
+            self.stdio.verbose('Execute Shell command on server {0}:{1}'.format(self.host_ip, cmd))
             stdin, stdout, stderr = self._ssh_fd.exec_command(cmd)
             err_text = stderr.read()
             if len(err_text):

conf/inner_config.yml

@@ -12,6 +12,9 @@ obdiag:
     log_level: INFO
     mode: obdiag
     stdout_handler_log_level: INFO
+    error_stream: sys.stdout
+  ssh_client:
+    remote_client_sudo: 0
 check:
   ignore_version: false
   work_path: "~/.obdiag/check"

config.py

@@ -73,6 +73,10 @@
             'log_level': 'INFO',
             'mode': 'obdiag',
             'stdout_handler_log_level': 'INFO',
+            'error_stream': 'sys.stdout',
+        },
+        'ssh_client': {
+            'remote_client_sudo': False,
         },
     },
     'check': {
@@ -257,7 +261,22 @@ def get_node_config(self, type, node_ip, config_item):
 
 class InnerConfigManager(Manager):
 
-    def __init__(self, stdio=None):
+    def __init__(self, stdio=None, inner_config_change_map=None):
+        if inner_config_change_map is None:
+            inner_config_change_map = {}
         inner_config_abs_path = os.path.abspath(INNER_CONFIG_FILE)
         super().__init__(inner_config_abs_path, stdio=stdio)
         self.config = self.load_config_with_defaults(DEFAULT_INNER_CONFIG)
+        if inner_config_change_map != {}:
+            self.config = self._change_inner_config(self.config, inner_config_change_map)
+
+    def _change_inner_config(self, conf_map, change_conf_map):
+        for key, value in change_conf_map.items():
+            if key in conf_map:
+                if isinstance(value, dict):
+                    self._change_inner_config(conf_map[key], value)
+                else:
+                    conf_map[key] = value
+            else:
+                conf_map[key] = value
+        return conf_map

core.py

@@ -62,7 +62,7 @@
 
 class ObdiagHome(object):
 
-    def __init__(self, stdio=None, config_path=os.path.expanduser('~/.obdiag/config.yml')):
+    def __init__(self, stdio=None, config_path=os.path.expanduser('~/.obdiag/config.yml'), inner_config_change_map=None):
         self._optimize_manager = None
         self.stdio = None
         self._stdio_func = None
@@ -71,7 +71,11 @@ def __init__(self, stdio=None, config_path=os.path.expanduser('~/.obdiag/config.
         self.namespaces = {}
         self.set_stdio(stdio)
         self.context = None
-        self.inner_config_manager = InnerConfigManager(stdio)
+        self.inner_config_manager = InnerConfigManager(stdio=stdio, inner_config_change_map=inner_config_change_map)
+        if self.inner_config_manager.config.get("obdiag") is not None and self.inner_config_manager.config.get("obdiag").get("basic") is not None and self.inner_config_manager.config.get("obdiag").get("basic").get("print_type") is not None:
+            stdio.set_err_stream(self.inner_config_manager.config.get("obdiag").get("logger").get("error_stream"))
+
+        self.set_stdio(stdio)
         self.config_manager = ConfigManager(config_path, stdio)
         if (
             self.inner_config_manager.config.get("obdiag") is not None

diag_cmd.py

@@ -29,7 +29,8 @@
 from common.version import get_obdiag_version
 from telemetry.telemetry import telemetry
 
-ROOT_IO = IO(1)
+# TODO when obdiag_version ≥ 3.0, the default value of err_stream will be changed to sys.stderr
+ROOT_IO = IO(1, error_stream=sys.stdout)
 OBDIAG_HOME_PATH = os.path.join(os.getenv('HOME'), 'oceanbase-diagnostic-tool')
 
 
@@ -111,13 +112,58 @@ def __init__(self, name, summary):
         self.is_init = False
         self.hidden = False
         self.has_trace = True
+        self.inner_config_change_map = {}
         self.parser = AllowUndefinedOptionParser(add_help_option=True)
         self.parser.add_option('-h', '--help', action='callback', callback=self._show_help, help='Show help and exit.')
         self.parser.add_option('-v', '--verbose', action='callback', callback=self._set_verbose, help='Activate verbose output.')
+        self.parser.add_option('--inner_config', action='callback', type="str", callback=self._inner_config_change, help='change inner config. ')
 
     def _set_verbose(self, *args, **kwargs):
         ROOT_IO.set_verbose_level(0xFFFFFFF)
 
+    def _inner_config_change(self, option, opt_str, value, parser):
+        """
+        Inner config change
+        """
+        try:
+            key, val = value.split('=')
+            if key is None or key == "":
+                return
+            m = self._inner_config_change_set(key, val)
+
+            def _change_inner_config(conf_map, change_conf_map):
+                for change_conf_map_key, change_conf_map_value in change_conf_map.items():
+                    if change_conf_map_key in conf_map:
+                        if isinstance(change_conf_map_value, dict):
+                            _change_inner_config(conf_map[change_conf_map_key], change_conf_map_value)
+                        else:
+                            conf_map[change_conf_map_key] = change_conf_map_value
+                    else:
+                        conf_map[change_conf_map_key] = change_conf_map_value
+                return conf_map
+
+            self.inner_config_change_map = _change_inner_config(self.inner_config_change_map, m)
+        except Exception as e:
+            raise Exception("Key or val ({1}) is illegal: {0}".format(e, value))
+
+    def _inner_config_change_set(self, key, val):
+        def recursion(change_map, key, val):
+            if key is None or key == "":
+                raise Exception("key is None")
+            if val is None or val == "":
+                raise Exception("val is None")
+            if key.startswith(".") or key.endswith("."):
+                raise Exception("Key starts or ends '.'")
+            if "." in key:
+                map_key = key.split(".")[0]
+                change_map[map_key] = recursion({}, key[len(map_key) + 1 :], val)
+                return change_map
+            else:
+                change_map[key] = val
+                return change_map
+
+        return recursion({}, key, val)
+
     def init(self, cmd, args):
         if self.is_init is False:
             self.prev_cmd = cmd
@@ -216,7 +262,7 @@ def do_command(self):
                 else:
                     ROOT_IO.error('The option you provided with -c: {0} is a non-existent configuration file path.'.format(custom_config))
                     return
-            obdiag = ObdiagHome(stdio=ROOT_IO, config_path=config_path)
+            obdiag = ObdiagHome(stdio=ROOT_IO, config_path=config_path, inner_config_change_map=self.inner_config_change_map)
             obdiag.set_options(self.opts)
             obdiag.set_cmds(self.cmds)
             ret = self._do_command(obdiag)

handler/gather/gather_log.py

@@ -281,9 +281,9 @@ def __get_log_name(self, ssh_client, node):
         home_path = node.get("home_path")
         log_path = os.path.join(home_path, "log")
         if self.scope == "observer" or self.scope == "rootservice" or self.scope == "election":
-            get_oblog = "ls -1 -F %s/*%s.log* | awk -F '/' '{print $NF}'" % (log_path, self.scope)
+            get_oblog = "ls -1 -F %s |grep %s | awk -F '/' '{print $NF}'" % (log_path, self.scope)
         else:
-            get_oblog = "ls -1 -F %s/observer.log* %s/rootservice.log* %s/election.log* | awk -F '/' '{print $NF}'" % (log_path, log_path, log_path)
+            get_oblog = "ls -1 -F %s |grep -E 'observer|rootservice|election'| awk -F '/' '{print $NF}'" % log_path
         log_name_list = []
         log_files = ssh_client.exec_cmd(get_oblog)
         if log_files:

stdio.py

@@ -13,6 +13,7 @@
 
 from __future__ import absolute_import, division, print_function
 
+import json
 import os
 import signal
 import sys
@@ -358,7 +359,7 @@ class IO(object):
     WARNING_PREV = FormtatText.warning('[WARN]')
     ERROR_PREV = FormtatText.error('[ERROR]')
 
-    def __init__(self, level, msg_lv=MsgLevel.DEBUG, use_cache=False, track_limit=0, root_io=None, input_stream=SysStdin, output_stream=sys.stdout):
+    def __init__(self, level, msg_lv=MsgLevel.DEBUG, use_cache=False, track_limit=0, root_io=None, input_stream=SysStdin, output_stream=sys.stdout, error_stream=sys.stdout):
         self.level = level
         self.msg_lv = msg_lv
         self.default_confirm = False
@@ -373,12 +374,15 @@ def __init__(self, level, msg_lv=MsgLevel.DEBUG, use_cache=False, track_limit=0,
         self.sync_obj = None
         self.input_stream = None
         self._out_obj = None
+        self._err_obj = None
         self._cur_out_obj = None
+        self._cur_err_obj = None
         self._before_critical = None
         self._output_is_tty = False
         self._input_is_tty = False
         self.set_input_stream(input_stream)
         self.set_output_stream(output_stream)
+        self.set_err_stream(error_stream)
 
     def isatty(self):
         if self._root_io:
@@ -400,6 +404,24 @@ def set_output_stream(self, output_stream):
         self._output_is_tty = output_stream.isatty()
         return True
 
+    def set_err_stream(self, error_stream):
+        if isinstance(error_stream, str):
+            error_stream = error_stream.strip().lower()
+            if error_stream == "sys.stderr":
+                error_stream = sys.stderr
+            elif error_stream == "sys.stdout":
+                error_stream = sys.stdout
+            else:
+                # TODO 3.X NEED CHANGE TO sys.stderr
+                error_stream = sys.stdout
+        if self._root_io:
+            return False
+        if self._cur_err_obj == self._err_obj:
+            self._cur_err_obj = error_stream
+        self._err_obj = error_stream
+        self._output_is_tty = error_stream.isatty()
+        return True
+
     def init_trace_logger(self, log_path, log_name=None, trace_id=None, recreate=False):
         if self._root_io:
             return False
@@ -417,7 +439,7 @@ def __getstate__(self):
         state = {}
         for key in self.__dict__:
             state[key] = self.__dict__[key]
-        for key in ['_trace_logger', 'input_stream', 'sync_obj', '_out_obj', '_cur_out_obj', '_before_critical']:
+        for key in ['_trace_logger', 'input_stream', 'sync_obj', '_out_obj', '_err_obj', '_cur_out_obj', '_cur_err_obj', '_before_critical']:
             state[key] = None
         return state
 
@@ -501,6 +523,11 @@ def get_input_stream(self):
             return self._root_io.get_input_stream()
         return self.input_stream
 
+    def get_cur_err_obj(self):
+        if self._root_io:
+            return self._root_io.get_cur_err_obj()
+        return self._cur_err_obj
+
     def get_cur_out_obj(self):
         if self._root_io:
             return self._root_io.get_cur_out_obj()
@@ -512,17 +539,24 @@ def _start_buffer_io(self):
         if self._cur_out_obj != self._out_obj:
             return False
         self._cur_out_obj = BufferIO()
+        self._cur_err_obj = BufferIO()
         return True
 
     def _stop_buffer_io(self):
         if self._root_io:
             return False
         if self._cur_out_obj == self._out_obj:
             return False
+        if self._cur_err_obj == self._err_obj:
+            return False
         text = self._cur_out_obj.read()
+        text_err = self._cur_err_obj.read()
         self._cur_out_obj = self._out_obj
+        self._cur_err_obj = self._err_obj
         if text:
             self.print(text)
+        if text_err:
+            self.error(text_err)
         return True
 
     @staticmethod
@@ -680,7 +714,10 @@ def _print(self, msg_lv, msg, *args, **kwargs):
             del kwargs['prev_msg']
         else:
             print_msg = msg
-        kwargs['file'] = self.get_cur_out_obj()
+        if msg_lv == MsgLevel.ERROR:
+            kwargs['file'] = self.get_cur_err_obj()
+        else:
+            kwargs['file'] = self.get_cur_out_obj()
         kwargs['file'] and print(self._format(print_msg, *args), **kwargs)
         del kwargs['file']
         self.log(msg_lv, msg, *args, **kwargs)
@@ -733,6 +770,16 @@ def verbose(self, msg, *args, **kwargs):
             return
         self._print(MsgLevel.VERBOSE, '%s %s' % (self._verbose_prefix, msg), *args, **kwargs)
 
+    def print_result_json(self, result):
+
+        if not result:
+            return
+        if isinstance(result, dict):
+            result = json.dumps(result, indent=4)
+        self.print(result)
+
+        pass
+
     if sys.version_info.major == 2:
 
         def exception(self, msg='', *args, **kwargs):