日志服务Python消费组实战（三）：实时跨域监测多日志库数据

时间 2019-12-05

原文原文链接

解决问题

使用日志服务进行数据处理与传递的过程当中，你是否遇到以下监测场景不能很好的解决：html

特定数据上传到日志服务中须要检查数据内的异常状况，而没有现成监控工具？
须要检索数据里面的关键字，但数据没有创建索引，没法使用日志服务的告警功能？
数据监测要求实时性（<5秒，例如Web访问500错误），而特定功能都有必定延迟（1分钟以上）？
存在多个域的多个日志库（例如每一个Region的错误文件对应的日志库），数据量不大，但监控逻辑相似，每一个目标都要监控与配置，比较繁琐？

若是是的，您能够考虑使用日志服务Python消费组进行跨域实时数据监控，本文主要介绍如何使用消费组实时监控多个域中的多个日志库中的异常数据，并进行下一步告警动做。能够很好解决以上问题，并利用消费组的特色，达到自动平衡、负载均衡和高可用性。python

基本概念

协同消费库（Consumer Library）是对日志服务中日志进行消费的高级模式，提供了消费组（ConsumerGroup）的概念对消费端进行抽象和管理，和直接使用SDK进行数据读取的区别在于，用户无需关心日志服务的实现细节，只须要专一于业务逻辑，另外，消费者之间的负载均衡、failover等用户也都无需关心。git

消费组（Consumer Group） - 一个消费组由多个消费者构成，同一个消费组下面的消费者共同消费一个logstore中的数据，消费者之间不会重复消费数据。
消费者（Consumer） - 消费组的构成单元，实际承担消费任务，同一个消费组下面的消费者名称必须不一样。github

在日志服务中，一个logstore下面会有多个shard，协同消费库的功能就是将shard分配给一个消费组下面的消费者，分配方式遵循如下原则：编程

每一个shard只会分配到一个消费者。
一个消费者能够同时拥有多个shard。
新的消费者加入一个消费组，这个消费组下面的shard从属关系会调整，以达到消费负载均衡的目的，可是上面的分配原则不会变，分配过程对用户透明。

协同消费库的另外一个功能是保存checkpoint，方便程序故障恢复时能接着从断点继续消费，从而保证数据不会被重复消费。跨域

使用消费组进行实时分发

这里咱们描述用Python使用消费组进行编程，实时跨域监测多个域的多个日志库，全文或特定字段检查
注意：本篇文章的相关代码可能会更新，最新版本在这里能够找到：Github样例.服务器

安装

环境微信

建议程序运行在靠近源日志库同Region下的ECS上，并使用局域网服务入口，这样好处是网络速度最快，其次是读取没有外网费用产生。
强烈推荐PyPy3来运行本程序，而不是使用标准CPython解释器。
日志服务的Python SDK能够以下安装：

pypy3 -m pip install aliyun-log-python-sdk -U

更多SLS Python SDK的使用手册，能够参考这里网络

程序配置

以下展现如何配置程序：并发

配置程序日志文件，以便后续测试或者诊断可能的问题（跳过，具体参考样例）。
基本的日志服务链接与消费组的配置选项。
目标Logstore的一些链接信息

请仔细阅读代码中相关注释并根据须要调整选项：

#encoding: utf8
def get_option():
    ##########################
    # 基本选项
    ##########################

    # 从环境变量中加载SLS参数与选项，endpoint、project、logstore能够多个并配对
    endpoints = os.environ.get('SLS_ENDPOINTS', '').split(";")  # ;分隔
    projects = os.environ.get('SLS_PROJECTS', '').split(";")    # ;分隔
    logstores = os.environ.get('SLS_LOGSTORES', '').split(";")  # ;分隔，同一个Project下的用,分隔
    accessKeyId = os.environ.get('SLS_AK_ID', '')
    accessKey = os.environ.get('SLS_AK_KEY', '')
    consumer_group = os.environ.get('SLS_CG', '')

    # 消费的起点。这个参数在第一次跑程序的时候有效，后续再次运行将从上一次消费的保存点继续。
    # 可使”begin“，”end“，或者特定的ISO时间格式。
    cursor_start_time = "2018-12-26 0:0:0"

    # 通常不要修改消费者名，尤为是须要并发跑时
    consumer_name = "{0}-{1}".format(consumer_group, current_process().pid)

      # 设定共享执行器
    exeuctor = ThreadPoolExecutor(max_workers=2)

    # 构建多个消费组（每一个logstore一个）
    options = []
    for i in range(len(endpoints)):
        endpoint = endpoints[i].strip()
        project = projects[i].strip()
        if not endpoint or not project:
            logger.error("project: {0} or endpoint {1} is empty, skip".format(project, endpoint))
            continue

        logstore_list = logstores[i].split(",")
        for logstore in logstore_list:
            logstore = logstore.strip()
            if not logstore:
                logger.error("logstore for project: {0} or endpoint {1} is empty, skip".format(project, endpoint))
                continue

            option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group,
                                  consumer_name, cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                                  cursor_start_time=cursor_start_time, shared_executor=exeuctor)
            options.append(option)

    # 设定检测目标字段与目标值，例如这里是检测status字段是否有500等错误
    keywords = {'status': r'5\d{2}'}

    return exeuctor, options, keywords

注意，配置了多个endpoint、project、logstore，须要用分号分隔，而且一一对应；若是一个project下有多个logstore须要检测，能够将他们直接用逗号分隔。以下是一个检测3个Region下的4个Logstore的配置：

