Aliyun Sls Skills

能力概述

目录结构

aliyun-sls-skills/
├── SKILL.md
├── .env.example
├── reference/                    # API 定义（YML），含入参与返回值
│   ├── list_project.yml
│   ├── list_logstores.yml
│   └── get_log_v2.yml
└── scripts/
    ├── sls_util.py               # 凭证与 Client 构造（供脚本复用）
    ├── list_project.py
    ├── list_logstores.py
    ├── get_logs_v2.py
    └── requirements.txt

• reference/：存放 API 相关信息（YML 片段），供查阅参数与响应含义。
• scripts/：可直接运行的示例（例如在 aliyun-sls-skills 目录下执行 python scripts/list_logstores.py ...）；也可阅读实现后改写。
• 依赖：pip install -r aliyun-sls-skills/scripts/requirements.txt。

本 skill 提供以下独立能力，可按需组合：

说明：

• 若用户不确定 project 名称，可先 ListProject 查看账号下所有项目。
• 若用户不确定 logstore 名称，可先 ListLogStores 再 GetLogsV2。
• GetLogsV2 时间区间为左闭右开 [from, to)，且须满足 from < to。
• Python SDK 中请求参数字段名以 alibabacloud_sls20201230.models 为准（例如 GetLogsV2Request 使用 from_、to，而非 from_time / to_time）。

项目配置文件（.aliyun-config.json）

将 SLS 相关的上下文信息保存到当前工作目录下的 .aliyun-config.json，便于后续操作复用已确认的 region、project、logstore 等配置，避免重复询问。支持两种格式：

单项目配置（对象格式）

{
  "version": 1,
  "sls": [
    {
      "region": "cn-hangzhou",
      "project": "prod-sls-project",
      "logstores": ["app-logs", "nginx-logs", "access-logs"],
      "default_logstore": "app-logs",
      "last_query_at": "2026-04-01T12:00:00Z"
    },
    {
      "region": "cn-beijing",
      "project": "backup-sls-project",
      "logstores": ["archive-logs"],
      "default_logstore": "archive-logs",
      "last_query_at": "2026-03-28T10:00:00Z"
    }
  ]
}

多项目配置（数组格式）

支持在同一仓库中管理多个应用/环境的 SLS 配置：

[
  {
    "project": "web-app",
    "version": 1,
    "sls": [
      {
        "region": "cn-hangzhou",
        "project": "web-prod",
        "logstores": ["frontend", "backend", "nginx"],
        "default_logstore": "backend",
        "last_query_at": "2026-04-01T12:00:00Z"
      },
      {
        "region": "cn-shanghai",
        "project": "web-dr",
        "logstores": ["frontend", "backend"],
        "default_logstore": "backend",
        "last_query_at": "2026-04-01T11:00:00Z"
      }
    ]
  },
  {
    "project": "data-platform",
    "version": 1,
    "sls": [
      {
        "region": "cn-beijing",
        "project": "etl-logs",
        "logstores": ["spark", "flink", "kafka"],
        "default_logstore": "spark",
        "last_query_at": "2026-04-01T10:00:00Z"
      }
    ]
  }
]

字段说明：

读取规则

执行 SLS 操作前，先检查当前工作目录是否存在 .aliyun-config.json：

import json
import os

config_path = os.path.join(os.getcwd(), ".aliyun-config.json")
sls_entries = []

if os.path.exists(config_path):
    with open(config_path, "r", encoding="utf-8") as f:
        config = json.load(f)
        # 数组格式（多项目）：提示用户选择项目
        if isinstance(config, list):
            # 展示项目列表供用户选择
            projects = [item.get("project", f"项目{i}") for i, item in enumerate(config)]
            # 用户选择后获取该项目的 sls 列表
            selected = config[selected_index]
            sls_entries = selected.get("sls", [])
        # 对象格式（单项目）：直接使用 sls 列表
        else:
            sls_entries = config.get("sls", [])

# sls_entries 是列表，包含多个 {region, project, logstores, ...}
# 若存在多个，提示用户选择具体的 project+logstore

配置复用逻辑：

1. 单条配置：若 sls 列表仅含一条配置，且字段完整，可直接复用。
2. 多条配置：若 sls 列表包含多条（多 Project 或多地域），提示用户选择具体的 region + project + logstore。
3. 多项目格式：先选择外层 project 字段，再在其 sls 列表中选择具体配置。
4. 若用户明确要求变更 region、project 或 logstore，须 重新确认 并更新配置。
5. 若配置文件不存在或字段缺失，需先通过 ListProject / ListLogStores 等 API 查询后，由用户选择或确认 再写入。

写入规则

在以下时机更新 .aliyun-config.json：

1. 用户首次确认 region、project、logstore 后：写入 sls 列表。
2. 每次成功执行查询后：更新对应条目的 last_query_at。
3. 用户变更目标资源时：更新或追加对应条目。
4. 新增项目配置时：数组格式下追加新项目。

单项目配置写入示例

import json
import os
from datetime import datetime, timezone

