提供trace_id参数时精确锁定节点去获取日志 (#442)

* gather scene run add --skip_type option

* close #279

* fix

* fix

* fix

common/command.py

@@ -449,7 +449,7 @@ def find_home_path_by_port(ssh_client, internal_port_str, stdio):
     for original_str in str_list:
         original_str = str(original_str)
         if original_str.endswith("/bin/observer") and not original_str.startswith('/[^\s]*'):
-            home_path = original_str.rstrip("/bin/observer")
+            home_path = original_str[: -len("/bin/observer")]
             break
     stdio.verbose("home_path:{0}".format(home_path))
     return home_path

common/tool.py

@@ -1168,15 +1168,23 @@ def parse_env(env_string, stdio=None):
         return env_dict
 
     @staticmethod
-    def get_observer_ip_from_trace_id(content, stdio=None):
-        if content[0] == 'Y' and len(content) >= 12:
-            sep = content.find('-')
-            uval = int(content[1:sep], 16)
+    def get_observer_ip_port_from_trace_id(trace_id):
+        if len(trace_id) >= 50:
+            raise ValueError(f"Trace_id({trace_id}) is invalid due to its length.")
+
+        if trace_id[0] == 'Y':
+            id_ = trace_id.split('-')[0].split('Y')[1]
+            uval = int(id_, 16)
             ip = uval & 0xFFFFFFFF
             port = (uval >> 32) & 0xFFFF
-            return "%d.%d.%d.%d:%d" % ((ip >> 24 & 0xFF), (ip >> 16 & 0xFF), (ip >> 8 & 0xFF), (ip >> 0 & 0xFF), port)
+            ip_str = f"{(ip >> 24) & 0xFF}.{(ip >> 16) & 0xFF}.{(ip >> 8) & 0xFF}.{ip & 0xFF}"
+            origin_ip_port = f"{ip_str}:{port}"
         else:
-            return ""
+            parts = trace_id.split('-')
+            processed_parts = [hex(int(t))[2:].upper().zfill(16 if idx == 1 else 0) for idx, t in enumerate(parts)]
+            s = 'Y' + '-'.join(processed_parts)
+            origin_ip_port = StringUtils.get_observer_ip_port_from_trace_id(s)
+        return origin_ip_port
 
     @staticmethod
     def parse_range_string(range_str, nu, stdio=None):

handler/gather/scenes/sql_problem.py

@@ -21,6 +21,8 @@
 from handler.gather.gather_obproxy_log import GatherObProxyLogHandler
 from handler.gather.gather_plan_monitor import GatherPlanMonitorHandler
 from common.tool import StringUtils
+from common.ssh_client.ssh import SshClient
+from common.command import find_home_path_by_port
 
 
 class SQLProblemScene(SafeStdio):
@@ -40,6 +42,7 @@ def __init__(self, context, scene_name, report_path, task_variable_dict=None, en
         self.scene_name = scene_name
         self.db_conn = {}
         self.trace_id = "FAKE_TRACE_ID"
+        self.task_nodes = []
 
     def execute(self):
         skip_type = self.context.get_variable("gather_skip_type", None)
@@ -52,10 +55,29 @@ def execute(self):
             if skip_type != "sql":
                 self.__gather_sql_info()
 
+    def __find_home_path_by_port(self, ip_str, internal_port_str):
+        for node in self.ob_nodes:
+            if node.get("ip") == ip_str:
+                remote_ip = node.get("ip")
+                remote_user = node.get("ssh_username")
+                try:
+                    ssh_client = SshClient(self.context, node)
+                    return find_home_path_by_port(ssh_client, internal_port_str, self.stdio)
+                except Exception as e:
+                    self.stdio.error("ssh {0}@{1}: failed, Please check the config".format(remote_user, remote_ip))
+
     def __gather_log(self):
         try:
+            ip_port_str = StringUtils.get_observer_ip_port_from_trace_id(self.trace_id)
+            ip_str, internal_port_str = ip_port_str.split(':')
+            home_path_str = self.__find_home_path_by_port(ip_str, internal_port_str)
+            for node in self.ob_nodes:
+                if node.get("ip") == ip_str and node.get("home_path") == home_path_str:
+                    self.task_nodes.append(node)
+                    break
             self.stdio.verbose("gather observer log start")
             handler = GatherLogHandler(self.context, self.report_path, is_scene=True)
+            self.context.set_variable('filter_nodes_list', self.task_nodes)
             self.context.set_variable('gather_grep', self.trace_id)
             handler.handle()
             self.stdio.verbose("gather observer log end")