把慢查询变成可行动的可靠性指标：OpenTelemetry 实践指南

python -m pip install \
  opentelemetry-api \
  opentelemetry-sdk

import sqlite3
import time
from contextlib import contextmanager

from opentelemetry import metrics, trace
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor

SLOW_QUERY_SECONDS = 0.05

resource = Resource.create({
    "service.name": "orders-api",
    "deployment.environment.name": "production",
})

tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer("orders-api.database")

metric_reader = PeriodicExportingMetricReader(
    ConsoleMetricExporter(),
    export_interval_millis=5_000,
)
metrics.set_meter_provider(MeterProvider(resource=resource, metric_readers=[metric_reader]))
meter = metrics.get_meter("orders-api.database")

query_duration = meter.create_histogram(
    "db.client.operation.duration",
    unit="s",
    description="Database operation duration in seconds",
)
slow_queries = meter.create_counter(
    "db.client.slow_queries",
    unit="{query}",
    description="Number of database operations over the slow-query threshold",
)


@contextmanager
def traced_query(connection, operation_name, sql, parameters=()):
    attributes = {
        "db.system": "sqlite",
        "db.operation.name": operation_name,
    }
    start = time.perf_counter()

    with tracer.start_as_current_span("sqlite " + operation_name) as span:
        span.set_attributes(attributes)
        # Keep SQL out of metric labels. In production, store only a sanitized
        # or parameterized statement in the span when it is needed for debugging.
        span.set_attribute("db.query.summary", operation_name)

        try:
            cursor = connection.execute(sql, parameters)
            yield cursor
        except Exception as exc:
            span.record_exception(exc)
            span.set_status(trace.StatusCode.ERROR, str(exc))
            raise
        finally:
            duration = time.perf_counter() - start
            query_duration.record(duration, attributes)
            if duration >= SLOW_QUERY_SECONDS:
                slow_queries.add(1, attributes)
                span.set_attribute("db.slow_query", True)
                span.set_attribute("db.slow_query.threshold_seconds", SLOW_QUERY_SECONDS)
            span.set_attribute("db.duration_seconds", duration)


if __name__ == "__main__":
    connection = sqlite3.connect(":memory:")
    connection.execute("create table orders (id integer primary key, status text)")
    connection.executemany(
        "insert into orders(status) values (?)",
        [("paid",), ("pending",), ("paid",)],
    )
    connection.commit()

    with traced_query(
        connection,
        "select_orders_by_status",
        "select id, status from orders where status = ?",
        ("paid",),
    ) as cursor:
        print(cursor.fetchall())

    # Keep the process alive long enough for the periodic metric reader to export.
    time.sleep(6)
    connection.close()