config_path = os.path.join(os.getcwd(), ".aliyun-config.json")

# 读取现有配置（若存在）
config = {}
if os.path.exists(config_path):
    with open(config_path, "r", encoding="utf-8") as f:
        config = json.load(f)

# 更新 sls 列表
config["version"] = config.get("version", 1)
config["sls"] = [
    {
        "region": "cn-hangzhou",
        "project": "prod-sls",
        "logstores": ["app", "nginx"],
        "default_logstore": "app",
        "last_query_at": datetime.now(timezone.utc).isoformat()
    },
    {
        "region": "cn-beijing",
        "project": "backup-sls",
        "logstores": ["archive"],
        "default_logstore": "archive",
        "last_query_at": "2026-03-28T10:00:00Z"
    }
]

# 写入
with open(config_path, "w", encoding="utf-8") as f:
    json.dump(config, f, indent=2, ensure_ascii=False)

多项目配置写入示例

import json
import os
from datetime import datetime, timezone

config_path = os.path.join(os.getcwd(), ".aliyun-config.json")

# 读取现有配置（若存在）
config = []
if os.path.exists(config_path):
    with open(config_path, "r", encoding="utf-8") as f:
        content = f.read().strip()
        if content:
            loaded = json.loads(content)
            config = loaded if isinstance(loaded, list) else [loaded]

# 查找或创建项目配置
project_name = "web-app"  # 用户指定的项目名
existing = next((item for item in config if item.get("project") == project_name), None)

new_entry = {
    "project": project_name,
    "version": 1,
    "sls": [
        {
            "region": "cn-hangzhou",
            "project": "web-prod",
            "logstores": ["frontend", "backend"],
            "default_logstore": "backend",
            "last_query_at": datetime.now(timezone.utc).isoformat()
        }
    ]
}

if existing:
    # 更新现有项目的 sls 列表
    existing["sls"] = new_entry["sls"]
    existing["version"] = new_entry["version"]
else:
    # 追加新项目配置
    config.append(new_entry)

# 写入
with open(config_path, "w", encoding="utf-8") as f:
    json.dump(config, f, indent=2, ensure_ascii=False)

与其他 skill 共享

.aliyun-config.json 可包含多个阿里云产品的配置（如 oss、cdn、domain 等）。本 skill 仅读写 sls 字段，不得覆盖或删除其他产品的配置段。

{
  "version": 1,
  "sls": { "region": "cn-hangzhou", "project": "..." },
  "oss": { "bucket": "...", "region": "..." },
  "cdn": { "domain": "..." }
}

安全提示

• 勿在配置文件中存储 AK/SK 等敏感凭证；凭证应通过环境变量或 .env 管理。
• 是否将 .aliyun-config.json 提交到版本库，由团队规范决定（建议加入 .gitignore）。

前置检查（必须首先执行）

调用 SLS 前须确认访问凭证已配置。

检查顺序

1. 环境变量

echo "AK_ID: ${ALIBABA_CLOUD_ACCESS_KEY_ID:+已设置}"
echo "AK_SECRET: ${ALIBABA_CLOUD_ACCESS_KEY_SECRET:+已设置}"

2. 当前目录 .env

find . -maxdepth 1 -name ".env" -exec cat {} \;

示例见 .env.example。

3. Python 中加载（与 scripts 行为一致）

from dotenv import load_dotenv
import os

access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "").strip()
access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "").strip()

if not access_key_id or not access_key_secret:
    load_dotenv()
    access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "").strip()
    access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "").strip()

if not access_key_id or not access_key_secret:
    raise RuntimeError("缺少 ALIBABA_CLOUD_ACCESS_KEY_ID / ALIBABA_CLOUD_ACCESS_KEY_SECRET")

未通过凭证检查时不应继续调用 API。

安装依赖

pip install -r aliyun-sls-skills/scripts/requirements.txt

或手动安装（与仓库其他 Tea SDK skill 一致）：

pip install alibabacloud_sls20201230 alibabacloud_credentials alibabacloud_tea_openapi alibabacloud_tea_util python-dotenv

创建 SLS Client

endpoint 格式为 {region}.log.aliyuncs.com。示例脚本通过 scripts/sls_util.py 的 create_sls_client(region) 使用 显式 AK/SK（与 aliyun-domain-skills/scripts/query_domainlist.py 相同）。

若需无 AK 链（ECS 角色、OIDC 等），可在自定义代码中改用 CredentialClient() 传入 open_api_models.Config(credential=credential)，并仍设置 endpoint。

from alibabacloud_sls20201230.client import Client as Sls20201230Client
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models

credential = CredentialClient()
config = open_api_models.Config(credential=credential)
config.endpoint = f"{region}.log.aliyuncs.com"
client = Sls20201230Client(config)

ListProject

列出当前账号下符合条件的 Project 信息。完整字段见 reference/list_project.yml。

常用参数（摘要）

核心流程

1. create_client() 创建 SLS Client（使用凭据链或显式 AK/SK）
2. 构建 ListProjectRequest
3. client.list_project_with_options(request, headers, runtime)
4. 使用 resp.body.to_map() 得到字典

