ログ分析スキル

DQLを使用してDynatraceログデータをクエリ、フィルタリング、分析し、トラブルシューティングと監視を実行します。

このスキルが対象とする内容

• 重要度、コンテンツ、エンティティによるログの取得とフィルタリング
• パターンマッチングを使用したログメッセージの検索
• エラー率と統計情報の計算
• ログパターンとトレンドの分析
• ディメンションによるログデータのグループ化と集約

クロスソース結合が必要: クエリがログをホスト属性 (OSタイプ、ホスト名、IPアドレス、クラウドプロバイダー) と組み合わせる必要がある場合 → クエリを作成する前に dt-dql-essentials/references/smartscape-topology-navigation.md も読んでください。

---

ユースケース

以下の場合にこのスキルを使用してください:

• 特定のログエントリを検索したい (例: "過去1時間のエラーログを表示")
• ログを重要度、プロセスグループ、またはコンテンツでフィルタリング
• ログから特定のキーワードまたはフレーズを検索
• エラー率またはログ統計を計算
• 一般的なエラーメッセージまたはパターンを特定
• 時系列でログトレンドを分析
• ログデータを使用して問題をトラブルシューティング

主要概念

ログデータモデル

• 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 を使用してraw コンテンツをダンプする代わりにフィールドを抽出してください:

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検出問題と相関