快速入门:接入日志服务SDK上报日志并分析

日志服务SDK支持通过接口手动上传日志。本文介绍使用Python语言SDK上报日志至Logstore,结合控制台进行可视化分析、配置日志异常告警等步骤,帮助您快速上手使用日志服务。

前提条件

  • 已有可用的ECS。更多信息,请参见云服务器ECS快速入门

  • 已开通日志服务。具体操作,请参见开通日志服务

  • 已安装PythonPython开发环境(例如PyCharm)。

    日志服务Python SDK支持以下Python版本。

    • Python2:2.7及以上版本。

    • Python3:3.7及以上版本。

    • Pypy2:2.7及以上版本。

    • Pypy3:3.7及以上版本。

方案概览

image

阅读本文您可以学习:

  • 创建Project(资源管理单元)和Logstore(日志存储单元)。

  • 使用Python SDK上传日志到Logstore进行存储、创建索引以及查询和分析。

  • 将分析结果通过仪表盘转换为可视化图形。

  • 通过配置告警规则实现告警监控。

  • 如何清理不再需要的资源以免产生额外费用。

1. 创建ProjectLogstore

1.1 创建Project

登录日志服务控制台Project列表区域,单击创建Project,在创建Project面板中,选择所属地域,输入Project名称。本文以华东1(杭州)地域,名称为aliyun-test-projectProject为例,其余参数保持默认。

image

1.2 创建Logstore

创建Project完成后,系统会提示您创建一个Logstore。在Logstore面板中,输入Logstore名称,其余参数保持默认。

image

2. 安装SDK

  1. 在命令行工具中,以管理员身份执行如下命令安装Python SDK。更多版本信息,请参见Aliyun Log Python Release

    pip install -U aliyun-log-python-sdk
  2. 安装SDK后,执行以下命令,验证已安装的日志服务 Python SDK。

    pip show aliyun-log-python-sdk

    返回以下信息,代表安装成功。

    Name: aliyun-log-python-sdk
    Version: 0.9.12
    Summary: Aliyun log service Python client SDK
    Home-page: https://github.com/aliyun/aliyun-log-python-sdk
    Author: Aliyun

3. 初始化客户端

LogClient是日志服务的客户端,提供创建ProjectLogstore、写入日志、读取日志等一系列方法。使用AK初始化(V1签名)方法需要以下步骤:

  1. 获取AccessKey:在阿里云控制台创建AccessKey,获取access_key_idaccess_key_secret

  2. 配置环境变量:根据操作系统(Linux/macOS/Windows)将密钥配置为环境变量(例如ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_ACCESS_KEY_SECRET)。

  3. 设置Endpoint:本文以华东1(杭州)地域为例,endpoint配置为 cn-hangzhou.log.aliyuncs.com(其他地域需替换为对应的Endpoint)。

# 引入sls包。
from aliyun.log import *

# 通过环境变量获取AccessKey时需要引入。
import os

# 从环境变量中获取 AccessKey ID 和 AccessKey Secret
access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')

# 日志服务的服务接入点
endpoint = "cn-hangzhou.log.aliyuncs.com"

# 创建 LogClient 实例
client = LogClient(endpoint, access_key_id, access_key_secret)

4. 上报日志

本示例中,通过调用put_logs接口主动上报日志数据。原始日志如下:

10.0.*.1 - - [14/Jul/2025:12:00:03 +0000] "POST /login HTTP/1.1" 302 0 "http://example.com/login.html" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.88 Safari/537.36"  

通过put_logs接口上传时,为了方便后续分析日志,建议您按字段设置日志内容并上传。

# Project名称。
project_name = "aliyun-test-project"
# Logstore名称
logstore_name = "aliyun-test-logstore"

