from __future__ import annotations import logging import os import re import sys import time from typing import Any, Dict, List, Optional from fastapi import FastAPI from fastapi.responses import PlainTextResponse from prometheus_client import ( CONTENT_TYPE_LATEST, CollectorRegistry, Counter, Gauge, Histogram, generate_latest, ) logger = logging.getLogger("telemetry") # Environment & Version Metadata (The Golden Triangle Invariant) SERVICE_NAME: str = os.getenv("OTEL_SERVICE_NAME", "api-backend") ENVIRONMENT: str = os.getenv("APP_ENV", os.getenv("DEPLOYMENT_ENVIRONMENT", "development")) VERSION: str = os.getenv("APP_VERSION", os.getenv("SERVICE_VERSION", "dev-latest")) OTLP_ENDPOINT: str = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://personal-lifecycle-otel-collector:4318") # Prometheus Registry REGISTRY = CollectorRegistry() # ----------------------------------------------------------------------------- # Prometheus Metric Instruments # ----------------------------------------------------------------------------- # Metadata / Build Info metric APP_INFO = Gauge( "app_info", "Application metadata containing deployment environment and service version", ["deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) APP_INFO.labels( deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).set(1) # 1. Golden Signals: HTTP Request Rate, Latency & Errors HTTP_REQUEST_DURATION = Histogram( "app_http_server_duration_milliseconds", "HTTP server request execution duration in milliseconds", ["http_route", "http_status_code", "http_method", "deployment_environment", "service_version", "service_name"], buckets=(5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0), registry=REGISTRY, ) HTTP_REQUESTS_TOTAL = Counter( "app_http_requests_total", "Total count of HTTP requests processed by the server", ["http_route", "http_status_code", "http_method", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) # 2. Product Metrics: Lifecycle Items (Renewals, Maintenance, Warranties) ITEMS_CREATED_TOTAL = Counter( "app_items_created_total", "Total count of lifecycle items created by users or groups", ["lifecycle_type", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) ITEM_CREATION_FAILURES_TOTAL = Counter( "app_item_creation_failures_total", "Total count of lifecycle item creation failures by lifecycle type", ["lifecycle_type", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) ITEMS_COMPLETED_TOTAL = Counter( "app_items_completed_total", "Total count of lifecycle items marked as complete", ["lifecycle_type", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) ITEMS_DELETED_TOTAL = Counter( "app_items_deleted_total", "Total count of lifecycle items permanently deleted", ["lifecycle_type", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) LIFECYCLE_ITEMS_CURRENT = Gauge( "app_lifecycle_items_current", "Instantaneous count of lifecycle items in the database by status and type", ["lifecycle_type", "status", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) # 3. Product Metrics: Accounts, Authentication & Shared Groups USER_SIGNUPS_TOTAL = Counter( "app_user_signups_total", "Total user registration events", ["deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) USER_LOGINS_TOTAL = Counter( "app_user_logins_total", "Total user login attempts by outcome status", ["status", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) GROUPS_CREATED_TOTAL = Counter( "app_groups_created_total", "Total shared groups created", ["deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) GROUP_MEMBERS_INVITED_TOTAL = Counter( "app_group_members_invited_total", "Total group members invited or added", ["deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) # 4. Product Metrics: Reminders & Daily Morning Email Digests DIGESTS_GENERATED_TOTAL = Counter( "app_digests_generated_total", "Total daily morning email digests generated or sent", ["status", "deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) REMINDERS_SUMMARY_REQUESTS_TOTAL = Counter( "app_reminders_summary_requests_total", "Total in-app reminder summaries requested by users", ["deployment_environment", "service_version", "service_name"], registry=REGISTRY, ) _current_lifecycle_counts: Dict[tuple[str, str], int] = {} def _init_default_series() -> None: """Pre-initialize key series so metric queries return series even before first traffic.""" for l_type in ("renewal", "maintenance", "warranty"): ITEMS_CREATED_TOTAL.labels(lifecycle_type=l_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) ITEMS_COMPLETED_TOTAL.labels(lifecycle_type=l_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) ITEMS_DELETED_TOTAL.labels(lifecycle_type=l_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) ITEM_CREATION_FAILURES_TOTAL.labels(lifecycle_type=l_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) for st in ("active", "completed", "overdue"): LIFECYCLE_ITEMS_CURRENT.labels(lifecycle_type=l_type, status=st, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME).set(0) _current_lifecycle_counts[(l_type, st)] = 0 ITEM_CREATION_FAILURES_TOTAL.labels(lifecycle_type="unknown", deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) USER_SIGNUPS_TOTAL.labels(deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) USER_LOGINS_TOTAL.labels(status="success", deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) USER_LOGINS_TOTAL.labels(status="failure", deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) DIGESTS_GENERATED_TOTAL.labels(status="success", deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) DIGESTS_GENERATED_TOTAL.labels(status="failure", deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) REMINDERS_SUMMARY_REQUESTS_TOTAL.labels(deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) GROUPS_CREATED_TOTAL.labels(deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) GROUP_MEMBERS_INVITED_TOTAL.labels(deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME) _init_default_series() # ----------------------------------------------------------------------------- # OpenTelemetry Module Imports & Global State # ----------------------------------------------------------------------------- try: from opentelemetry import metrics as otel_metrics from opentelemetry import trace as otel_trace from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter 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, SimpleSpanProcessor from opentelemetry.semconv.resource import ResourceAttributes HAS_OTEL = True except ImportError: otel_trace = None otel_metrics = None HAS_OTEL = False _tracer_provider: Optional[Any] = None _meter_provider: Optional[Any] = None _otel_meter: Optional[Any] = None # OTel Metric Instruments _otel_http_requests: Optional[Any] = None _otel_http_duration: Optional[Any] = None _otel_items_created: Optional[Any] = None _otel_item_creation_failures: Optional[Any] = None _otel_items_completed: Optional[Any] = None _otel_items_deleted: Optional[Any] = None _otel_user_signups: Optional[Any] = None _otel_user_logins: Optional[Any] = None _otel_groups_created: Optional[Any] = None _otel_group_members_invited: Optional[Any] = None _otel_digests_generated: Optional[Any] = None _otel_reminders_requests: Optional[Any] = None # ----------------------------------------------------------------------------- # Metric Recording Helper Functions (Dual Prometheus + OpenTelemetry) # ----------------------------------------------------------------------------- def record_http_request(method: str, route: str, status_code: int | str, duration_ms: float) -> None: code_str = str(status_code) HTTP_REQUEST_DURATION.labels( http_route=route, http_status_code=code_str, http_method=method, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).observe(duration_ms) HTTP_REQUESTS_TOTAL.labels( http_route=route, http_status_code=code_str, http_method=method, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() attrs = { "http_route": route, "http_status_code": code_str, "http_method": method, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, } if _otel_http_duration is not None: try: _otel_http_duration.record(duration_ms, attrs) except Exception: pass if _otel_http_requests is not None: try: _otel_http_requests.add(1, attrs) except Exception: pass def record_item_created(lifecycle_type: str) -> None: ITEMS_CREATED_TOTAL.labels( lifecycle_type=lifecycle_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_items_created is not None: try: _otel_items_created.add(1, { "lifecycle_type": lifecycle_type, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_item_creation_failure(lifecycle_type: str = "unknown") -> None: """Record a failed lifecycle item creation attempt.""" ITEM_CREATION_FAILURES_TOTAL.labels( lifecycle_type=lifecycle_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_item_creation_failures is not None: try: _otel_item_creation_failures.add(1, { "lifecycle_type": lifecycle_type, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_item_completed(lifecycle_type: str) -> None: ITEMS_COMPLETED_TOTAL.labels( lifecycle_type=lifecycle_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_items_completed is not None: try: _otel_items_completed.add(1, { "lifecycle_type": lifecycle_type, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_item_deleted(lifecycle_type: str) -> None: ITEMS_DELETED_TOTAL.labels( lifecycle_type=lifecycle_type, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_items_deleted is not None: try: _otel_items_deleted.add(1, { "lifecycle_type": lifecycle_type, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_user_signup() -> None: USER_SIGNUPS_TOTAL.labels( deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_user_signups is not None: try: _otel_user_signups.add(1, { "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_user_login(success: bool) -> None: status_label = "success" if success else "failure" USER_LOGINS_TOTAL.labels( status=status_label, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_user_logins is not None: try: _otel_user_logins.add(1, { "status": status_label, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_group_created() -> None: GROUPS_CREATED_TOTAL.labels( deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_groups_created is not None: try: _otel_groups_created.add(1, { "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_group_member_invited() -> None: GROUP_MEMBERS_INVITED_TOTAL.labels( deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_group_members_invited is not None: try: _otel_group_members_invited.add(1, { "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_digest_generated(success: bool) -> None: status_label = "success" if success else "failure" DIGESTS_GENERATED_TOTAL.labels( status=status_label, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_digests_generated is not None: try: _otel_digests_generated.add(1, { "status": status_label, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def record_reminders_summary_requested() -> None: REMINDERS_SUMMARY_REQUESTS_TOTAL.labels( deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).inc() if _otel_reminders_requests is not None: try: _otel_reminders_requests.add(1, { "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }) except Exception: pass def update_lifecycle_items_gauge(lifecycle_type: str, status: str, count: int) -> None: _current_lifecycle_counts[(lifecycle_type, status)] = count LIFECYCLE_ITEMS_CURRENT.labels( lifecycle_type=lifecycle_type, status=status, deployment_environment=ENVIRONMENT, service_version=VERSION, service_name=SERVICE_NAME, ).set(count) # ----------------------------------------------------------------------------- # OpenTelemetry SDK Integration (Traces & Metrics OTLP Exporter) # ----------------------------------------------------------------------------- def _app_info_obs_callback(options: Any) -> List[Any]: if not HAS_OTEL: return [] from opentelemetry.metrics import Observation return [ Observation( value=1, attributes={ "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }, ) ] def _lifecycle_items_obs_callback(options: Any) -> List[Any]: if not HAS_OTEL: return [] from opentelemetry.metrics import Observation obs = [] for (l_type, st), cnt in _current_lifecycle_counts.items(): obs.append( Observation( value=cnt, attributes={ "lifecycle_type": l_type, "status": st, "deployment_environment": ENVIRONMENT, "service_version": VERSION, "service_name": SERVICE_NAME, }, ) ) return obs def init_opentelemetry(app: FastAPI) -> None: """Initialize OpenTelemetry tracer and meter providers with Golden Triangle resource attributes.""" global _tracer_provider, _meter_provider, _otel_meter global _otel_http_requests, _otel_http_duration, _otel_items_created, _otel_item_creation_failures global _otel_items_completed, _otel_items_deleted, _otel_user_signups, _otel_user_logins global _otel_groups_created, _otel_group_members_invited, _otel_digests_generated, _otel_reminders_requests is_test_env = "pytest" in sys.modules or os.getenv("TESTING", "").lower() in ("1", "true") if is_test_env or not HAS_OTEL: return otlp_endpoint = (os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT") or "").strip() if not otlp_endpoint or (not otlp_endpoint.startswith("http://") and not otlp_endpoint.startswith("https://")): logger.info("Remote OTLP exporter skipped (no valid http/https endpoint configured)") return try: endpoint_base = otlp_endpoint.rstrip("/") resource = Resource.create( { ResourceAttributes.SERVICE_NAME: SERVICE_NAME, ResourceAttributes.DEPLOYMENT_ENVIRONMENT: ENVIRONMENT, ResourceAttributes.SERVICE_VERSION: VERSION, } ) headers: Dict[str, str] = {} raw_headers = (os.getenv("OTEL_EXPORTER_OTLP_HEADERS") or "").strip() if raw_headers: for part in raw_headers.split(","): if "=" in part: k, v = part.split("=", 1) headers[k.strip()] = v.strip() # Tracing Setup _tracer_provider = TracerProvider(resource=resource) traces_endpoint = ( f"{endpoint_base}/v1/traces" if not endpoint_base.endswith("/v1/traces") else endpoint_base ) otlp_timeout = int(os.getenv("OTEL_EXPORTER_OTLP_TIMEOUT", "2")) span_exporter = OTLPSpanExporter( endpoint=traces_endpoint, headers=headers if headers else None, timeout=otlp_timeout, ) force_sync = os.getenv("OTEL_FORCE_SYNC_EXPORT", "false").lower() in ("true", "1", "yes") is_lambda = bool(os.getenv("AWS_LAMBDA_FUNCTION_NAME") or os.getenv("AWS_EXECUTION_ENV")) if is_lambda and force_sync: _tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter)) else: _tracer_provider.add_span_processor( BatchSpanProcessor( span_exporter, max_queue_size=2048, schedule_delay_millis=500, max_export_batch_size=512, export_timeout_millis=2000, ) ) otel_trace.set_tracer_provider(_tracer_provider) # Metrics Setup metrics_endpoint = ( f"{endpoint_base}/v1/metrics" if not endpoint_base.endswith("/v1/metrics") else endpoint_base ) metric_exporter = OTLPMetricExporter( endpoint=metrics_endpoint, headers=headers if headers else None, timeout=otlp_timeout, ) metric_interval = int(os.getenv("OTEL_METRIC_EXPORT_INTERVAL", "5000")) metric_reader = PeriodicExportingMetricReader( metric_exporter, export_interval_millis=metric_interval, export_timeout_millis=2000, ) _meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader]) otel_metrics.set_meter_provider(_meter_provider) # Create OTel Metric Instruments _otel_meter = otel_metrics.get_meter("api-backend") _otel_meter.create_observable_gauge( "app_info", callbacks=[_app_info_obs_callback], description="Application metadata containing deployment environment and service version", ) _otel_meter.create_observable_gauge( "app_lifecycle_items_current", callbacks=[_lifecycle_items_obs_callback], description="Instantaneous count of lifecycle items in the database by status and type", ) _otel_http_requests = _otel_meter.create_counter( "app_http_requests", unit="1", description="Total count of HTTP requests processed by the server", ) _otel_http_duration = _otel_meter.create_histogram( "app_http_server_duration_milliseconds", unit="ms", description="HTTP server request execution duration in milliseconds", ) _otel_items_created = _otel_meter.create_counter( "app_items_created", unit="1", description="Total count of lifecycle items created by users or groups", ) _otel_item_creation_failures = _otel_meter.create_counter( "app_item_creation_failures", unit="1", description="Total count of lifecycle item creation failures by lifecycle type", ) _otel_items_completed = _otel_meter.create_counter( "app_items_completed", unit="1", description="Total count of lifecycle items marked as complete", ) _otel_items_deleted = _otel_meter.create_counter( "app_items_deleted", unit="1", description="Total count of lifecycle items permanently deleted", ) _otel_user_signups = _otel_meter.create_counter( "app_user_signups", unit="1", description="Total user registration events", ) _otel_user_logins = _otel_meter.create_counter( "app_user_logins", unit="1", description="Total user login attempts by outcome status", ) _otel_groups_created = _otel_meter.create_counter( "app_groups_created", unit="1", description="Total shared groups created", ) _otel_group_members_invited = _otel_meter.create_counter( "app_group_members_invited", unit="1", description="Total group members invited or added", ) _otel_digests_generated = _otel_meter.create_counter( "app_digests_generated", unit="1", description="Total daily morning email digests generated or sent", ) _otel_reminders_requests = _otel_meter.create_counter( "app_reminders_summary_requests", unit="1", description="Total in-app reminder summaries requested by users", ) logger.info( "OpenTelemetry initialized (service=%s, env=%s, ver=%s, otlp=%s)", SERVICE_NAME, ENVIRONMENT, VERSION, otlp_endpoint, ) except Exception as exc: logger.warning("OpenTelemetry initialization deferred or optional: %s", exc) # ----------------------------------------------------------------------------- # Pure ASGI Metrics Middleware & Route Normalization # ----------------------------------------------------------------------------- UUID_REGEX = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") EMAIL_REGEX = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+") MEMBERS_PATH_REGEX = re.compile(r"(/members/)[^/?#]+") EXCLUDED_PATHS = {"/metrics", "/favicon.ico"} def normalize_scope_path(scope: dict) -> str: """Extract low-cardinality route template or sanitize dynamic IDs to prevent metric cardinality explosion.""" raw_path = scope.get("path", "") sanitized = UUID_REGEX.sub("{id}", raw_path) sanitized = EMAIL_REGEX.sub("{email}", sanitized) sanitized = MEMBERS_PATH_REGEX.sub(r"\1{memberEmail}", sanitized) return sanitized class MetricsMiddleware: """High-performance pure ASGI middleware that records HTTP duration and request counts.""" def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": return await self.app(scope, receive, send) path = scope.get("path", "") if path in EXCLUDED_PATHS: return await self.app(scope, receive, send) start_time = time.perf_counter() status_code = 500 async def send_wrapper(message): nonlocal status_code if message["type"] == "http.response.start": status_code = message["status"] await send(message) method = scope.get("method", "GET") route_path = normalize_scope_path(scope) tracer = otel_trace.get_tracer("api-backend") if (HAS_OTEL and otel_trace is not None) else None if tracer is not None: with tracer.start_as_current_span( f"{method} {route_path}", attributes={ "http.method": method, "http.route": route_path, "service.name": SERVICE_NAME, "deployment.environment": ENVIRONMENT, }, ) as span: try: await self.app(scope, receive, send_wrapper) finally: duration_ms = (time.perf_counter() - start_time) * 1000.0 span.set_attribute("http.status_code", status_code) record_http_request( method=method, route=route_path, status_code=status_code, duration_ms=duration_ms, ) else: try: await self.app(scope, receive, send_wrapper) finally: duration_ms = (time.perf_counter() - start_time) * 1000.0 record_http_request( method=method, route=route_path, status_code=status_code, duration_ms=duration_ms, ) def setup_telemetry_middleware(app: FastAPI) -> None: """Attach HTTP metrics tracking middleware and /metrics scraping endpoint to FastAPI app.""" app.add_middleware(MetricsMiddleware) @app.get("/metrics", include_in_schema=False) async def metrics_endpoint() -> PlainTextResponse: return PlainTextResponse(generate_latest(REGISTRY), media_type=CONTENT_TYPE_LATEST)