备案控制台
探索云世界
开发者社区 云存储 文章 正文

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

2019-01-03 4182
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
对象存储 OSS，20GB 3个月
推荐场景：
基于PAI-EAS挂载OSS部署AIGC服务 ossutil工具管理OSS
对象存储 OSS，恶意文件检测 1000次 1年
对象存储 OSS，内容安全 1000次 1年
简介： 本文主要介绍如何使用消费组实时监控多个域中的多个日志库中的异常数据，并进行下一步告警动作。具备配置简单、逻辑灵活、支持跨域多Region、实时监测，无需配置索引等特点，并且性能可靠、自动负载均衡与保持高可用性。

解决问题

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

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

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

image

基本概念

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

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

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

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

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

使用消费组进行实时分发

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

安装

环境

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

更多SLS Python SDK的使用手册，可以参考这里

程序配置

如下展示如何配置程序：

  1. 配置程序日志文件，以便后续测试或者诊断可能的问题（跳过，具体参考样例）。
  2. 基本的日志服务连接与消费组的配置选项。
  3. 目标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的机器会自动信任此证书。如果某些特殊情况，机器不信任此证书，可以参考这里下载并安装此证书。

更多案例

文章标签：
日志服务
监控
Python
关键词：
Python库
日志服务数据
Python数据
Python实战
Python库数据
相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
成喆
目录
相关文章
李振良_阿良
|
1天前
|
JSON Shell 数据格式
第十章 Python常用标准库使用（必会）
第十章 Python常用标准库使用（必会）
李振良_阿良
4 0
L&ecirc;v&iuml;
|
1天前
|
开发框架 前端开发 数据库
Python从入门到精通：3.3.2 深入学习Python库和框架：Web开发框架的探索与实践
Python从入门到精通：3.3.2 深入学习Python库和框架：Web开发框架的探索与实践
L&ecirc;v&iuml;
7 1
L&ecirc;v&iuml;
|
1天前
|
数据采集 数据可视化 数据处理
Python从入门到精通的文章3.3.1 深入学习Python库和框架：数据处理与可视化的利器
Python从入门到精通的文章3.3.1 深入学习Python库和框架：数据处理与可视化的利器
L&ecirc;v&iuml;
3 0
小Lee
|
1天前
|
JSON 测试技术 API
Python的Api自动化测试使用HTTP客户端库发送请求
【4月更文挑战第18天】在Python中进行HTTP请求和API自动化测试有多个库可选：1) `requests`是最流行的选择，支持多种请求方法和内置JSON解析；2) `http.client`是标准库的一部分，适合需要低级别控制的用户；3) `urllib`提供URL操作，适用于复杂请求；4) `httpx`拥有类似`requests`的API，提供现代特性和异步支持。根据具体需求选择，如多数情况`requests`已足够。
小Lee
7 3
同学小张
|
2天前
|
人工智能 Python
【AI大模型应用开发】【LangChain系列】实战案例1：用LangChain写Python代码并执行来生成答案
【AI大模型应用开发】【LangChain系列】实战案例1：用LangChain写Python代码并执行来生成答案
同学小张
6 0
Python
17 0
HannYang
|
2天前
|
JSON 数据格式 索引
python 又一个点运算符操作的字典库：Munch
python 又一个点运算符操作的字典库：Munch
HannYang
14 0
东方睿赢
|
2天前
|
数据挖掘 数据处理 索引
如何使用Python的Pandas库进行数据筛选和过滤？
Pandas是Python数据分析的核心库，提供DataFrame数据结构。基本步骤包括导入库、创建DataFrame及进行数据筛选。示例代码展示了如何通过布尔索引、`query()`和`loc[]`方法筛选`Age`大于19的记录。
东方睿赢
10 0
众所周知
|
3天前
|
数据处理 Python
如何使用Python的Pandas库进行数据排序和排名
【4月更文挑战第22天】Pandas Python库提供数据排序和排名功能。使用`sort_values()`按列进行升序或降序排序，如`df.sort_values(by=&#39;A&#39;, ascending=False)`。`rank()`函数用于计算排名，如`df[&#39;A&#39;].rank(ascending=False)`。多列操作可传入列名列表，如`df.sort_values(by=[&#39;A&#39;, &#39;B&#39;], ascending=[True, False])`和分别对&#39;A&#39;、&#39;B&#39;列排名。
众所周知
13 2
龙大吉
|
4天前
|
算法 Python
请解释Python中的关联规则挖掘以及如何使用Sklearn库实现它。
使用Python的mlxtend库，可以通过Apriori算法进行关联规则挖掘。首先导入TransactionEncoder和apriori等模块，然后准备数据集（如购买行为列表）。对数据集编码并转换后，应用Apriori算法找到频繁项集（设置最小支持度）。最后，生成关联规则并计算置信度（设定最小置信度阈值）。通过调整这些参数可以优化结果。
龙大吉
25 9

云存储

热门文章

最新文章

  • 1
    5分钟搭建网站实时分析：Grafana+日志服务实战
  • 2
    如何打造千万级Feed流系统
  • 3
    高并发IM系统架构优化实践
  • 4
    日志服务数据加工：控制台操作
  • 5
    OSS跨域资源共享(CORS)错误及排除
  • 6
    开放分布式追踪（OpenTracing）入门与 Jaeger 实现
  • 7
    使用阿里云极速型NAS构建高可用的GitLab
  • 8
    日志服务（原SLS）新功能发布(9)--Logtail配置支持主题（Topic）设置功能
  • 9
    使用日志服务LogHub替换Kafka
  • 10
    解密OpenTSDB的表存储优化
  • 1
    邮件发送API的方法和步骤
    9
  • 2
    邮件API接口使用的方法和步骤
    6
  • 3
    Java NIO和IO之间的区别
    7
  • 4
    邮箱API发送邮件调试的方法和步骤
    9
  • 5
    免费邮箱API发送邮件测试调试的方法和步骤
    9
  • 6
    IMAP邮箱API接收收取邮件的方法和步骤
    12
  • 7
    【qemu虚拟化】将img镜像文件转换为VMware虚拟机
    38
  • 8
    SMTP邮件邮箱API发送邮件的方法和步骤
    6
  • 9
    Relay邮件邮箱API发送邮件的方法和步骤
    15
  • 10
    邮件中继中转邮箱API发送邮件的方法和步骤
    17

    • 相关产品

  • 日志服务
    文档详情 产品详情

    • 相关课程

    更多
  • Python爬虫实战
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 相关实验场景

    更多
  • 通过日志服务实现云资源OSS的安全审计
  • 如何将OSS数据导入至SLS中进行分析
  • 日志服务之使用Nginx模式采集日志
  • Python新手入门
  • Python入门
  • 通过FastMR自动拉起大数据集群并运行TPCDS任务
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）