프레임워크 가이드

pip install "sophonz-opentelemetry[flask,psycopg]" flask gunicorn "psycopg[binary,pool]"

myservice/
  telemetry.py      # init()
  wsgi.py           # telemetry를 먼저 import한 뒤 앱을 만듭니다
  app.py            # create_app(): 헬퍼, 라우트, 에러 핸들러
gunicorn.conf.py

# myservice/telemetry.py
import os
 
from sophonz.opentelemetry import init
 
SERVICE_NAME = os.getenv("OTEL_SERVICE_NAME", "checkout-api")
 
 
def setup_telemetry() -> bool:
    return init(
        service=SERVICE_NAME,
        api_key=os.getenv("SOPHONZ_API_KEY"),
        instrumentations={
            "psycopg": {"enable_commenter": True},
        },
    )

# myservice/wsgi.py
import logging
 
from myservice.telemetry import setup_telemetry
 
setup_telemetry()
 
logging.basicConfig(level=logging.INFO)
 
from myservice.app import create_app  # noqa: E402
 
app = create_app()

# myservice/app.py
from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException, InternalServerError
from werkzeug.middleware.proxy_fix import ProxyFix
 
from sophonz.opentelemetry.flask import setup_flask_error_handler, sophonz_flask_middleware
 
 
def create_app() -> Flask:
    app = Flask(__name__)
 
    # 리버스 프록시 뒤에서: request.remote_addr, 즉 http.client.ip가
    # X-Forwarded-For에서 옵니다.
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
 
    sophonz_flask_middleware(app)
    setup_flask_error_handler(app)
 
    @app.get("/api/error")
    def error():
        raise RuntimeError("Intentional error")
 
    @app.errorhandler(InternalServerError)
    def internal_error(exc):
        original = exc.original_exception or exc
        return jsonify(error=str(original)), 500
 
    @app.errorhandler(HTTPException)
    def http_error(exc):
        return jsonify(error=exc.description), exc.code
 
    return app

# gunicorn.conf.py
import os
 
bind = f"0.0.0.0:{os.getenv('PORT', '8000')}"
workers = int(os.getenv("WEB_CONCURRENCY", "2"))
worker_class = "gthread"
threads = int(os.getenv("GUNICORN_THREADS", "4"))
timeout = 30
graceful_timeout = 20
# preload_app 없음: 각 워커가 fork된 뒤 wsgi.py를 import하고 SDK를 시작합니다.
 
 
def worker_exit(server, worker):
    # 여기서 import합니다: 마스터 프로세스는 SDK를 시작하지 않습니다.
    from sophonz.opentelemetry import shutdown
 
    shutdown()

gunicorn -c gunicorn.conf.py myservice.wsgi:app

instrumentations={
    "psycopg": {"enable_commenter": True},
    "flask": {"enable_commenter": False},
}

pip install "sophonz-opentelemetry[fastapi,psycopg]" fastapi uvicorn "psycopg[binary,pool]"

# myservice/main.py
import logging
 
from myservice.telemetry import setup_telemetry
 
setup_telemetry()
 
logging.basicConfig(level=logging.INFO)
 
from myservice.app import create_app  # noqa: E402
 
app = create_app()

# myservice/app.py
import contextlib
 
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from psycopg_pool import AsyncConnectionPool
 
from sophonz.opentelemetry.fastapi import setup_fastapi_error_handler, sophonz_fastapi_middleware
 
 
def create_app() -> FastAPI:
    pool = AsyncConnectionPool("postgresql://app:app@localhost:5432/app", open=False)
 
    @contextlib.asynccontextmanager
    async def lifespan(_app: FastAPI):
        await pool.open()
        try:
            yield
        finally:
            await pool.close()
 
    app = FastAPI(lifespan=lifespan)
 
    setup_fastapi_error_handler(app)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["https://app.example.com"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["traceparent", "tracestate", "baggage", "content-type"],
    )
    sophonz_fastapi_middleware(app)
 
    @app.get("/api/error")
    async def error():
        raise RuntimeError("Intentional error")
 
    @app.exception_handler(Exception)
    async def unhandled(request: Request, exc: Exception):
        headers = {}
        if origin := request.headers.get("origin"):
            headers = {"Access-Control-Allow-Origin": origin, "Vary": "Origin"}
        return JSONResponse({"error": str(exc)}, status_code=500, headers=headers)
 
    return app