示例脚本：scripts/list_project.py

命令行示例

cd aliyun-sls-skills
python scripts/list_project.py

代码示例

from alibabacloud_sls20201230.client import Client as Sls20201230Client
from alibabacloud_sls20201230 import models as sls_models
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
import json

credential = CredentialClient()
config = open_api_models.Config(credential=credential)
config.endpoint = "cn-hangzhou.log.aliyuncs.com"
client = Sls20201230Client(config)

request = sls_models.ListProjectRequest(project_name="test", size=10)
runtime = util_models.RuntimeOptions()
resp = client.list_project_with_options(request, {}, runtime)
print(json.dumps(resp.body.to_map(), default=str, indent=2, ensure_ascii=False))

响应（body，常见字段）：total、count、projects 等，以实际返回为准。

ListLogStores

列出指定 project 下的 LogStore。完整字段见 reference/list_logstores.yml。

常用参数（摘要）

核心流程

1. create_sls_client(region)
2. 构建 ListLogStoresRequest
3. client.list_log_stores_with_options(project, request, {}, runtime)
4. 使用 resp.body.to_map() 得到字典

示例脚本：scripts/list_logstores.py

命令行示例

cd aliyun-sls-skills
python scripts/list_logstores.py --region cn-hangzhou --project your-project
python scripts/list_logstores.py --region cn-hangzhou --project your-project --logstore-name app

代码示例

from alibabacloud_sls20201230 import models as sls_models
from alibabacloud_tea_util import models as util_models
import json

from sls_util import create_sls_client

client = create_sls_client("cn-hangzhou")
request = sls_models.ListLogStoresRequest(logstore_name="error")
runtime = util_models.RuntimeOptions()
resp = client.list_log_stores_with_options("your-project", request, {}, runtime)
print(json.dumps(resp.body.to_map(), default=str, indent=2, ensure_ascii=False))

响应（body，常见字段）：total、count、logstores 等，以实际返回为准。

GetLogsV2

在指定 project、logstore 上按时间范围查询。完整说明见 reference/get_log_v2.yml。

常用参数（摘要）

若 query 为分析语句，line/offset 往往不生效，请用 SQL LIMIT 等分页（见官方文档）。

核心流程

1. 构建 GetLogsV2Request(from_=..., to=..., query=..., ...)
2. headers = GetLogsV2Headers(accept_encoding="lz4")（或 gzip）
3. client.get_logs_v2with_options(project, logstore, request, headers, runtime)
4. resp.body.to_map() — 通常含 meta、data

示例脚本：scripts/get_logs_v2.py

命令行示例

python scripts/get_logs_v2.py --region cn-hangzhou --project your-project --logstore your-logstore --recent-minutes 15 --query "error"
python scripts/get_logs_v2.py --region cn-hangzhou --project your-project --logstore your-logstore \
  --from-time 1700000000 --to-time 1700003600 --query '* | SELECT count(*) AS cnt'
python scripts/get_logs_v2.py --region cn-hangzhou --project your-project --logstore your-logstore \
  --from-time 1700000000 --to-time 1700003600 --query '*' --session "mode=scan"

代码示例

import time
from alibabacloud_sls20201230 import models as sls_models
from alibabacloud_tea_util import models as util_models
import json

from sls_util import create_sls_client

client = create_sls_client("cn-hangzhou")
now = int(time.time())
request = sls_models.GetLogsV2Request(
    query="error",
    from_=now - 900,
    to=now,
    line=100,
    offset=0,
    reverse=False,
)
headers = sls_models.GetLogsV2Headers(accept_encoding="lz4")
runtime = util_models.RuntimeOptions()
resp = client.get_logs_v2with_options("your-project", "your-logstore", request, headers, runtime)
print(json.dumps(resp.body.to_map(), default=str, indent=2, ensure_ascii=False))

错误处理

try:
    resp = client.get_logs_v2with_options(project, logstore, request, headers, runtime)
except Exception as error:
    print(getattr(error, "message", str(error)))
    if hasattr(error, "data") and error.data:
        print(error.data.get("Recommend", ""))

脚本使用 scripts/sls_util.py 中的 print_operation_error 做同类输出。

使用流程小结

1. 检查环境变量或 .env 中的 AK/SK。
2. 安装 scripts/requirements.txt。
3. 检查 .aliyun-config.json 是否已有 SLS 配置（sls 列表），若存在且与意图一致可直接复用。
4. 若配置包含多条 sls 条目或多项目配置，提示用户选择具体的 region + project + logstore。
5. 若不确定 project 名称，先 ListProject 查看账号下项目。
6. 确认 region、project；不确定 logstore 时先 ListLogStores。
7. GetLogsV2 传入合法时间区间与 query，解析 body 中 meta / data。
8. 操作成功后更新 .aliyun-config.json 中对应条目的 last_query_at。

注意事项

• reference/ 用于查契约；scripts/ 用于验证环境与调用方式。
• 业务与示例不符时，以 SLS 官方文档 与当前 SDK 模型为准。