Python SDK

telemetry-sh 패키지는 동기식 애플리케이션용 Telemetry와 asyncio 서비스용 TelemetryAsync를 제공합니다. 두 클라이언트 모두 각 메서드 호출을 즉시 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하지 마십시오. 이벤트 생성자의 경우 수집 전용 키를 사용하고 쿼리 전용 작업의 경우 읽기 범위 키를 사용하여 프로세스당 한 번 초기화합니다.

동기적으로 이벤트 보내기

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
    """)

이러한 메서드는 대화형 쿼리 끝점을 사용합니다. 장기 실행 JSON 또는 Parquet 내보내기에는 비동기 쿼리 API를 직접 사용하세요.

재시도 및 실패 정책

변경되지 않은 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 통합, 수집 문제 해결을 계속 진행하세요.

코딩 에이전트와 함께 이 문서를 사용하세요.