Aliyun Sls Log Query
- 96 installs
- 396 repo stars
- Updated July 18, 2026
- cinience/alicloud-skills
Query and troubleshoot logs in Alibaba Cloud Log Service (SLS) using query|analysis syntax and the Python SDK for time-bounded search and root-cause analysis.
About
Uses Alibaba Cloud SLS query|analysis syntax and the Python SDK to run time-bounded log search, filtering, and analytics. A developer uses it for error investigation and root-cause analysis over logs.
- query|analysis syntax with the aliyun-log Python SDK
- Time-bounded search across one or more logstores
Aliyun Sls Log Query by the numbers
- 96 all-time installs (skills.sh)
- Ranked #567 of 1,039 Cloud & Infrastructure skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/cinience/alicloud-skills --skill aliyun-sls-log-queryAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 96 |
|---|---|
| repo stars | ★ 396 |
| Last updated | July 18, 2026 |
| Repository | cinience/alicloud-skills ↗ |
What it does
Query and troubleshoot logs in Alibaba Cloud Log Service (SLS) using query|analysis syntax and the Python SDK for time-bounded search and root-cause analysis.
Files
Category: service
SLS Log Query and Troubleshooting
Use SLS query|analysis syntax and Python SDK for log search, filtering, and analytics.
Prerequisites
- Install SDK (virtual environment recommended to avoid PEP 668 restrictions):
python3 -m venv .venv
. .venv/bin/activate
python -m pip install -U aliyun-log-python-sdk- Configure environment variables:
ALIBABACLOUD_ACCESS_KEY_IDALIBABACLOUD_ACCESS_KEY_SECRETSLS_ENDPOINT(e.g.cn-hangzhou.log.aliyuncs.com)SLS_PROJECTSLS_LOGSTORE(supports a single value or comma-separated values)
Query Composition
- Query clause: filters logs (e.g.
status:500). - Analysis clause: statistical aggregation, format
query|analysis. - Example:
* | SELECT status, count(*) AS pv GROUP BY status
See references/query-syntax.md for full syntax.
Quickstart (Python SDK)
import os
import time
from aliyun.log import LogClient, GetLogsRequest
client = LogClient(
os.environ["SLS_ENDPOINT"],
os.environ["ALIBABACLOUD_ACCESS_KEY_ID"],
os.environ["ALIBABACLOUD_ACCESS_KEY_SECRET"],
)
project = os.environ["SLS_PROJECT"]
logstore = os.environ["SLS_LOGSTORE"]
query = "status:500"
start_time = int(time.time()) - 15 * 60
end_time = int(time.time())
request = GetLogsRequest(project, logstore, start_time, end_time, query=query)
response = client.get_logs(request)
for log in response.get_logs():
print(log.contents)Script quickstart
python skills/observability/sls/aliyun-sls-log-query/scripts/query_logs.py \
--query "status:500" \
--last-minutes 15Optional args: --project, --logstore(repeatable, or comma-separated values), --endpoint, --start, --end, --last-minutes, --limit, --parallel.
Troubleshooting script
python skills/observability/sls/aliyun-sls-log-query/scripts/troubleshoot.py \
--group-field status \
--last-minutes 30 \
--limit 20Optional args: --error-query, --group-field, --limit, --logstore(repeatable, or comma-separated values), --parallel, plus the time range args above.
Workflow
1) Ensure Logstore indexing is enabled (queries/analysis fail without index). 2) Write query clause and append analysis clause when needed. 3) Execute with SDK/script and inspect results. 4) Control returned rows with limit; narrow time range when needed.
Validation
mkdir -p output/aliyun-sls-log-query
for f in skills/observability/sls/aliyun-sls-log-query/scripts/*.py; do
python3 -m py_compile "$f"
done
echo "py_compile_ok" > output/aliyun-sls-log-query/validate.txtPass criteria: command exits 0 and output/aliyun-sls-log-query/validate.txt is generated.
Output And Evidence
- Save artifacts, command outputs, and API response summaries under
output/aliyun-sls-log-query/. - Include key parameters (region/resource id/time range) in evidence files for reproducibility.
References
- Syntax and examples:
references/query-syntax.md - Python SDK initialization and queries:
references/python-sdk.md - Troubleshooting templates:
references/templates.md
- Source list:
references/sources.md
interface:
display_name: "Alibaba Cloud Observability SLS Log Query"
short_description: "Log query and troubleshooting workflows"
default_prompt: "Use $aliyun-sls-log-query to complete this observability/sls task on Alibaba Cloud."
Python SDK 基本调用
安装
python3 -m venv .venv
. .venv/bin/activate
python -m pip install -U aliyun-log-python-sdk初始化与查询
from aliyun.log import LogClient, GetLogsRequest
client = LogClient(endpoint, access_key_id, access_key_secret)
request = GetLogsRequest(project, logstore, from_time, to_time, query=query)
response = client.get_logs(request)
logs = response.get_logs()SLS 查询与分析语法要点
基本格式
- 分析语句必须配合查询语句使用，格式为：
查询语句|分析语句
查询语句
- 用于过滤日志，例如：
status:200 *表示不做过滤
分析语句
- 用于统计聚合，例如：
* | SELECT status, count(*) AS PV GROUP BY status
结果与限制
- 使用
LIMIT控制返回行数与分析结果规模。
官方文档来源（用于后续更新） ============================
- （暂无外部文档链接）
SLS Troubleshooting Query Templates
Top errors by status
status:5* | SELECT status, count(*) AS total GROUP BY status ORDER BY total DESC LIMIT 20Top error messages
level:error | SELECT message, count(*) AS total GROUP BY message ORDER BY total DESC LIMIT 20Top trace IDs with errors
level:error AND trace_id:* | SELECT trace_id, count(*) AS total GROUP BY trace_id ORDER BY total DESC LIMIT 20Slow requests (adjust field names)
* | SELECT request_path, avg(latency_ms) AS p50, max(latency_ms) AS max_ms GROUP BY request_path ORDER BY max_ms DESC LIMIT 20Timeout keywords
message:*timeout* | SELECT message, count(*) AS total GROUP BY message ORDER BY total DESC LIMIT 20import argparse
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from aliyun.log import LogClient, GetLogsRequest
def get_env(*names: str, default: str | None = None) -> str:
for name in names:
value = os.getenv(name)
if value:
return value
if default is not None:
return default
print(f"Missing env var: {' / '.join(names)}", file=sys.stderr)
sys.exit(1)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="SLS log query")
parser.add_argument("--query", default="*")
parser.add_argument("--project", default=os.getenv("SLS_PROJECT"))
parser.add_argument(
"--logstore",
action="append",
default=None,
help="logstore name (repeatable, or comma-separated values)",
)
parser.add_argument("--endpoint", default=os.getenv("SLS_ENDPOINT"))
parser.add_argument("--start", type=int, default=None, help="epoch seconds")
parser.add_argument("--end", type=int, default=None, help="epoch seconds")
parser.add_argument("--last-minutes", type=int, default=15)
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--parallel", type=int, default=4, help="max concurrent logstore queries")
return parser.parse_args()
def parse_logstores(raw_values: list[str] | None, fallback: str | None) -> list[str]:
values = raw_values or []
if not values and fallback:
values = [fallback]
out: list[str] = []
seen: set[str] = set()
for raw in values:
for item in (part.strip() for part in raw.split(",")):
if not item or item in seen:
continue
seen.add(item)
out.append(item)
if not out:
print("Missing env var or arg: SLS_LOGSTORE / --logstore", file=sys.stderr)
sys.exit(1)
return out
def fetch_logs(
client: LogClient,
project: str,
logstore: str,
start_time: int,
end_time: int,
query: str,
limit: int,
) -> list[dict]:
request = GetLogsRequest(
project,
logstore,
start_time,
end_time,
query=query,
line=limit,
)
response = client.get_logs(request)
rows: list[dict] = []
for log in response.get_logs():
item = dict(log.contents)
item["__project"] = project
item["__logstore"] = logstore
rows.append(item)
return rows
def main() -> None:
args = parse_args()
project = args.project or get_env("SLS_PROJECT")
logstores = parse_logstores(args.logstore, os.getenv("SLS_LOGSTORE"))
endpoint = args.endpoint or get_env("SLS_ENDPOINT")
start_time = args.start
end_time = args.end
if end_time is None:
end_time = int(time.time())
if start_time is None:
start_time = end_time - args.last_minutes * 60
client = LogClient(
endpoint,
get_env("ALIBABACLOUD_ACCESS_KEY_ID", "ALIBABA_CLOUD_ACCESS_KEY_ID", "ALICLOUD_ACCESS_KEY_ID"),
get_env("ALIBABACLOUD_ACCESS_KEY_SECRET", "ALIBABA_CLOUD_ACCESS_KEY_SECRET", "ALICLOUD_ACCESS_KEY_SECRET"),
)
max_workers = max(1, min(args.parallel, len(logstores)))
failures: list[str] = []
all_rows: list[dict] = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {
pool.submit(
fetch_logs,
client,
project,
logstore,
start_time,
end_time,
args.query,
args.limit,
): logstore
for logstore in logstores
}
for fut in as_completed(futures):
logstore = futures[fut]
try:
all_rows.extend(fut.result())
except Exception as exc: # noqa: BLE001
failures.append(f"{logstore}: {exc}")
for row in all_rows:
print(json.dumps(row, ensure_ascii=False))
if failures:
print("partial failures:", file=sys.stderr)
for msg in failures:
print(f"- {msg}", file=sys.stderr)
sys.exit(2)
if __name__ == "__main__":
main()
import argparse
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from aliyun.log import LogClient, GetLogsRequest
def get_env(*names: str, default: str | None = None) -> str:
for name in names:
value = os.getenv(name)
if value:
return value
if default is not None:
return default
print(f"Missing env var: {' / '.join(names)}", file=sys.stderr)
sys.exit(1)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="SLS troubleshooting query helper")
parser.add_argument("--project", default=os.getenv("SLS_PROJECT"))
parser.add_argument(
"--logstore",
action="append",
default=None,
help="logstore name (repeatable, or comma-separated values)",
)
parser.add_argument("--endpoint", default=os.getenv("SLS_ENDPOINT"))
parser.add_argument("--start", type=int, default=None, help="epoch seconds")
parser.add_argument("--end", type=int, default=None, help="epoch seconds")
parser.add_argument("--last-minutes", type=int, default=30)
parser.add_argument("--limit", type=int, default=20)
parser.add_argument("--group-field", default="status")
parser.add_argument("--error-query", default="status:5* or level:error or level:ERROR")
parser.add_argument("--parallel", type=int, default=4, help="max concurrent logstore queries")
return parser.parse_args()
def parse_logstores(raw_values: list[str] | None, fallback: str | None) -> list[str]:
values = raw_values or []
if not values and fallback:
values = [fallback]
out: list[str] = []
seen: set[str] = set()
for raw in values:
for item in (part.strip() for part in raw.split(",")):
if not item or item in seen:
continue
seen.add(item)
out.append(item)
if not out:
print("Missing env var or arg: SLS_LOGSTORE / --logstore", file=sys.stderr)
sys.exit(1)
return out
def fetch_logs(
client: LogClient,
project: str,
logstore: str,
start_time: int,
end_time: int,
query: str,
limit: int,
) -> list[dict]:
request = GetLogsRequest(
project,
logstore,
start_time,
end_time,
query=query,
line=limit,
)
response = client.get_logs(request)
rows: list[dict] = []
for log in response.get_logs():
item = dict(log.contents)
item["__project"] = project
item["__logstore"] = logstore
rows.append(item)
return rows
def main() -> None:
args = parse_args()
project = args.project or get_env("SLS_PROJECT")
logstores = parse_logstores(args.logstore, os.getenv("SLS_LOGSTORE"))
endpoint = args.endpoint or get_env("SLS_ENDPOINT")
start_time = args.start
end_time = args.end
if end_time is None:
end_time = int(time.time())
if start_time is None:
start_time = end_time - args.last_minutes * 60
query = (
f"{args.error_query} | "
f"SELECT {args.group_field} AS key, count(*) AS total "
f"GROUP BY {args.group_field} "
f"ORDER BY total DESC LIMIT {args.limit}"
)
client = LogClient(
endpoint,
get_env("ALIBABACLOUD_ACCESS_KEY_ID", "ALIBABA_CLOUD_ACCESS_KEY_ID", "ALICLOUD_ACCESS_KEY_ID"),
get_env("ALIBABACLOUD_ACCESS_KEY_SECRET", "ALIBABA_CLOUD_ACCESS_KEY_SECRET", "ALICLOUD_ACCESS_KEY_SECRET"),
)
max_workers = max(1, min(args.parallel, len(logstores)))
failures: list[str] = []
all_rows: list[dict] = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {
pool.submit(
fetch_logs,
client,
project,
logstore,
start_time,
end_time,
query,
args.limit,
): logstore
for logstore in logstores
}
for fut in as_completed(futures):
logstore = futures[fut]
try:
all_rows.extend(fut.result())
except Exception as exc: # noqa: BLE001
failures.append(f"{logstore}: {exc}")
for row in all_rows:
print(json.dumps(row, ensure_ascii=False))
if failures:
print("partial failures:", file=sys.stderr)
for msg in failures:
print(f"- {msg}", file=sys.stderr)
sys.exit(2)
if __name__ == "__main__":
main()
Related skills
Azure AiIntegrates Azure AI Content Safety, Document Intelligence, Speech, and Search services into Java-based agents and applications.495k1.4kAzure PrepareGenerate the exact Azure infrastructure files, Dockerfiles, and azure.yaml configuration needed before deploying any new or modernized application.494k1.4kAzure StorageConnect agents and applications to Azure Blob Storage, File Shares, Queues, Tables, and Data Lake without leaving the coding environment.494k1.4kAppinsights InstrumentationAutomatically instrument web applications running on Azure App Service with Application Insights for observability without manual configuration.493k1.4kAzure Resource LookupInstantly list, query, and discover any Azure resources across subscriptions without leaving the agent chat.493k1.4kAzure AigatewayConfigure Azure API Management as a secure, governed gateway for routing traffic to LLMs, MCP servers, and agent tools.493k1.4k