Добавлен OTLP-провайдер, реализовано отбрасывание метрик и трейсов в observability + добавлен перезапуск nginx при пересборке контейнеров (ошибка, когда докер меняет адреса сервисов)

2026-07-29 15:15:40 +03:00
parent 3ed7239efa
commit 41e19005fb
43 changed files with 3016 additions and 83 deletions
@@ -31,6 +31,7 @@ from app.service import (
    read_message,
 )
 from app.settings import get_settings
+from app.telemetry import add_trace_context, init_telemetry, instrument_fastapi

 log = structlog.get_logger()

@@ -40,6 +41,7 @@ def configure_logging(level: str) -> None:
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
+            add_trace_context,
            structlog.processors.TimeStamper(fmt="iso", utc=True, key="timestamp"),
            structlog.stdlib.add_log_level,
            structlog.processors.JSONRenderer(),
@@ -50,11 +52,16 @@ def configure_logging(level: str) -> None:
@asynccontextmanager
 async def lifespan(app: FastAPI):
    settings = get_settings()
+    telemetry = init_telemetry()
    configure_logging(settings.log_level)
    app.state.settings = settings
    app.state.db = Database(settings.database_url)
-    yield
-    await app.state.db.close()
+    try:
+        yield
+    finally:
+        await app.state.db.close()
+        if telemetry:
+            telemetry.shutdown()


 app = FastAPI(
@@ -77,7 +84,6 @@ async def request_context(request: Request, call_next: Any) -> Response:
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        method=request.method,
-        route=request.url.path,
        **{"service.name": "sms-service"},
    )
    response = await call_next(request)
@@ -86,6 +92,7 @@ async def request_context(request: Request, call_next: Any) -> Response:
    response.headers["Cache-Control"] = "no-store"
    log.info(
        "request.complete",
+        route=getattr(request.scope.get("route"), "path", request.url.path),
        status_code=response.status_code,
        duration_ms=round((time.monotonic() - started) * 1000, 2),
    )
@@ -312,6 +319,9 @@ def datetime_now():
    return datetime.now(UTC)


