Python SDK
の telemetry-sh パッケージが提供する Telemetry 同期アプリケーションの場合と TelemetryAsync のために asyncio サービス。両方のクライアントは、各メソッド呼び出しを Telemetry HTTP API にただちに送信します。 API キーをサーバー側の構成に保持します。
この例は架空の値を送信して SDK のデータ送信を確認します。telemetry_quickstart テーブルを使ってください。実行中のアプリケーションを計装する例ではありません。実際の利用では、選んだ処理の完了後に観測値をその処理用のテーブルに記録します。実際の処理時間を計測し、Telemetry 呼び出しの時間とは区別してください。
インストールと初期化
python -m pip install telemetry-sh
同期コード:
import os
from telemetry_sh import Telemetry
telemetry = Telemetry()
telemetry.init(os.environ.get("TELEMETRY_API_KEY"))
非同期コード:
import os
from telemetry_sh import TelemetryAsync
telemetry = TelemetryAsync()
telemetry.init(os.environ.get("TELEMETRY_API_KEY"))
init これは両方のクライアントにとって通常の方法です。しないでください await それ。イベント プロデューサーの場合は取り込み専用キー、クエリ専用ジョブの場合は読み取りスコープのキーを使用して、プロセスごとに 1 回初期化します。
イベントを同期的に送信する
import uuid
event_id = str(uuid.uuid4())
try:
response = telemetry.log("telemetry_quickstart", {
"event_id": event_id,
"job_name": "invoice_sync",
"status": "success",
"duration_ms": 912,
"attempt": 1,
"release": os.environ.get("APP_RELEASE"),
})
except Exception:
print({
"event_id": event_id,
"error_type": "telemetry_delivery_failed",
})
同期クライアントが使用するのは、 requests リクエストが完了するまでブロックされます。公開されたクライアントは、タイムアウト引数、セッション挿入、または自動再試行を公開しません。レイテンシと接続プールの要件が厳しいサービスの場合は、明示的なタイムアウトを使用してアプリケーション所有のクライアントを介して HTTP API を呼び出すか、バインドされたワーカーで SDK 呼び出しを分離します。
非同期クライアントを使用する
import asyncio
import uuid
async def record_job():
await telemetry.log("telemetry_quickstart", {
"event_id": str(uuid.uuid4()),
"job_name": "invoice_sync",
"status": "success",
"duration_ms": 912,
"attempt": 1,
})
asyncio.run(record_job())
TelemetryAsync.log を開きます aiohttp リクエストのセッションを終了し、その後セッションを閉じます。これによりイベント ループのブロックは回避されますが、現在のパッケージでは共有セッション、バックグラウンド キュー、再試行ポリシー、またはフラッシュ メソッドが公開されていません。
互換性のあるバッチを送信する
どちらのクライアントも辞書のリストを受け入れます。
import asyncio
async def send_batch():
await telemetry.log("telemetry_quickstart", [
{
"event_id": "evt_201",
"route_template": "/api/projects/:id",
"status": "success",
"status_code": 200,
"latency_ms": 84,
},
{
"event_id": "evt_202",
"route_template": "/api/projects/:id",
"status": "failed",
"status_code": 503,
"latency_ms": 904,
"error_type": "dependency_unavailable",
},
])
asyncio.run(send_batch())
すべての項目と同じテーブル スキーマとの互換性を維持します。アプリケーションが所有するバッチ キューをバインドし、そのオーバーフローとシャットダウン ポリシーを文書化します。
SQLを実行します
同期:
result = telemetry.query("""
SELECT
status,
COUNT(*) AS sample_events
FROM telemetry_quickstart
WHERE timestamp_utc >= now() - INTERVAL '24 hours'
GROUP BY status
ORDER BY sample_events DESC
""")
for row in result.get("data", []):
print(row["status"], row["sample_events"])
非同期:
async def load_summary():
return await telemetry.query("""
SELECT status, COUNT(*) AS sample_events
FROM telemetry_quickstart
GROUP BY status
ORDER BY sample_events DESC
""")
これらのメソッドは対話型クエリ エンドポイントを使用します。を使用します。 非同期クエリ API 長時間実行される JSON または Parquet エクスポートの場合は直接。
再試行および失敗ポリシー
変更されていないものを再試行しないでください 400 リクエスト。一時的な接続障害の場合は、 429, 502, 503、そして 504, ジッター付き指数バックオフを使い、アプリケーション側で上限付きの再試行を行ってください.同じものを再利用する event_id 論理イベントの場合。
ほとんどのアプリケーション分析では、テレメトリの停止によって顧客の正常な応答が置き換えられるべきではありません。請求または承認された監査ワークフローには、耐久性のあるアプリケーション所有の送信ボックスを使用します。レビュー イベント配信と冪等性.
確認とトラブルシューティングを行う
既知の成功フィクスチャと失敗フィクスチャを送信し、クエリを実行します。
SELECT timestamp_utc, event_id, job_name, status, duration_ms
FROM telemetry_quickstart
ORDER BY timestamp_utc DESC
LIMIT 20;
タイプ、タイムスタンプ、機密データの境界を確認します。よくある失敗:
API key is not initialized: 電話をかけるinit空ではないサーバー側キーを使用します。401または403: キーを置き換えるか、そのスコープを修正してください。- JSON デコード例外: HTTP ステータスと SDK の外側の安全な応答コンテキストを検査します。
- 遅い同期リクエスト: タイムアウトのある非同期クライアントまたはアプリケーション所有の HTTP クライアントに移動します。
- プロセスのシャットダウン: 追跡された呼び出しを待つか、必要なイベントを保持します。どちらのクライアントもフラッシュ可能なキューを維持しません。
続けて、 FastAPIの統合, Django と Celery の統合、そして 取り込みのトラブルシューティング.