ログ
デプロイのログタブでは、OpenTelemetry（OTel）標準形式でモデルおよびエージェントワークフローからログを受信し、アプリケーションのパフォーマンスやエラーに関する詳細な分析、トラブルシューティング、理解のために、関連するログ情報を一元化します。
収集されたログは期間指定によるフィルター機能を提供し、OTel Logs APIを利用することで同様のフィルター機能を備えたログをプログラムでエクスポートすることが可能です。 ログはOTelに準拠しているため、サードパーティ製のオブザーバビリティツール（Datadogなど）へのエクスポートのために標準化されています。
アクセスと保持
ログはすべてのデプロイおよびターゲットタイプで利用可能です。 デプロイで「オーナー」ロールと「ユーザー」ロールを持つユーザーのみが、これらのログを表示できます。 ログは30日間保持され、その後自動的に削除されます。
デプロイのログにアクセスするには、デプロイタブでデプロイを検索してクリックし、アクティビティログタブ、ログの順にクリックします。
使用できるログレベルは INFO、DEBUG、WARN、ERROR です。 ログをフィルターするには、範囲（UTC）ドロップダウンを使用して、直近15分、直近1時間、直近1日、またはカスタム範囲を選択します。
OTelログのエクスポート
以下のコード例では、OTel Logs APIを使用してデプロイのOpenTelemetry互換ログを取得し、プレビューとERRORログの数を表示した後、ログを出力ファイルに書き込みます。 コードを実行する前に、entity_id変数にデプロイIDを設定し、<DEPLOYMENT_ID>をデプロイの概要タブまたはURLから取得したデプロイIDに置き換えます。 さらに、export_logs_to_json関数を、ターゲットとするオブザーバビリティサービスで想定される形式に合わせて変更できます。
| Export OTel logs to JSON |
|---|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78 | import json
import requests
from typing import Dict, List, Any
import datarobot as dr
# Initialize DataRobot client for authentication
dr_client = dr.Client()
# Get authentication details from the DataRobot client
api_token: str = dr_client.token
base_url: str = dr_client.endpoint
# Provide DataRobot entity information
entity_type: str = "deployment"
entity_id: str = "<DEPLOYMENT_ID>" # From the deployment Overview tab or URL
# Set output filename
output_filename: str = "datarobot_otel_logs.json"
def get_otel_logs(entity_type: str, entity_id: str, **params: Any) -> Dict[str, Any]:
"""Get OTel logs from the DataRobot API"""
url: str = f"{base_url}/otel/{entity_type}/{entity_id}/logs/"
headers: Dict[str, str] = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
def export_logs_to_json(logs: List[Dict[str, Any]], entity_type: str, entity_id: str, filename: str) -> None:
"""Export logs to a JSON file"""
log_entries = []
for log in logs:
log_payload = {
"timestamp": log.get('timestamp', ''),
"level": log.get('level', ''),
"message": log.get('message', ''),
"entity_type": entity_type,
"entity_id": entity_id,
"stacktrace": log.get('stacktrace', '')
}
log_entries.append(log_payload)
with open(filename, 'w') as f:
json.dump(log_entries, f, indent=2)
# Get and export OTel logs
try:
print(f"Getting logs for {entity_type} {entity_id}...")
logs_data: Dict[str, Any] = get_otel_logs(entity_type, entity_id)
logs: List[Dict[str, Any]] = logs_data.get('data', [])
print(f"Found {len(logs)} logs")
# Preview first few logs
for log in logs[:5]:
timestamp = log.get('timestamp', 'N/A')
level = log.get('level', 'N/A')
message = log.get('message', 'N/A')
print(f"{timestamp} - {level:8} - {message}")
# Example: Filter for ERROR logs only
error_logs = [log for log in logs if log.get('level', '').upper() == 'ERROR']
print(f"Found {len(error_logs)} ERROR logs")
# Export to JSON file
print(f"Writing logs to {output_filename}...")
export_logs_to_json(logs, entity_type, entity_id, output_filename)
print(f"Exported {len(logs)} logs to {output_filename}")
except requests.exceptions.RequestException as e:
print(f"API Error: {e}")
except FileNotFoundError as e:
print(f"File Error: {e}")
except json.JSONEncodeError as e:
print(f"JSON Error: {e}")
except Exception as e:
print(f"Unexpected Error: {e}")
|