# 向Logstore写入数据。
def put_logs():
    print("ready to put logs for %s" % logstore_name)
    log_group = []
    for i in range(0, 100):
        log_item = LogItem()
        # 按字段设置日志内容
        contents = [
            ('remote_addr', '192.168.0.%d' % (i % 255)),
            ('remote_user', 'user%d' % i),
            ('time_local', time.strftime('%d/%b/%Y:%H:%M:%S +0000', time.gmtime())),
            ('request_method', 'GET' if i % 2 == 0 else 'POST'),
            ('request_uri', '/index.html' if i % 3 == 0 else '/api/data'),
            ('status', str(200 + (i % 100))),  # 状态码在200-299之间
            ('body_bytes_sent', str(1024 + i)),
            ('http_referer', 'http://example.com/page%d' % (i // 10)),
            ('http_user_agent',
             'Mozilla/5.0 AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.%d Safari/537.36' % (i * 10))
        ]
        log_item.set_contents(contents)
        log_group.append(log_item)
    request = PutLogsRequest(project_name, logstore_name, "", "", log_group, compress=False)
    client.put_logs(request)
    print("put logs for %s success " % logstore_name)
    time.sleep(60)

5. 创建索引

上传的日志需要配置索引才能进行查询。设置remote_addrremote_usertime_localrequest_methodrequest_uristatusbody_bytes_senthttp_refererhttp_user_agent作为索引字段。

# Project名称。
project_name = "aliyun-test-project"
# Logstore名称
logstore_name = "aliyun-test-logstore"
# 索引。
logstore_index = {'line': {
    'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t',
              '\r'], 'caseSensitive': False, 'chn': False},
    'keys': {'remote_addr': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'remote_addr', 'doc_value': True, 'chn': False},
            'remote_user': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'remote_user', 'doc_value': True, 'chn': False},
            'time_local': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'time_local', 'doc_value': True, 'chn': False},
            'request_method': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'request_method', 'doc_value': True, 'chn': False},
            'request_uri': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'request_uri', 'doc_value': True, 'chn': False},
            'status': {'type': 'long', 'alias': 'status', 'doc_value': True},
            'body_bytes_sent': {'type': 'long', 'alias': 'body_bytes_sent', 'doc_value': True},
            'http_referer': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'http_referer', 'doc_value': True, 'chn': False},
            'http_user_agent': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'http_user_agent', 'doc_value': True, 'chn': False}}}

# 创建索引。
def create_index():
    print("ready to create index for %s" % logstore_name)
    index_config = IndexConfig()
    index_config.from_json(logstore_index)
    client.create_index(project_name, logstore_name, index_config)
    print("create index for %s success " % logstore_name)
    time.sleep(60 * 2)

索引创建成功之后,查询日志结果如下:

image

6. 查询与分析日志

手动上报日志,保证有新日志产生。配置*| select request_method,status from aliyun-test-logstore查询分析语句来检索日志。查询与分析功能,请参考索引模式查询与分析进行学习。

# Project名称。
project_name = "aliyun-test-project"
# Logstore名称
logstore_name = "aliyun-test-logstore"
# 查询语句。
query = "*| select request_method,status from " + logstore_name
# from_time和to_time表示查询日志的时间范围,Unix时间戳格式。
from_time = int(time.time()) - 3600
to_time = time.time() + 3600

# 通过SQL查询日志。
def get_logs():
    print("ready to query logs from logstore %s" % logstore_name)
    request = GetLogsRequest(project_name, logstore_name, from_time, to_time, query=query)
    response = client.get_logs(request)
    for log in response.get_logs():
        for k, v in log.contents.items():
            print("%s : %s" % (k, v))
        print("*********************")

响应结果如下:

ready to query logs from logstore aliyun-test-logstore
request_method : GET
status : 200
*********************
request_method : POST
status : 201
*********************
request_method : GET
status : 202
*********************
request_method : POST
status : 203
*********************
Process finished with exit code 0

7. 仪表盘可视化数据

  1. Project页面左侧导航栏中找到仪表盘下的仪表盘列表,单击添加仪表盘

    image

  2. 添加到仪表盘对话框,选择您希望的仪表盘布局模式,默认使用网格布局,填写仪表盘名称后单击确认

    image

  3. 在新生成的仪表盘中点击添加新图表后进入图表创建页面。

    image

  4. 在编辑图表页面中,左下角查询分析面板处下拉选择Logstore(SQL),然后下拉选择对应的Logstore,输入* | SELECT request_method,status,COUNT(*) AS request_count GROUP BY request_method,status ORDER BY request_count DESC LIMIT 10;,表示统计某段时间内各方法和状态码的请求次数(TOP 10)。由于示例是统计字段,因此在右侧图表类型中选择表格Pro,然后点击应用,即可展现下图效果。如果可视化效果符合预期,单击确定,然后在仪表盘页面中单击右上角的保存。仪表盘支持多种图表类型,支持数据筛选过滤与对接第三方工具,更多使用请参考可视化概述学习。

    image

8. 告警与监控

