tencentcloud-cls

Tencent Cloud CLS（日志服务）— 搜索与分析

在 Tencent Cloud CLS 日志主题中搜索并运行 CQL / SQL 分析。

设置： 请参阅 tencentcloud 身份验证。SDK 会从环境中读取 TENCENTCLOUD_SECRET_ID / TENCENTCLOUD_SECRET_KEY / TENCENTCLOUD_REGION。

配套技能： 使用 tencentcloud-cls-alarm 管理告警策略 / 通知组 / 屏蔽。本技能仅用于搜索日志内容。

CLI（推荐）

本技能提供 scripts/cls.py — 一个涵盖最常见操作的独立 CLI。

# $SKILL_DIR can point at another skill loaded this turn — anchor on our own
# script (re-run this at the top of every fresh-shell Bash block).
CLS="$SKILL_DIR/scripts/cls.py"; [ -f "$CLS" ] || CLS=$(find /tmp -maxdepth 8 -path '*/skills/*/scripts/cls.py' 2>/dev/null | head -1)
[ -f "$CLS" ] || { echo "tencentcloud-cls script not found (SKILL_DIR=$SKILL_DIR)" >&2; exit 1; }

python3 $CLS topics                                              # list topics
python3 $CLS search --topic <topic-id> --query 'level:ERROR' --time 1h
python3 $CLS search --topic <topic-id> --trace-id <uuid>
python3 $CLS search --topic <topic-id> --time 1d \
    --query '* | SELECT api_name, count(*) AS cnt GROUP BY api_name ORDER BY cnt DESC LIMIT 20' \
    --format json

--time 接受 30m / 1h / 6h / 1d / 7d。--query 默认为 CQL；传入 --lucene 可切换方言。在查询后追加 | SELECT ... GROUP BY ...，即可进行 SQL 分析。

如果需要执行 CLI 未提供的操作（自定义字段投影、原始分页遍历、基于 Context 的实时跟踪），请改用下方的 SDK 调用。

适用场景

• 查找某项服务近期的错误 / 5xx 响应
• 跨多个主题按 trace ID 查询单个请求
• 运行 CQL 筛选条件（status_code:>=500 AND service:"openai"）
• 运行 SQL 分析（SELECT api_name, COUNT(*) GROUP BY api_name）
• 串联跟踪日志，还原请求生命周期
• 提取特定字段，用于账单 / 审计报告

依赖项

pip install tencentcloud-sdk-python

快速开始

import os
import json
from tencentcloud.common import credential
from tencentcloud.cls.v20201016 import cls_client, models

cred = credential.EnvironmentVariableCredential().get_credential()
client = cls_client.ClsClient(cred, os.environ["TENCENTCLOUD_REGION"])

工作流

发现主题

req = models.DescribeTopicsRequest()
resp = client.DescribeTopics(req)
for t in resp.Topics:
    print(t.TopicId, t.TopicName, t.LogsetId)

提示：请预先向用户询问主题 ID。主题 ID 类似 751a7350-dc5d-41c3-a6ca-c178bae05807，无法根据服务名称推测。

搜索日志（CQL）

import time

req = models.SearchLogRequest()
req.TopicId = "<topic-id>"
req.From = int((time.time() - 3600) * 1000)   # 1 hour ago, ms epoch
req.To = int(time.time() * 1000)
req.Query = 'status_code:>=500 AND service:"openai"'
req.Limit = 100
req.Sort = "desc"
req.SyntaxRule = 1                             # 1 = CQL, 0 = Lucene

resp = client.SearchLog(req)
for line in resp.Results:
    fields = {f.Key: f.Value for f in line.LogJson and []}  # see below
    print(line.Time, line.PkgLogId, fields.get("status_code"), fields.get("api_name"))

CLS 返回的 Results 中，每条记录的 LogJson 都是 JSON 字符串。使用 json.loads(line.LogJson) 解码，可获取主题存储的所有索引字段组成的扁平字典。

跨多个主题查询 trace ID

TRACE_ID = "34341776-0835-422a-956d-ac8d5b404db1"
TOPICS = {
    "trace": "<trace-topic-id>",
    "api-usages": "<api-usages-topic-id>",
}

for label, topic_id in TOPICS.items():
    req = models.SearchLogRequest()
    req.TopicId = topic_id
    req.From = int((time.time() - 86400) * 1000)
    req.To = int(time.time() * 1000)
    req.Query = f'trace_id:"{TRACE_ID}"'
    req.Limit = 100
    req.SyntaxRule = 1
    resp = client.SearchLog(req)
    print(f"--- {label}: {len(resp.Results)} records ---")
    for line in resp.Results:
        print(line.Time, line.LogJson)

SQL 分析

CLS 支持用于聚合的日志 SQL 子集。在 CQL 筛选条件后追加 | <SQL>：

req.Query = '* | SELECT api_name, count(*) AS cnt GROUP BY api_name ORDER BY cnt DESC LIMIT 20'
req.SyntaxRule = 1
resp = client.SearchLog(req)
# When the query has a `|`, results land in resp.Analysis (rows of column→value)
for row in resp.AnalysisResults or []:
    print(row)

实时跟踪近期日志

# Poll every 5s for new lines after the last cursor
last_cursor = None
while True:
    req = models.SearchLogRequest()
    req.TopicId = topic_id
    req.From = int((time.time() - 30) * 1000)
    req.To = int(time.time() * 1000)
    req.Query = 'level:ERROR'
    req.Limit = 100
    req.Sort = "asc"
    req.SyntaxRule = 1
    if last_cursor:
        req.Context = last_cursor
    resp = client.SearchLog(req)
    for line in resp.Results:
        print(line.Time, line.LogJson)
    last_cursor = resp.Context
    time.sleep(5)

CQL 速查表

模式	含义
status_code:500	精确匹配
status_code:>=500	范围
status_code:[500 TO 599]	带边界的范围
service:"openai"	带引号的短语（值中包含空格时使用）
NOT level:DEBUG	否定
service:openai AND status_code:>=400	合取
(api:foo OR api:bar)	分组
`*	SELECT ... GROUP BY ...`

分页

CLS 每次调用最多返回 1000 条记录。对于更大的结果集，请使用 Context 迭代：

all_records = []
context = None
while True:
    req = models.SearchLogRequest()
    req.TopicId = topic_id
    req.From = ...
    req.To = ...
    req.Query = '...'
    req.Limit = 1000
    req.SyntaxRule = 1
    if context:
        req.Context = context
    resp = client.SearchLog(req)
    all_records.extend(resp.Results)
    context = resp.Context
    if not context or len(resp.Results) < 1000:
        break

错误模式

症状	可能原因
InvalidParameter.QueryError	用引号括起包含 : 或空格的短语；检查 SyntaxRule 是否与查询匹配
LimitExceeded	并发的 SearchLog 调用超过 30-QPS 配额 — 请退避后重试
OperationDenied.AccountIsolated	CLS 服务因账单问题而暂停 — 请检查控制台
Results 为空，但控制台中可见日志	时间范围有误（From / To 使用毫秒级 epoch，而非秒），或者主题的字段名称与查询预期不同

控制台链接

• CLS 控制台：https://console.cloud.tencent.com/cls/topic
• CQL 语法：https://www.tencentcloud.com/document/product/614/47044
• SQL 分析语法：https://www.tencentcloud.com/document/product/614/58978