dt-obs-logs

日志分析技能

使用 DQL 查询、筛选和分析 Dynatrace 日志数据，以便进行故障排查和监控。

此技能涵盖的内容

• 按严重程度、内容和实体获取及筛选日志
• 使用模式匹配搜索日志消息
• 计算错误率和统计数据
• 分析日志模式和趋势
• 按维度对日志数据分组和聚合

需要跨数据源联接： 如果查询必须将日志与主机属性（OS 类型、主机名、IP 地址、云提供商）结合 → 在编写查询前，还需阅读 dt-dql-essentials/references/smartscape-topology-navigation.md。

---

使用场景

在用户希望执行以下操作时使用此技能：

• 查找特定日志条目（例如，“显示过去一小时的错误日志”）
• 按严重程度、进程组或内容筛选日志
• 在日志中搜索特定关键词或短语
• 计算错误率或日志统计数据
• 找出常见错误消息或模式
• 分析日志随时间变化的趋势
• 使用日志数据排查问题

关键概念

日志数据模型

• timestamp：日志条目的创建时间
• content：日志消息文本
• status：日志级别（ERROR、FATAL、WARN、INFO 等）
• dt.process_group.id：关联的进程组实体
• dt.process_group.detected_name：将进程组 ID 解析为易读名称

查询模式

• fetch logs：访问日志数据的主要命令
• 时间范围：使用 from:now() - <duration> 设置时间窗口
• 筛选：应用严重程度、内容和实体筛选条件
• 聚合：对日志数据分组并汇总
• 模式检测：使用 matchesPhrase() 和 contains() 搜索内容

常见操作

• 按严重程度筛选（单个或多个级别）
• 搜索内容（简单搜索和全文搜索）
• 按实体筛选（进程组）
• 时间序列分析（分桶、排序）
• 计算错误率
• 模式分析（异常、超时等）

核心工作流

1. 搜索日志

按时间、严重程度和内容查找特定日志条目。

典型步骤：

1. 定义时间范围
2. 按严重程度筛选（可选）
3. 搜索内容中的关键词
4. 选择相关字段
5. 排序并限制结果数量

示例：

fetch logs, from:now() - 1h
| filter status == "ERROR"
| fields timestamp, content, process_group = dt.process_group.detected_name
| sort timestamp desc
| limit 100

2. 筛选日志

使用多个条件（严重程度、实体、内容）缩小日志范围。

典型步骤：

1. 获取指定时间范围内的日志
2. 应用严重程度筛选条件
3. 按实体（process_group）筛选
4. 应用内容筛选条件
5. 格式化并排序输出

示例：

fetch logs, from:now() - 2h
| filter in(status, {"ERROR", "FATAL", "WARN"})
| summarize count(), by: {dt.process_group.id, dt.process_group.detected_name}
| fieldsAdd process_group = dt.process_group.detected_name
| sort `count()` desc

3. 模式分析

识别日志数据中的模式、趋势和异常。

典型步骤：

1. 获取指定时间范围内的日志
2. 添加模式检测字段
3. 按实体或时间聚合
4. 计算统计数据和比率
5. 按频次或速率排序

示例：

fetch logs, from:now() - 2h
| filter status == "ERROR"
| fieldsAdd
    has_exception = if(matchesPhrase(content, "exception"), true, else: false),
    has_timeout = if(matchesPhrase(content, "timeout"), true, else: false)
| summarize
    count(),
    exception_count = countIf(has_exception == true),
    timeout_count = countIf(has_timeout == true),
    by: {process_group = dt.process_group.detected_name}

关键函数

筛选

• filter status == "ERROR" - 按状态级别筛选
• in(status, {"ERROR", "FATAL", "WARN"}) - 多状态筛选（字面值集合请使用花括号）
• contains(content, "keyword") - 简单子字符串搜索
• matchesPhrase(content, "exact phrase") - 全文短语搜索

实体操作

• dt.process_group.detected_name - 获取易读的进程组名称
• filter process_group == "service-name" - 按指定实体筛选

聚合

• count() - 统计所有日志条目
• countIf(condition) - 按条件计数
• by: {dimension} - 按实体或时间段分组
• bin(timestamp, 5m) - 按时间分桶以分析趋势

字段操作

• fields timestamp, content, status - 选择指定字段
• fieldsAdd name = expression - 添加计算字段
• if(condition, true_value, else: false_value) - 条件逻辑

