예제

myservice/
  __init__.py       # 빈 파일
  telemetry.py
  app.py
  wsgi.py
gunicorn.conf.py

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

# myservice/telemetry.py
import os
 
from sophonz.opentelemetry import init
 
SERVICE_NAME = os.getenv("OTEL_SERVICE_NAME", "example-flask")
 
# SQL 주석에 traceparent만 남깁니다. pg_tracing은 다른 값을 읽지 않습니다.
PSYCOPG = {
    "enable_commenter": True,
    "commenter_options": {
        "db_driver": False,
        "dbapi_threadsafety": False,
        "dbapi_level": False,
        "libpq_version": False,
        "driver_paramstyle": False,
    },
}
 
 
def setup_telemetry() -> bool:
    return init(
        service=SERVICE_NAME,
        api_key=os.getenv("SOPHONZ_API_KEY"),
        advanced_network_capture=True,
        instrumentations={
            "psycopg": PSYCOPG,
            # Flask가 주석에 framework/controller/route 태그를 더하지 않게 합니다.
            "flask": {"enable_commenter": False},
        },
    )

# myservice/app.py
import logging
import os
 
import psycopg
from flask import Flask, jsonify, request
from flask_cors import CORS
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
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
 
logger = logging.getLogger("myservice")
 
 
def create_app() -> Flask:
    app = Flask(__name__)
    app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
 
    CORS(
        app,
        origins=os.getenv("CORS_ORIGINS", "http://localhost:3000").split(","),
        supports_credentials=True,
        allow_headers=["content-type", "traceparent", "tracestate", "baggage"],
    )
 
    # 요청의 모든 스팬에 baggage.*, session.id, 클라이언트 IP, user agent를 붙이고,
    # 라우트에서 빠져나간 예외를 한 번 기록합니다.
    sophonz_flask_middleware(app)
    setup_flask_error_handler(app)
 
    # gunicorn 워커마다 풀 하나. SDK가 시작된 뒤에 만듭니다.
    pool = ConnectionPool(
        os.environ["DATABASE_URL"],
        min_size=1,
        max_size=int(os.getenv("DB_POOL_SIZE", "5")),
        kwargs={"autocommit": True, "row_factory": dict_row},
        open=False,
    )
 
    def connection():
        if pool.closed:
            pool.open(wait=False)
        return pool.connection()
 
    @app.get("/health")
    def health():
        return jsonify(status="ok")
 
    @app.get("/api/users")
    def list_users():
        with connection() as conn:
            users = conn.execute("SELECT id, email, name FROM users ORDER BY id").fetchall()
        logger.info("listed users", extra={"count": len(users)})
        return jsonify(users)
 
    @app.get("/api/users/<int:user_id>")
    def get_user(user_id):
        with connection() as conn:
            user = conn.execute(
                "SELECT id, email, name FROM users WHERE id = %s", (user_id,)
            ).fetchone()
            if user is None:
                return jsonify(error="User not found"), 404
            posts = conn.execute(
                "SELECT id, title, body FROM posts WHERE author_id = %s ORDER BY id", (user_id,)
            ).fetchall()
        return jsonify({**user, "posts": posts})
 
    @app.post("/api/users")
    def create_user():
        payload = request.get_json(silent=True) or {}
        if not payload.get("email") or not payload.get("name"):
            return jsonify(error="email and name are required"), 400
        try:
            with connection() as conn:
                user = conn.execute(
                    "INSERT INTO users (email, name) VALUES (%s, %s) RETURNING id, email, name",
                    (payload["email"], payload["name"]),
                ).fetchone()
        except psycopg.errors.UniqueViolation:
            return jsonify(error="email already exists"), 409
        logger.info("created user", extra={"user_id": user["id"]})
        return jsonify(user), 201
 
    @app.get("/api/error")
    def error():
        raise RuntimeError("Intentional error from /api/error")
 
    # got_request_exception 이후에 실행되므로 스팬에는 이미 예외가 있습니다.
    @app.errorhandler(InternalServerError)
    def internal_error(exc):
        return jsonify(error=str(exc.original_exception or exc)), 500
 
    @app.errorhandler(HTTPException)
    def http_error(exc):
        return jsonify(error=exc.description), exc.code
 
    return app

# myservice/wsgi.py
import logging
import os
 
from myservice.telemetry import setup_telemetry
 
# 각 gunicorn 워커에서, Flask와 psycopg를 import하기 전에.
setup_telemetry()
 
logging.basicConfig(
    level=os.getenv("LOG_LEVEL", "INFO").upper(),
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
 
from myservice.app import create_app  # noqa: E402
 
app = create_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
accesslog = "-"
 
 
def worker_exit(server, worker):
    from sophonz.opentelemetry import shutdown
 
    shutdown()

export SOPHONZ_API_KEY=sk_...
export DATABASE_URL=postgresql://app:app@localhost:5432/app
export OTEL_METRICS_EXPORTER=none
 
gunicorn -c gunicorn.conf.py myservice.wsgi:app