阿里云日志服务（SLS）是一种一站式日志管理服务，提供日志采集、存储、查询、分析、告警等全流程功能，广泛应用于应用日志监控、运维排查、业务数据分析等场景。相较于传统的日志文件存储，SLS具备弹性扩容、低成本、快速查询等优势，能够轻松应对海量日志数据的处理需求。本文将使用Python SDK实现SLS的日志采集、查询、分析等核心操作，附上完整代码示例，帮助开发者快速搭建日志管理体系。
首先完成开发准备：1. 在阿里云控制台创建SLS Project和Logstore，Project是SLS的资源管理单元，Logstore用于存储具体的日志数据；2. 获取SLS的访问凭证（AccessKey ID、AccessKey Secret）以及Project的Endpoint；3. 在Python环境中安装SLS SDK，通过pip命令安装：pip install aliyun-log-python-sdk。
首先实现日志采集功能，SLS支持通过SDK直接上传日志数据，适用于应用程序主动上报日志的场景。本次实战将模拟一个Python应用，采集应用运行日志（如INFO、ERROR级别日志），并上传到SLS Logstore。以下是日志采集的核心代码：
import logging
from aliyun.log import LogClient, PutLogsRequest, LogItem
import time
import random
配置SLS凭证与资源信息
SLS_CONFIG = {
"access_key_id": "your_access_key_id", # 替换为你的AccessKey ID
"access_key_secret": "your_access_key_secret", # 替换为你的AccessKey Secret
"endpoint": "cn-hangzhou.log.aliyuncs.com", # 替换为你的Project Endpoint
"project": "your-project-name", # 替换为你的SLS Project名称
"logstore": "your-logstore-name" # 替换为你的Logstore名称
}
初始化SLS客户端
client = LogClient(SLS_CONFIG["endpoint"], SLS_CONFIG["access_key_id"], SLS_CONFIG["access_key_secret"])
def init_logger():
"""初始化本地日志记录器（可选）"""
logger = logging.getLogger("sls_demo")
logger.setLevel(logging.DEBUG)
# 创建控制台处理器
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def upload_logs_to_sls(log_level, message, logger=None):
"""
上传日志到SLS
:param log_level: 日志级别（INFO/ERROR/WARN/DEBUG）
:param message: 日志内容
:param logger: 本地日志记录器
"""
try:
# 构造日志条目
log_item = LogItem()
log_time = int(time.time()) # 日志时间戳（秒级）
log_item.set_time(log_time)
# 设置日志内容（键值对形式）
log_item.set_contents({
"log_level": log_level,
"message": message,
"app_name": "python-sls-demo",
"host_ip": "127.0.0.1", # 模拟主机IP
"request_id": f"req-{random.randint(10000, 99999)}" # 模拟请求ID
})
# 构造批量上传请求
request = PutLogsRequest(
project=SLS_CONFIG["project"],
logstore=SLS_CONFIG["logstore"],
log_items=[log_item],
topic="python-app-log", # 日志主题，用于分类管理
source="python-script" # 日志来源
)
# 执行日志上传
response = client.put_logs(request)
if response.is_success():
if logger:
logger.info(f"日志上传成功，请求ID：{response.get_request_id()}")
return True
else:
err_msg = f"日志上传失败，错误码：{response.get_error_code()}，错误信息：{response.get_error_message()}"
if logger:
logger.error(err_msg)
return False
except Exception as e:
err_msg = f"日志上传异常：{str(e)}"
if logger:
logger.error(err_msg)
return False
模拟应用运行，上传日志
if name == "main":
local_logger = init_logger()
# 模拟上传不同级别的日志
upload_logs_to_sls("INFO", "应用启动成功，开始处理业务请求", local_logger)
upload_logs_to_sls("DEBUG", "用户登录请求，用户名：zhangsan", local_logger)
# 模拟异常日志
try:
1 / 0
except ZeroDivisionError as e:
upload_logs_to_sls("ERROR", f"业务处理异常：{str(e)}，异常位置：demo.py:50", local_logger)
upload_logs_to_sls("WARN", "数据库连接池连接数即将达到上限，当前连接数：8/10", local_logger)
upload_logs_to_sls("INFO", "应用处理完所有请求，正常退出", local_logger)
日志上传完成后，需要实现日志查询与分析功能。SLS支持通过SDK执行日志查询，支持关键词查询、时间范围查询、字段过滤等，同时支持SQL分析，能够对日志数据进行统计、聚合等操作。以下是日志查询与SQL分析的代码示例：
from aliyun.log import GetLogsRequest, GetLogStoreLogsRequest, SqlQueryRequest
import time
初始化SLS客户端（同日志采集）
client = LogClient(SLS_CONFIG["endpoint"], SLS_CONFIG["access_key_id"], SLS_CONFIG["access_key_secret"])
def query_logs_by_keyword(keyword, start_time, end_time):
"""
根据关键词查询日志
:param keyword: 查询关键词
:param start_time: 开始时间（Unix时间戳，秒级）
:param end_time: 结束时间（Unix时间戳，秒级）
:return: 日志列表
"""
try:
# 构造日志查询请求
request = GetLogsRequest(
project=SLS_CONFIG["project"],
logstore=SLS_CONFIG["logstore"],
from_time=start_time,
to_time=end_time,
query=keyword, # 查询关键词
line=100, # 最多返回100条日志
offset=0 # 查询偏移量
)
# 执行查询
response = client.get_logs(request)
if response.is_success():
logs = response.get_logs()
print(f"查询到日志数量：{len(logs)}")
return logs
else:
print(f"日志查询失败，错误码：{response.get_error_code()}，错误信息：{response.get_error_message()}")
return []
except Exception as e:
print(f"日志查询异常：{str(e)}")
return []
def sql_analysis_logs(sql_statement, start_time, end_time):
"""
使用SQL分析日志数据
:param sql_statement: SQL分析语句
:param start_time: 开始时间（Unix时间戳，秒级）
:param end_time: 结束时间（Unix时间戳，秒级）
:return: 分析结果
"""
try:
# 构造SQL分析请求
request = SqlQueryRequest(
project=SLS_CONFIG["project"],
logstore=SLS_CONFIG["logstore"],
sql=sql_statement,
from_time=start_time,
to_time=end_time
)
# 执行SQL分析
response = client.sql_query(request)
if response.is_success():
analysis_result = response.get_results()
print("SQL分析结果：")
for row in analysis_result:
print(row)
return analysis_result
else:
print(f"SQL分析失败，错误码：{response.get_error_code()}，错误信息：{response.get_error_message()}")
return []
except Exception as e:
print(f"SQL分析异常：{str(e)}")
return []
测试日志查询与分析
if name == "main":
# 定义查询时间范围（近1小时）
end_time = int(time.time())
start_time = end_time - 3600
# 1. 根据关键词查询ERROR级别日志
error_logs = query_logs_by_keyword("ERROR", start_time, end_time)
for log in error_logs[:5]: # 打印前5条ERROR日志
print(f"日志时间：{log.get_time()}，日志内容：{log.get_contents()}")
# 2. 使用SQL分析不同日志级别的数量
sql = """
SELECT log_level, COUNT(*) AS log_count
FROM log
GROUP BY log_level
ORDER BY log_count DESC;
"""
sql_analysis_logs(sql, start_time, end_time)
# 3. 分析不同应用的日志数量
sql2 = """
SELECT app_name, COUNT(*) AS app_log_count
FROM log
WHERE log_level = 'INFO'
GROUP BY app_name;
"""
sql_analysis_logs(sql2, start_time, end_time)
在实际开发中，还可以对日志采集与分析功能进行优化：1. 实现日志批量上传，减少API调用次数，提升采集效率；2. 配置SLS的自动采集规则，通过Logtail采集服务器日志、应用日志，无需手动调用SDK上传；3. 结合SLS的告警功能，设置日志告警规则（如ERROR日志数量超过阈值时触发告警），及时发现问题；4. 使用SLS的可视化功能，创建日志仪表盘，实时监控日志数据；5. 配置日志生命周期管理，自动清理过期日志，降低存储成本。
需要注意的是，SLS的查询与分析性能与日志索引配置相关，建议为常用的查询字段（如log_level、app_name、request_id）创建索引，提升查询速度。同时，为了保证日志数据的安全性，建议通过RAM子账号配置SLS的访问权限，限制子账号的操作范围（如仅允许上传日志、查询日志，禁止删除Logstore）。
通过本文的实战代码，开发者可以快速实现日志数据的采集、查询与分析，依托阿里云SLS构建完善的日志管理体系。SLS能够轻松应对海量日志数据的处理需求，帮助开发者快速排查应用问题、分析业务数据，提升运维效率与业务决策能力。后续可结合阿里云其他产品（如函数计算、云监控），构建自动化的日志监控与运维体系。