常见模式

搜索内容

简单子字符串搜索：

fetch logs, from:now() - 1h
| filter contains(content, "database")
| fields timestamp, content, status

全文短语搜索：

fetch logs, from:now() - 1h
| filter matchesPhrase(content, "connection timeout")
| fields timestamp, content, process_group = dt.process_group.detected_name

计算错误率

计算随时间变化的错误率：

fetch logs, from:now() - 2h
| summarize
    total_logs = count(),
    error_logs = countIf(status == "ERROR"),
    by: {time_bucket = bin(timestamp, 5m)}
| fieldsAdd error_rate = (error_logs * 100.0) / total_logs
| sort time_bucket asc

错误消息排行

查找最常见的错误：

fetch logs, from:now() - 24h
| filter status == "ERROR"
| summarize error_count = count(), by: {content}
| sort error_count desc
| limit 20

特定进程组的日志

按进程组筛选日志：

fetch logs, from:now() - 1h
| fieldsAdd process_group = dt.process_group.detected_name
| filter process_group == "payment-service"
| filter status == "ERROR"
| fields timestamp, content, status
| sort timestamp desc

结构化 / JSON 日志解析

许多应用会输出 JSON 格式的日志行。使用 parse 提取字段，而不是直接输出原始内容：

fetch logs, from:now() - 1h
| filter status == "ERROR"
| parse content, "JSON:log"
| fieldsAdd level = log[level], message = log[msg], error = log[error]
| fields timestamp, level, message, error
| sort timestamp desc
| limit 50

按解析后的字段聚合：

fetch logs, from:now() - 4h
| filter status == "ERROR"
| parse content, "JSON:log"
| fieldsAdd message = log[msg]
| summarize error_count = count(), by: {message}
| sort error_count desc
| limit 20

注意：

• parse content, "JSON:log" 会创建记录字段 log — 使用 log[key] 访问嵌套值
• 在 parse 之前使用 contains() 过滤日志，以减少解析开销
• 适用于任何 JSON 结构化字段，而不仅限于 content

最佳实践

1. 始终指定时间范围 - 使用 from:now() - <duration> 限制数据
2. 尽早应用过滤条件 - 在聚合之前按严重程度和实体过滤
3. 使用适当的搜索方法 - 简单搜索使用 contains()，精确搜索使用 matchesPhrase()
4. 限制结果数量 - 添加 | limit 100，避免输出过多
5. 按有意义的方式排序 - 最近日志按时间戳排序，错误排行按计数排序
6. 为实体命名 - 使用 dt.process_group.detected_name 或 getNodeName() 输出易读的名称
7. 使用时间分桶分析趋势 - 使用 bin(timestamp, 5m) 进行时间序列分析

集成点

• 实体模型：使用 dt.process_group.id 关联服务
• 时间序列：通过 bin() 和时间范围支持时序分析
• 内容搜索：通过 matchesPhrase() 提供全文搜索能力
• 聚合：使用 summarize 和条件函数进行统计分析

限制与注意事项

• 日志是否可用取决于 OneAgent 配置和日志摄取情况
• 对大型数据集进行全文搜索（matchesPhrase）可能会影响性能
• 要解析实体名称，需要 OneAgent 正确监控实体
• 时间范围应保持合理（避免不设上限的查询）

故障排除

问题	原因	解决方案
未返回日志	缺少时间范围或范围过窄	扩大 from: 时间窗口；确认日志摄取已启用
getNodeName() 返回 null	OneAgent 未监控该实体，或实体尚未解析	确认已部署 OneAgent 且已发现该实体；也可使用 dt.process_group.detected_name 作为可靠的替代方案
matchesPhrase() 在大型数据集上运行缓慢	未预先过滤就进行全文搜索	在 matchesPhrase() 之前添加 filter status == "ERROR"
字段名错误 log.level	常见错误	使用 loglevel（不带点）表示严重程度；参见 dt-dql-essentials
content 字段为空	日志行为空或未被摄取	检查 OneAgent 中的日志源配置

相关技能

• dt-dql-essentials - 日志查询的核心 DQL 语法和查询结构
• dt-obs-tracing - 使用跟踪 ID 将日志与分布式跟踪关联
• dt-obs-problems - 将日志与 DAVIS 检测到的问题关联