+instrument_fastapi(app)
+
+
 def run() -> None:
    settings = get_settings()
    uvicorn.run(
@@ -18,6 +18,7 @@ class Settings(BaseSettings):
    callback_password: SecretStr = Field(alias="IDGTL_SMS_CALLBACK_PASSWORD")
    log_level: str = Field(default="INFO", alias="LOG_LEVEL")
    api_port: int = Field(default=8080, alias="SMS_API_PORT", ge=1, le=65535)
+    metrics_port: int = Field(default=9464, alias="SMS_METRICS_PORT", ge=1, le=65535)

    @field_validator(
        "service_token",
@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any
+
+from fastapi import FastAPI
+from opentelemetry import metrics, trace
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
+from opentelemetry.propagate import set_global_textmap
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.sdk.trace.sampling import ALWAYS_ON
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+
+
+@dataclass(slots=True)
+class TelemetryRuntime:
+    tracer_provider: TracerProvider
+    meter_provider: MeterProvider
+
+    def shutdown(self) -> None:
+        self.meter_provider.shutdown()
+        self.tracer_provider.shutdown()
+
+
+_runtime: TelemetryRuntime | None = None
+
+
+def _resource(service_name: str) -> Resource:
+    return Resource.create(
+        {
+            "service.name": service_name,
+            "service.namespace": "han-chat",
+            "service.version": os.getenv("RELEASE_VERSION", "unknown"),
+            "deployment.environment": os.getenv("APP_ENV", "production-like"),
+        }
+    )
+
+
+def init_telemetry(service_name: str | None = None) -> TelemetryRuntime | None:
+    global _runtime
+    if _runtime is not None:
+        return _runtime
+    endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "").strip()
+    if not endpoint:
+        return None
+
+    resource = _resource(service_name or os.getenv("OTEL_SERVICE_NAME", "sms-service"))
+    insecure = endpoint.startswith("http://")
+    set_global_textmap(TraceContextTextMapPropagator())
+
+    tracer_provider = TracerProvider(resource=resource, sampler=ALWAYS_ON)
+    tracer_provider.add_span_processor(
+        BatchSpanProcessor(
+            OTLPSpanExporter(endpoint=endpoint, insecure=insecure, timeout=3),
+            max_queue_size=2048,
+            schedule_delay_millis=5000,
+            max_export_batch_size=512,
+            export_timeout_millis=3000,
+        )
+    )
+    trace.set_tracer_provider(tracer_provider)
+
+    metric_reader = PeriodicExportingMetricReader(
+        OTLPMetricExporter(endpoint=endpoint, insecure=insecure, timeout=3),
+        export_interval_millis=30000,
+        export_timeout_millis=3000,
+    )
+    meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
+    metrics.set_meter_provider(meter_provider)
+
+    HTTPXClientInstrumentor().instrument()
+    SQLAlchemyInstrumentor().instrument(enable_commenter=False)
+    _runtime = TelemetryRuntime(tracer_provider, meter_provider)
+    return _runtime
+
+
+def instrument_fastapi(app: FastAPI) -> None:
+    FastAPIInstrumentor.instrument_app(
+        app,
+        excluded_urls="/health/live,/health/ready",
+    )
+
+
+def add_trace_context(
+    _logger: Any,
+    _method_name: str,
+    event_dict: dict[str, Any],
+) -> dict[str, Any]:
+    context = trace.get_current_span().get_span_context()
+    if context.is_valid:
+        event_dict["trace_id"] = format(context.trace_id, "032x")
+        event_dict["span_id"] = format(context.span_id, "016x")
+    return event_dict
@@ -9,6 +9,8 @@ from datetime import UTC, datetime, timedelta

 import httpx
 import structlog
+from opentelemetry import trace
+from prometheus_client import start_http_server
 from sqlalchemy import and_, func, or_, select, update

 from app.db import Database, SendStatus, SmsOutboundMessage
@@ -23,6 +25,7 @@ from app.metrics import (
 from app.provider import IdgtlClient, IdgtlConfig
 from app.service import RuntimeSettings, load_runtime_settings
 from app.settings import Settings, get_settings
+from app.telemetry import add_trace_context, init_telemetry

 log = structlog.get_logger()
 MAX_CONNECT_ATTEMPTS = 3
@@ -32,6 +35,8 @@ def configure_logging(level: str) -> None:
    logging.basicConfig(level=level, format="%(message)s")
    structlog.configure(
        processors=[
+            structlog.contextvars.merge_contextvars,
+            add_trace_context,
            structlog.processors.TimeStamper(fmt="iso", utc=True, key="timestamp"),
            structlog.stdlib.add_log_level,
            structlog.processors.JSONRenderer(),
@@ -169,7 +174,14 @@ async def update_queue_metrics(db: Database) -> None:

 async def worker_loop(stop: asyncio.Event) -> None:
    settings = get_settings()
+    telemetry = init_telemetry("sms-worker")
    configure_logging(settings.log_level)
+    structlog.contextvars.bind_contextvars(**{"service.name": "sms-worker"})
+    metrics_server, metrics_thread = start_http_server(
+        settings.metrics_port,
+        addr="0.0.0.0",  # noqa: S104 - internal Docker-network listener
+    )
+    tracer = trace.get_tracer("han.sms.worker")
    db = Database(settings.database_url)
    async with httpx.AsyncClient() as http:
        try:
@@ -179,16 +191,30 @@ async def worker_loop(stop: asyncio.Event) -> None:
                    async with db.sessions() as session:
                        runtime = await load_runtime_settings(session)
                    SETTINGS_VALID.set(1)
+                    claim_started_ns = time.time_ns()
                    message = await lease_message(db, runtime)
+                    claim_finished_ns = time.time_ns()
                    if message is None:
                        await update_queue_metrics(db)
                        await asyncio.wait_for(stop.wait(), timeout=runtime.poll_interval_ms / 1000)
                        continue
-                    client = IdgtlClient(http, provider_config(settings, runtime))
-                    started = time.monotonic()
-                    result = await client.send(message)
-                    PROVIDER_LATENCY.labels("idgtl").observe(time.monotonic() - started)
-                    await save_result(db, message.id, result, message.attempt_count)
+                    process_span = tracer.start_span("sms.process", start_time=claim_started_ns)
+                    with trace.use_span(process_span, end_on_exit=True):
+                        claim_span = tracer.start_span("sms.claim", start_time=claim_started_ns)
+                        claim_span.set_attribute("sms.claimed", True)
+                        claim_span.end(end_time=claim_finished_ns)
+                        span = trace.get_current_span()
+                        span.set_attribute("messaging.operation.name", "send")
+                        span.set_attribute("messaging.system", "idgtl")
+                        span.set_attribute("sms.attempt", message.attempt_count)
+                        client = IdgtlClient(http, provider_config(settings, runtime))
+                        with tracer.start_as_current_span("sms.provider"):
+                            started = time.monotonic()
+                            result = await client.send(message)
+                        PROVIDER_LATENCY.labels("idgtl").observe(time.monotonic() - started)
+                        span.set_attribute("sms.outcome", result.send_status.value)
+                        with tracer.start_as_current_span("sms.save_result"):
+                            await save_result(db, message.id, result, message.attempt_count)
                except TimeoutError:
                    continue
                except Exception:
@@ -200,6 +226,10 @@ async def worker_loop(stop: asyncio.Event) -> None:
                        pass
        finally:
            await db.close()
+            metrics_server.shutdown()
+            metrics_thread.join(timeout=5)
+            if telemetry:
+                telemetry.shutdown()


 def run() -> None:
@@ -8,6 +8,12 @@ dependencies = [
  "asyncpg>=0.30,<1",
  "fastapi>=0.116,<1",
  "httpx>=0.28,<1",
+  "opentelemetry-api>=1.44,<2",
+  "opentelemetry-exporter-otlp-proto-grpc>=1.44,<2",
+  "opentelemetry-instrumentation-fastapi>=0.65b0,<1",
+  "opentelemetry-instrumentation-httpx>=0.65b0,<1",
+  "opentelemetry-instrumentation-sqlalchemy>=0.65b0,<1",
+  "opentelemetry-sdk>=1.44,<2",
  "phonenumbers>=9,<10",
  "prometheus-client>=0.22,<1",
  "pydantic-settings>=2.10,<3",
@@ -0,0 +1,21 @@
+from opentelemetry.sdk.trace import TracerProvider
+
+from app import telemetry
+
+
+def test_telemetry_is_fail_open_without_endpoint(monkeypatch) -> None:
+    monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
+    monkeypatch.setattr(telemetry, "_runtime", None)
+
+    assert telemetry.init_telemetry("sms-test") is None
+
+
+def test_structlog_processor_adds_active_trace_context() -> None:
+    tracer = TracerProvider().get_tracer("test")
+
+    with tracer.start_as_current_span("sms.operation"):
+        result = telemetry.add_trace_context(None, "info", {"event": "sms.safe"})
+
+    assert result["event"] == "sms.safe"
+    assert len(result["trace_id"]) == 32
+    assert len(result["span_id"]) == 16