export SLS_ENDPOINTS=cn-hangzhou.log.aliyuncs.com;cn-beijing.log.aliyuncs.com;cn-qingdao.log.aliyuncs.com
export SLS_PROJECTS=project1;project2;project3
export SLS_LOGSTORES=logstore1;logstore2;logstore3_1,logstore3_2

数据监测

以下代码展现如何构建一个关键字检测器，针对数据中的目标字段进行检测，您也能够修改逻辑设定为符合须要的场景（例如多个字段的组合关系等）。

class KeywordMonitor(ConsumerProcessorBase):
    """
    this consumer will keep monitor with k-v fields. like {"content": "error"}
    """
    def __init__(self, keywords=None, logstore=None):
        super(KeywordMonitor, self).__init__()  # remember to call base init

        self.keywords = keywords
        self.kw_check = {}
        for k, v in self.keywords.items():
            self.kw_check[k] = re.compile(v)
        self.logstore = logstore

    def process(self, log_groups, check_point_tracker):
        logs = PullLogResponse.loggroups_to_flattern_list(log_groups)
        match_count = 0
        sample_error_log = ""
        for log in logs:
            m = None
            for k, c in self.kw_check.items():
                if k in log:
                    m = c.search(log[k])
                    if m:
                        logger.debug('Keyword detected for shard "{0}" with keyword: "{1}" in field "{2}", log: {3}'
                                    .format(self.shard_id, log[k], k, log))
            if m:
                match_count += 1
                sample_error_log = log

        if match_count:
            logger.info("Keyword detected for shard {0}, count: {1}, example: {2}".format(self.shard_id, match_count, sample_error_log))
                    
            # TODO： 这里添加通知下游的代码
    
        else:
            logger.debug("No keyword detected for shard {0}".format(self.shard_id))

        self.save_checkpoint(check_point_tracker)

控制逻辑

以下展现如何控制多个消费者，并管理退出命令：

def main():
    exeuctor, options, keywords = get_monitor_option()

    logger.info("*** start to consume data...")
    workers = []

    for option in options:
        worker = ConsumerWorker(KeywordMonitor, option, args=(keywords,) )
        workers.append(worker)
        worker.start()

    try:
        for i, worker in enumerate(workers):
            while worker.is_alive():
                worker.join(timeout=60)
            logger.info("worker project: {0} logstore: {1} exit unexpected, try to shutdown it".format(
                options[i].project, options[i].logstore))
            worker.shutdown()
    except KeyboardInterrupt:
        logger.info("*** try to exit **** ")
        for worker in workers:
            worker.shutdown()

        # wait for all workers to shutdown before shutting down executor
        for worker in workers:
            while worker.is_alive():
                worker.join(timeout=60)

    exeuctor.shutdown()


if __name__ == '__main__':
    main()

启动

假设程序命名为"monitor_keyword.py"，能够以下启动：

export SLS_ENDPOINTS=cn-hangzhou.log.aliyuncs.com;cn-beijing.log.aliyuncs.com;cn-qingdao.log.aliyuncs.com
export SLS_PROJECTS=project1;project2;project3
export SLS_LOGSTORES=logstore1;logstore2;logstore3_1,logstore3_2

export SLS_AK_ID=<YOUR AK ID>
export SLS_AK_KEY=<YOUR AK KEY>
export SLS_CG=<消费组名，能够简单命名为"dispatch_data">

pypy3 monitor_keyword.py

性能考虑

启动多个消费者

若是您的目标logstore存在多个shard，或者您的目标监测日志库较多，您能够进行必定划分并并启动屡次程序：

# export SLS_ENDPOINTS, SLS_PROJECTS, SLS_LOGSTORES
nohup pypy3 dispatch_data.py &

# export SLS_ENDPOINTS, SLS_PROJECTS, SLS_LOGSTORES
nohup pypy3 dispatch_data.py &

# export SLS_ENDPOINTS, SLS_PROJECTS, SLS_LOGSTORES
nohup pypy3 dispatch_data.py &
...

注意:
全部消费者使用了同一个消费组的名字和不一样的消费者名字（由于消费者名以进程ID为后缀）。
但数据量较大或者目标日志库较多时，单个消费者的速度可能没法知足需求，且由于Python的GIL的缘由，只能用到一个CPU核。强烈建议您根据目标日志库的Shard数以及CPU的数量进行划分，启动屡次以便重复利用CPU资源。

性能吞吐

基于测试，在没有带宽限制、接收端速率限制（如Splunk端）的状况下，以推动硬件用pypy3运行上述样例，单个消费者占用大约10%的单核CPU下能够消费达到5 MB/s原始日志的速率。所以，理论上能够达到50 MB/s原始日志每一个CPU核，也就是每一个CPU核天天能够消费4TB原始日志。

注意: 这个数据依赖带宽、硬件参数等。

高可用性

消费组会将检测点（check-point）保存在服务器端，当一个消费者中止，另一个消费者将自动接管并从断点继续消费。

能够在不一样机器上启动消费者，这样当一台机器中止或者损坏的清下，其余机器上的消费者能够自动接管并从断点进行消费。

理论上，为了备用，也能够启动大于shard数量的消费者。

其余

限制与约束

每个日志库（logstore）最多能够配置10个消费组，若是遇到错误ConsumerGroupQuotaExceed则表示遇到限制，建议在控制台端删除一些不用的消费组。

监测

Https

若是服务入口（endpoint）配置为https://前缀，如https://cn-beijing.log.aliyuncs.com，程序与SLS的链接将自动使用HTTPS加密。

服务器证书*.aliyuncs.com是GlobalSign签发，默认大多数Linux/Windows的机器会自动信任此证书。若是某些特殊状况，机器不信任此证书，能够参考这里下载并安装此证书。