Alicloud Observability Sls Log Query
- 340 installs
- 396 repo stars
- Updated July 18, 2026
- cinience/alicloud-skills
alicloud-observability-sls-log-query is an agent skill that queries Alibaba Cloud Log Service using query|analysis syntax and the Python SDK for developers who need production log search and incident investigation withou
About
alicloud-observability-sls-log-query is a cinience/alicloud-skills observability skill for Alibaba Cloud Log Service (SLS). It composes SLS query clauses like `status:500` with optional analysis clauses such as `* | SELECT status, count(*) AS pv GROUP BY status`, then executes searches via the Python SDK or bundled scripts. Configuration requires five environment variables: ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET, SLS_ENDPOINT, SLS_PROJECT, and SLS_LOGSTORE. Two helper scripts—query_logs.py and troubleshoot.py—support `--last-minutes`, `--limit`, and parallel logstore queries. Developers reach for this skill during AliCloud production incidents when they need agent-driven error aggregation, audit log review, or latency investigation instead of manual SLS console queries.
- SLS log store navigation
- Query syntax and time-window filters
- Error and latency investigation patterns
- Audit and access log retrieval
- Incident triage query templates
Alicloud Observability Sls Log Query by the numbers
- 340 all-time installs (skills.sh)
- Ranked #403 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 alicloud-observability-sls-log-queryAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 340 |
|---|---|
| repo stars | ★ 396 |
| Last updated | July 18, 2026 |
| Repository | cinience/alicloud-skills ↗ |
How do you query Alibaba Cloud SLS logs from code?
Query Alibaba Cloud Log Service (SLS) to trace errors, audit access, and investigate latency or outage signals without leaving the agent session.
Who is it for?
Developers operating AliCloud workloads who need agent-assisted SLS log search, error aggregation, and root-cause analysis during incidents.
Skip if: Teams on AWS CloudWatch, GCP Logging, or Datadog who do not run workloads on Alibaba Cloud Log Service.
When should I use this skill?
User mentions Alibaba Cloud SLS, Log Service queries, status:500 errors, log troubleshooting, or query|analysis syntax on AliCloud.
What you get
Filtered log rows, query|analysis aggregation tables, and troubleshooting summaries from SLS projects and logstores.
- filtered log results
- aggregation tables
- troubleshooting summaries
By the numbers
- Requires 5 environment variables for SLS authentication and targeting
- Bundles 2 Python scripts: query_logs.py and troubleshoot.py
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:
ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_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["ALIBABA_CLOUD_ACCESS_KEY_ID"],
os.environ["ALIBABA_CLOUD_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/alicloud-observability-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/alicloud-observability-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/alicloud-observability-sls-log-query
for f in skills/observability/sls/alicloud-observability-sls-log-query/scripts/*.py; do
python3 -m py_compile "$f"
done
echo "py_compile_ok" > output/alicloud-observability-sls-log-query/validate.txtPass criteria: command exits 0 and output/alicloud-observability-sls-log-query/validate.txt is generated.
Output And Evidence
- Save artifacts, command outputs, and API response summaries under
output/alicloud-observability-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 $alicloud-observability-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(name: str, default: str | None = None) -> str:
value = os.getenv(name, default)
if not value:
print(f"Missing env var: {name}", file=sys.stderr)
sys.exit(1)
return value
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("ALIBABA_CLOUD_ACCESS_KEY_ID"),
get_env("ALIBABA_CLOUD_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(name: str, default: str | None = None) -> str:
value = os.getenv(name, default)
if not value:
print(f"Missing env var: {name}", file=sys.stderr)
sys.exit(1)
return value
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("ALIBABA_CLOUD_ACCESS_KEY_ID"),
get_env("ALIBABA_CLOUD_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.4kHow it compares
Pick this skill for AliCloud SLS log queries inside agents; use generic observability MCPs when the stack is not on Alibaba Cloud Log Service.
FAQ
What environment variables does alicloud-observability-sls-log-query need?
alicloud-observability-sls-log-query requires ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET, SLS_ENDPOINT, SLS_PROJECT, and SLS_LOGSTORE, with logstore supporting single or comma-separated values.
How does SLS query|analysis syntax work in this skill?
alicloud-observability-sls-log-query uses a query clause to filter logs (e.g. status:500) and an optional analysis clause after a pipe for SQL-style aggregation like SELECT status, count(*) GROUP BY status.
Which scripts ship with alicloud-observability-sls-log-query?
alicloud-observability-sls-log-query bundles query_logs.py for targeted searches and troubleshoot.py for grouped error investigation, both supporting --last-minutes, --limit, and parallel logstore args.