feature: 作业平台插件提取变量支持多行文本提取

pipeline_plugins/components/collections/sites/open/job/base.py

@@ -74,29 +74,25 @@ def get_sops_var_dict_from_log_text(log_text, service_logger):
     {"key1": "value1", "key2": "value2"}
     """
     sops_var_dict = {}
-    # 逐行匹配以便打印全局变量所在行
+    # 支持跨行匹配全局变量
     service_logger.info("search log var with config: {}".format(LOG_VAR_SEARCH_CONFIGS))
-    for index, log_line in enumerate(log_text.splitlines(), 1):
-        for var_search_config in LOG_VAR_SEARCH_CONFIGS:
-            reg = var_search_config["re"]
-            excape_reg = reg.replace("<", "&lt;").replace(">", "&gt;")
-            kv_sep = var_search_config["kv_sep"]
-
-            sops_key_val_list = re.findall(reg, log_line)
-            sops_key_val_list.extend(re.findall(excape_reg, log_line))
-            if len(sops_key_val_list) == 0:
+    for var_search_config in LOG_VAR_SEARCH_CONFIGS:
+        reg = var_search_config["re"]
+        excape_reg = reg.replace("<", "&lt;").replace(">", "&gt;")
+        kv_sep = var_search_config["kv_sep"]
+
+        sops_key_val_list = re.findall(reg, log_text, re.DOTALL)
+        sops_key_val_list.extend(re.findall(excape_reg, log_text, re.DOTALL))
+        if len(sops_key_val_list) == 0:
+            continue
+        for sops_key_val in sops_key_val_list:
+            if kv_sep not in sops_key_val:
                 continue
-            for sops_key_val in sops_key_val_list:
-                if kv_sep not in sops_key_val:
-                    continue
-                sops_key, sops_val = sops_key_val.split(kv_sep, 1)
-                # 限制变量名不为空
-                if len(sops_key) == 0:
-                    continue
-                sops_var_dict.update({sops_key: sops_val})
-            service_logger.info(
-                _("[{group}]提取日志中全局变量，匹配行[{index}]：[{line}]").format(group=__group_name__, index=index, line=log_line)
-            )
+            sops_key, sops_val = sops_key_val.split(kv_sep, 1)
+            # 限制变量名不为空
+            if len(sops_key) == 0:
+                continue
+            sops_var_dict.update({sops_key: sops_val})
     return sops_var_dict
 
 
@@ -234,12 +230,7 @@ def get_job_tagged_ip_dict(
     result = client.jobv3.get_job_instance_status(kwargs)
 
     if not result["result"]:
-        message = handle_api_error(
-            __group_name__,
-            "jobv3.get_job_instance_status",
-            kwargs,
-            result,
-        )
+        message = handle_api_error(__group_name__, "jobv3.get_job_instance_status", kwargs, result,)
         service_logger.warning(message)
         return False, message
 
@@ -339,12 +330,7 @@ def get_job_tagged_ip_dict_complex(
     result = client.jobv3.get_job_instance_status(kwargs)
 
     if not result["result"]:
-        message = handle_api_error(
-            __group_name__,
-            "jobv3.get_job_instance_status",
-            kwargs,
-            result,
-        )
+        message = handle_api_error(__group_name__, "jobv3.get_job_instance_status", kwargs, result,)
         service_logger.warning(message)
         return False, message
 
@@ -520,9 +506,7 @@ def schedule(self, data, parent_data, callback_data=None):
 
                 if not global_var_result["result"]:
                     message = job_handle_api_error(
-                        "jobv3.get_job_instance_global_var_value",
-                        get_var_kwargs,
-                        global_var_result,
+                        "jobv3.get_job_instance_global_var_value", get_var_kwargs, global_var_result,
                     )
                     self.logger.error(message)
                     data.outputs.ex_data = message
@@ -762,9 +746,7 @@ def schedule(self, data, parent_data, callback_data=None):
 
                 if not global_var_result["result"]:
                     message = job_handle_api_error(
-                        "jobv3.get_job_instance_global_var_value",
-                        get_var_kwargs,
-                        global_var_result,
+                        "jobv3.get_job_instance_global_var_value", get_var_kwargs, global_var_result,
                     )
                     self.logger.error(message)
                     data.outputs.ex_data = message
@@ -921,12 +903,7 @@ def get_job_history_result(self, data, parent_data):
         job_result = client.jobv3.get_job_instance_status(job_kwargs)
 
         if not job_result["result"]:
-            message = handle_api_error(
-                __group_name__,
-                "jobv3.get_job_instance_status",
-                job_kwargs,
-                job_result,
-            )
+            message = handle_api_error(__group_name__, "jobv3.get_job_instance_status", job_kwargs, job_result,)
             self.logger.error(message)
             data.outputs.ex_data = message
             self.logger.info(data.outputs)
@@ -946,10 +923,7 @@ def get_job_history_result(self, data, parent_data):
             return True
 
         get_job_sops_var_dict_return = get_job_sops_var_dict(
-            client,
-            self.logger,
-            job_success_id,
-            data.get_one_of_inputs("biz_cc_id", parent_data.inputs.biz_cc_id),
+            client, self.logger, job_success_id, data.get_one_of_inputs("biz_cc_id", parent_data.inputs.biz_cc_id),
         )
         if not get_job_sops_var_dict_return["result"]:
             self.logger.error(

pipeline_plugins/tests/components/collections/sites/open/job_test/test_get_sops_var_dict_from_log_text.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+import logging
+
+from django.test import TestCase
+
+from pipeline_plugins.components.collections.sites.open.job.base import get_sops_var_dict_from_log_text
+
+logger = logging.getLogger(__name__)
+
+
+MULTI_LINE_TEXT1 = """
+<SOPS_VAR>key1:value1</SOPS_VAR>
+gsectl -rwxr-xr-x 1 root
+<SOPS_VAR>key2:value2</SOPS_VAR>
+gsectl -rwxr-xr-x 1 root
+"""
+
+MULTI_LINE_TEXT2 = """
+<SOPS_VAR>key1:
+value1</SOPS_VAR>
+abcd
+<SOPS_VAR>key
+2:value2</SOPS_VAR>
+"""
+
+MULTI_LINE_TEXT3 = """<SOPS_VAR>
+key1: value1
+key2: value2
+</SOPS_VAR>
+"""
+
+
+MULTI_LINE_TEXT4 = """
+<SOPS_VAR>
+key1: value1 key2:value2
+</SOPS_VAR>
+"""
+
+
+MULTI_LINE_TEXT5 = "<SOPS_VAR> key1: value1\r\nkey2:value2 </SOPS_VAR>"
+
+
+class GetSOPSVarDictFromLogText(TestCase):
+    def test_single_line_log(self):
+        text = "<SOPS_VAR>key1:value1</SOPS_VAR> gsectl -rwxr-xr-x 1 root<SOPS_VAR>key2:value2</SOPS_VAR>"
+        result = get_sops_var_dict_from_log_text(text, logger)
+        self.assertEqual(result, {"key1": "value1", "key2": "value2"})
+
+        text2 = "<SOPS_VAR> key1 : value1 </SOPS_VAR> gsectl -rwxr-xr-x 1 root<SOPS_VAR>key2:value2</SOPS_VAR>"
+        result2 = get_sops_var_dict_from_log_text(text2, logger)
+        self.assertEqual(result2, {" key1 ": " value1 ", "key2": "value2"})
+
+    def test_multiple_line_log(self):
+        test_cases = [
+            (MULTI_LINE_TEXT1, {"key1": "value1", "key2": "value2"}),
+            (MULTI_LINE_TEXT2, {"key1": "\nvalue1", "key\n2": "value2"}),
+            (MULTI_LINE_TEXT3, {"\nkey1": " value1\nkey2: value2\n"}),
+            (MULTI_LINE_TEXT4, {"\nkey1": " value1 key2:value2\n"}),
+            (MULTI_LINE_TEXT5, {" key1": " value1\r\nkey2:value2 "}),
+        ]
+        for log, expected_result in test_cases:
+            result = get_sops_var_dict_from_log_text(log, logger)
+            self.assertEqual(result, expected_result)