8.1 创建行动策略

  1. Project左侧导航栏中找到告警,单击通知策略 > 行动策略下的创建

    image

  2. 添加行动策略弹框,创建一个标识符test-action-policy名称新告警规则-我的行动策略test的告警策略,当告警触发时,使用钉钉通知,行动组具体配置如下:

    • 渠道:通过钉钉-自定义进行告警。

    • 请求地址:设置具体的钉钉机器人Webhook,比如https://oapi.dingtalk.com/robot/send?access_token=4dff******6bfe

    • 提醒方式:钉钉群收到告警时不提醒用户。

    • 内容模板:告警内容通过SLS内置内容模板进行展示。

    • 发送时段任意时间段发生告警通知。

    image

日志服务提供管理用户、管理用户组、管理Webhook集成、管理行动策略、管理告警策略和管理内容模板SDK接口,更多信息,请参见管理告警资源数据

8.2 创建告警规则

  1. Project左侧导航栏中找到告警,单击告警规则下的新建告警

    image

  2. 在新建告警页面,配置告警规则。当日志服务监测到有数据时,每隔15分钟触发钉钉告警通知,具体配置如下:

    • 规则名称新告警规则

    • 检查频率固定间隔15分钟

    • 查询统计

      • 类型日志库

      • 区域华东1(杭州)

      • 项目aliyun-test-project

      • 日志库aliyun-test-logstore

      • 查询语句* | select *

      • 查询区间今天

    • 分组评估不分组

    • 触发条件:当有数据时,严重度的告警信息。

    • 添加标注:key分别为titledesc,描述均为${alert_name}告警触发

    • 输出目标:SLS通知,行动策略配置为创建的test-action-policy

    image

  3. 单击新建的告警规则,在告警历史区域中发现告警信息如下。在仪表盘可视化中统计数据是5,但是告警配置后第一次并未触发,手动在access.log中添加两条背景信息中的日志样例后,统计数据为7并触发了告警。原因是检查频率与查询统计的查询区间均设置15分钟,造成只有增量日志时才触发告警的表象,您可以通过修改时间范围调整。如果您想深入了解告警功能,请参考告警概述以及快速设置日志告警进行学习。image

9. 删除日志资源

在您完成以上快速入门教程后,此时您可能不再需要新手教程中创建的资源,但保留这些资源将会产生计费。因为只要Logstore存在,无论是否使用都会产生活跃Shard租用费用。因此,您需要手动删除资源。执行如下代码,通过删除Project来删除Project的所有资源。

# Project名称。
project_name = "aliyun-test-project"

# 删除指定project
def main():
    try:
        response = client.delete_project(project_name)
        response.log_print()
    except Exception as error:
        print(error)

附录:完整示例代码

from aliyun.log import LogClient, PutLogsRequest, LogItem, GetLogsRequest, IndexConfig
import time
import os

# 本示例从环境变量中获取AccessKey ID和AccessKey Secret。
accessKeyId = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
accessKey = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
# 日志服务的服务接入点。此处以杭州为例,其它地域请根据实际情况填写。
endpoint = "cn-hangzhou.log.aliyuncs.com"

# 创建日志服务Client。
client = LogClient(endpoint, accessKeyId, accessKey)

# Project名称。
project_name = "aliyun-test-project"
# Logstore名称
logstore_name = "aliyun-test-logstore"
# 查询语句。
query = "*| select request_method,status from " + logstore_name
# from_time和to_time表示查询日志的时间范围,Unix时间戳格式。
from_time = int(time.time()) - 3600
to_time = time.time() + 3600
# 索引。
logstore_index = {'line': {
    'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t',
              '\r'], 'caseSensitive': False, 'chn': False},
    'keys': {'remote_addr': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'remote_addr', 'doc_value': True, 'chn': False},
            'remote_user': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'remote_user', 'doc_value': True, 'chn': False},
            'time_local': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'time_local', 'doc_value': True, 'chn': False},
            'request_method': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'request_method', 'doc_value': True, 'chn': False},
            'request_uri': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'request_uri', 'doc_value': True, 'chn': False},
            'status': {'type': 'long', 'alias': 'status', 'doc_value': True},
            'body_bytes_sent': {'type': 'long', 'alias': 'body_bytes_sent', 'doc_value': True},
            'http_referer': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'http_referer', 'doc_value': True, 'chn': False},
            'http_user_agent': {'type': 'text', 'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t', '\r'], 'caseSensitive': False, 'alias': 'http_user_agent', 'doc_value': True, 'chn': False}}}

