从日志管理痛点说起
谈及日志管理，阿里云SLS（日志服务）堪称业界一站式解决方案——覆盖采集、存储、查询、分析及告警全链路。相比传统在磁盘上堆积日志的做法，SLS的优势非常明显：弹性扩容、低成本、秒级查询，即便海量日志也能轻松应对。本文直接上手Python SDK，完整演示日志采集、查询、分析等核心操作，代码均可直接复制，帮助开发者快速搭建自己的日志管理体系。
开发准备：三件事先搞定
动手之前，需要先搭好环境。三个步骤：第一，在阿里云控制台创建一个SLS Project和Logstore——Project是资源管理单元，Logstore用于存放具体日志。第二，获取访问凭证（AccessKey ID和Secret），以及Project的Endpoint。第三，在Python环境中安装SDK，一行命令搞定：pip install aliyun-log-python-sdk。完成这些准备工作，就可以正式开始了。
日志采集：让应用日志“飞”上云
SLS支持通过SDK主动上传日志，非常适合应用程序实时上报的场景。接下来模拟一个Python应用，采集INFO、ERROR级别的运行日志，并上传到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提供灵活的支持：可以按关键词检索，按时间范围过滤，甚至直接使用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)落地优化：让SLS真正好用
仅具备基础功能还不够，实际生产环境中还有几个优化点值得关注：
- 批量上传：将多条日志合并一次性上传，能有效减少API调用次数，显著提升采集效率。
- Logtail自动采集：如果不想每个应用都手动调用SDK，可以配置阿里云Logtail自动采集服务器上的日志文件，省时省力。
- 告警规则：例如ERROR日志数量超过阈值时自动触发告警，第一时间发现异常。
- 可视化仪表盘：将日志数据转化为实时图表，监控状态一目了然。
- 生命周期管理：自动清理过期日志，降低存储成本。
另外需要特别留意，查询和分析的性能与索引配置密切相关。建议为常用字段（如log_level、app_name、request_id）创建索引，否则检索速度会变慢。安全方面，强烈建议使用RAM子账号操作，只授予上传和查询权限，避免子账号误删Logstore，防止意外操作。
整篇文章的实战代码，基本覆盖了从日志采集到查询分析的全链路。借助阿里云SLS，海量日志处理不再困难，运维排查与业务分析的效率都能显著提升。进一步还可以与函数计算、云监控等产品联动，搭建一套自动化日志监控运维体系——这才是真正的高级玩法。
#阿里云SLS #日志服务 #Python SDK #日志采集 #SQL分析
