Python SDK
telemetry-sh 패키지는 동기식 애플리케이션용 Telemetry와 asyncio 서비스용 TelemetryAsync를 제공합니다. 두 클라이언트 모두 각 메서드 호출을 즉시 Telemetry HTTP API로 보냅니다. 서버 측 구성에서 API 키를 유지하십시오.
설치 및 초기화
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하지 마십시오. 이벤트 생성자의 경우 쓰기 범위 키를 사용하고 쿼리 전용 작업의 경우 읽기 범위 키를 사용하여 프로세스당 한 번 초기화합니다.
동기적으로 이벤트 보내기
import time
import uuid
started_at = time.perf_counter()
event_id = str(uuid.uuid4())
try:
response = telemetry.log("job_completed", {
"event_id": event_id,
"job_name": "invoice_sync",
"status": "success",
"duration_ms": round((time.perf_counter() - started_at) * 1000),
"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 time
import uuid
async def record_job():
started_at = time.perf_counter()
await telemetry.log("job_completed", {
"event_id": str(uuid.uuid4()),
"job_name": "invoice_sync",
"status": "success",
"duration_ms": round((time.perf_counter() - started_at) * 1000),
"attempt": 1,
})
asyncio.run(record_job())
TelemetryAsync.log는 요청에 대해 aiohttp 세션을 열고 나중에 닫습니다. 이벤트 루프 차단을 방지하지만 현재 패키지는 공유 세션, 백그라운드 큐, 재시도 정책 또는 플러시 메서드를 노출하지 않습니다.
호환되는 배치 보내기
두 클라이언트 모두 사전 목록을 허용합니다.
await telemetry.log("api_request_completed", [
{
"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",
},
])
모든 항목이 동일한 테이블 스키마와 호환되도록 유지하세요. 모든 애플리케이션 소유 배치 큐를 바인딩하고 오버플로 및 종료 정책을 문서화합니다.
SQL 실행
동기식:
result = telemetry.query("""
SELECT
status,
COUNT(*) AS jobs
FROM job_completed
WHERE timestamp_utc >= now() - INTERVAL '24 hours'
GROUP BY status
ORDER BY jobs DESC
""")
for row in result.get("data", []):
print(row["status"], row["jobs"])
비동기식:
async def load_summary():
return await telemetry.query("""
SELECT status, COUNT(*) AS jobs
FROM job_completed
GROUP BY status
ORDER BY jobs DESC
""")
이러한 메서드는 대화형 쿼리 끝점을 사용합니다. 장기 실행 JSON 또는 Parquet 내보내기에는 비동기 쿼리 API를 직접 사용하세요.
재시도 및 실패 정책
변경되지 않은 400 요청을 재시도하지 마세요. 일시적인 연결 오류인 429, 502, 503 및 504의 경우 지수 백오프 및 지터를 사용하여 제한된 애플리케이션 재시도를 사용합니다. 논리적 이벤트에 대해 동일한 event_id를 재사용합니다.
대부분의 애플리케이션 분석에서는 텔레메트리 중단이 성공적인 고객 응답을 대체해서는 안 됩니다. 청구 또는 승인된 감사 워크플로의 경우 내구성이 뛰어난 애플리케이션 소유의 아웃박스을 사용하세요. 이벤트 전달 및 멱등성을 검토하세요.
확인 및 문제 해결
알려진 성공 및 실패 픽스처를 보낸 후 다음을 쿼리합니다.
SELECT timestamp_utc, event_id, job_name, status, duration_ms, error_type
FROM job_completed
ORDER BY timestamp_utc DESC
LIMIT 20;
유형, 타임스탬프, 민감한 데이터 경계를 확인하세요. 일반적인 오류:
API key is not initialized: 비어 있지 않은 서버 측 키를 사용하여init를 호출합니다.401또는403: 키를 교체하거나 범위를 수정하세요.- JSON-디코딩 예외: HTTP 상태와 SDK 외부의 안전한 응답 컨텍스트를 검사합니다.
- 느린 동기 요청: 시간 초과가 있는 비동기 클라이언트 또는 애플리케이션 소유 HTTP 클라이언트로 이동합니다.
- 프로세스 종료: 추적된 호출을 기다리거나 필수 이벤트를 지속합니다. 어느 클라이언트도 플러시 가능한 대기열을 유지하지 않습니다.
FastAPI 통합, Django 및 Celery 통합, 수집 문제 해결을 계속 진행하세요.