import asyncio
 
 
@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    async def find_user():
        async with pool.connection() as conn:
            cur = await conn.execute("SELECT id, email FROM users WHERE id = %s", (user_id,))
            return await cur.fetchone()
 
    async def find_posts():
        async with pool.connection() as conn:
            cur = await conn.execute("SELECT id, title FROM posts WHERE author_id = %s", (user_id,))
            return await cur.fetchall()
 
    user, posts = await asyncio.gather(find_user(), find_posts())
    return {"user": user, "posts": posts}

uvicorn myservice.main:app --host 0.0.0.0 --port 8000 \
  --loop asyncio --proxy-headers --forwarded-allow-ips "*" \
  --timeout-graceful-shutdown 20

instrumentations={"fastapi": {"exclude_spans": ["receive", "send"]}}

pip install "sophonz-opentelemetry[django,psycopg]" django gunicorn "psycopg[binary]"

# mysite/telemetry.py
import os
 
from sophonz.opentelemetry import init, is_initialized
 
 
def setup_telemetry() -> bool:
    if is_initialized():
        return False
    return init(
        service=os.getenv("OTEL_SERVICE_NAME", "mysite"),
        api_key=os.getenv("SOPHONZ_API_KEY"),
        instrumentations={"psycopg": {"enable_commenter": True}},
    )

# mysite/wsgi.py — gunicorn mysite.wsgi
import os
 
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
 
from mysite.telemetry import setup_telemetry  # noqa: E402
 
setup_telemetry()
 
from django.core.wsgi import get_wsgi_application  # noqa: E402
 
application = get_wsgi_application()

# mysite/asgi.py — uvicorn mysite.asgi:application
import os
 
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
 
from mysite.telemetry import setup_telemetry  # noqa: E402
 
setup_telemetry()
 
from django.core.asgi import get_asgi_application  # noqa: E402
 
application = get_asgi_application()

# manage.py
import os
import sys
 
 
def main():
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
    # 요청을 처리하는 것은 개발 서버뿐입니다. migrate 등 다른 명령은
    # SDK 없이 실행됩니다.
    if len(sys.argv) > 1 and sys.argv[1] == "runserver":
        from mysite.telemetry import setup_telemetry
 
        setup_telemetry()
 
    from django.core.management import execute_from_command_line
 
    execute_from_command_line(sys.argv)
 
 
if __name__ == "__main__":
    main()

# mysite/settings.py (발췌)
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.middleware.common.CommonMiddleware",
    # ... REMOTE_ADDR을 바꾸는 미들웨어는 이 줄보다 위에 둡니다
    "sophonz.opentelemetry.django.SophonzMiddleware",
]
 
SOPHONZ_MIDDLEWARE = {
    "attributes": "mysite.telemetry.request_attributes",  # 또는 dict
}
 
LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {"console": {"class": "logging.StreamHandler"}},
    "root": {"handlers": ["console"], "level": "INFO"},
}

# mysite/telemetry.py (이어서)
def request_attributes(request):
    user = getattr(request, "user", None)
    if user is None or not user.is_authenticated:
        return {}
    return {"enduser.id": str(user.pk)}

import os
 
from sophonz.opentelemetry import init, shutdown, trace
 
init(service="nightly-sync", api_key=os.getenv("SOPHONZ_API_KEY"))
 
tracer = trace.get_tracer("nightly-sync")
 
try:
    with tracer.start_as_current_span("sync-inventory"):
        run_sync()
finally:
    shutdown()