# 创建索引。
def create_index():
    print("ready to create index for %s" % logstore_name)
    index_config = IndexConfig()
    index_config.from_json(logstore_index)
    client.create_index(project_name, logstore_name, index_config)
    print("create index for %s success " % logstore_name)
    time.sleep(60 * 2)


# 向Logstore写入数据。
def put_logs():
    print("ready to put logs for %s" % logstore_name)
    log_group = []
    for i in range(0, 100):
        log_item = LogItem()
        # 按字段设置日志内容
        contents = [
            ('remote_addr', '192.168.0.%d' % (i % 255)),
            ('remote_user', 'user%d' % i),
            ('time_local', time.strftime('%d/%b/%Y:%H:%M:%S +0000', time.gmtime())),
            ('request_method', 'GET' if i % 2 == 0 else 'POST'),
            ('request_uri', '/index.html' if i % 3 == 0 else '/api/data'),
            ('status', str(200 + (i % 100))),  # 状态码在200-299之间
            ('body_bytes_sent', str(1024 + i)),
            ('http_referer', 'http://example.com/page%d' % (i // 10)),
            ('http_user_agent',
             'Mozilla/5.0 AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.%d Safari/537.36' % (i * 10))
        ]
        log_item.set_contents(contents)
        log_group.append(log_item)
    request = PutLogsRequest(project_name, logstore_name, "", "", log_group, compress=False)
    client.put_logs(request)
    print("put logs for %s success " % logstore_name)
    time.sleep(60)


# 通过SQL查询日志。
def get_logs():
    print("ready to query logs from logstore %s" % logstore_name)
    request = GetLogsRequest(project_name, logstore_name, from_time, to_time, query=query)
    response = client.get_logs(request)
    for log in response.get_logs():
        for k, v in log.contents.items():
            print("%s : %s" % (k, v))
        print("*********************")


if __name__ == '__main__':
    # 创建索引。
    create_index()
    # 向Logstore写入数据。
    put_logs()
    # 通过SQL查询日志。
    get_logs()

扩展场景

  • 日志服务支持多种语言的SDK日志采集,具体请参考SDK参考概述

  • 阿里云OpenAPI开发者门户提供调试、SDK、示例和配套文档。通过OpenAPI,您无需手动封装请求和签名操作,就可以快速对日志服务API进行调试。更多信息,请参见OpenAPI开发者门户

  • 为满足越来越多的自动化日志服务配置需求,日志服务提供命令行工具CLI。更多信息,请参见CLI概述

  • 使用SDK、OpenAPI开发者门户和日志服务CLI产生的费用和使用控制台产生的费用一致。更多信息,请参见计费概述

常见问题

日志服务SDK都支持哪些功能?

日志服务SDK已经实现日志服务大部分功能,包括日志采集、创建索引、查询和分析、数据加工、日志消费、日志投递管理、告警、定时SQL等。若您在SDK调试中发现未实现功能,建议您升级到最新版本SDK重试或关注后续SDK版本更新。

使用日志服务SDK的基本流程是什么?

日志服务SDK提供全流程的日志管理,其使用流程和控制台使用流程基本相似。其使用基本流程大致如下:

  1. 开通日志服务。

  2. 获取访问密钥。

  3. 创建项目Project和日志库Logstore。

  4. 日志采集并存储至Logstore。

  5. 为日志创建索引。

  6. 查询和分析日志,可视化展示。

  7. 对日志数据进行加工、投递和告警等操作。

日志服务提供界面化操作控制台,操作更简单。更多信息,请参考日志服务快速入门

SDK调试常见报错如何处理?

日志服务SDK提供错误处理逻辑。SDK可能出现的异常错误可以分成如下几类:

  • 由日志服务端返回的错误。这类错误由日志服务端返回并由SDK处理。关于这类错误的详细信息可以参见具体的API接口说明、API错误码。关于错误码的更多信息,请参见错误码

  • SDK在向服务端发出请求时出现的网络错误。这类错误包括网络连接不通,服务端返回超时等。

  • SDK自身产生的、与平台及语言相关的错误,如内存溢出等。

更多信息,请参见错误处理

在使用日志服务SDK过程中,您可能遇到日志采集、索引、查询和分析、加工等各类报错,您可以参考日志服务常见问题进行处理。更多信息,请参见常见问题

使用日志服务SDK是否存在限制?

日志服务对基础资源(例如Project个数、Logstore个数、Shard个数、LogItem大小)设定了合理的限制。建议您在使用前阅读使用限制文档,了解基础资源的使用限制。更多信息,请参见基础资源使用限制