feat: add logtail pipeline config (#344)

* feat: add logtail pipeline config support

* fix warning

* bump version 0.9.35

* add sample

Author: crimson-gao

aliyun/log/__init__.py

@@ -7,6 +7,8 @@
 from .listtopicsrequest import ListTopicsRequest
 from .listlogstoresrequest import ListLogstoresRequest
 from .logtail_config_detail import *
+from .logtail_pipeline_config_detail import *
+from .logtail_pipeline_config_response import *
 from .machine_group_detail import MachineGroupDetail
 from .putlogsrequest import PutLogsRequest
 from .shipper_config import ShipperTask, OssShipperConfig, OdpsShipperConfig

aliyun/log/etl_core/trans_comp/trans_kv.py

@@ -32,12 +32,14 @@ class kv_transformer(trans_comp_check_mdoe_base):
     @staticmethod
     @cached
     def _get_kv_ptn(sep, quote, escape):
-        p1 = u'(?!{0})([\u4e00-\u9fa5\u0800-\u4e00\\w\\.\\-]+)\\s*{0}\\s*([\u4e00-\u9fa5\u0800-\u4e00\\w\\.\\-]+)'
+        # Use Unicode characters directly to avoid escape sequence warnings
+        # \u4e00-\u9fa5 = Chinese characters, \u0800-\u4e00 = other Unicode range
+        p1 = r'(?!{0})([' + '\u4e00-\u9fa5\u0800-\u4e00' + r'\w\.\-]+)\s*{0}\s*([' + '\u4e00-\u9fa5\u0800-\u4e00' + r'\w\.\-]+)'
         if not escape:
-            p2 = u'(?!{0})([\u4e00-\u9fa5\u0800-\u4e00\\w\\.\\-]+)\\s*{0}\\s*{1}\s*([^{1}]*?)\s*{1}'
+            p2 = r'(?!{0})([' + '\u4e00-\u9fa5\u0800-\u4e00' + r'\w\.\-]+)\s*{0}\s*{1}\s*([^{1}]*?)\s*{1}'
         else:
-            p2 = u'(?!{0})([\u4e00-\u9fa5\u0800-\u4e00\\w\\.\\-]+)\\s*{0}\\s*{1}\s*((?:[^{1}]|\\\\{1})*?[^\\\\]){1}'
-        ps = u'|'.join([p1, p2]).format(sep, quote)
+            p2 = r'(?!{0})([' + '\u4e00-\u9fa5\u0800-\u4e00' + r'\w\.\-]+)\s*{0}\s*{1}\s*((?:[^{1}]|\\{1})*?[^\\]){1}'
+        ps = '|'.join([p1, p2]).format(sep, quote)
 
         logger.info(u"trans_comp_kv: get ptn: {0}".format(ps))
         return re.compile(ps)

aliyun/log/logclient.py

@@ -41,6 +41,7 @@
 from .logstore_config_response import *
 from .substore_config_response import *
 from .logtail_config_response import *
+from .logtail_pipeline_config_response import *
 from .machinegroup_response import *
 from .rebuild_index_response import *
 from .project_response import *
@@ -2328,6 +2329,132 @@ def list_logtail_config(self, project_name, logstore=None, config=None, offset=0
         (resp, header) = self._send("GET", project_name, None, resource, params, headers)
         return ListLogtailConfigResponse(resp, header)
 
+    def create_logtail_pipeline_config(self, project_name, config_detail):
+        """ create logtail pipeline config in a project
+        Unsuccessful operation will cause an LogException.
+
+        :type project_name: string
+        :param project_name: the Project name
+
+        :type config_detail: LogtailPipelineConfigDetail
+        :param config_detail: the logtail pipeline config detail
+
+        :return: CreateLogtailPipelineConfigResponse
+
+        :raise: LogException
+        """
+        headers = {}
+        params = {}
+        resource = "/pipelineconfigs"
+        headers['Content-Type'] = 'application/json'
+        body = six.b(json.dumps(config_detail.to_json()))
+        headers['x-log-bodyrawsize'] = str(len(body))
+        (resp, headers) = self._send("POST", project_name, body, resource, params, headers)
+        return CreateLogtailPipelineConfigResponse(headers, resp)
+
+    def update_logtail_pipeline_config(self, project_name, config_detail):
+        """ update logtail pipeline config in a project
+        Unsuccessful operation will cause an LogException.
+
+        :type project_name: string
+        :param project_name: the Project name
+
+        :type config_detail: LogtailPipelineConfigDetail
+        :param config_detail: the logtail pipeline config detail
+
+        :return: UpdateLogtailPipelineConfigResponse
+
+        :raise: LogException
+        """
+        headers = {}
+        params = {}
+        resource = "/pipelineconfigs/" + config_detail.config_name
+        headers['Content-Type'] = 'application/json'
+        body = six.b(json.dumps(config_detail.to_json()))
+        headers['x-log-bodyrawsize'] = str(len(body))
+        (resp, headers) = self._send("PUT", project_name, body, resource, params, headers)
+        return UpdateLogtailPipelineConfigResponse(headers, resp)
+
+    def delete_logtail_pipeline_config(self, project_name, config_name):
+        """ delete logtail pipeline config in a project
+        Unsuccessful operation will cause an LogException.
+
+        :type project_name: string
+        :param project_name: the Project name
+
+        :type config_name: string
+        :param config_name: the logtail pipeline config name
+
+        :return: DeleteLogtailPipelineConfigResponse
+
+        :raise: LogException
+        """
+        headers = {}
+        params = {}
+        resource = "/pipelineconfigs/" + config_name
+        (resp, headers) = self._send("DELETE", project_name, None, resource, params, headers)
+        return DeleteLogtailPipelineConfigResponse(headers, resp)
+
+    def get_logtail_pipeline_config(self, project_name, config_name):
+        """ get logtail pipeline config in a project
+        Unsuccessful operation will cause an LogException.
+
+        :type project_name: string
+        :param project_name: the Project name
+
+        :type config_name: string
+        :param config_name: the logtail pipeline config name
+
+        :return: GetLogtailPipelineConfigResponse
+
+        :raise: LogException
+        """
+        headers = {}
+        params = {}
+        resource = "/pipelineconfigs/" + config_name
+        (resp, headers) = self._send("GET", project_name, None, resource, params, headers)
+        return GetLogtailPipelineConfigResponse(resp, headers)
+
+    def list_logtail_pipeline_config(self, project_name, config_name=None, logstore_name=None, offset=0, size=100):
+        """ list logtail pipeline config names in a project
+        Unsuccessful operation will cause an LogException.
+
+        :type project_name: string
+        :param project_name: the Project name
+
+        :type config_name: string
+        :param config_name: config name to filter
+
+        :type logstore_name: string
+        :param logstore_name: logstore name to filter
+
+        :type offset: int
+        :param offset: the offset of all config names
+
+        :type size: int
+        :param size: the max return names count, -1 means all
+
+        :return: ListLogtailPipelineConfigResponse
+
+        :raise: LogException
+        """
+        # need to use extended method to get more
+        if int(size) == -1 or int(size) > MAX_LIST_PAGING_SIZE:
+            return list_more(self.list_logtail_pipeline_config, int(offset), int(size), MAX_LIST_PAGING_SIZE,
+                           project_name, config_name, logstore_name)
+
+        headers = {}
+        params = {}
+        resource = "/pipelineconfigs"
+        params['offset'] = str(offset)
+        params['size'] = str(size)
+        if config_name:
+            params['configName'] = config_name
+        if logstore_name:
+            params['logstoreName'] = logstore_name
+        (resp, header) = self._send("GET", project_name, None, resource, params, headers)
+        return ListLogtailPipelineConfigResponse(resp, header)
+
     def create_machine_group(self, project_name, group_detail):
         """ create machine group in a project
         Unsuccessful operation will cause an LogException.

aliyun/log/logtail_pipeline_config_detail.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python
+# encoding: utf-8
+
+# Copyright (C) Alibaba Cloud Computing
+# All rights reserved.
+
+"""
+Logtail Pipeline Config Detail Classes
+
+This module provides classes for managing Logtail pipeline configurations.
+"""
+
+from __future__ import print_function
+import json
+
+__all__ = ['LogtailPipelineConfigDetail']
+
+
+class LogtailPipelineConfigDetail(object):
+    """Logtail Pipeline Config Detail
+    
+    :type config_name: string
+    :param config_name: config name, can only contain lowercase letters, numbers, - and _, 
+                        must start and end with lowercase letter or number, length between 2-128 characters
+    
+    :type log_sample: string
+    :param log_sample: log sample, supports multiple log entries
+    
+    :type global_config: dict
+    :param global_config: global configuration
+    
+    :type inputs: list
+    :param inputs: input plugin list, currently only allows 1 input plugin
+    
+    :type processors: list
+    :param processors: processor plugin list
+    
+    :type aggregators: list
+    :param aggregators: aggregator plugin list, maximum 1 aggregator plugin allowed
+    
+    :type flushers: list
+    :param flushers: flusher plugin list, currently only allows 1 flusher_sls plugin
+    """
+    
+    def __init__(self, config_name, inputs, flushers, log_sample=None, 
+                 global_config=None, processors=None, aggregators=None):
+        """Initialize a Logtail Pipeline Config Detail
+        
+        :type config_name: string
+        :param config_name: config name
+        
+        :type inputs: list
+        :param inputs: input plugin list
+        
+        :type flushers: list
+        :param flushers: flusher plugin list
+        
+        :type log_sample: string
+        :param log_sample: log sample (optional)
+        
+        :type global_config: dict
+        :param global_config: global configuration (optional)
+        
+        :type processors: list
+        :param processors: processor plugin list (optional)
+        
+        :type aggregators: list
+        :param aggregators: aggregator plugin list (optional)
+        """
+        self.config_name = config_name
+        self.log_sample = log_sample
+        self.global_config = global_config if global_config is not None else {}
+        self.inputs = inputs if inputs is not None else []
+        self.processors = processors if processors is not None else []
+        self.aggregators = aggregators if aggregators is not None else []
+        self.flushers = flushers if flushers is not None else []
+    
+    def to_json(self):
+        """Convert to JSON object
+        
+        :return: dict object ready for JSON serialization
+        """
+        json_value = {
+            "configName": self.config_name
+        }
+        
+        # Add optional fields
+        if self.log_sample:
+            json_value["logSample"] = self.log_sample
+        
+        if self.global_config:
+            json_value["global"] = self.global_config
+        
+        # Required fields
+        json_value["inputs"] = self.inputs
+        json_value["flushers"] = self.flushers
+        
+        # Optional fields
+        if self.processors:
+            json_value["processors"] = self.processors
+        
+        if self.aggregators:
+            json_value["aggregators"] = self.aggregators
+        
+        return json_value
+    
+    @staticmethod
+    def from_json(json_value):
+        """Create LogtailPipelineConfigDetail from JSON object
+        
+        :type json_value: dict
+        :param json_value: JSON object
+        
+        :return: LogtailPipelineConfigDetail object
+        """
+        config_name = json_value.get("configName", "")
+        log_sample = json_value.get("logSample", None)
+        global_config = json_value.get("global", None)
+        inputs = json_value.get("inputs", [])
+        processors = json_value.get("processors", None)
+        aggregators = json_value.get("aggregators", None)
+        flushers = json_value.get("flushers", [])
+        
+        return LogtailPipelineConfigDetail(
+            config_name=config_name,
+            inputs=inputs,
+            flushers=flushers,
+            log_sample=log_sample,
+            global_config=global_config,
+            processors=processors,
+            aggregators=aggregators
+        )
+
+
+